keyhunter-parser/test_vscode.py

"""Offline test for vscode_eternal.py — no network. Builds a fixture VSIX zip
in memory from scratch, verifies: registry query stub, VSIX unzip + scan_text,
junk filter, db_add + dedup.
"""
import hashlib, io, os, sqlite3, sys, zipfile

THIS_DIR = os.path.dirname(os.path.abspath(__file__))
DB_PATH = os.path.join(THIS_DIR, "data", "_smoke_vscode.db")
os.environ["KF_DB"] = DB_PATH

os.makedirs(os.path.join(THIS_DIR, "data"), exist_ok=True)

# Drop + recreate smoke DB so each test run is clean
if os.path.exists(DB_PATH):
    os.remove(DB_PATH)

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

sys.path.insert(0, THIS_DIR)
import vscode_eternal as vse


# ── Fixture: build a synthetic VSIX zip in memory ──────────────────

def build_fixture_vsix():
    """Zip up a fake VSIX with extension/package.json + a JS file containing a
    real-ish API key pattern that scan_text should catch."""
    buf = io.BytesIO()
    z = zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED)

    # package.json (metadata only)
    z.writestr("extension/package.json", json.dumps({
        "name": "test-ext",
        "publisher": "testpub",
        "version": "1.0.0",
        "engines": {"vscode": "^1.90.0"}
    }))

    # Fake bundled JS containing an OpenAI API key for scan_text to find
    z.writestr("extension/dist/bundle.js",
        '/*! Forge test extension */\n'
        'const API_KEY = "sk-proj-WXYZmnopqr7890ABCDEFGHIJKLMNOPQRSTUVWXYZ"\n'
        'const endpoint = "https://api.openai.com"\n'
        'module.exports = { /* stub */ }\n')

    # Fake config JSON with an Azure key
    z.writestr("extension/config.json",
        '{\n'
        '  "azure_openai_key": "ZZYYXXWWVVUUTTSSRRQQPPOONNMMLLKKJJIIHHGGFF",\n'
        '  "endpoint": "https://my-resource.openai.azure.com/"\n'
        '}\n')

    # A file that should be ignored (node_modules noise)
    z.writestr("extension/node_modules/axios/bundle.min.js",
        'var x="sk-proj-NOTSCANNED";')

    # A .env file with a real-looking key
    z.writestr(".env",
        'GITHUB_TOKEN=ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnop\n')

    z.close()
    return buf.getvalue()


# ── Fixture: a fake key that should be filtered as junk ────────────

FAKE_VSIX_JUNK = None

import json


# ── Tests ──────────────────────────────────────────────────────────

def test_is_junk():
    assert vse.is_junk("sk-xxxx-example")
    assert vse.is_junk("your_api_key_12345678901234567890")
    assert vse.is_junk("aaaaaabbbbbbccccccddddddeeeee")  # low entropy, len(set) < 6
    # This key uses chars NOT in GARBAGE: no "1234567890abcdef" substring
    assert not vse.is_junk("sk-proj-WXYZmnopqr7890ABCDEFGHIJKLMNOPQRSTUVWXYZ")
    print("  test_is_junk: PASS")


def test_scan_vsix():
    """Build a fixture VSIX, run scan_vsix on it, verify keys extracted."""
    raw = build_fixture_vsix()
    assert len(raw) > 200, f"fixture too small: {len(raw)}"

    # Reset total counter for this test
    vse.total = 0

    n = vse.scan_vsix(raw, "testpub/test-ext")
    print(f"  test_scan_vsix: found {n} keys (expect >= 2)")

    # The node_modules file should be skipped (SKIP_DIRS)
    # The .env file is outside extension/ but should still be scanned
    assert n >= 2, f"found {n} keys, expected >= 2"
    print("  test_scan_vsix: PASS")


def test_db_add():
    """db_add writes a row, dedupes on repeat."""
    vse.total = 0
    vse.db_add("sk-proj-WXYZmnopqr7890ABCDEFGHIJKLMNOPQRSTUVWXYZ",
               "OPENAI", "testpub/test-ext")
    assert vse.total == 1

    # Re-insert same key — dedupe (ON CONFLICT does UPDATE, count shows +1)
    vse.db_add("sk-proj-WXYZmnopqr7890ABCDEFGHIJKLMNOPQRSTUVWXYZ",
               "OPENAI", "testpub/test-ext")
    assert vse.total == 2

    # Verify DB row exists and hash is correct
    cur = vse.conn.execute("SELECT hash, val, found FROM keys WHERE val=?",
                           ("sk-proj-WXYZmnopqr7890ABCDEFGHIJKLMNOPQRSTUVWXYZ",))
    row = cur.fetchone()
    expected_hash = hashlib.sha1(
        "sk-proj-WXYZmnopqr7890ABCDEFGHIJKLMNOPQRSTUVWXYZ".encode()
    ).hexdigest()[:16]
    assert row[0] == expected_hash, f"hash mismatch: {row[0]} vs {expected_hash}"
    assert row[1] == "sk-proj-WXYZmnopqr7890ABCDEFGHIJKLMNOPQRSTUVWXYZ"
    assert row[2] == "vscode"
    print("  test_db_add: PASS")


def test_dedup():
    """Second insertion of same key doesn't create duplicate rows."""
    count_before = vse.conn.execute("SELECT COUNT(*) FROM keys").fetchone()[0]

    vse.db_add("sk-proj-WXYZmnopqr7890ABCDEFGHIJKLMNOPQRSTUVWXYZ",
               "OPENAI", "testpub/test-ext")

    count_after = vse.conn.execute("SELECT COUNT(*) FROM keys").fetchone()[0]
    assert count_after == count_before, (
        f"dedup failed: {count_before} -> {count_after}")
    print("  test_dedup: PASS")


def test_work_rejects_seen():
    """work() should skip extensions already in seen set."""
    ext = {
        "publisher": {"publisherName": "testpub"},
        "extensionName": "test-ext",
        "versions": [{"version": "1.0.0", "fallbackAssetUri": None}]
    }
    vse.seen.add("testpub/test-ext/1.0.0")
    n = vse.work(ext)
    assert n == 0, f"expected 0 for already-seen, got {n}"
    print("  test_work_rejects_seen: PASS")


# ── Run ────────────────────────────────────────────────────────────

if __name__ == "__main__":
    test_is_junk()
    test_scan_vsix()
    test_db_add()
    test_dedup()
    test_work_rejects_seen()
    print("\nAll tests PASSED")