import unittest

from app.validator.mx_batch import CatchAllCache, MxBatchVerifier
from app.validator.smtp_client import RcptOutcome


class FakeConnection:
    """Records probes and answers from a scripted map.

    Keyed by lowercased address; a "*" entry is the default answer, which is how
    a catch-all domain behaves (accepts everything).
    """

    registry: dict[str, dict] = {}
    opened: list[str] = []
    closed: list[int] = []

    def __init__(self, host, **kwargs):
        self.host = host
        self.answers = FakeConnection.registry.get(host, {})
        self.ready = True
        self.opened.append(host)

    def open(self):
        self.ready = True
        return True

    def probe(self, email):
        # Catch-all probe uses a gibberish local part -> "*" rule applies.
        answer = self.answers.get(email.lower(), self.answers.get("*"))
        if answer is None:
            return RcptOutcome(code=550, message="550 5.1.1 no such user")
        if isinstance(answer, RcptOutcome):
            return answer
        return RcptOutcome(code=answer, message=f"{answer} response")

    def quit(self):
        FakeConnection.closed.append(1)

    def close(self):
        pass


class MxBatchVerifierTests(unittest.TestCase):
    def setUp(self):
        FakeConnection.registry = {}
        FakeConnection.opened = []
        FakeConnection.closed = []

    def verifier(self, **kw):
        return MxBatchVerifier(
            connection_factory=FakeConnection,
            max_connections_per_host=kw.pop("max_connections_per_host", 2),
            **kw,
        )

    def test_groups_by_mx_and_opens_one_session_per_host(self):
        # google host: two addresses; corp host: one address
        FakeConnection.registry = {
            "gmail-smtp-in.l.google.com": {"a@gmail.com": 250, "b@gmail.com": 550, "*": 550},
            "mx.corp.test": {"c@corp.test": 250, "*": 550},
        }
        v = self.verifier()
        out = v.verify_batch(
            [
                ("a@gmail.com", "gmail-smtp-in.l.google.com"),
                ("b@gmail.com", "gmail-smtp-in.l.google.com"),
                ("c@corp.test", "mx.corp.test"),
            ]
        )
        self.assertEqual(out["a@gmail.com"].verified, True)
        self.assertEqual(out["b@gmail.com"].verified, False)
        self.assertEqual(out["c@corp.test"].verified, True)
        # One connection opened per MX host (not per address).
        self.assertEqual(sorted(FakeConnection.opened), ["gmail-smtp-in.l.google.com", "mx.corp.test"])

    def test_catch_all_domain_does_not_certify_mailbox(self):
        # Microsoft 365 behaviour: accepts everything at RCPT.
        FakeConnection.registry = {
            "x.mail.protection.outlook.com": {"*": 250},
        }
        v = self.verifier()
        out = v.verify_batch([("anyone@contoso.com", "x.mail.protection.outlook.com")])
        result = out["anyone@contoso.com"]
        # Accepted by an accept-all host, so NOT verified.
        self.assertIsNone(result.verified)
        self.assertIn("catch_all", (result.error or "").lower())

    def test_catch_all_probe_is_cached_per_domain(self):
        FakeConnection.registry = {"mx.corp.test": {"*": 250}}
        v = self.verifier()
        v.verify_batch([("a@corp.test", "mx.corp.test")])
        probes_after_first = len(FakeConnection.opened)
        # Second address in the same domain must reuse the cached catch-all flag.
        v.verify_batch([("b@corp.test", "mx.corp.test")])
        self.assertEqual(len(FakeConnection.opened), probes_after_first)

    def test_transport_error_marks_unknown_not_invalid(self):
        class DeadConn(FakeConnection):
            def open(self):
                self.ready = False
                return False

        FakeConnection.registry = {"mx.dead.test": {"*": 250}}
        v = MxBatchVerifier(connection_factory=DeadConn)
        out = v.verify_batch([("a@dead.test", "mx.dead.test")])
        r = out["a@dead.test"]
        self.assertIsNone(r.verified)
        self.assertIsNotNone(r.error)

    def test_greylisted_address_is_reported_retryable(self):
        FakeConnection.registry = {
            "mx.corp.test": {"a@corp.test": RcptOutcome(code=451, message="451 greylisted"), "*": 550}
        }
        v = self.verifier()
        out = v.verify_batch([("a@corp.test", "mx.corp.test")])
        self.assertIsNone(out["a@corp.test"].verified)

    def test_empty_batch_returns_empty(self):
        v = self.verifier()
        self.assertEqual(v.verify_batch([]), {})


class CatchAllCacheTests(unittest.TestCase):
    def test_stores_and_returns_within_ttl(self):
        c = CatchAllCache(ttl_seconds=60)
        c.put("corp.test", True)
        self.assertIs(c.get("corp.test"), True)

    def test_expires_after_ttl(self):
        c = CatchAllCache(ttl_seconds=0)
        c.put("corp.test", True)
        self.assertIsNone(c.get("corp.test"))

    def test_unknown_is_not_cached_as_false(self):
        c = CatchAllCache(ttl_seconds=60)
        c.put("corp.test", None)
        # None (probe failed) must not be cached as a decision.
        self.assertIsNone(c.get("corp.test"))

    def test_distinguishes_false_from_missing(self):
        c = CatchAllCache(ttl_seconds=60)
        c.put("strict.test", False)
        self.assertIs(c.get("strict.test"), False)
        self.assertIsNone(c.get("never-seen.test"))


