"""Offline tests for commoncrawl_eternal: CDX parse, WARC Range-record
inflate, WARC/HTTP header stripping, config-URL + junk filters, db_add dedup.

No network. Builds a gzip'd WARC record fixture whose HTTP body carries a fake
key, feeds it through _inflate -> warc_http_body -> scan_text, asserts the key
is extracted and written exactly once (dedup on repeat).
"""
import gzip, os, sqlite3, sys

THIS_DIR = os.path.dirname(os.path.abspath(__file__))
DB_PATH = os.path.join(THIS_DIR, "data", "_smoke_commoncrawl.db")
os.environ["KF_DB"] = DB_PATH
os.makedirs(os.path.join(THIS_DIR, "data"), exist_ok=True)
if os.path.exists(DB_PATH):
    os.remove(DB_PATH)

c = sqlite3.connect(DB_PATH)
c.execute("""
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)
""")
c.commit()
c.close()

sys.path.insert(0, THIS_DIR)
import commoncrawl_eternal as cc

FAKE_KEY = "sk-0a1b2c3d4e5f60718293a4b5c6d7e8f9"

# ---- Fixture: one gzipped WARC/1.0 response record ----
_warc_hdr = (
    b"WARC/1.0\r\n"
    b"WARC-Type: response\r\n"
    b"WARC-Date: 2026-09-04T23:33:41Z\r\n"
    b"WARC-Target-URI: https://example.com/.env\r\n"
    b"Content-Length: 200\r\n"       # deliberately wrong -> extractor must clamp to body
    b"Content-Type: application/http; msgtype=response\r\n"
    b"\r\n"
)
_http = (
    b"HTTP/1.1 200 OK\r\n"
    b"Content-Type: text/plain\r\n"
    b"Content-Length: 51\r\n"
    b"\r\n"
    b"OPENAI_API_KEY=" + FAKE_KEY.encode() + b"\n"
)
_warc_rec = _warc_hdr + _http
_warc_gz = gzip.compress(_warc_rec)


def test_parse_cdx_rows():
    text = ('{"url":"https://example.com/.env","status":"200",'
            '"offset":"100","length":"51","filename":"a.warc.gz"}\n'
            '<html>504 Gateway Time-out</html>\n'
            '{ not-json }\n'
            '{"url":"https://example.com/index.html","status":"404"}\n')
    rows = cc.parse_cdx_rows(text)
    assert len(rows) == 2, rows
    assert rows[0]["filename"] == "a.warc.gz"
    print("  test_parse_cdx_rows: PASS")


def test_inflate_and_strip():
    rec = cc._inflate(_warc_gz)
    assert rec.startswith(b"WARC/"), rec[:10]
    body = cc.warc_http_body(rec)
    assert body == b"OPENAI_API_KEY=" + FAKE_KEY.encode() + b"\n", body
    # Content-Length clamp honoured (51 bytes exactly).
    assert len(body) == 51, len(body)
    print("  test_inflate_and_strip: PASS")


def test_truncated_gzip_partial():
    # Range that cuts mid-member: _inflate must still return what decoded.
    partial = _warc_gz[: len(_warc_gz) // 2]
    # A half gzip stream usually yields nothing; just assert no exception.
    out = cc._inflate(partial)
    assert isinstance(out, (bytes, bytearray))
    print("  test_truncated_gzip_partial: PASS")


def test_is_config_url():
    assert cc.is_config_url("https://x.com/.env")
    assert cc.is_config_url("https://x.com/a/b/.env.production")
    assert cc.is_config_url("https://x.com/wp-config.php")
    assert cc.is_config_url("https://x.com/secrets.yaml")
    assert cc.is_config_url("https://x.com/.env?v=1")
    assert not cc.is_config_url("https://x.com/index.html")
    assert not cc.is_config_url("https://x.com/env")
    print("  test_is_config_url: PASS")


def test_is_junk():
    assert cc.is_junk("sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx")
    assert cc.is_junk("sk-your_api_key_here_placeholder")
    assert cc.is_junk("aaaaaaaaaaaaaaaaaaaa")
    assert cc.is_junk("short")                      # <30, no digit
    assert not cc.is_junk(FAKE_KEY)                 # real-shaped 35-char hex
    print("  test_is_junk: PASS")


def test_scan_and_db_dedup():
    body = cc.warc_http_body(cc._inflate(_warc_gz)).decode("utf-8", "replace")
    hits = [(p, k) for p, k in cc.scan_text(body) if not cc.is_junk(k)]
    assert any(k == FAKE_KEY for _, k in hits), hits
    before = cc.total
    for prov, key in hits:
        cc.db_add(key, prov, "https://example.com/.env @ a.warc.gz:100")
    after = cc.total
    assert after > before, (before, after)
    # dedup: re-adding same key must not create a second row (hash UNIQUE)
    cur = sqlite3.connect(DB_PATH).execute(
        "SELECT COUNT(*) FROM keys WHERE val=?", (FAKE_KEY,))
    n1 = cur.fetchone()[0]
    assert n1 == 1, n1
    for prov, key in hits:
        cc.db_add(key, prov, "again")
    cur = sqlite3.connect(DB_PATH).execute(
        "SELECT COUNT(*) FROM keys WHERE val=?", (FAKE_KEY,))
    assert cur.fetchone()[0] == 1, "dedup failed"
    print(f"  test_scan_and_db_dedup: PASS (rows for key stayed 1)")


if __name__ == "__main__":
    test_parse_cdx_rows()
    test_inflate_and_strip()
    test_truncated_gzip_partial()
    test_is_config_url()
    test_is_junk()
    test_scan_and_db_dedup()
    print("\nAll commoncrawl tests passed.")