"""Engine unit tests. Run: python3 tests/unit/test_engine.py"""

import os
import sys
import tempfile
import unittest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "scripts"))

_TMP = tempfile.mkdtemp(prefix="ctxguard-test-")
os.environ["CTXGUARD_HOME"] = _TMP

from ctxguard import detect, policy as policy_mod, substitute
from ctxguard.store import Store, base26

SLUG = "test-project"


def engine(entities=()):
    store = Store(SLUG)
    pol = policy_mod.Policy()
    for value, type_, match, hint in entities:
        store.add(value, type_, match=match, hint=hint, prefix=pol.alias_prefix)
    return substitute.Engine(store, pol), store


class TestSecrets(unittest.TestCase):
    def test_vendor_keys_are_replaced_by_markers(self):
        eng, _ = engine()
        cases = [
            "AKIA1234567890ABCDEF",
            "ghp_Rf2Lp9Zt4Wm1Nb8Kd3Vy6Qs5Hj7Xc0Ae1Bg9",
            "sk-ant-api03-Rf2Lp9Zt4Wm1Nb8Kd3Vy6Qs5Hj7Xc0",
            "glpat-Rf2Lp9Zt4Wm1Nb8Kd3Vy",
            "xoxb-123456789012-Rf2Lp9Zt4Wm1",
        ]
        for secret in cases:
            with self.subTest(secret=secret[:8]):
                out = eng.redact(f"key = {secret} end")
                self.assertNotIn(secret, out.text)
                self.assertIn("<SECRET:", out.text)

    def test_marker_is_stable_and_carries_no_value(self):
        eng, _ = engine()
        text = "a = AKIA1234567890ABCDEF and b = AKIA1234567890ABCDEF"
        out = eng.redact(text)
        markers = substitute.MARKER_RE.findall(out.text)
        self.assertEqual(len(markers), 2)
        self.assertEqual(markers[0], markers[1], "same secret must map to the same marker")

    def test_private_key_block(self):
        eng, _ = engine()
        out = eng.redact("-----BEGIN RSA PRIVATE KEY-----\nMIIE...\n")
        self.assertIn("<SECRET:private-key-block:", out.text)

    def test_connection_string_password(self):
        eng, _ = engine()
        out = eng.redact("DSN=postgres://admin:sup3rS3cret@db.internal:5432/app")
        self.assertNotIn("sup3rS3cret", out.text)

    def test_local_dev_connection_strings_are_not_secrets(self):
        """Regression from real data: 265 of 265 credential-URL hits in one session
        were user==password on loopback. Firing there gets the plugin disabled."""
        eng, _ = engine()
        for dsn in (
            "postgres://service:service@127.0.0.1:5432/app",
            "postgres://app:app@db:5432/app",
            "mysql://root:root@localhost/test",
            "redis://user:password@127.0.0.1:6379",
            "postgres://svc:postgres@db/app",
        ):
            with self.subTest(dsn=dsn):
                self.assertEqual(eng.redact(dsn).secret_count, 0)

    def test_real_connection_string_still_caught_on_loopback(self):
        eng, _ = engine()
        out = eng.redact("postgres://svc:Xq7Rf2Lp9Zt4Wm@127.0.0.1:5432/app")
        self.assertEqual(out.secret_count, 1)

    def test_documented_examples_are_not_blocked(self):
        eng, _ = engine()
        out = eng.redact("use AKIAIOSFODNN7EXAMPLE in the docs")
        self.assertEqual(out.secret_count, 0)

    def test_allowlist_entry_inside_a_live_token_does_not_launder_it(self):
        """Regression: a bidirectional allowlist check let 'example.com' inside a
        real token suppress the finding."""
        eng, _ = engine()
        secret = "AKIAEXAMPLECOM12345Z"
        out = eng.redact(f"key = {secret}")
        self.assertNotIn(secret, out.text)

    def test_low_entropy_assignment_is_not_a_secret(self):
        eng, _ = engine()
        out = eng.redact('password = "aaaaaaaaaaaaaaaaaaaa"')
        self.assertEqual(out.secret_count, 0)

    def test_prefixed_env_var_secrets_are_caught(self):
        """Regression from a real .env: `_` is a word character, so `\\bapi_key\\b`
        never fired inside SERVICE_API_KEY and four of five common shapes were missed."""
        eng, _ = engine()
        for line in (
            "SERVICE_API_KEY=Xq7Rf2Lp9Zt4W",
            "SECRET_KEY=Xq7Rf2Lp9Zt4W",
            "APP_DATABASE_PASSWORD=Xq7Rf2Lp9Zt4W",
            "MY_AUTH_TOKEN=Xq7Rf2Lp9Zt4W",
            "api_key=Xq7Rf2Lp9Zt4W",
        ):
            with self.subTest(line=line):
                self.assertEqual(eng.redact(line).secret_count, 1)

    def test_short_secret_is_caught_despite_entropy_ceiling(self):
        """Entropy is bounded by log2(len), so a 13-char key can never reach 3.5.
        Character-class diversity has to carry short values."""
        eng, _ = engine()
        self.assertEqual(eng.redact("SERVICE_API_KEY=ab3-xy9-kd2m1").secret_count, 1)

    def test_empty_config_key_does_not_capture_the_next_line(self):
        """Regression from a real .env: `\\s*` after the separator crossed newlines, so
        an empty key swallowed the following line as its value."""
        eng, _ = engine()
        text = "CONTOSO_API_KEY=\nCONTOSO_SECRET_KEY=\nTIMEOUT_MAX_MS=200\n"
        self.assertEqual(eng.redact(text).secret_count, 0)

    def test_code_references_are_not_secrets(self):
        """Regression: `SET password_hash = EXCLUDED.password_hash` in real SQL."""
        eng, _ = engine()
        for line in (
            "SET password_hash = EXCLUDED.password_hash,",
            "let token = config.auth_token;",
            "api_key: settings.routing.api_key",
            "password = ${DB_PASSWORD}",
            "api_key = <your-key>",
        ):
            with self.subTest(line=line):
                self.assertEqual(eng.redact(line).secret_count, 0)

    def test_dev_password_assignment_is_not_a_secret(self):
        eng, _ = engine()
        self.assertEqual(eng.redact("POSTGRES_PASSWORD: service").secret_count, 0)

    def test_high_entropy_assignment_is_a_secret(self):
        eng, _ = engine()
        out = eng.redact('api_key = "Xq7Rf2Lp9Zt4Wm1Nb8Kd3Vy6"')
        self.assertEqual(out.secret_count, 1)

    def test_own_markers_are_not_detected_as_secrets(self):
        """Regression: the word SECRET inside `<SECRET:rule:hash>` matched the keyword
        rule, so the leak metric counted the engine's own redactions and could never
        reach zero."""
        eng, _ = engine()
        marked = "SERVICE_API_KEY: <SECRET:generic-assignment:cafda980>"
        self.assertEqual(eng.redact(marked).secret_count, 0)
        self.assertEqual(eng.redact(marked).text, marked)

    def test_rule_names_in_prose_are_not_secrets(self):
        """A marker quoted with the hash elided -- `<SECRET:generic-assignment:...>` --
        does not match MARKER_RE, so the rule name inside read as a credential."""
        eng, _ = engine()
        for text in ("a handful of `<SECRET:generic-assignment:…>` markers",
                     "rules fired: secret:aws-access-key"):
            with self.subTest(text=text):
                self.assertEqual(eng.redact(text).secret_count, 0)

    def test_our_own_report_format_is_not_self_detected(self):
        """`secret:generic-assignment=85` read as keyword-colon-value, so piping a
        report through the shell filter redacted its own labels."""
        from ctxguard.cli import _render
        eng, _ = engine()
        report = _render({"secret:generic-assignment": 85, "secret:aws-access-key": 3})
        self.assertEqual(eng.redact(report).secret_count, 0)
        self.assertEqual(eng.redact(report).text, report)

    def test_redacting_twice_is_stable(self):
        eng, _ = engine()
        once = eng.redact("k = AKIA2X7QP4LMZ9KDVR3T").text
        twice = eng.redact(once).text
        self.assertEqual(once, twice)

    def test_secrets_are_irreversible(self):
        eng, _ = engine()
        secret = "AKIA1234567890ABCDEF"
        redacted = eng.redact(f"k={secret}").text
        restored, _ = eng.rehydrate(redacted)
        self.assertNotIn(secret, restored)
        self.assertTrue(eng.has_secret_marker(restored))


