"""Fixture tests for the merged key_patterns module (no network).

Run: python tests/test_key_patterns.py   (exit 1 on any failure)

Covers both public APIs that exist in the two trees:
  root lanes  -> PATTERNS / scan_text
  src modules -> KEY_PATTERNS / ENV_RE / KIMI_PATTERN / extract_all_keys
"""
import os
import re
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.dirname(HERE))
import key_patterns as K  # noqa: E402

DS = "sk-3f9a1c7e5b2d8460a1c3e5f7091b2d4c"          # 32 hex, high entropy
DS2 = "sk-7d2f4a6b8c0e1d3f5a7b9c1e3d5f7a9b"
KIMI = "sk-" + "Qw3rTy7uIo9pAs2dFg6hJk1lZx5cVb8nMq4wEr0tYu3iOp1a"  # 48 alnum
ZAI = "9c1f7a3e5b8d2046f1a3c5e7b9d1f3a5.RtY7uI2oP4aS6dF8"
LOW_ENTROPY = "sk-aabbccddeeff00112233445566778899"   # passes _is_junk today
CODING_PLAN = "sk-sp-Kf83Jd92Ls04Mx71Qw63Er25Ty88Ua40"

FAILS = []


def check(name, cond, detail=""):
    if cond:
        print(f"ok   {name}")
    else:
        print(f"FAIL {name} {detail}")
        FAILS.append(name)


def providers(text, fn):
    return {p for p, _ in fn(text)}


def values(text, fn):
    return {k for _, k in fn(text)}


def test_both_apis_exist():
    check("api.PATTERNS", isinstance(K.PATTERNS, list) and len(K.PATTERNS) > 100,
          f"len={len(getattr(K, 'PATTERNS', []))}")
    check("api.scan_text", callable(getattr(K, "scan_text", None)))
    check("api.KEY_PATTERNS", isinstance(getattr(K, "KEY_PATTERNS", None), dict)
          and len(K.KEY_PATTERNS) > 50, f"len={len(getattr(K,'KEY_PATTERNS',[]) or [])}")
    check("api.extract_all_keys", callable(getattr(K, "extract_all_keys", None)))
    check("api.ENV_RE", getattr(K, "ENV_RE", None) is not None)
    check("api.KIMI_PATTERN", getattr(K, "KIMI_PATTERN", None) is not None)


def test_dashscope_env_context():
    text = f"# .env\nDASHSCOPE_API_KEY={DS}\nOTHER=1\n"
    check("dashscope.env.scan_text", DS in values(text, K.scan_text),
          str(sorted(values(text, K.scan_text)))[:120])
    check("dashscope.env.extract_all", DS in values(text, K.extract_all_keys))
    check("dashscope.env.prov", "DASHSCOPE" in providers(text, K.scan_text))


def test_dashscope_bare():
    text = '{"model":"qwen","api_key":"%s"}' % DS2
    check("dashscope.bare.scan_text", DS2 in values(text, K.scan_text))
    check("dashscope.bare.extract_all", DS2 in values(text, K.extract_all_keys))


def test_placeholder_and_junk_dropped():
    text = ("DASHSCOPE_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\n"
            "KIMI_API_KEY=sk-your_api_key_here_1234567890abcdefghij\n")
    v = values(text, K.scan_text) | values(text, K.extract_all_keys)
    check("junk.placeholder_dropped", not any("xxxx" in x for x in v), str(v)[:120])
    check("junk.your_key_dropped", not any("your_api_key" in x for x in v))


def test_low_entropy_dropped():
    text = "DASHSCOPE_API_KEY=%s" % LOW_ENTROPY
    v = values(text, K.scan_text) | values(text, K.extract_all_keys)
    check("template.pair_repeat_dropped", LOW_ENTROPY not in v,
          "low-entropy key accepted (need Shannon prefilter)")


def test_coding_plan_pattern():
    text = '{"coding_plan_key": "%s"}' % CODING_PLAN
    v = values(text, K.scan_text) | values(text, K.extract_all_keys)
    check("pattern.sk-sp_coding_plan", CODING_PLAN in v,
          "sk-sp- Coding Plan shape not matched")


