116 lines
4.2 KiB
Plaintext
116 lines
4.2 KiB
Plaintext
"""Test jenkins_eternal: fixture console text, parse + junk filter + db_add."""
|
|
import os, sys, hashlib, sqlite3, tempfile
|
|
|
|
# Point DB to temporary smoke db
|
|
BASE = os.path.dirname(os.path.abspath(__file__))
|
|
os.environ["KF_DB"] = os.path.join(BASE, "data", "_smoke_jenkins.db")
|
|
|
|
# Patch sys.modules so the lane can be imported cleanly
|
|
import jenkins_eternal as lane
|
|
|
|
# --- fixture: realistic Jenkins console text with embedded keys ---
|
|
FIXTURE = """\
|
|
Started by an SCM change
|
|
Running as SYSTEM
|
|
[EnvInject] - Loading node environment variables.
|
|
Building remotely on build-agent-03 in workspace /var/lib/jenkins/workspace/deploy-job
|
|
[EnvInject] - Injecting as environment variables the properties content
|
|
OPENAI_API_KEY=sk-proj-K1r4nT3stA1b2C3d4E5f6G7h8I9j0Klmn
|
|
ANTHROPIC_API_KEY=sk-ant-K1r4nT3stA1b2C3d4E5f6G7h8I9j0Klmn
|
|
AWS_ACCESS_KEY_ID=AKIAIOSFODNN7TESTKEY
|
|
GEMINI_API_KEY=AIzaSyAK1r4nT3stA1b2C3d4E5f6G7h8I9j0Klmn
|
|
[EnvInject] - Variables injected successfully.
|
|
Cloning repository https://github.com/org/test-repo.git
|
|
Fetching upstream changes from https://github.com/org/test-repo.git
|
|
Checking out Revision abc123def456 (refs/remotes/origin/main)
|
|
+ make test
|
|
+ make deploy
|
|
All tests passed.
|
|
----------------------------------------------------------------------
|
|
This is junk: the_key_dummy_value=CHANGEME
|
|
this_is_all_aaa = aaaaaa
|
|
test_value = placeholder_key_here
|
|
"""
|
|
|
|
# Keys we expect to find (5 real-looking hits, 3 junk).
|
|
# NOTE: AWS_SECRET_ACCESS_KEY with "/" chars not matched by ENV value char class.
|
|
# Junk lines (dummy_value=CHANGEME, aaaaaa, placeholder_key_here) are NOT expected.
|
|
EXPECTED_KEYS = {
|
|
("OPENAI", "sk-proj-K1r4nT3stA1b2C3d4E5f6G7h8I9j0Klmn"),
|
|
("ANTHROPIC", "sk-ant-K1r4nT3stA1b2C3d4E5f6G7h8I9j0Klmn"),
|
|
("GOOGLE", "AIzaSyAK1r4nT3stA1b2C3d4E5f6G7h8I9j0Klmn"),
|
|
("AWS", "AKIAIOSFODNN7TESTKEY"),
|
|
}
|
|
|
|
|
|
def test_parsing():
|
|
"""raw scan_text yields expected key candidates before junk filter."""
|
|
hits = lane.scan_text(FIXTURE)
|
|
assert len(hits) >= 5, f"expected >=5 hits, got {len(hits)}: {hits}"
|
|
# Every expected key should appear at least once
|
|
for prov, key in EXPECTED_KEYS:
|
|
assert (prov, key) in hits, f"missing ({prov}, {key})"
|
|
|
|
|
|
def test_junk_filter():
|
|
"""is_junk rejects garbage."""
|
|
assert lane.is_junk("dummy_placeholder_CHANGEME")
|
|
assert lane.is_junk("aaaaaa")
|
|
assert lane.is_junk("sk-dummy-placeholder-key")
|
|
assert not lane.is_junk("sk-proj-FakeKeyA1b2C3d4E5f6G7h8I9j0Klmnop")
|
|
|
|
|
|
def test_scan_write():
|
|
"""scan + db_add writes to DB and dedupes."""
|
|
smoke_db = os.environ["KF_DB"]
|
|
# Close any prior connection
|
|
try:
|
|
lane.conn.close()
|
|
except Exception:
|
|
pass
|
|
try:
|
|
os.remove(smoke_db)
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
# Re-init connection
|
|
lane.conn = sqlite3.connect(os.environ["KF_DB"], timeout=120,
|
|
check_same_thread=False, isolation_level=None)
|
|
lane.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)""")
|
|
|
|
lane.total = 0
|
|
|
|
# Scan and add
|
|
hits = lane.scan_text(FIXTURE)
|
|
for prov, key in hits:
|
|
if lane.is_junk(key):
|
|
continue
|
|
lane.db_add(key, prov, "http://example.com/job/test")
|
|
|
|
# Verify rows written
|
|
rows = lane.conn.execute("SELECT COUNT(*) FROM keys").fetchone()[0]
|
|
# We should have at least as many as expected unique (maybe more from ENV)
|
|
assert rows >= len(EXPECTED_KEYS), f"expected >= {len(EXPECTED_KEYS)} rows, got {rows}"
|
|
|
|
# Verify dedup: adding same keys again does not increase row count
|
|
for prov, key in hits:
|
|
if lane.is_junk(key):
|
|
continue
|
|
lane.db_add(key, prov, "http://example.com/job/test")
|
|
|
|
rows2 = lane.conn.execute("SELECT COUNT(*) FROM keys").fetchone()[0]
|
|
assert rows2 == rows, f"dedup failed: {rows} -> {rows2}"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_parsing()
|
|
print("PASS test_parsing")
|
|
test_junk_filter()
|
|
print("PASS test_junk_filter")
|
|
test_scan_write()
|
|
print("PASS test_scan_write")
|
|
print(f"ALL PASS ({'' if os.path.exists(os.environ['KF_DB']) else 'DB_ERR'})") |