"""Behavior checks for dashscope_hunt.py — no network, plain asserts.

Covers the three changes made on 2026-10-06:
  1. classify_http: 429/400-arrears/403-model are NOT "dead"
  2. validate_key: DEAD only when every probe rejected; provider recorded is the
     one the probe proved (regex cannot tell DASHSCOPE from DEEPSEEK)
  3. fleet sharding: KF_DORK_MOD/KF_DORK_SHARD slices are disjoint and complete

Run: python test_dashscope_hunt.py
"""
import importlib.util
import os
import shutil
import sqlite3
import subprocess
import sys
import tempfile
import uuid

HERE = os.path.dirname(os.path.abspath(__file__))
TARGET = os.path.join(HERE, "dashscope_hunt.py")
# pre-fix snapshot kept by the portability pass; used to prove the old behaviour
OLD_CANDIDATES = [
    r"C:\Users\User\tmp\kf_backup_20261006_003939\dashscope_hunt.py",
    r"C:\Users\User\tmp\kf_backup_20261006_012235\dashscope_hunt.py",
]
SCHEMA = ("CREATE TABLE IF NOT EXISTS keys (id INTEGER PRIMARY KEY AUTOINCREMENT,"
          " hash TEXT UNIQUE, val TEXT, prov TEXT, status TEXT DEFAULT 'NEW',"
          " plan TEXT, price TEXT, remaining TEXT, found TEXT, repo TEXT)")


def load(path, db, extra_path=None):
    os.environ["KF_DB"] = db
    if extra_path and extra_path not in sys.path:
        sys.path.insert(0, extra_path)
    if HERE not in sys.path:
        sys.path.insert(0, HERE)
    name = "dsh_" + str(abs(hash(path)))
    spec = importlib.util.spec_from_file_location(name, path)
    assert spec is not None and spec.loader is not None, f"cannot load {path}"
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


def fresh_db(tmp):
    db = os.path.join(tmp, f"keys_test_{uuid.uuid4().hex[:8]}.db")
    c = sqlite3.connect(db)
    c.execute(SCHEMA)
    c.execute("INSERT INTO keys (hash, val, prov, status, found, repo) "
              "VALUES ('h1', 'sk-0123456789abcdef0123456789abcdef', "
              "'DASHSCOPE', 'NEW', 'test', 'o/r')")
    c.commit()
    c.close()
    return db


def row(db):
    c = sqlite3.connect(db)
    r = c.execute("SELECT status, prov, plan FROM keys WHERE hash='h1'").fetchone()
    c.close()
    return r


def patch(mod, ds, deep, moon, sili):
    setattr(mod, "probe_dashscope", lambda _k: ds)
    setattr(mod, "probe_deepseek", lambda _k: deep)
    setattr(mod, "probe_moonshot", lambda _k: moon)
    setattr(mod, "probe_siliconflow", lambda _k: sili)


def test_classify(mod):
    assert mod.classify_http(429, "Requests rate limit exceeded") == ("throttled", 429)
    assert mod.classify_http(400, "Access denied, please make sure your account"
                             " is in good standing. Type: Arrearage")[0] == "arrearage"
    assert mod.classify_http(400, "账户欠费")[0] == "arrearage"
    assert mod.classify_http(403, "Model not authorized")[0] == "no_model"
    assert mod.classify_http(404, "Model does not exist")[0] == "no_model"
    assert mod.classify_http(401, "Incorrect API key provided") == ("reject", 401)
    print("ok classify_http: 429->throttled 400/欠费->arrearage 403/404-model->no_model 401->reject")