def test_kimi_and_zai():
    text = "KIMI_API_KEY=%s\nZHIPU_API_KEY=%s\n" % (KIMI, ZAI)
    v = values(text, K.scan_text) | values(text, K.extract_all_keys)
    check("kimi.51char_found", KIMI in v)
    check("zai.dot_form_found", ZAI in v)


def test_grouped_patterns_do_not_crash():
    """A PATTERNS entry with a capture group makes findall() yield tuples;
    scan_text must flatten them instead of raising on .strip()."""
    grouped = [(p, rx) for p, rx in K.PATTERNS if getattr(rx, "groups", 0)]
    if not grouped:
        print("skip grouped_patterns (no grouped pattern in PATTERNS)")
        return
    prov, rx = grouped[0]
    sample = None
    for cand in (
        'SUNO_API_KEY = "sk-Ab3xK9mQ2wE5rT7yU1iO"',
        'token = "pk-lf-Ab3xK9mQ2wE5rT7yU1iO4pA8sD2fG6hJ"',
        'sk-proj-Ab3xK9mQ2wE5rT7yU1iO4pA8sD2fG6hJk0Lz9Xc8Vb7Nm6Qa5Ws4Ed3Rf2Tg1Yh0',
    ):
        try:
            if rx.search(cand):
                sample = cand
                break
        except Exception:
            continue
    if sample is None:
        print(f"skip grouped_patterns fixture (no sample matched {prov})")
        return
    try:
        out = K.scan_text(sample)
        ok = all(isinstance(k, str) for _, k in out)
        check("grouped.no_crash_and_str_keys", ok, str(out)[:120])
    except Exception as e:
        check("grouped.no_crash_and_str_keys", False, f"{type(e).__name__}: {e}")


def test_lowercase_env_attribution():
    """IGNORECASE ENV_RE must still attribute the provider, not 'ENV'."""
    text = "dashscope_api_token=Ab3xK9mQ2wE5rT7yU1iO4pA8sD2fG6hJzX7cV9bN\n"
    hits = K.scan_text(text)
    provs = {p for p, _ in hits}
    check("env.lowercase_provider_not_ENV", "DASHSCOPE" in provs,
          str(hits)[:120])
    check("env.lowercase_no_bare_ENV", "ENV" not in provs, str(hits)[:120])


def test_key_patterns_dict_coverage():
    """KEY_PATTERNS[prov] must match EVERY alternative of that provider."""
    check("dict.DASHSCOPE_coding_plan",
          bool(K.KEY_PATTERNS["DASHSCOPE"].search(CODING_PLAN)))
    check("dict.DASHSCOPE_hex32", bool(K.KEY_PATTERNS["DASHSCOPE"].search(DS)))
    kimi_rx = K.KEY_PATTERNS.get("KIMI")
    check("dict.KIMI_narrow48", kimi_rx is not None
          and bool(kimi_rx.fullmatch("sk-" + "aB3xK9mQ2wE5rT7yU1iO4pA8sD2fG6hJzX7cV9bNmN2qR4tY"))
          and not kimi_rx.search("sk-short123"))


def test_one_pair_per_value():
    """A value matching two same-shape providers (KIMI/SILICONFLOW sk-48)
    must come back exactly once — first (specific-first) provider wins."""
    val = "sk-" + "aB3xK9mQ2wE5rT7yU1iO4pA8sD2fG6hJzX7cV9bNmN2qR4tY"
    hits = K.scan_text("SOME_CONTEXT '%s'" % val)
    same = [h for h in hits if h[1] == val]
    check("dedup.one_pair_per_value", len(same) == 1, str(hits)[:140])


# ── v6.4 placeholder junk fixture tests ─────────────────────────────

NPM_PLACEHOLDER = "npm_XXXX_XXXX_XXXX_XXXX_XXXX"
INSERT_YOUR = "ghp_INSERT_YOUR_TOKEN_HERE_1234567890abcdef"
SK_UPPERCASE_X = "sk-" + "X" * 32  # 32 uppercase X
REAL_SK = "sk-3f9a1c7e5b2d8460a1c3e5f7091b2d4c"        # 32 hex, same as DS
REAL_GHP = "ghp_" + "aB3xK9mQ2wE5rT7yU1iO4pA8sD2fG6hJzX7cV9bN"  # valid shape


