computer-playbook/tests/unit/python/plugins/lookup/test_email.py

# nocheck: comments-valid  explanatory WHY-comments predate the stricter lint
from __future__ import annotations

import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

from ansible.errors import AnsibleError

from plugins.lookup.email import LookupModule
from utils.cache import _reset_cache_for_tests
from utils.cache import base as cache_base
from utils.cache import users as cache_users
from utils.cache.yaml import dump_yaml_str
from utils.roles.mapping import ROLE_FILE_META_SERVICES


def _write_role_config(base_dir: Path, role_name: str, payload: dict) -> None:
    config_path = base_dir / "roles" / role_name / ROLE_FILE_META_SERVICES
    config_path.parent.mkdir(parents=True, exist_ok=True)
    config_path.write_text(dump_yaml_str(payload), encoding="utf-8")


class TestEmailLookup(unittest.TestCase):
    @classmethod
    def setUpClass(cls) -> None:
        _reset_cache_for_tests()
        cls.addClassCleanup(_reset_cache_for_tests)

    def setUp(self) -> None:
        cache_base._reset()
        self.lookup = LookupModule()
        self.lookup._templar = None
        self._cwd = str(Path.cwd())
        self._tmpdir = tempfile.TemporaryDirectory()
        self._tmp = Path(self._tmpdir.name)
        (self._tmp / "roles").mkdir(parents=True, exist_ok=True)
        os.chdir(self._tmp)
        self._tokens_store_patcher = patch.object(
            cache_users, "_load_store_users", return_value={}
        )
        self._tokens_store_patcher.start()

    def tearDown(self) -> None:
        self._tokens_store_patcher.stop()
        os.chdir(self._cwd)
        self._tmpdir.cleanup()

    def test_returns_plugin_defaults_when_no_vars(self) -> None:
        result = self.lookup.run([], variables={"inventory_hostname": "host1"})[0]
        self.assertTrue(result["enabled"])
        self.assertEqual(result["host"], "localhost")
        self.assertEqual(result["port"], 25)
        self.assertEqual(result["from"], "root@host1.localdomain")
        self.assertEqual(result["username"], "root@host1.localdomain")
        self.assertEqual(result["password"], "")
        self.assertFalse(result["tls"])
        self.assertFalse(result["auth"])
        self.assertTrue(result["smtp"])

    def test_host_is_localhost_when_not_external_under_tls(self) -> None:
        result = self.lookup.run(
            [],
            variables={"inventory_hostname": "host1", "TLS_ENABLED": True},
        )[0]
        self.assertEqual(result["host"], "localhost")
        self.assertFalse(result["auth"])
        self.assertFalse(result["tls"])

    def test_keys_are_lowercased_without_prefix(self) -> None:
        result = self.lookup.run([], variables={})[0]
        for key in result:
            self.assertFalse(key.startswith("SYSTEM_EMAIL_"))
            self.assertEqual(key, key.lower())

    def test_group_var_overrides_plugin_default(self) -> None:
        variables = {
            "SYSTEM_EMAIL_HOST": "smtp.example.org",
            "SYSTEM_EMAIL_PORT": 465,
            "SYSTEM_EMAIL_TLS": True,
            "inventory_hostname": "host1",
        }
        result = self.lookup.run([], variables=variables)[0]
        self.assertEqual(result["host"], "smtp.example.org")
        self.assertEqual(result["port"], 465)
        self.assertTrue(result["tls"])
        self.assertEqual(result["from"], "root@host1.localdomain")

    def test_onion_domain_disables_tls_but_keeps_auth(self) -> None:
        """An onion mail host has no TLS certificate, so the relay speaks
        plaintext — Tor already encrypts the transport. AUTH must survive that:
        without credentials the message reaches Mailu unauthenticated with a
        local envelope sender, and rspamd's antispoof rule rejects it with
        554 5.7.1. Plaintext submission is port 587; 25 is the inbound MX and
        offers no AUTH at all."""
        variables = {
            "groups": {"web-app-mailu": ["host1"]},
            "SYSTEM_EMAIL_HOST": "mail.abc123.onion",
            "TLS_ENABLED": True,
            "DOMAIN_PRIMARY": "abc123.onion",
            "inventory_hostname": "host1",
        }
        result = self.lookup.run([], variables=variables)[0]
        self.assertFalse(result["tls"])
        self.assertTrue(result["auth"])
        self.assertEqual(result["port"], 587)

    def test_clearnet_external_keeps_tls(self) -> None:
        variables = {
            "groups": {"web-app-mailu": ["host1"]},
            "SYSTEM_EMAIL_HOST": "mail.example.org",
            "TLS_ENABLED": True,
            "DOMAIN_PRIMARY": "example.org",
            "inventory_hostname": "host1",
        }
        result = self.lookup.run([], variables=variables)[0]
        self.assertTrue(result["tls"])
        self.assertEqual(result["port"], 465)

    def test_the_axis_matrix_stays_where_it_was_measured(self) -> None:
        """Decoupling auth from tls may only move the two cases it was meant to.

        The clearnet default and the local relay must render byte-identically,
        or a change aimed at the onion has silently reached every deploy."""
        cases = [
            ("mail.example.org", True, True, {"tls": True, "auth": True, "port": 465}),
            (
                "mail.example.org",
                False,
                True,
                {"tls": False, "auth": True, "port": 587},
            ),
            ("mail.x.onion", True, True, {"tls": False, "auth": True, "port": 587}),
            (
                "mail.example.org",
                True,
                False,
                {"tls": False, "auth": False, "port": 25},
            ),
        ]
        for host, tls_enabled, external, expected in cases:
            with self.subTest(host=host, tls=tls_enabled, external=external):
                variables = {
                    "groups": {"web-app-mailu": ["host1"]} if external else {},
                    "SYSTEM_EMAIL_HOST": host,
                    "TLS_ENABLED": tls_enabled,
                    "DOMAIN_PRIMARY": host.split(".", 1)[1],
                    "inventory_hostname": "host1",
                }
                result = self.lookup.run([], variables=variables)[0]
                for key, value in expected.items():
                    self.assertEqual(
                        bool(result[key]) if key != "port" else result[key], value
                    )

    def test_auth_mechanism_is_named_only_where_msmtp_would_refuse(self) -> None:
        """msmtp aborts with EX_UNAVAILABLE when it has to pick a mechanism
        itself on a cleartext link, so the onion relay needs PLAIN spelled out.
        The TLS and the no-relay cases must keep rendering on and off, or the
        change reaches deploys it was never meant to touch."""
        cases = [
            ("mail.example.org", True, True, "on"),
            ("mail.x.onion", True, True, "plain"),
            ("mail.example.org", True, False, "off"),
        ]
        for host, tls_enabled, external, expected in cases:
            with self.subTest(host=host, external=external):
                variables = {
                    "groups": {"web-app-mailu": ["host1"]} if external else {},
                    "SYSTEM_EMAIL_HOST": host,
                    "TLS_ENABLED": tls_enabled,
                    "DOMAIN_PRIMARY": host.split(".", 1)[1],
                    "inventory_hostname": "host1",
                }
                result = self.lookup.run([], variables=variables)[0]
                self.assertEqual(result["auth_mechanism"], expected)

    def test_empty_string_falls_back_to_plugin_default(self) -> None:
        variables = {
            "SYSTEM_EMAIL_HOST": "",
            "inventory_hostname": "host1",
        }
        result = self.lookup.run([], variables=variables)[0]
        self.assertEqual(result["host"], "localhost")

    def test_application_override_wins_over_defaults(self) -> None:
        # Per the file root of meta/services.yml IS the services map
        # (no `compose.services` envelope).
        _write_role_config(
            self._tmp,
            "web-app-x",
            {"email": {"host": "smtp.app.org", "port": 587}},
        )
        variables = {
            "SYSTEM_EMAIL_HOST": "smtp.global.org",
            "SYSTEM_EMAIL_PORT": 25,
            "inventory_hostname": "host1",
        }
        result = self.lookup.run(
            ["web-app-x"],
            variables=variables,
            roles_dir=str(self._tmp / "roles"),
        )[0]
        self.assertEqual(result["host"], "smtp.app.org")
        self.assertEqual(result["port"], 587)
        self.assertEqual(result["from"], "root@host1.localdomain")

    def test_missing_application_returns_defaults(self) -> None:
        variables = {
            "SYSTEM_EMAIL_HOST": "smtp.global.org",
            "inventory_hostname": "host1",
        }
        result = self.lookup.run(
            ["web-app-unknown"],
            variables=variables,
            roles_dir=str(self._tmp / "roles"),
        )[0]
        self.assertEqual(result["host"], "smtp.global.org")

    def test_application_without_email_service_returns_defaults(self) -> None:
        _write_role_config(
            self._tmp,
            "web-app-nomail",
            {"logout": {"enabled": True}},
        )
        variables = {
            "SYSTEM_EMAIL_HOST": "smtp.global.org",
            "inventory_hostname": "host1",
        }
        result = self.lookup.run(
            ["web-app-nomail"],
            variables=variables,
            roles_dir=str(self._tmp / "roles"),
        )[0]
        self.assertEqual(result["host"], "smtp.global.org")

    def test_too_many_terms_raises(self) -> None:
        with self.assertRaises(AnsibleError):
            self.lookup.run(["a", "b"], variables={})

    def test_computed_defaults_are_templated(self) -> None:
        self.lookup._templar = _DummyTemplar(
            {"DOMAIN_PRIMARY_RESOLVED": "mail.example.org"}
        )
        variables = {
            "DOMAIN_PRIMARY": "{{ DOMAIN_PRIMARY_RESOLVED }}",
            "inventory_hostname": "host1",
        }
        result = self.lookup.run([], variables=variables)[0]
        self.assertEqual(result["domain"], "mail.example.org")


class _DummyTemplar:
    def __init__(self, available_variables: dict[str, str]) -> None:
        self.available_variables = available_variables

    def template(self, value, fail_on_undefined=False):
        if isinstance(value, str):
            rendered = value
            for key, replacement in self.available_variables.items():
                rendered = rendered.replace(f"{{{{ {key} }}}}", str(replacement))
            return rendered
        return value


if __name__ == "__main__":
    unittest.main()