class TestEntities(unittest.TestCase):
    def setUp(self):
        self.eng, self.store = engine([
            ("Acme Corp", "company", "exact", "retail customer"),
            ("Иван Петров", "person", "stem", "backend developer"),
            ("db.acme.internal", "host", "exact", "prod database"),
        ])

    def test_company_alias(self):
        out = self.eng.redact("Acme Corp asked for a report")
        self.assertIn("CTXG_COMPANY_A", out.text)
        self.assertNotIn("Acme", out.text)

    def test_homoglyph_evasion_is_caught(self):
        out = self.eng.redact("Aсme Corp")  # Cyrillic 'с'
        self.assertIn("CTXG_COMPANY_A", out.text)

    def test_separator_variants(self):
        for variant in ("acme-corp", "AcmeCorp", "ACME_CORP"):
            with self.subTest(variant=variant):
                self.assertIn("CTXG_COMPANY_A", self.eng.redact(variant).text)

    def test_russian_declensions(self):
        for form in ("Иван Петров", "Ивана Петрова", "Ивану Петрову", "Иваном Петровым"):
            with self.subTest(form=form):
                out = self.eng.redact(f"Задачу закрыл {form}.")
                self.assertIn("CTXG_PERSON_A", out.text)
                self.assertNotIn("Петров", out.text)

    def test_longest_match_wins(self):
        self.store.add("Acme", "company", prefix="CTXG_")
        eng = substitute.Engine(self.store, policy_mod.Policy())
        out = eng.redact("Acme Corp")
        self.assertIn("CTXG_COMPANY_A", out.text)
        self.assertNotIn("CTXG_COMPANY_B", out.text)

    def test_roundtrip(self):
        original = "Иван Петров из Acme Corp правит db.acme.internal"
        redacted = self.eng.redact(original).text
        restored, n = self.eng.rehydrate(redacted)
        self.assertEqual(restored, original)
        self.assertEqual(n, 3)

    def test_glossary_never_leaks_values(self):
        blob = repr(self.store.glossary())
        for forbidden in ("Acme", "Петров", "db.acme.internal"):
            self.assertNotIn(forbidden, blob)

    def test_hint_repeating_the_value_is_rejected(self):
        with self.assertRaises(ValueError):
            self.store.add("Globex", "company", hint="Globex is a customer")