def test_npm_placeholder_rejected():
    """npm_XXXX_XXXX_... must be rejected by _is_junk (v6.4)."""
    text = f'NPM_TOKEN={NPM_PLACEHOLDER}\n'
    v = values(text, K.scan_text) | values(text, K.extract_all_keys)
    check("junk.npm_placeholder_rejected",
          NPM_PLACEHOLDER not in v, str(v)[:120])


def test_insert_your_token_rejected():
    """INSERT_YOUR_TOKEN_HERE must be rejected by _is_junk (v6.4)."""
    text = f'GITHUB_TOKEN={INSERT_YOUR}\n'
    v = values(text, K.scan_text) | values(text, K.extract_all_keys)
    check("junk.insert_your_token_rejected",
          INSERT_YOUR not in v, str(v)[:120])


def test_sk_uppercase_x_rejected():
    """sk-XXXX... (all uppercase X) must be rejected by _is_junk (v6.4)."""
    text = f'DASHSCOPE_API_KEY={SK_UPPERCASE_X}\n'
    v = values(text, K.scan_text) | values(text, K.extract_all_keys)
    check("junk.sk_uppercase_x_rejected",
          SK_UPPERCASE_X not in v, str(v)[:120])


def test_real_sk_accepted():
    """Real sk-<32hex> key must NOT be rejected."""
    text = f'DASHSCOPE_API_KEY={REAL_SK}\n'
    v = values(text, K.scan_text) | values(text, K.extract_all_keys)
    check("real.sk_accepted", REAL_SK in v, str(v)[:120])


def test_real_ghp_accepted():
    """Real ghp_ token must NOT be rejected."""
    text = f'GITHUB_TOKEN={REAL_GHP}\n'
    v = values(text, K.scan_text) | values(text, K.extract_all_keys)
    check("real.ghp_accepted", REAL_GHP in v, str(v)[:120])


import hashlib as _hl

def _filler(seed, n, alpha="0123456789abcdef"):
    out = ""
    i = 0
    while len(out) < n:
        out += _hl.sha256((seed + str(i)).encode()).hexdigest()
        i += 1
    return "".join(ch for ch in out if ch in alpha)[:n]

_BECH32 = "QPZRY9X8GF2TVDW0S3JN54KHCE6MUA7L"
def _bech_filler(n):
    dg = b""
    i = 0
    while len(dg) < n:
        dg += _hl.sha256(b"age" + str(i).encode()).digest(); i += 1
    return "".join(_BECH32[b % 32] for b in dg[:n])

NEW_PAT_FIXTURES = [
    ("GROK",          "xai-" + _filler("xai", 40, "0123456789abcdefghijklmnopqrstuvwxyz") + "_" + _filler("xai2", 39, "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ")),
    ("LANGSMITH",     "lsv2_pt_" + _filler("lsp1", 32) + "_" + _filler("lsp2", 10)),
    ("OPENAI_ADMIN",  "sk-admin-" + _filler("oa1", 58, "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ_-") + "T3BlbkFJ" + _filler("oa2", 58, "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ_-")),
    ("DATABRICKS",    "dapi" + _filler("db", 32)),
    ("DOPPLER",       "dp.pt." + _filler("dp", 43, "0123456789abcdefghijklmnopqrstuvwxyz")),
    ("ARTIFACTORY",   "AKCp" + _filler("art", 69, "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ")),
    ("ALIBABA_LTAI",  "LTAI" + _filler("ltai", 20, "0123456789abcdefghijklmnopqrstuvwxyz")),
    ("AGE_KEY",       "AGE-SECRET-KEY-1" + _bech_filler(58)),
    ("ONEPASSWORD",   "A3-7K2M9X-4Q8W2E7R1T5-B3N6M-K9P2Q-7X4Z8"),
    ("AZURE_AD",      "k7Q" + "4Q~" + _filler("az", 32, "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ_~.-")),
]


def test_v65_merged_patterns():
    """v6.5: 10 rules merged from gitleaks v8 + trufflehog detectors."""
    for prov, sample in NEW_PAT_FIXTURES:
        text = "export SOME_CTX='%s'\n" % sample
        v = values(text, K.scan_text) | values(text, K.extract_all_keys)
        p = providers(text, K.scan_text)
        check("v65." + prov, sample in v and prov in p,
              "in_values=%s provs=%s" % (sample in v, sorted(p)[:4]))