def test_validate(mod, label):
    tmp = tempfile.mkdtemp(prefix="kf_t_")
    try:
        # a) proved provider wins over the regex label
        db = fresh_db(tmp)
        m2 = load(mod.__file__, db, HERE)
        patch(m2, ("reject", 401), ("WORKING", "deepseek $1.5"),
              ("net", None), ("reject", 401))
        st, info = m2.validate_key("h1", "sk-0123456789abcdef0123456789abcdef",
                                   "DASHSCOPE", "o/r")
        assert st == "WORKING", st
        assert row(db) == ("WORKING", "DEEPSEEK", "DEEPSEEK: deepseek $1.5"), row(db)

        # b) one transient probe => row stays NEW, never DEAD
        db = fresh_db(tmp)
        m3 = load(mod.__file__, db, HERE)
        patch(m3, ("reject", 401), ("throttled", 429), ("reject", 401), ("net", None))
        st, _ = m3.validate_key("h1", "sk-x", "DASHSCOPE", "o/r")
        assert st == "NEW", st
        assert row(db)[0] == "NEW", row(db)

        # c) arrears => LIMITED with the reason recorded
        db = fresh_db(tmp)
        m4 = load(mod.__file__, db, HERE)
        patch(m4, ("arrearage", 400), ("reject", 401), ("reject", 401), ("reject", 401))
        st, _ = m4.validate_key("h1", "sk-x", "DASHSCOPE", "o/r")
        assert st == "LIMITED", st
        assert row(db)[:2] == ("LIMITED", "DASHSCOPE"), row(db)

        # d) every probe rejected => DEAD
        db = fresh_db(tmp)
        m5 = load(mod.__file__, db, HERE)
        patch(m5, ("reject", 401), ("reject", 401), ("reject", 401), ("reject", 401))
        st, _ = m5.validate_key("h1", "sk-x", "DASHSCOPE", "o/r")
        assert st == "DEAD", st
        assert row(db)[0] == "DEAD", row(db)

        # e) auth-blind /models is transient BY DESIGN: CDN edges intermittently
        #    serve a cached public catalog, so the row stays NEW and gets retried.
        #    An earlier revision of this suite demanded a terminal LIMITED here;
        #    that was dropped on evidence - plan keys now get a real verdict from
        #    the tier-2 coding-plan probe, and for sk- keys the cache clears by
        #    itself. What must not regress is the note: it has to name the class,
        #    otherwise the retry reason is invisible in every node log.
        db = fresh_db(tmp)
        m6 = load(mod.__file__, db, HERE)
        patch(m6, ("UNVERIFIED_AUTHBLIND", "beijing"), ("reject", 401),
              ("reject", 401), ("reject", 401))
        st, note = m6.validate_key("h1", "sk-x", "DASHSCOPE", "o/r")
        assert st == "NEW", st
        assert note == "retry:UNVERIFIED_AUTHBLIND", note
        assert row(db)[0] == "NEW", row(db)

        # f) a shape this hunt cannot judge must never be written DEAD: 401 from
        #    every sk- endpoint only proves we asked the wrong host.
        db = fresh_db(tmp)
        m7 = load(mod.__file__, db, HERE)
        st, note = m7.validate_key("h1", "nvapi-" + "Kx7Q" * 16, "NVIDIA", "o/r")
        assert st == "NEW", st
        assert note.startswith("no-probe-for-shape"), note
        assert row(db)[0] == "NEW", row(db)
        print(f"ok validate_key [{label}]: provider attribution / throttled->NEW / "
              f"arrearage->LIMITED / all-reject->DEAD / auth-blind->NEW+named-retry / "
              f"unprobeable shape->NEW (never DEAD)")
    finally:
        shutil.rmtree(tmp, ignore_errors=True)


def test_old_behaviour_was_wrong():
    old = next((p for p in OLD_CANDIDATES if os.path.exists(p)), None)
    if not old:
        print("skip RED proof: no pre-fix snapshot found")
        return
    tmp = tempfile.mkdtemp(prefix="kf_red_")
    try:
        fixture = os.path.join(tmp, "old")
        os.makedirs(fixture, exist_ok=True)
        shutil.copy2(old, os.path.join(fixture, "dashscope_hunt.py"))
        shutil.copy2(os.path.join(HERE, "key_patterns.py"),
                     os.path.join(fixture, "key_patterns.py"))
        db = os.path.join(tmp, "keys_old.db")
        c = sqlite3.connect(db); c.execute(SCHEMA)
        c.execute("INSERT INTO keys (hash,val,prov,status) VALUES ('h1','sk-x','DASHSCOPE','NEW')")
        c.commit(); c.close()
        m = load(os.path.join(fixture, "dashscope_hunt.py"), db, fixture)
        # the pre-fix snapshot ignores KF_DB (it hardcodes the production path),
        # so redirect its connection: this test must never write to the real DB
        m.conn = sqlite3.connect(db, isolation_level=None)  # module expects autocommit
        # throttled key, old code: every non-200 was ("reject", code)
        setattr(m, "probe_dashscope", lambda _k: ("reject", 429))
        setattr(m, "probe_deepseek", lambda _k: ("reject", 429))
        setattr(m, "probe_moonshot", lambda _k: ("reject", 429))
        setattr(m, "probe_siliconflow", lambda _k: ("reject", 429))
        st, _ = m.validate_key("h1", "sk-x", "DASHSCOPE", "o/r")
        c = sqlite3.connect(db)
        got = c.execute("SELECT status, prov FROM keys WHERE hash='h1'").fetchone()
        c.close()
        assert st == "DEAD" and got[0] == "DEAD", f"expected old code to kill it, got {st} {got}"
        print(f"ok RED proof: pre-fix code marked a 429-throttled key DEAD "
              f"(status={got[0]}, prov={got[1]}) — that is the bug being fixed")
    finally:
        shutil.rmtree(tmp, ignore_errors=True)