class TestIdentMode(unittest.TestCase):
    """Regression from a live run: the real vendor name leaked through code
    identifiers, where boundary-anchored patterns cannot match."""

    def setUp(self):
        self.eng, self.store = engine([("Contoso", "company", "ident", "address vendor")])

    def test_matches_inside_identifiers(self):
        for text in ("ContosoClient", "struct ContosoConfig {", "CONTOSO_API_KEY=x",
                     "src/clients/contoso.rs", "use crate::contoso::Client;"):
            with self.subTest(text=text):
                out = self.eng.redact(text).text
                self.assertNotIn("adata", out, f"real name survived in {out!r}")

    def test_exact_mode_would_have_missed_these(self):
        store = Store(SLUG)
        store.add("Contoso", "company", match="exact", prefix="CTXG_")
        eng = substitute.Engine(store, policy_mod.Policy())
        self.assertIn("ContosoClient", eng.redact("ContosoClient").text)

    def test_case_survives_the_round_trip(self):
        for original in ("ContosoClient", "CONTOSO_API_KEY", "contoso.rs",
                         "Contoso and CONTOSO and contoso"):
            with self.subTest(original=original):
                red = self.eng.redact(original).text
                restored, n = self.eng.rehydrate(red)
                self.assertEqual(restored, original)
                self.assertGreater(n, 0)

    def test_cli_exposes_the_mode(self):
        """The engine supported ident before the CLI did, so it was unreachable."""
        from ctxguard.cli import build_parser
        args = build_parser().parse_args(
            ["entity", "add", "Contoso", "--type", "company", "--match", "ident"])
        self.assertEqual(args.match, "ident")

    def test_short_values_are_refused(self):
        with self.assertRaises(ValueError):
            self.store.add("ACL", "project", match="ident")