def test_elevenlabs_boundary():
    """Boundary: XI_API_KEY matches, PERPLEXITY/complexity/maxi_token do not."""
    import hashlib, string
    def _body(seed, n, alpha=string.ascii_letters + string.digits):
        out, i = '', 0
        while len(out) < n:
            for b in hashlib.sha256(f'{seed}:{i}'.encode()).digest():
                out += alpha[b % len(alpha)]
                if len(out) == n: return out
            i += 1

    h32 = _body('el', 32, '0123456789abcdef')  # pure hex for xi key
    pplx = _body('pplx', 56, string.ascii_lowercase + string.digits)  # exactly 56 lowercase alnum

    # Positive: XI_API_KEY=<hex32> → ELEVENLABS
    pos_txt = f"XI_API_KEY={h32}"
    check("EL.xi.positive", any(
        prov == 'ELEVENLABS' for prov, key in K.scan_text(pos_txt)
    ), f"XI_API_KEY={h32[:6]}...")

    # Negative: PERPLEXITY_API_KEY=<32 hex> → not ELEVENLABS
    neg1 = f"PERPLEXITY_API_KEY={h32}"
    el_hits = [prov for prov, key in K.scan_text(neg1) if prov == 'ELEVENLABS']
    check("EL.perplexity.not_xi", len(el_hits) == 0,
          f"Got ELEVENLABS hits on PERPLEXITY: {el_hits}")

    # Negative: pplx-<56 lnum> → PERPLEXITY, not ELEVENLABS
    ppx_txt = f"PERPLEXITY_API_KEY=pplx-{pplx}"
    hits = list(K.scan_text(ppx_txt))
    provs = {p for p, _ in hits}
    check("EL.pplx.is_perplexity", 'PERPLEXITY' in provs,
          f"PERPLEXITY not in {provs}")
    check("EL.pplx.not_elevenlabs", 'ELEVENLABS' not in provs,
          f"ELEVENLABS falsely detected: {provs}")

    # Negative: word-internal xi → no match
    for word in ('complexity', 'maxi_token'):
        neg_txt = f"{word}: {h32}"
        hits = [prov for _, prov, _ in K.scan_text(neg_txt) if prov == 'ELEVENLABS']
        check(f"EL.{word}.not_xi", len(hits) == 0,
              f"FP on '{word}': {hits}")

    # Drift guard: key_patterns ELEVENLABS regex == eternal_v10 PROVIDERS regex
    # Textual grep avoids importing eternal_v10 (which shells out to gh, connects sockets, etc.)
    EV10 = os.path.join(os.path.dirname(HERE), 'src', 'eternal_v10.py')
    kp_pat = dict(K.PATTERNS)['ELEVENLABS'].pattern
    if not os.path.exists(EV10):
        check("EL.drift_guard", True, f"skip: no {EV10}")
    else:
        # Extract the ELEVENLABS regex string from eternal_v10 by finding the
        # block line-by-line, avoiding the nested-quote escaping problem.
        lines = open(EV10, encoding='utf-8').readlines()
        ev_pat = None
        for i, line in enumerate(lines):
            if '"ELEVENLABS"' not in line:
                continue
            for j in range(i, min(i + 30, len(lines))):
                m = re.search(r'_re\(\s*r[\'"](.+?)[\'"]\s*,?\s*$', lines[j])
                if m:
                    ev_pat = m.group(1)
                    break
            break
        if ev_pat is None:
            check("EL.drift_guard", False, "could not extract PROVIDERS['ELEVENLABS']['regex']")
        else:
            check("EL.drift_guard", ev_pat == kp_pat,
                  f"PATTERNS={kp_pat!r} PROVIDERS={ev_pat!r}")