def test_shards(mod):
    full = list(mod.DORKS)
    assert len(full) == len(set(full)), "duplicate dorks"
    assert any("\u767e\u70bc" in d or "\u901a\u4e49\u5343\u95ee" in d for d in full), \
        "Chinese-language dorks missing"
    assert any("MODELSCOPE" in d.upper() for d in full), "ModelScope dorks missing"
    for fam in ("MOONSHOT_API_KEY", "DEEPSEEK_API_KEY", "SILICONFLOW_API_KEY", "KIMI_API_KEY"):
        assert any(fam in d for d in full), f"{fam} dork missing"
    N = 4
    env_base = dict(os.environ, KF_DB=os.path.join(tempfile.gettempdir(), "kf_shard.db"))
    slices = []
    for i in range(N):
        env = dict(env_base, KF_DORK_MOD=str(N), KF_DORK_SHARD=str(i))
        code = ("import sys; sys.path.insert(0, r'%s');"
                "import dashscope_hunt as d; sys.stdout.write('\\x1e'.join(d.DORKS))" % HERE)
        out = subprocess.run([sys.executable, "-c", code], env=env,
                             capture_output=True, text=True, timeout=120)
        assert out.returncode == 0, out.stderr[-300:]
        slices.append(set(out.stdout.split("\x1e")) - {""})
    union = set().union(*slices)
    assert union == set(full), f"shards lost dorks: {len(union)} vs {len(full)}"
    for a in range(N):
        for b in range(a + 1, N):
            assert not (slices[a] & slices[b]), f"shards {a},{b} overlap"
    assert all(slices), "empty shard"
    print(f"ok sharding: {len(full)} dorks -> {N} disjoint slices "
          f"{[len(s) for s in slices]}, union complete")


class _Resp:
    def __init__(self, status_code, text="", payload=None):
        self.status_code = status_code
        self.text = text
        self._payload = payload or {}

    def json(self):
        return self._payload