class TestPII(unittest.TestCase):
    def test_email_is_reversible(self):
        eng, _ = engine()
        original = "write to ivan.petrov@acme.ru please"
        red = eng.redact(original)
        self.assertNotIn("ivan.petrov@acme.ru", red.text)
        self.assertIn("CTXG_EMAIL_", red.text)
        restored, n = eng.rehydrate(red.text)
        self.assertEqual(restored, original)
        self.assertEqual(n, 1)

    def test_same_email_same_alias(self):
        eng, _ = engine()
        out = eng.redact("a@b.ru and a@b.ru").text
        aliases = {t for t in out.split() if t.startswith("CTXG_EMAIL_")}
        self.assertEqual(len(aliases), 1)

    def test_russian_phone(self):
        eng, _ = engine()
        for form in ("+7 916 123-45-67", "+79161234567", "8 (916) 123-45-67"):
            with self.subTest(form=form):
                self.assertIn("CTXG_PHONE_", eng.redact(f"тел {form}").text)

    def test_card_requires_luhn(self):
        eng, _ = engine()
        self.assertIn("CTXG_ID_", eng.redact("card 4111111111111111").text)
        self.assertNotIn("CTXG_ID_", eng.redact("id 4111111111111112").text)

    def test_card_requires_a_real_issuer_prefix(self):
        """Regression: Luhn alone passes 1 in 10 random digit runs, so lockfile
        checksums and base64 blobs were reported as payment cards."""
        eng, _ = engine()
        # Luhn-valid but no issuer ever started with 9 or 1
        self.assertNotIn("CTXG_ID_", eng.redact("n 9999999999999995").text)
        self.assertNotIn("CTXG_ID_", eng.redact("n 1234567890128").text)
        for card in ("4111111111111111", "5500005555555559", "371449635398431"):
            with self.subTest(card=card):
                self.assertIn("CTXG_ID_", eng.redact(f"card {card}").text)

    def test_digits_inside_a_hex_token_are_not_a_card(self):
        eng, _ = engine()
        self.assertNotIn("CTXG_ID_", eng.redact("sha=ab4111111111111111cd").text)

    def test_inn_needs_context_keyword(self):
        eng, _ = engine()
        self.assertIn("CTXG_ID_", eng.redact("ИНН 7712345678").text)
        self.assertNotIn("CTXG_ID_", eng.redact("count 7712345678").text)


