"""Docs: docs/website/content/bulk-email-language-series.md
Run native CLIs as subprocesses: no SDK imports, keys, or network access needed.
"""
import json
from pathlib import Path
import subprocess
import tempfile
import unittest

ROOT = Path(__file__).resolve().parent
RUNNERS = [("node", "node.mjs"), ("python3", "python.py"), ("php", "php.php")]


class BulkTests(unittest.TestCase):
    def run_cli(self, runner: tuple[str, str], ledger: Path, *args: str, expected: int = 0) -> dict:
        """Execute a native mock CLI and read its persisted checkpoint."""
        result = subprocess.run([runner[0], str(ROOT / runner[1]), "--ledger", str(ledger), *args], capture_output=True, text=True)
        self.assertEqual(result.returncode, expected, result.stderr)
        return json.loads(ledger.read_text())

    def test_mixed_resume_selective_retry(self) -> None:
        """Acceptance survives restart; only the allowlisted failed row is retried."""
        for runner in RUNNERS:
            with self.subTest(language=runner[0]), tempfile.TemporaryDirectory() as directory:
                path = Path(directory) / "ledger.json"
                first = self.run_cli(runner, path)
                self.assertEqual([len(r) for r in first["requests"]], [50, 3])
                self.assertEqual(sum(r["state"] == "sent" for r in first["rows"]), 51)
                resumed = self.run_cli(runner, path)
                self.assertEqual(first, resumed)
                retried = self.run_cli(runner, path, "--retry-failed")
                self.assertEqual(retried["requests"][-1], ["person1@example.com"])
                self.assertEqual(retried["rows"][1]["state"], "sent")
                self.assertEqual(retried["rows"][2]["state"], "failed")
                for i, row in enumerate(retried["rows"]):
                    self.assertEqual(row["attempts"], 2 if i == 1 else 1)

    def test_unknown_response_is_not_retried(self) -> None:
        """Network uncertainty and invalid responses hold the complete attempted chunk."""
        for runner in RUNNERS:
            for fault in ("timeout", "short", "reordered", "badstatus"):
                with self.subTest(language=runner[0], fault=fault), tempfile.TemporaryDirectory() as directory:
                    path = Path(directory) / "ledger.json"
                    first = self.run_cli(runner, path, "--scenario", fault)
                    self.assertEqual(sum(r["state"] == "unknown" for r in first["rows"]), 50)
                    self.assertEqual(len(first["requests"]), 1)
                    after = self.run_cli(runner, path, "--retry-failed")
                    self.assertEqual(after["requests"][-1], [f"person{i}@example.com" for i in range(50, 53)])
                    self.assertTrue(all(r["attempts"] == 1 for r in after["rows"]))

    def test_pre_request_checkpoint_and_restart(self) -> None:
        """A pre-call process exit preserves intent and never reselects those rows."""
        for runner in RUNNERS:
            with self.subTest(language=runner[0]), tempfile.TemporaryDirectory() as directory:
                path = Path(directory) / "ledger.json"
                crashed = self.run_cli(runner, path, "--scenario", "crash", expected=75)
                self.assertEqual(sum(r["state"] == "inflight" for r in crashed["rows"]), 50)
                after = self.run_cli(runner, path)
                self.assertEqual(sum(r["state"] == "unknown" for r in after["rows"]), 50)
                self.assertEqual(sum(r["state"] == "sent" for r in after["rows"]), 3)

    def test_scheduled_is_accepted_not_resent(self) -> None:
        """Parked messages are excluded from subsequent dispatch selections."""
        for runner in RUNNERS:
            with self.subTest(language=runner[0]), tempfile.TemporaryDirectory() as directory:
                path = Path(directory) / "ledger.json"
                first = self.run_cli(runner, path, "--scenario", "scheduled")
                self.assertEqual(sum(r["state"] == "scheduled" for r in first["rows"]), 51)
                after = self.run_cli(runner, path)
                self.assertEqual(first, after)

    def test_job_mismatch_fails_closed(self) -> None:
        """A different job identity cannot overwrite or reuse a saved audience ledger."""
        for runner in RUNNERS:
            with self.subTest(language=runner[0]), tempfile.TemporaryDirectory() as directory:
                path = Path(directory) / "ledger.json"
                ledger = self.run_cli(runner, path)
                ledger["job"] = "different job"
                path.write_text(json.dumps(ledger))
                result = subprocess.run([runner[0], str(ROOT / runner[1]), "--ledger", str(path)], capture_output=True)
                self.assertNotEqual(result.returncode, 0)
                self.assertEqual(json.loads(path.read_text()), ledger)

    def test_programmatic_dedupe_completion_and_cancel(self) -> None:
        """Repeated facts enroll once; paid exits and direct cancellation stop waits."""
        runner = ("node", "programmatic.mjs")
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "demo.json"
            first = self.run_cli(runner, path)
            self.assertEqual(len(first["events"]), 2)
            self.assertEqual(len(first["enrollments"]), 1)
            self.assertEqual(first["enrollments"][0]["status"], "completed")
            self.assertEqual(first, self.run_cli(runner, path))
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "cancel.json"
            # Create the mock active sequence through demo, then reset mock state for a separate scenario.
            ledger = self.run_cli(runner, path)
            ledger.update(operations={}, events={}, enrollments=[])
            path.write_text(json.dumps(ledger))
            self.run_cli(runner, path, "--action", "failed")
            canceled = self.run_cli(runner, path, "--action", "cancel")
            self.assertEqual(canceled["enrollments"][0]["status"], "canceled")
            self.assertEqual(canceled["operations"]["cancel"]["result"]["stopped"], 1)


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