def test_dashscope_regions(mod):
    # documented Model Studio error matrix
    assert mod.classify_ds(200)[0] == "WORKING"
    assert mod.classify_ds(429, '{"code":"Throttling.RateQuota"}')[0] == "throttled"
    assert mod.classify_ds(429, 'Too Many Requests')[0] == "throttled"
    assert mod.classify_ds(429, '{"code":"Throttling.AllocationQuota"}')[0] == "WORKING"
    assert mod.classify_ds(403, '{"code":"Model.AccessDenied"}')[0] == "WORKING"
    assert mod.classify_ds(403, '{"code":"AllocationQuota.FreeTierOnly"}')[0] == "free_tier"
    assert mod.classify_ds(403, '{"code":"AccessDenied.Unpurchased"}')[0] == "free_tier"
    assert mod.classify_ds(400, '{"code":"Arrearage"}')[0] == "arrearage"
    assert mod.classify_ds(401, '{"code":"InvalidApiKey"}') == ("reject", 401)
    # Regression: a SECOND DS_ALIVE_WORDS assignment used to shadow the real one
    # and add "throttling"/"rate limit"/"too many requests", so any WAF or
    # secondary-limit body classified as a live key. None of these bodies says
    # anything about the key, so none may ever reach WORKING.
    for code, body in ((403, '{"message":"You have exceeded a secondary rate limit"}'),
                       (403, '{"message":"API rate limit exceeded"}'),
                       (503, 'Too Many Requests'),
                       (400, 'Requests rate limit exceeded')):
        assert mod.classify_ds(code, body)[0] != "WORKING", (code, body)
    assert mod.DS_ALIVE_WORDS == ("model.accessdenied",), mod.DS_ALIVE_WORDS

    real_get = mod.requests.get
    # a key that only lives in Singapore must be found, not marked dead
    calls = []

    def fake_get(url, headers=None, timeout=None):
        calls.append(url)
        # _key_attributed() re-GETs the same URL with GARBAGE_KEY. An endpoint that
        # 200s for the control serves a cached public catalog and proves nothing
        # about THIS key, so the fake must reject the control - otherwise every
        # real 200 reads as auth-blind and the probe can never return WORKING.
        if mod.GARBAGE_KEY in (headers or {}).get("Authorization", ""):
            return _Resp(401, '{"code":"InvalidApiKey"}')
        if "dashscope-intl" in url:
            return _Resp(200, "", {"data": [{"id": "qwen-plus"}] * 7})
        return _Resp(401, '{"code":"InvalidApiKey"}')

    setattr(mod.requests, "get", fake_get)
    try:
        st, info = mod.probe_dashscope("sk-0123456789abcdef0123456789abcdef")
    finally:
        setattr(mod.requests, "get", real_get)
    assert st == "WORKING", (st, info)
    assert "intl" in info and "models=7" in info, info
    # stop early: exactly one call past the first live region, and that call must
    # be the auth-blind control re-GET of the SAME url - not another region.
    live_idx = next(i for i, u in enumerate(calls) if "dashscope-intl" in u)
    assert len(calls) == live_idx + 2, f"must stop at the first live region, called {calls}"
    assert calls[-1] == calls[live_idx], f"extra call is not the control: {calls}"
    assert all("chat/completions" not in u for u in calls), "validation must not spend tokens"

    # free-tier exhaustion is not "dead", and sk-sp- keys also try the plan hosts
    calls2 = []

    def fake_get2(url, headers=None, timeout=None):
        calls2.append(url)
        return _Resp(403, '{"code":"AllocationQuota.FreeTierOnly"}')

    setattr(mod.requests, "get", fake_get2)
    try:
        st2, info2 = mod.probe_dashscope("sk-sp-plankey")
    finally:
        setattr(mod.requests, "get", real_get)
    assert st2 == "free_tier", (st2, info2)
    assert any("token-plan" in u for u in calls2), "sk-sp- must try plan endpoints"
    print(f"ok dashscope probe: {len(mod.DS_REGIONS)} regions + "
          f"{len(mod.DS_PLAN_REGIONS)} plan hosts, zero-token /v1/models, "
          f"429/Model.AccessDenied stay WORKING, FreeTierOnly != dead")

def test_endpoint_error_guard(mod):
    """A missing/HTML endpoint must never be counted as a dead key."""
    assert mod.classify_ds(404, '{"code":"ModelNotFound"}')[0] == "endpoint_error"
    assert mod.classify_ds(404, "<html><body>404 Not Found</body></html>")[0] == "endpoint_error"
    assert mod.classify_ds(None, "")[0] == "endpoint_error"
    assert mod.classify_ds(401, '{"code":"InvalidApiKey"}')[0] == "reject"
    assert mod.ds_model_count({"output": {"total": 519, "models": [1]}}) == 519
    assert mod.ds_model_count({"data": [{"id": "a"}, {"id": "b"}]}) == 2

    real_get = mod.requests.get

    def all_404(url, headers=None, timeout=None):
        return _Resp(404, "<html>nginx 404</html>")

    setattr(mod.requests, "get", all_404)
    try:
        st, info = mod.probe_dashscope("sk-0123456789abcdef0123456789abcdef")
    finally:
        setattr(mod.requests, "get", real_get)
    assert st == "net", f"every-endpoint-404 must read as net, got {st} {info}"
    print("ok endpoint guard: 404/HTML/None -> endpoint_error, all-404 probe -> net "
          "(row stays NEW, never DEAD); model count parsed from native and compat shapes")