class TestStructural(unittest.TestCase):
    def setUp(self):
        self.pol = policy_mod.Policy()

    def test_env_file_denied(self):
        for path in ("/srv/app/.env", ".env", "/srv/app/.env.production", "/x/id_rsa"):
            with self.subTest(path=path):
                self.assertIsNotNone(detect.denied_path(path, self.pol))

    def test_documented_templates_are_readable(self):
        """Regression from a live run: the .env refusal told the agent to read
        .env.example, and the next rule blocked that too. Contradictory guidance is
        how an agent learns to ignore guidance."""
        for path in ("/srv/app/.env.example", ".env.sample", "/srv/app/.env.template"):
            with self.subTest(path=path):
                self.assertIsNone(detect.denied_path(path, self.pol))

    def test_cat_env_example_is_allowed_but_cat_env_is_not(self):
        self.assertIsNone(detect.denied_command("cat .env.example", self.pol))
        self.assertIsNone(detect.denied_command("cat .env.sample", self.pol))
        self.assertIsNotNone(detect.denied_command("cat .env", self.pol))
        self.assertIsNotNone(detect.denied_command("cat config/.env", self.pol))

    def test_ordinary_source_allowed(self):
        for path in ("/srv/app/main.py", "README.md", "/srv/app/env.example.md"):
            with self.subTest(path=path):
                self.assertIsNone(detect.denied_path(path, self.pol))

    def test_self_protection(self):
        for path in (
            "/home/u/.claude/settings.json",
            "/home/u/.claude/ctx-guard/projects/x/entities.json",
            "/repo/plugins/ctxguard/scripts/ctxguard/detect.py",
        ):
            with self.subTest(path=path):
                self.assertIsNotNone(detect.self_protected(path, self.pol))

    def test_credential_dumping_commands_denied(self):
        for cmd in ("env", "printenv", "gh auth token", "kubectl get secret app -o yaml",
                    "cat .env", "aws configure get aws_secret_access_key", "op read op://a/b",
                    "env | grep TOKEN", "printenv SERVICE_API_KEY", "sudo env",
                    "ls; env", "cat config/.env"):
            with self.subTest(cmd=cmd):
                self.assertIsNotNone(detect.denied_command(cmd, self.pol))

    def test_ordinary_commands_allowed(self):
        for cmd in ("ls -la", "git status", "npm test", "grep -r foo src/",
                    "cat .env.example", "ls -la .env", "cp .env.example .env",
                    "grep ROUTING src/config.rs"):
            with self.subTest(cmd=cmd):
                self.assertIsNone(detect.denied_command(cmd, self.pol))


class TestPolicyPersistence(unittest.TestCase):
    """Regression: `init` wrote a full snapshot of every rule list, so a later plugin
    upgrade shipped fixed rules the hook never read. The fix appeared not to work and
    only a direct comparison of file vs code showed why."""

    def test_defaults_persist_as_a_diff(self):
        import json
        stored = json.loads(policy_mod.Policy().to_json())
        self.assertEqual(stored, {"version": policy_mod.POLICY_VERSION})

    def test_customisation_persists(self):
        import json
        pol = policy_mod.Policy()
        pol.mode = "observe"
        pol.allowlist = pol.allowlist + ["MY-OWN-EXAMPLE"]
        stored = json.loads(pol.to_json())
        self.assertEqual(stored["mode"], "observe")
        self.assertIn("MY-OWN-EXAMPLE", stored["allowlist"])
        self.assertNotIn("deny_command_patterns", stored)

    def test_stale_v1_rule_lists_are_discarded(self):
        stale = {
            "version": 1,
            "mode": "observe",
            "deny_command_patterns": [r"(?<![\w-])(env|printenv)(?![\w-])(?!.*\|)"],
            "deny_read_paths": ["**/.env", "**/.env.*"],
        }
        pol = policy_mod.Policy.from_dict(stale)
        self.assertEqual(pol.mode, "observe", "preferences must survive")
        self.assertEqual(pol.version, policy_mod.POLICY_VERSION)
        # the fixed rules must apply, not the snapshot
        self.assertIsNone(detect.denied_command("cat .env.example", pol))
        self.assertIsNone(detect.denied_command("ls -la .env", pol))
        self.assertIsNotNone(detect.denied_command("env", pol))
        self.assertIsNone(detect.denied_path("/srv/.env.example", pol))

    def test_dev_credentials_can_be_made_strict(self):
        pol = policy_mod.Policy()
        pol.dev_credentials_are_secrets = True
        store = Store(SLUG)
        eng = substitute.Engine(store, pol)
        self.assertEqual(
            eng.redact("postgres://app:app@127.0.0.1/db").secret_count, 1)


