keyhunter-parser/test_dockerhub.py.disabled

"""Test dockerhub_eternal lane: offline fixture parsing + db_add + dedup."""
import hashlib
import os
import sqlite3
import sys
import tempfile

# Point DB to temp file
os.environ["KF_DB"] = tempfile.mktemp(suffix=".db")

# Import after setting env
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import dockerhub_eternal as dh

# --- fixture: realistic config JSON with Env containing secrets ---
FIXTURE_CONFIG = {
    "config": {
        "Env": [
            "PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
            "OPENAI_API_KEY=sk-proj-AbCdEfGhIjKlMnOpQrStUvWxYz1234567890ABCDEFGHIJKLMNOPQRSTUVWXYZ",
            "BOT_TOKEN=1234567890:AAHjKlMnOpQrStUvWxYzAbCdEfGhIjKlMnO",
            "REDIS_URL=redis://:password@redis:6379",
            "PGPASS=postgresql://user:randompassword123@localhost:5432/db",
            "ANTHROPIC_KEY=sk-ant-api03-AbCdEfGhIjKlMnOpQrStUvWxYz1234567890ABCDEFGHIJKLMNOPQRSTUV-abCdEfGh",
            "PLACEHOLDER=sk-your-secret-key-here-1234567890",
            "TEST_VAR=test1234",
            "SHORT=abc",
            "DASHSCOPE_KEY=sk-aabbccddee0011223344556677889900",
            "HUGGINGFACE_TOKEN=hf_AbCdEfGhIjKlMnOpQrStUvWxYz1234567890AbCd",
            "ANY_SECRET=hf_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
        ]
    }
}

DB_CONTRACT_SQL = """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 setup_db():
    conn = sqlite3.connect(os.environ["KF_DB"])
    conn.execute(DB_CONTRACT_SQL)
    conn.commit()
    conn.close()


def test_scan_env_whole_text():
    """Config Env scanning finds keys and rejects junk."""
    total_before = dh.total
    n = dh.scan_env_whole_text(FIXTURE_CONFIG, "docker:library/test:latest")
    assert n >= 4, f"Expected >=4 keys, got {n}"

    # Verify DB rows
    conn = sqlite3.connect(os.environ["KF_DB"])
    rows = conn.execute("SELECT val, prov, found, repo FROM keys").fetchall()
    conn.close()

    vals = {r[0] for r in rows}
    provs = {r[1] for r in rows}

    # Should have OPENAI key
    assert any("sk-proj-AbCdEfGhIjKlMnOpQrStU" in v for v in vals), \
        "Missing OpenAI key"
    assert "OPENAI" in provs, f"Missing OPENAI provider, got {provs}"

    # Should have TELEGRAM bot token
    assert any("AAHjKlMnOpQrStUvWxYz" in v for v in vals), \
        "Missing Telegram token"
    assert "TELEGRAM" in provs, f"Missing TELEGRAM provider, got {provs}"

    # Should have ANTHROPIC key
    assert any("sk-ant-api03" in v for v in vals), \
        "Missing Anthropic key"
    assert "ANTHROPIC" in provs, f"Missing ANTHROPIC provider, got {provs}"

    # Should have DASHSCOPE key
    assert any("sk-aabbccddee0011223344556677889900" in v for v in vals), \
        "Missing DashScope key"
    assert "DASHSCOPE" in provs, f"Missing DASHSCOPE provider, got {provs}"

    assert "HUGGINGFACE" in provs, f"Missing HUGGINGFACE provider, got {provs}"

    # Junk filter should reject placeholder key
    for r in rows:
        assert "your" not in r[0].lower(), f"Junk leaked: {r[0]}"

    # Junk filter should reject test keys
    for r in rows:
        assert "test" not in r[0].lower() or "test" in r[2].lower(), \
            f"Junk leaked: {r[0]}"

    print(f"  Found {len(rows)} total keys, {len(set(vals))} unique")

    # Verify found tag
    for r in rows:
        assert r[3] == "docker:library/test:latest", f"Wrong repo: {r[3]}"
        assert r[2] == "dockerhub", f"Wrong found tag: {r[2]}"


def test_dedup():
    """Running scan_env_whole_text again should not duplicate (DB dedup)."""
    conn = sqlite3.connect(os.environ["KF_DB"])
    before = conn.execute("SELECT COUNT(*) FROM keys").fetchone()[0]
    conn.close()
    dh.scan_env_whole_text(FIXTURE_CONFIG, "docker:library/test:latest")
    conn = sqlite3.connect(os.environ["KF_DB"])
    after = conn.execute("SELECT COUNT(*) FROM keys").fetchone()[0]
    conn.close()
    assert after == before, f"Rows changed: {before} -> {after}"
    print(f"  Dedup OK: {after} rows unchanged")


def test_db_add_direct():
    """db_add writes a row with correct schema."""
    test_val = "sk-proj-AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
    dh.db_add(test_val, "OPENAI", "docker:test/test:latest")
    conn = sqlite3.connect(os.environ["KF_DB"])
    row = conn.execute(
        "SELECT val, prov, found, repo, hash FROM keys WHERE val=?",
        (test_val,)
    ).fetchone()
    conn.close()
    assert row is not None, "Row not found"
    assert row[2] == "dockerhub", f"Expected found='dockerhub', got '{row[2]}'"
    assert row[3] == "docker:test/test:latest"
    expected_hash = hashlib.sha1(row[0].encode()).hexdigest()[:16]
    assert row[4] == expected_hash, "Hash mismatch"
    print(f"  db_add row: prov={row[1]} hash={row[4]}")


def test_db_add_dedup():
    """db_add with same key updates existing row (no duplicate)."""
    key = "sk-proj-BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"
    dh.db_add(key, "OPENAI", "docker:test/a:latest")
    conn = sqlite3.connect(os.environ["KF_DB"])
    count_before = conn.execute("SELECT COUNT(*) FROM keys").fetchone()[0]
    conn.close()

    dh.db_add(key, "OPENAI", "docker:test/a:latest")
    conn = sqlite3.connect(os.environ["KF_DB"])
    count_after = conn.execute("SELECT COUNT(*) FROM keys").fetchone()[0]
    conn.close()
    assert count_before == count_after, f"Dedup failed: {count_before} -> {count_after}"
    print(f"  db_add dedup OK: {count_before} rows")


def test_is_junk():
    assert dh.is_junk("sk-your-secret"), "Did not catch 'your'"
    assert dh.is_junk("testtest12345"), "Did not catch 'testtest'"
    assert dh.is_junk("aaaaaa"), "Did not catch low-entropy"
    assert dh.is_junk("dummy_value_here"), "Did not catch 'DUMMY'"
    assert not dh.is_junk("sk-proj-AbCdEfGhIjKlMnOpQrStUvWxYz123456"), "False junk"
    print("  Junk filter OK")


def test_scan_config_env():
    """Legacy scan_config_env also works (non-ENV_RE patterns)."""
    n = dh.scan_config_env(FIXTURE_CONFIG, "docker:test/legacy:latest")
    # scan_config_env only processes individual env values as direct text patterns
    vals_sofar = set()
    conn = sqlite3.connect(os.environ["KF_DB"])
    rows = conn.execute("SELECT val FROM keys").fetchall()
    conn.close()
    print(f"  scan_config_env contributed {n} keys")


if __name__ == "__main__":
    setup_db()
    test_is_junk()
    test_scan_env_whole_text()
    test_dedup()
    test_db_add_direct()
    test_db_add_dedup()
    test_scan_config_env()
    print("\nALL PASS")