def test_authblind_and_plan_tier2(mod):
    """A cached catalog must not discard a real verdict, and a plan key must
    resolve by spending one token instead of looping as NEW forever."""
    real_get, real_post = mod.requests.get, mod.requests.post
    calls = []

    def fake_get(url, headers=None, timeout=None):
        calls.append(url)
        garbage = mod.GARBAGE_KEY in (headers or {}).get("Authorization", "")
        if "/api/v1/models" in url:   # native shape: CDN-cached public catalog
            return _Resp(200, "", {"output": {"total": 519}})
        if garbage:                   # compat shape: honest, key-attributed
            return _Resp(401, '{"code":"InvalidApiKey"}')
        return _Resp(200, "", {"data": [{"id": "qwen-plus"}] * 5})

    setattr(mod.requests, "get", fake_get)
    try:
        st, info = mod.probe_dashscope("sk-0123456789abcdef0123456789abcdef")
    finally:
        setattr(mod.requests, "get", real_get)
    assert st == "WORKING", (st, info)
    assert "models=5" in info, info
    assert any("/api/v1/models" in u for u in calls), "native shape not tried"
    assert all("chat/completions" not in u for u in calls), "tier 1 must stay free"

    posts = []

    def blind_get(url, headers=None, timeout=None):
        return _Resp(200, "", {"data": [{"id": "x"}]})   # 200 even for the control

    def fake_post(url, headers=None, json=None, timeout=None):
        posts.append(url)
        return _Resp(200, "")

    setattr(mod.requests, "get", blind_get)
    setattr(mod.requests, "post", fake_post)
    try:
        st2, info2 = mod.probe_dashscope("sk-sp-plankey")
    finally:
        setattr(mod.requests, "get", real_get)
        setattr(mod.requests, "post", real_post)
    assert st2 == "WORKING", (st2, info2)
    assert "coding-plan" in info2, info2
    assert posts == ["https://coding.dashscope.aliyuncs.com/v1/chat/completions"], posts
    print("ok auth-blind: cached native catalog falls through to the honest compat "
          "shape (models=5, zero tokens); sk-sp- resolved by ONE token on the "
          "coding-plan host instead of looping NEW forever")


def test_recency_columns(mod):
    """first_seen/last_seen must exist and behave: the old schema has neither, so
    this also proves the idempotent ALTER migration path on a live-shape DB."""
    tmp = tempfile.mkdtemp(prefix="kf_rec_")
    try:
        db = fresh_db(tmp)                     # SCHEMA has no recency columns
        m = load(mod.__file__, db, HERE)
        cols_before = [r[1] for r in sqlite3.connect(db).execute("PRAGMA table_info(keys)")]
        assert "first_seen" not in cols_before, cols_before
        h = m.db_upsert("sk-0123456789abcdef0123456789abcdef", "DASHSCOPE", "o/r")
        c = sqlite3.connect(db)
        cols = [r[1] for r in c.execute("PRAGMA table_info(keys)")]
        assert "first_seen" in cols and "last_seen" in cols, cols
        row1 = c.execute("SELECT first_seen, last_seen, status, found FROM keys WHERE hash=?", (h,)).fetchone()
        assert row1[0] and row1[0] == row1[1], row1
        # re-find: last_seen is the measurement, first_seen must NOT move,
        # and a WORKING row must not be demoted to NEW by the re-find
        c.execute("UPDATE keys SET status='WORKING', first_seen='2020-01-01T00:00:00Z' WHERE hash=?", (h,))
        c.commit(); c.close()
        m.db_upsert("sk-0123456789abcdef0123456789abcdef", "DASHSCOPE", "o/r2")
        c = sqlite3.connect(db)
        row2 = c.execute("SELECT first_seen, last_seen, status, repo FROM keys WHERE hash=?", (h,)).fetchone()
        assert row2[0] == "2020-01-01T00:00:00Z", row2
        assert row2[1] != "2020-01-01T00:00:00Z", row2
        assert row2[2] == "WORKING", ("re-find must not demote", row2)
        assert row2[3] == "o/r", ("repo must not move for non-ERR rows", row2)
        print("ok recency: ALTER migration on an old schema, first_seen stable, "
              "last_seen bumped on re-find, WORKING/ERR semantics unchanged")
    finally:
        shutil.rmtree(tmp, ignore_errors=True)


