sashatrask
72 lines
2.5 KiB
Python
72 lines
2.5 KiB
Python
import sys
|
|
|
|
sys.dont_write_bytecode = True
|
|
|
|
import os
|
|
import sqlite3
|
|
import tempfile
|
|
|
|
from keycheckers.keycheck_common import (
|
|
append_checked,
|
|
append_status,
|
|
record_cached_keycheck_occurrence,
|
|
)
|
|
from keycheck_runner import ingest_keycheck_results_to_db
|
|
from scanner_db import ScannerDB
|
|
from runtime_security import ensure_private_directory
|
|
|
|
|
|
def main():
|
|
for key in ('SCANNER_DB_URL', 'DATABASE_URL', 'TRUF_MANAGED_POSTGRES_DSN', 'KEYCHECK_DB_URL'):
|
|
os.environ.pop(key, None)
|
|
with tempfile.TemporaryDirectory(prefix="keycheck-accounting-") as tmp:
|
|
db_path = os.path.join(tmp, "scanner.db")
|
|
output_dir = os.path.join(tmp, "keychecks", "openai")
|
|
ensure_private_directory(output_dir, reject_reparse=True)
|
|
|
|
db = ScannerDB(db_path=db_path)
|
|
db.close()
|
|
|
|
alive_file = os.path.join(output_dir, "openaiAlive.txt")
|
|
checked_file = os.path.join(output_dir, "openaiChecked.txt")
|
|
key = "sk-test-keycheck-accounting-1234567890"
|
|
append_status(alive_file, key, "ALIVE", "fixture", "smoke")
|
|
append_checked(checked_file, key, "ALIVE")
|
|
|
|
os.environ["KEYCHECK_DB_PATH"] = db_path
|
|
os.environ["KEYCHECK_OUTPUT_DIR"] = output_dir
|
|
os.environ["KEYCHECK_SERVICE"] = "openai"
|
|
|
|
finding = {"DetectorName": "OpenAI", "Raw": key}
|
|
ok = record_cached_keycheck_occurrence("openai", key, "ALIVE", "fixture.jsonl:1", finding, "OpenAI")
|
|
if not ok:
|
|
raise SystemExit("cached occurrence write returned false")
|
|
|
|
inserted = ingest_keycheck_results_to_db(
|
|
{"database_path": db_path, "keycheck_dir": os.path.join(tmp, "keychecks")},
|
|
["openai"],
|
|
max_rows=10,
|
|
)
|
|
if inserted != 1:
|
|
raise SystemExit(f"unexpected ingest count: {inserted}")
|
|
|
|
conn = sqlite3.connect(db_path)
|
|
try:
|
|
row = conn.execute(
|
|
"SELECT service, status, status_group, metadata_json FROM keycheck_results"
|
|
).fetchone()
|
|
finally:
|
|
conn.close()
|
|
if not row:
|
|
raise SystemExit("missing keycheck_results row")
|
|
service, status, status_group, metadata_json = row
|
|
if (service, status, status_group) != ("openai", "ALIVE", "alive"):
|
|
raise SystemExit(f"unexpected row status: {(service, status, status_group)}")
|
|
if "cached_status" not in metadata_json:
|
|
raise SystemExit("missing cached_status metadata")
|
|
print("keycheck accounting smoke ok")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|