class EgressCircuitBreakerTests(unittest.TestCase):
    """When outbound port 25 is blocked (HF Spaces, most PaaS) every connect
    burns the full timeout. A bulk job must detect that once and fail fast
    instead of spending timeout × N — 10k rows at 10s/row is ~27 hours."""

    def setUp(self):
        FakeConnection.registry = {}
        FakeConnection.opened = []

    def test_circuit_opens_after_threshold_and_stops_connecting(self):
        class DeadConn(FakeConnection):
            def open(self):
                self.ready = False
                return False

        FakeConnection.registry = {"mx.a.test": {"*": 250}}
        v = MxBatchVerifier(connection_factory=DeadConn, fail_threshold=2, cooldown_seconds=60)

        v.verify_batch([("a@a.test", "mx.a.test")])
        v.verify_batch([("b@b.test", "mx.b.test")])
        self.assertTrue(v.circuit_open)

        before = len(FakeConnection.opened)
        out = v.verify_batch([("c@c.test", "mx.c.test")])
        # No new connection attempted while the circuit is open.
        self.assertEqual(len(FakeConnection.opened), before)
        self.assertIsNone(out["c@c.test"].verified)
        self.assertIn("unreachable", (out["c@c.test"].error or "").lower())

    def test_successful_connection_resets_failure_count(self):
        class FlakyConn(FakeConnection):
            fail_next = True

            def open(self):
                if FlakyConn.fail_next:
                    self.ready = False
                    return False
                self.ready = True
                return True

        FakeConnection.registry = {"mx.a.test": {"*": 250}}
        v = MxBatchVerifier(connection_factory=FlakyConn, fail_threshold=2, cooldown_seconds=60)
        v.verify_batch([("a@a.test", "mx.a.test")])
        FlakyConn.fail_next = False
        v.verify_batch([("b@b.test", "mx.b.test")])
        self.assertFalse(v.circuit_open)

    def test_default_threshold_is_conservative(self):
        v = MxBatchVerifier(connection_factory=FakeConnection)
        self.assertGreaterEqual(v.fail_threshold, 2)


class ParallelDomainTests(unittest.IsolatedAsyncioTestCase):
    """Domains must be probed CONCURRENTLY (with a cap).

    Sequential probing means wall-clock = sum of every domain's latency; on a
    blocked host that is timeout x domains. Commercial validators get bulk
    throughput from parallelism across domains, not speed per domain.
    """

    async def test_domains_are_probed_in_parallel(self):
        import asyncio
        import time

        active = {"now": 0, "peak": 0}

        class SlowConn(FakeConnection):
            def open(self):
                active["now"] += 1
                active["peak"] = max(active["peak"], active["now"])
                time.sleep(0.25)
                active["now"] -= 1
                self.ready = True
                return True

        FakeConnection.registry = {f"mx{i}.test": {"*": 550} for i in range(8)}
        v = MxBatchVerifier(
            connection_factory=SlowConn, bulk_concurrency=8, max_connections_per_host=1
        )
        pairs = [(f"u{i}@dom{i}.test", f"mx{i}.test") for i in range(8)]

        t0 = time.time()
        out = await v.verify_batch_async(pairs)
        elapsed = time.time() - t0

        self.assertEqual(len(out), 8)
        # 8 x 0.25s sequential = 2.0s; parallel should be well under 1s.
        self.assertLess(elapsed, 1.2, f"domains ran sequentially ({elapsed:.2f}s)")
        self.assertGreater(active["peak"], 2, "no real concurrency observed")

    async def test_concurrency_cap_is_respected(self):
        import time

        active = {"now": 0, "peak": 0}

        class SlowConn(FakeConnection):
            def open(self):
                active["now"] += 1
                active["peak"] = max(active["peak"], active["now"])
                time.sleep(0.15)
                active["now"] -= 1
                self.ready = True
                return True

        FakeConnection.registry = {f"mx{i}.test": {"*": 550} for i in range(10)}
        v = MxBatchVerifier(
            connection_factory=SlowConn, bulk_concurrency=3, max_connections_per_host=1
        )
        pairs = [(f"u{i}@dom{i}.test", f"mx{i}.test") for i in range(10)]
        out = await v.verify_batch_async(pairs)

        self.assertEqual(len(out), 10)
        self.assertLessEqual(active["peak"], 3, "exceeded the concurrency cap")

    async def test_sync_verify_batch_still_works(self):
        FakeConnection.registry = {"mx.corp.test": {"a@corp.test": 250, "*": 550}}
        v = MxBatchVerifier(connection_factory=FakeConnection)
        out = v.verify_batch([("a@corp.test", "mx.corp.test")])
        self.assertTrue(out["a@corp.test"].verified)


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