def test_dashscope_ws_extraction():
    """sk-ws- keys (115-200 chars, dots/dashes) → DASHSCOPE (v11.1)."""
    import hashlib
    def _ws_body(n):
        out, i = '', 0
        alpha = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_."
        while len(out) < n:
            for b in hashlib.sha256(f'ws:{i}'.encode()).digest():
                out += alpha[b % len(alpha)]
                if len(out) == n: return out
            i += 1

    ws_body = _ws_body(108)  # 108 chars after sk-ws- prefix (total 115)
    ws_key = f"sk-ws-{ws_body}"
    text = f"DASHSCOPE_API_KEY={ws_key}"
    hits = list(K.scan_text(text))
    provs = {p for p, _ in hits}
    vals = {k for _, k in hits}
    check("DS.ws.extracted", ws_key in vals,
          f"sk-ws- key not in values, got: {sorted(vals)}")
    check("DS.ws.attributed", 'DASHSCOPE' in provs,
          f"DASHSCOPE not in providers, got: {provs}")


def test_dashscope_long_hex_not_truncated():
    """40-hex sk- key must NOT be truncated to 32 chars (v11.1 boundary)."""
    import hashlib, string
    def _body(seed, n, alpha):
        out, i = '', 0
        while len(out) < n:
            for b in hashlib.sha256(f'{seed}:{i}'.encode()).digest():
                out += alpha[b % len(alpha)]
                if len(out) == n: return out
            i += 1

    long_hex = _body('ds40', 40, '0123456789abcdef')
    full_key = f"sk-{long_hex}"
    assert len(full_key) == 43, f"expected 43, got {len(full_key)}"  # sk- + 40 hex
    text = f"DASHSCOPE_API_KEY={full_key}"

    hits = list(K.scan_text(text))
    vals = {k for _, k in hits}

    # The full 40-hex key should NOT appear (doesn't match 32-hex pattern)
    # BUT it should NOT produce a truncated 32-char fake
    truncated = full_key[:35]  # sk- + 32 hex
    fake_hits = [v for v in vals if len(v) == 35 and v.startswith('sk-')]
    check("DS.notrunc.no32fake", len(fake_hits) == 0,
          f"Truncated 32-hex fake extracted: {fake_hits}")

    # sk-ws- pattern should NOT match (this is hex, not ws format)
    ws_hits = [v for v in vals if v.startswith('sk-ws-')]
    check("DS.notrunc.no_ws_misattrib", len(ws_hits) == 0,
          f"sk-ws- misattributed on hex key: {ws_hits}")

    # Known limitation: 40-hex key not captured by any DASHSCOPE alt
    # (newer shapes >32 hex need a separate probe path via endpoint auto-detection)
    check("DS.notrunc.accepted_limitation", len(vals) == 0,
          f"40-hex key unexpectedly matched: {vals} (known: >32 hex needs probe path)")


def main():
    for fn in (test_both_apis_exist, test_dashscope_env_context,
               test_dashscope_bare, test_placeholder_and_junk_dropped,
               test_low_entropy_dropped, test_coding_plan_pattern,
               test_kimi_and_zai, test_grouped_patterns_do_not_crash,
               test_lowercase_env_attribution,
               test_key_patterns_dict_coverage,
               test_one_pair_per_value,
               test_npm_placeholder_rejected,
               test_insert_your_token_rejected,
               test_sk_uppercase_x_rejected,
               test_real_sk_accepted,
               test_real_ghp_accepted,
               test_v65_merged_patterns,
               test_elevenlabs_boundary,
               test_dashscope_ws_extraction,
               test_dashscope_long_hex_not_truncated):
        fn()
    _nextra = len(getattr(K, 'EXTRA_PATTERNS', {}))
    print(f"\nPATTERNS={len(K.PATTERNS)} KEY_PATTERNS={len(K.KEY_PATTERNS)} "
          f"EXTRA={_nextra}")
    # loud gate: пустой EXTRA-hook (нет patterns_extra.py рядом с key_patterns)
    # НЕ должен проходить молча — batch-2 и kh_* семьи исчезнут из lanes.
    check("extras-loaded", _nextra >= 17,
          f"EXTRA_PATTERNS={_nextra} — patterns_extra.py не загружается рядом с этим деревом")
    if FAILS:
        print(f"RED: {len(FAILS)} failing -> {FAILS}")
        return 1
    
    print("GREEN: all key_patterns fixture tests passed")
    return 0


if __name__ == "__main__":
    sys.exit(main())