keyhunter-parser/test_iascholar.py

"""Test iascholar_eternal — offline with fixtures."""
import os, sys, tempfile, sqlite3, hashlib

BASE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, BASE)

# --- Fixtures --------------------------------------------------------------
# Simulated IA Solr metadata (title + description with keys)
SOLR_DOCS = [
    {
        "identifier": "APIKEYyy",
        "title": "SCH",
        "description": (
            "sk-proj-AEl35fZ3UVhxVxOD1EUUl0tiruoMN49D4LnPPFNDGv9mqOSq85bHBeHn-"
            "EXlxi76j7uEOVCf4PT3BlbkFJ9ZlPDKcCdpBJYM-hjMtlB7n147k5L8LNjD1TJ2okoxkhAe2-"
            "9plma9PdUNdEFHsq5xsiWVHVgA "
            "sk-proj-X71ckomAq4JU_9M_p5QbmQVB_CPvs53VcVwQ8buMVqvSV0cM_gCi3MjVjz4EvIjg6NP"
            "leahJTaT3BlbkFJ6dMojS8X7ipFN-pFpDjJX2Y4KlJ3Bbaz9OUvpByzq5AaRFemyOMwO6FhdrS"
            "HALli6GjurXMKwA"
        ),
        "mediatype": "software",
    },
]

# Simulated _djvu.txt content with keys that match key_patterns.PATTERNS
# OPENAI: sk-proj-[A-Za-z0-9_-]{60,200}  → need 60-200 chars after prefix
# GITHUB_PAT: ghp_[A-Za-z0-9]{36}
_OPENAI_KEY = 'sk-proj-' + 'A' * 80
_GITHUB_PAT = 'ghp_' + 'A' * 36
DJVU_CONTENT = (
    "Some OCR text from a research paper. The API key used was: "
    + _OPENAI_KEY + "\n"
    "Also a GitHub PAT: " + _GITHUB_PAT + "\n"
    "And a google key: GOOGLE_API_KEY=AIzaSyDummyKey123ForTestingPurposesOnly\n"
)

# Junk that should be filtered
JUNK_CONTENT = (
    "This file has: sk-proj-... doc example and ghp_your_token_here\n"
    "Also: AKIA1234567890ABCDEF which is a placeholder\n"
    "And: xxxx and test patterns everywhere.\n"
)

# OSF fixture — single downloadable file
_OSF_OPENAI = 'sk-proj-' + 'B' * 80
_OSF_GH_TOKEN = 'ghp_' + 'C' * 36
OSF_FILE_CONTENT = (
    '{\n'
    '  "openai_api_key": "' + _OSF_OPENAI + '",\n'
    '  "github_token": "' + _OSF_GH_TOKEN + '"\n'
    '}\n'
)


# --- Patch module globals --------------------------------------------------
def _test_db():
    """Create temp DB, patch iascholar_eternal globals, return conn."""
    fh, path = tempfile.mkstemp(suffix='.db')
    os.close(fh)
    conn = sqlite3.connect(path, timeout=120, check_same_thread=False,
                           isolation_level=None)
    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)
    """)
    return conn, path


def test_metadata_scan():
    """Solr description scan finds keys from IA metadata."""
    import iascholar_eternal as m
    conn, db_path = _test_db()
    old_conn = m.conn
    m.conn = conn

    try:
        for doc in SOLR_DOCS:
            m.ia_process(doc)

        rows = conn.execute("SELECT val, prov, found FROM keys").fetchall()
        # Should have found both sk-proj keys
        keys_found = [r for r in rows if r[2] == 'ia-text']
        assert len(keys_found) >= 2, "Expected >=2 keys from Solr metadata, got %d" % len(keys_found)

        provs = [r[1] for r in keys_found]
        # First key is sk-proj (OPENAI)
        assert any('sk-proj' in r[0] for r in keys_found), "Expected sk-proj key"
        print("  metadata_scan: %d keys found (provs: %s)" % (len(keys_found), provs[:5]))
    finally:
        m.conn = old_conn
        conn.close()
        os.unlink(db_path)


def test_text_scan():
    """_djvu.txt content scan finds keys via scan_text directly."""
    from key_patterns import scan_text
    hits = scan_text(DJVU_CONTENT)
    assert len(hits) >= 3, "Expected >=3 key hits from djvu text, got %d: %s" % (
        len(hits), hits)
    print("  text_scan: %d key hits from djvu fixture" % len(hits))


def test_junk_filter():
    """Junk keys are correctly filtered."""
    import iascholar_eternal as m
    assert m.is_junk("ghp_your_token_here12345678")
    assert m.is_junk("sk-proj-YOUR_KEY_HERE_12345678")
    assert m.is_junk("AKIA1234567890abcdef")  # lowercase variant matches GARBAGE
    assert m.is_junk("example_xxxx_token_12345678")
    assert not m.is_junk("sk-proj-AEl35fZ3UVhxVxOD1EUUl0tiruoMN49")
    assert not m.is_junk("ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890")
    print("  junk_filter: OK")


def test_db_add_dedup():
    """db_add inserts and deduplicates."""
    import iascholar_eternal as m
    conn, db_path = _test_db()
    old_conn = m.conn
    m.conn = conn

    try:
        key = "sk-proj-FAKEKEY1234567890FAKEKEY1234567890"
        m.db_add(key, "OPENAI", "test")
        m.db_add(key, "OPENAI", "test")
        rows = conn.execute("SELECT count(*) FROM keys").fetchone()[0]
        assert rows == 1, "Expected 1 row after dedup, got %d" % rows
        print("  db_add_dedup: OK (1 row after 2 inserts)")
    finally:
        m.conn = old_conn
        conn.close()
        os.unlink(db_path)


def test_osf_scan():
    """OSF file content scanning works via scan_text."""
    from key_patterns import scan_text
    hits = scan_text(OSF_FILE_CONTENT)
    assert len(hits) >= 2, "Expected >=2 keys from OSF fixture, got %d: %s" % (len(hits), hits)
    print("  osf_scan: %d key hits from OSF fixture" % len(hits))


def test_seen_set():
    """Seen set prevents re-processing."""
    import iascholar_eternal as m
    m.seen.clear()
    ident = "test-item-001"
    m.mark_seen(ident)
    assert ident in m.seen
    # Processing a seen item should return 0
    result = m.ia_process({"identifier": ident, "title": "", "description": ""})
    assert result == 0, "Seen item should return 0, got %d" % result
    print("  seen_set: OK")


if __name__ == "__main__":
    # Clean up any existing seen file
    seen_path = os.path.join(BASE, "data", "iascholar_seen.txt")
    if os.path.exists(seen_path):
        os.remove(seen_path)

    test_metadata_scan()
    test_text_scan()
    test_junk_filter()
    test_db_add_dedup()
    test_osf_scan()
    test_seen_set()
    print("\nALL TESTS PASSED")