class TestDictionaryHandover(unittest.TestCase):
    """Aliases cannot be reproduced: they are assigned in registration order, and the
    salt is per-store. Two people who each ran `entity add` would mean different
    companies by CTXG_COMPANY_A, and their secret markers would not correlate. So the
    dictionary has to be copied, and copying it has to reproduce output exactly."""

    def test_aliases_are_order_dependent(self):
        a = Store("proj")
        a.add("Globex", "company", prefix="CTXG_")
        a.add("Initech", "company", prefix="CTXG_")
        b = Store("proj")
        b.add("Initech", "company", prefix="CTXG_")
        b.add("Globex", "company", prefix="CTXG_")
        self.assertNotEqual(
            a.find("Globex").alias, b.find("Globex").alias,
            "if this ever passes, the handover warning can be relaxed",
        )

    def test_salt_differs_per_store(self):
        self.assertNotEqual(Store("p").salt, Store("p").salt)

    def test_copied_dictionary_reproduces_output_exactly(self):
        a = Store("proj")
        a.add("Globex", "company", match="ident", prefix="CTXG_")
        a.add("Иван Петров", "person", match="stem", prefix="CTXG_")

        # what export/import moves: salt, counters, entities
        b = Store("proj")
        b.salt = a.salt
        b.counters = dict(a.counters)
        b.entities = list(a.entities)
        b._invalidate()

        pol = policy_mod.Policy()
        text = ("Globex и GlobexClient, закрыл Иваном Петровым, "
                "key=AKIA2X7QP4LMZ9KDVR3T, mail ivan@globex.ru")
        out_a = substitute.Engine(a, pol).redact(text).text
        out_b = substitute.Engine(b, pol).redact(text).text
        self.assertEqual(out_a, out_b)
        self.assertIn("<SECRET:aws-access-key:", out_a)
        self.assertIn("CTXG_EMAIL_", out_a)

    def test_export_refuses_to_land_in_a_git_repo(self):
        from ctxguard.cli import _inside_git_worktree
        self.assertTrue(_inside_git_worktree(Path(__file__)))


class TestProjectIdentity(unittest.TestCase):
    """Regression: keying the dictionary by cwd meant a session started in a
    subdirectory got a different, empty dictionary -- protection that looked enabled
    and applied to nothing."""

    def test_subdirectories_share_one_dictionary(self):
        import subprocess
        from ctxguard import paths as pmod
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp) / "repo"
            (root / "src" / "deep").mkdir(parents=True)
            subprocess.run(["git", "init", "-q", str(root)], check=True,
                           capture_output=True)
            pmod._TOPLEVEL_CACHE.clear()
            slugs = {pmod.slug(root), pmod.slug(root / "src"),
                     pmod.slug(root / "src" / "deep")}
            self.assertEqual(len(slugs), 1, f"expected one slug, got {slugs}")

    def test_non_repository_falls_back_to_the_directory(self):
        from ctxguard import paths as pmod
        with tempfile.TemporaryDirectory() as tmp:
            a, b = Path(tmp) / "a", Path(tmp) / "b"
            a.mkdir(); b.mkdir()
            pmod._TOPLEVEL_CACHE.clear()
            self.assertNotEqual(pmod.slug(a), pmod.slug(b))


class TestTranscriptScope(unittest.TestCase):
    """Regression: scanning every project against the current dictionary reported 802
    leaks, nearly all from unrelated sessions. The one metric the design leans on has
    to be scoped to mean anything."""

    def test_project_dir_encoding(self):
        from ctxguard.transcript import project_transcript_dir
        got = project_transcript_dir("/home/u/Work/sandbox/demo-service").name
        self.assertEqual(got, "-home-u-Work-sandbox-demo-service")

    def test_missing_project_dir_yields_nothing(self):
        from ctxguard.transcript import find_transcripts
        self.assertEqual(find_transcripts(cwd="/nonexistent/xyz-project"), [])

    def test_all_projects_is_opt_in(self):
        from ctxguard.cli import build_parser
        args = build_parser().parse_args(["scan-transcript"])
        self.assertFalse(args.all_projects)


class TestHelpers(unittest.TestCase):
    def test_base26(self):
        self.assertEqual([base26(n) for n in (1, 26, 27, 52)], ["A", "Z", "AA", "AZ"])

    def test_luhn(self):
        self.assertTrue(detect.luhn_ok("4111111111111111"))
        self.assertFalse(detect.luhn_ok("4111111111111112"))

    def test_withheld_carries_no_value(self):
        f = detect.Finding("aws-access-key", "secret", 0, 20, "AKIA1234567890ABCDEF", "<x>")
        self.assertNotIn("AKIA1234567890ABCDEF", repr(detect.withheld(f)))


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