def test_recency_fallback(mod):
    """A failed migration must cost recency data, never the harvest.

    The ALTER loop is non-atomic: first_seen can land while last_seen hits a lock
    (run_imba.py documents 11 writers on this one WAL file). If db_upsert then
    emitted the 8-column INSERT unconditionally, every call would raise "table keys
    has no column named last_seen", work() would swallow it per dork, and the run
    would print "HUNT done: +0 keys" and exit 0 - a silent fleet-wide zero-yield
    failure that node-health gating reports as 1390/1390 ok.
    """
    tmp = tempfile.mkdtemp(prefix="kf_rec_fb_")
    try:
        db = fresh_db(tmp)                 # old schema: no recency columns
        m = load(mod.__file__, db, HERE)

        class NoAlter:
            def __init__(self, real):
                self._real = real
            def execute(self, sql, *a):
                if sql.strip().upper().startswith("ALTER"):
                    raise sqlite3.OperationalError("database is locked")
                return self._real.execute(sql, *a)
            def __getattr__(self, name):
                return getattr(self._real, name)

        real_conn = sqlite3.connect(db, isolation_level=None)
        m.conn = NoAlter(real_conn)
        m._recency_ready[0] = None
        h = m.db_upsert("sk-0123456789abcdef0123456789abcdef", "DASHSCOPE", "o/r")
        assert h, "upsert must still return a hash"
        cols = [r[1] for r in sqlite3.connect(db).execute("PRAGMA table_info(keys)")]
        assert "last_seen" not in cols, cols
        row = sqlite3.connect(db).execute(
            "SELECT status, prov, found, repo FROM keys WHERE hash=?", (h,)).fetchone()
        assert row == ("NEW", "DASHSCOPE", "ds-hunt", "o/r"), row
        assert m._recency_ready[0] is False, m._recency_ready
        # and the verified path must still be chosen once the columns do exist
        m2 = load(mod.__file__, db, HERE)
        m2.db_upsert("sk-0123456789abcdef0123456789abcdee", "DASHSCOPE", "o/r")
        cols2 = [r[1] for r in sqlite3.connect(db).execute("PRAGMA table_info(keys)")]
        assert {"first_seen", "last_seen"} <= set(cols2), cols2
        print("ok recency fallback: locked ALTER -> 6-column upsert, row still written, "
              "flag verified False; healthy DB -> columns added and used")
    finally:
        shutil.rmtree(tmp, ignore_errors=True)


def test_recency_concurrent_probe(mod):
    """8 workers racing the migration must all end up instrumented.

    _ensure_recency_columns() is called OUTSIDE dblock by db_upsert and work()
    runs under a ThreadPoolExecutor, so without double-checked locking one thread
    can read PRAGMA table_info between the two ALTERs (isolation_level=None makes
    each one commit immediately), see only first_seen, and latch
    _recency_ready[0]=False for the whole process lifetime - under --loop that is
    days of a node silently running the 6-column fallback on a DB that HAS the
    columns.
    """
    import threading
    tmp = tempfile.mkdtemp(prefix="kf_rec_race_")
    try:
        db = fresh_db(tmp)                  # old schema, no recency columns
        m = load(mod.__file__, db, HERE)
        m._recency_ready[0] = None
        errors = []

        def hit(i):
            try:
                m.db_upsert("sk-%032x" % i, "DASHSCOPE", "o/r")
            except Exception as e:          # any raise here = the failure mode
                errors.append(repr(e))

        threads = [threading.Thread(target=hit, args=(i,)) for i in range(8)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()
        assert not errors, errors
        cols = [r[1] for r in sqlite3.connect(db).execute("PRAGMA table_info(keys)")]
        assert {"first_seen", "last_seen"} <= set(cols), cols
        assert m._recency_ready[0] is True, m._recency_ready
        # fresh_db() seeds one row ('h1'), so count only what these threads wrote
        n = sqlite3.connect(db).execute(
            "SELECT COUNT(*) FROM keys WHERE found='ds-hunt'").fetchone()[0]
        assert n == 8, n
        stamped = sqlite3.connect(db).execute(
            "SELECT COUNT(*) FROM keys WHERE found='ds-hunt' AND last_seen IS NOT NULL"
        ).fetchone()[0]
        assert stamped == 8, ("instrumented write must stamp last_seen", stamped)
        print("ok recency race: 8 threads, no error, columns present, flag True, 8 rows")
    finally:
        shutil.rmtree(tmp, ignore_errors=True)


def main():
    tmpdb = os.path.join(tempfile.gettempdir(), "kf_test_main.db")
    mod = load(TARGET, tmpdb, HERE)
    print(f"loaded {TARGET} | DORKS={len(mod.DORKS)} PATS={len(mod.POOL.tokens)} "
          f"RAW_BUDGET={mod.RAW_BUDGET} PAGES={mod.PAGES}")
    test_classify(mod)
    test_dashscope_regions(mod)
    test_validate(mod, "current")
    test_old_behaviour_was_wrong()
    test_shards(mod)
    test_endpoint_error_guard(mod)
    test_authblind_and_plan_tier2(mod)
    test_recency_columns(mod)
    test_recency_fallback(mod)
    test_recency_concurrent_probe(mod)
    print("\nALL TESTS PASS")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())