tests/test_mailbox_provider.py

"""Mailbox-layer tests after the v3.0.0 autonomy refactor.

The MillionVerifier delegation was removed BY DESIGN: the engine must run with
no vendor account, key, or per-check fee, and a stray key must never be able
to re-route the recipient list through a third party. These tests pin that
decision: no MV symbol is importable, no settings knob re-enables it, and the
explicit 5xx mailbox-rejection path (previously exercised through the MV
adapter) still hard-fails to score 0 through a plain provider.
"""
import unittest
from unittest.mock import AsyncMock, patch

from app.config import settings
from app.models import SMTPResult
from app.validator import mailbox_provider as mp
from app.validator.mailbox_provider import (
    MailboxProvider,
    LocalSmtpProvider,
    AutonomousSmtpProvider,
    RemoteSmtpWorkerProvider,
    resolve_provider,
)


class AutonomyInvariantTests(unittest.TestCase):
    """The engine is autonomous: no third-party verification API exists."""

    def test_millionverifier_is_gone(self):
        self.assertFalse(hasattr(mp, "MillionVerifierProvider"))
        self.assertFalse(hasattr(mp, "map_mv_result"))

    def test_config_has_no_mv_knobs(self):
        from app.config import Settings

        for field in ("mv_api_key", "mv_timeout", "prefer_api_provider"):
            self.assertFalse(
                hasattr(settings, field),
                f"settings.{field} must not exist after the autonomy refactor",
            )
            self.assertFalse(
                field in Settings.model_fields,
                f"Settings.{field} must not be declared",
            )

    def test_api_mode_is_no_longer_a_valid_choice(self):
        # Any leftover "api" configuration degrades to the built-in engine
        # instead of raising or silently selecting a vendor.
        with patch.object(settings, "mailbox_provider", "api"), patch.object(
            settings, "smtp_worker_url", ""
        ):
            provider = resolve_provider()
        self.assertIsInstance(provider, (AutonomousSmtpProvider, RemoteSmtpWorkerProvider))


class ResolveProviderTests(unittest.TestCase):
    def test_local_forced_is_autonomous(self):
        with patch.object(settings, "mailbox_provider", "local"), patch.object(
            settings, "smtp_worker_url", "https://worker.example.com"
        ):
            self.assertIsInstance(resolve_provider(), AutonomousSmtpProvider)

    def test_auto_without_worker_is_autonomous(self):
        with patch.object(settings, "mailbox_provider", "auto"), patch.object(
            settings, "smtp_worker_url", ""
        ):
            self.assertIsInstance(resolve_provider(), AutonomousSmtpProvider)

    def test_auto_with_worker_resolves_remote(self):
        with patch.object(settings, "mailbox_provider", "auto"), patch.object(
            settings, "smtp_worker_url", "https://worker.example.com"
        ):
            self.assertIsInstance(resolve_provider(), RemoteSmtpWorkerProvider)

    def test_remote_forced_without_url_degrades_to_autonomous(self):
        with patch.object(settings, "mailbox_provider", "remote"), patch.object(
            settings, "smtp_worker_url", ""
        ):
            self.assertIsInstance(resolve_provider(), AutonomousSmtpProvider)

    def test_providers_satisfy_protocol(self):
        for cls in (AutonomousSmtpProvider, RemoteSmtpWorkerProvider, LocalSmtpProvider):
            self.assertTrue(hasattr(cls, "verify"))
            self.assertTrue(hasattr(cls, "catch_all"))
            self.assertTrue(issubclass(cls, MailboxProvider))


class EngineProviderVerdictTests(unittest.IsolatedAsyncioTestCase):
    """End-to-end: an explicit 5xx mailbox rejection must hard-fail to score 0
    (Invalid) through the whole engine — the invariant the MV adapter used to
    guarantee, now pinned against a plain provider."""

    async def test_explicit_rejection_scores_zero(self):
        from app.models import DNSResult, DisposableResult, SyntaxResult
        from app.validator.engine import EmailValidationEngine

        class DeadMailboxProvider:
            async def verify(self, email, mx_host):
                return SMTPResult(
                    verified=False,
                    smtp_code=550,
                    smtp_message="550 mailbox does not exist",
                )

            async def catch_all(self, domain, mx_host):
                return (False, 0.9)

        eng = EmailValidationEngine(mailbox=DeadMailboxProvider())
        with (
            patch("app.validator.engine.validate_syntax", return_value=SyntaxResult(
                passed=True, normalized_email="a@example.test",
                domain="example.test", local_part="a",
            )),
            patch("app.validator.engine.check_dns", new=AsyncMock(return_value=DNSResult(
                domain_exists=True, has_mx_records=True,
                mx_records=[{"host": "mx.example.test", "priority": 10}],
            ))),
            patch("app.validator.engine.check_disposable", return_value=DisposableResult(
                is_disposable=False, is_free_provider=False, is_role_based=False,
            )),
        ):
            result = await eng.validate("a@example.test", deep=True)

        self.assertEqual(result.scoring.score, 0)
        self.assertEqual(result.status.value, "invalid")
        self.assertEqual(result.reason_code, "rcpt_5xx")
        # The normalized verdict agrees with the deterministic gauntlet.
        self.assertEqual(result.deterministic_verdict, "invalid")


