193 lines
7.7 KiB
Python
193 lines
7.7 KiB
Python
"""Test sourcegraph_eternal: SSE parsing, scan_text integration, junk filter, db_add+dedup.
|
|
|
|
Sets KF_DB to a smoke DB before importing the module so the global conn
|
|
points to the test DB. Creates the keys table. No network calls.
|
|
"""
|
|
import hashlib, os, sqlite3, sys
|
|
|
|
# --- Point KF_DB to a smoke DB before module import ---
|
|
THIS_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
DB_PATH = os.path.join(THIS_DIR, "data", "_smoke_sourcegraph.db")
|
|
os.environ["KF_DB"] = DB_PATH
|
|
|
|
os.makedirs(os.path.join(THIS_DIR, "data"), exist_ok=True)
|
|
|
|
# Drop + recreate the smoke DB so each test run is clean
|
|
if os.path.exists(DB_PATH):
|
|
os.remove(DB_PATH)
|
|
|
|
boot_conn = sqlite3.connect(DB_PATH)
|
|
boot_conn.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_conn.commit()
|
|
boot_conn.close()
|
|
|
|
# --- Now import the lane (reads KF_DB) ---
|
|
sys.path.insert(0, THIS_DIR)
|
|
import sourcegraph_eternal as sg
|
|
|
|
|
|
# ── Fixture: a realistic SSE response with fake keys ──────────────────
|
|
|
|
# Real Sourcegraph format: single-line compact JSON per data line
|
|
FIXTURE_SSE = (
|
|
'event: filters\n'
|
|
'data: [{"value":"archived:yes","label":"Include archived repos","count":68586,"exhaustive":false,"kind":"utility"}]\n'
|
|
'\n'
|
|
'event: progress\n'
|
|
'data: {"done":false,"matchCount":1024,"durationMs":312}\n'
|
|
'\n'
|
|
'event: matches\n'
|
|
'data: [{"type":"content","path":".env.production","repository":"github.com/octocat/hello-world","repoStars":42,"commit":"abc123def456","lineMatches":[{"line":"OPENAI_API_KEY=sk-proj-XyZ1aB2cD3eF4gH5iJ6kL7mN8oP9qR0sTuVwXyZaBcDeFgHiJkLmNoPqRsTuVw","lineNumber":10,"offsetAndLengths":[[0,7]]},{"line":"ANTHROPIC_API_KEY=sk-ant-api03-ABCdefGHIjklMNOpqrSTUvwxYZ12345678","lineNumber":11,"offsetAndLengths":[[0,9]]}],"language":"JavaScript"},{"type":"content","path":"config/secrets.yml","repository":"github.com/user/myapp","repoStars":7,"commit":"deadbeef","lineMatches":[{"line":" GITHUB_TOKEN: ghp_ABCdefGHIjklMNOpqrSTUvwxYZ1234567890","lineNumber":5,"offsetAndLengths":[[14,4]]},{"line":" HUGGINGFACE_TOKEN: hf_ABCdefGHIjklMNOpqrSTUvwxYZ12345678","lineNumber":6,"offsetAndLengths":[[14,20]]}],"language":"YAML"},{"type":"content","path":"readme.md","repository":"github.com/user/empty","repoStars":0,"commit":"f00f00","lineMatches":[{"line":"Just a README with no keys","lineNumber":1,"offsetAndLengths":[]}],"language":"Markdown"}]\n'
|
|
'\n'
|
|
'event: progress\n'
|
|
'data: {"done":true,"matchCount":1024,"durationMs":812}\n'
|
|
'\n'
|
|
'event: done\n'
|
|
'data: {}\n'
|
|
)
|
|
|
|
|
|
# ── Tests ─────────────────────────────────────────────────────────────
|
|
|
|
def test_parse_sse():
|
|
"""Proof: SSE parser extracts events correctly."""
|
|
events = list(sg.parse_sse_events(FIXTURE_SSE))
|
|
# events: filters, progress, matches, progress, done
|
|
assert len(events) == 5, f"Expected 5 events, got {len(events)}"
|
|
assert events[0][0] == "filters"
|
|
assert "archived" in events[0][1]
|
|
assert events[1][0] == "progress"
|
|
assert events[2][0] == "matches"
|
|
assert "OPENAI_API_KEY" in events[2][1]
|
|
assert events[3][0] == "progress"
|
|
assert events[3][1] == '{"done":true,"matchCount":1024,"durationMs":812}'
|
|
assert events[4][0] == "done"
|
|
print(" test_parse_sse: PASS")
|
|
|
|
|
|
def test_is_junk():
|
|
"""Proof: junk filter catches sample/placeholder values."""
|
|
junk_values = [
|
|
"sk-your-test-key-xxxxx",
|
|
"sk-EXAMPLEkey12345678901234567890",
|
|
"sk-your-test-api-key-CHANGEME",
|
|
"ghp_testKey1234567890abcdef1234567",
|
|
"sk-n0tr3al-1234567890123456789012345",
|
|
"AAAAAA", # < 6 unique chars
|
|
]
|
|
for v in junk_values:
|
|
assert sg.is_junk(v), f"Expected junk: {v}"
|
|
|
|
clean_values = [
|
|
"sk-proj-real-key-1234567890123456789012345678901234567890",
|
|
"ghp_realToken1234567890123456789012345678",
|
|
"sk-ant-api03-real-test-key-12345678901234567890",
|
|
"hf_zvWqKpLmRtYnBgHjUcXdEsFkAoPqRwMzXcVbNmLk",
|
|
"sk-or-v1-aBcDeFgHiJkLmNoPqRsTuVwXyZaBcDeFgHiJkLmNo",
|
|
]
|
|
for v in clean_values:
|
|
assert not sg.is_junk(v), f"Expected clean: {v}"
|
|
|
|
print(" test_is_junk: PASS")
|
|
|
|
|
|
def test_scan_and_db():
|
|
"""Proof: parse fixture SSE, extract keys, write DB, dedupe."""
|
|
# Clean state
|
|
sg.conn.execute("DELETE FROM keys")
|
|
sg.total = 0
|
|
sg.seen = set()
|
|
|
|
# Process each matches event through the same logic as fetch_and_scan
|
|
import json
|
|
seen_keys = set()
|
|
for evt, data_str in sg.parse_sse_events(FIXTURE_SSE):
|
|
if evt != "matches":
|
|
continue
|
|
matches = json.loads(data_str)
|
|
for m in matches:
|
|
repo = m.get("repository", "")
|
|
path = m.get("path", "")
|
|
commit = m.get("commit", "")
|
|
sig = f"{repo}:{path}:{commit}"
|
|
if sig in sg.seen:
|
|
continue
|
|
snippets = []
|
|
for lm in m.get("lineMatches", []):
|
|
ln = lm.get("line", "")
|
|
if ln:
|
|
snippets.append(ln)
|
|
sg.mark_seen(sig)
|
|
if snippets:
|
|
text_block = "\n".join(snippets)
|
|
for prov, key in sg.scan_text(text_block):
|
|
if sg.is_junk(key) or len(key) < 30:
|
|
continue
|
|
sg.db_add(key, prov, f"{repo}:{path}")
|
|
seen_keys.add(key)
|
|
|
|
# Verify we got keys from the two real files (the readme has no keys)
|
|
assert len(seen_keys) >= 3, (
|
|
f"Expected >=3 unique keys, got {len(seen_keys)}: {seen_keys}"
|
|
)
|
|
|
|
cur = sg.conn.execute(
|
|
"SELECT val, prov, found, repo FROM keys"
|
|
).fetchall()
|
|
assert len(cur) > 0, "No rows in DB"
|
|
for val, prov, found, repo in cur:
|
|
assert found == "sourcegraph", f"found tag mismatch: {found}"
|
|
assert len(val) >= 28, f"key too short: {val[:20]}"
|
|
|
|
print(f" db_add wrote {len(cur)} rows with found='sourcegraph': PASS")
|
|
|
|
# Dedup test: process the same fixture again with a fresh seen set
|
|
sg.total = 0
|
|
sg.seen = set()
|
|
for evt, data_str in sg.parse_sse_events(FIXTURE_SSE):
|
|
if evt != "matches":
|
|
continue
|
|
matches = json.loads(data_str)
|
|
for m in matches:
|
|
repo = m.get("repository", "")
|
|
path = m.get("path", "")
|
|
commit = m.get("commit", "")
|
|
sig = f"{repo}:{path}:{commit}"
|
|
if sig in sg.seen:
|
|
continue
|
|
snippets = []
|
|
for lm in m.get("lineMatches", []):
|
|
ln = lm.get("line", "")
|
|
if ln:
|
|
snippets.append(ln)
|
|
sg.mark_seen(sig)
|
|
if snippets:
|
|
text_block = "\n".join(snippets)
|
|
for prov, key in sg.scan_text(text_block):
|
|
if sg.is_junk(key) or len(key) < 30:
|
|
continue
|
|
sg.db_add(key, prov, f"{repo}:{path}")
|
|
|
|
cur2 = sg.conn.execute(
|
|
"SELECT val, prov, found, repo FROM keys"
|
|
).fetchall()
|
|
assert len(cur2) == len(cur), (
|
|
f"Dedup failed: rows before={len(cur)} after={len(cur2)}"
|
|
)
|
|
print(f" dedup on repeat: {len(cur2)} rows unchanged: PASS")
|
|
|
|
|
|
# ── Run ──────────────────────────────────────────────────────────────
|
|
|
|
if __name__ == "__main__":
|
|
test_parse_sse()
|
|
test_is_junk()
|
|
test_scan_and_db()
|
|
print("\nAll tests passed.") |