class RemoteSmtpWorkerProviderTests(unittest.IsolatedAsyncioTestCase):
    """The remote worker path (spec §3.3): orchestrator delegates port-25."""

    def _provider(self):
        return RemoteSmtpWorkerProvider(
            url="https://worker.example.com", token="s3cret", timeout=5,
            fail_threshold=2, cooldown_seconds=60,
        )

    async def test_verify_posts_to_worker(self):
        p = self._provider()
        payload = {"verified": True, "smtp_code": 250, "smtp_message": "ok"}
        with patch("app.validator.mailbox_provider.httpx.AsyncClient") as client_cls:
            resp = AsyncMock()
            resp.raise_for_status = lambda: None
            resp.json = lambda: payload
            client = AsyncMock()
            client.post = AsyncMock(return_value=resp)
            client_cls.return_value.__aenter__ = AsyncMock(return_value=client)
            client_cls.return_value.__aexit__ = AsyncMock(return_value=False)
            result = await p.verify("a@corp.test", "mx.corp.test")
        self.assertTrue(result.verified)
        args, kwargs = client.post.call_args
        self.assertEqual(args[0], "https://worker.example.com/smtp/probe")
        self.assertEqual(kwargs["headers"]["X-Worker-Token"], "s3cret")

    async def test_worker_unreachable_is_inconclusive_and_opens_circuit(self):
        p = self._provider()
        with patch("app.validator.mailbox_provider.httpx.AsyncClient") as client_cls:
            client_cls.return_value.__aenter__ = AsyncMock(
                side_effect=RuntimeError("boom")
            )
            client_cls.return_value.__aexit__ = AsyncMock(return_value=False)
            r1 = await p.verify("a@corp.test", "mx.corp.test")
            r2 = await p.verify("b@corp.test", "mx.corp.test")
        self.assertIsNone(r1.verified)
        self.assertIn("unreachable", (r1.error or ""))
        # Threshold 2 reached → circuit open, next call fails fast.
        self.assertTrue(p.circuit_open)
        r3 = await p.verify("c@corp.test", "mx.corp.test")
        self.assertIn("circuit open", (r3.error or "").lower())

    async def test_catch_all_roundtrip(self):
        p = self._provider()
        payload = {"is_catch_all": True, "confidence": 0.95}
        with patch("app.validator.mailbox_provider.httpx.AsyncClient") as client_cls:
            resp = AsyncMock()
            resp.raise_for_status = lambda: None
            resp.json = lambda: payload
            client = AsyncMock()
            client.post = AsyncMock(return_value=resp)
            client_cls.return_value.__aenter__ = AsyncMock(return_value=client)
            client_cls.return_value.__aexit__ = AsyncMock(return_value=False)
            is_ca, confidence = await p.catch_all("corp.test", "mx.corp.test")
        self.assertTrue(is_ca)
        self.assertGreater(confidence, 0.5)

    async def test_verify_many_batch_roundtrip(self):
        p = self._provider()
        payload = {
            "results": {
                "a@corp.test": {"verified": True, "smtp_code": 250},
                "b@corp.test": {"verified": False, "smtp_code": 550},
            },
            "catch_all": {"corp.test": False},
        }
        with patch("app.validator.mailbox_provider.httpx.AsyncClient") as client_cls:
            resp = AsyncMock()
            resp.raise_for_status = lambda: None
            resp.json = lambda: payload
            client = AsyncMock()
            client.post = AsyncMock(return_value=resp)
            client_cls.return_value.__aenter__ = AsyncMock(return_value=client)
            client_cls.return_value.__aexit__ = AsyncMock(return_value=False)
            results, catch_all = await p.verify_many(
                [("a@corp.test", "mx"), ("b@corp.test", "mx")]
            )
        self.assertTrue(results["a@corp.test"].verified)
        self.assertFalse(results["b@corp.test"].verified)
        self.assertEqual(catch_all, {"corp.test": False})


class CircuitBreakerTests(unittest.TestCase):
    """Blocked egress must fail fast instead of burning smtp_timeout per row."""

    def test_local_provider_opens_circuit_after_threshold(self):
        p = LocalSmtpProvider(fail_threshold=2, cooldown_seconds=60)
        p.record_failure()
        self.assertFalse(p.circuit_open)
        p.record_failure()
        self.assertTrue(p.circuit_open)

    def test_open_circuit_returns_immediately_unavailable(self):
        import asyncio

        p = LocalSmtpProvider(fail_threshold=1, cooldown_seconds=60)
        p.record_failure()
        result = asyncio.run(p.verify("a@example.com", "mx.example.com"))
        self.assertIsNone(result.verified)
        self.assertIn("unavailable", (result.error or "").lower())

    def test_circuit_closes_after_cooldown(self):
        import time as _t

        p = LocalSmtpProvider(fail_threshold=1, cooldown_seconds=60)
        p.record_failure()
        self.assertTrue(p.circuit_open)
        # Simulate the cooldown elapsing: the open-window is now in the past.
        p._open_until = _t.monotonic() - 1
        self.assertFalse(p.circuit_open)
        self.assertFalse(p.circuit_open_after_cooldown())

    def test_zero_cooldown_never_stays_open(self):
        # cooldown_seconds=0 opens and closes in the same instant — there is no
        # window, so it must not read as open on the next call.
        p = LocalSmtpProvider(fail_threshold=1, cooldown_seconds=0)
        p.record_failure()
        self.assertFalse(p.circuit_open)


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