130 lines
4.4 KiB
Python
130 lines
4.4 KiB
Python
|
|
#!/usr/bin/env python3
|
||
|
|
"""Tests for scripts/knowledge_staleness_check.py — 8 tests."""
|
||
|
|
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
import tempfile
|
||
|
|
|
||
|
|
sys.path.insert(0, os.path.dirname(__file__) or ".")
|
||
|
|
import importlib.util
|
||
|
|
spec = importlib.util.spec_from_file_location("ks", os.path.join(os.path.dirname(__file__) or ".", "knowledge_staleness_check.py"))
|
||
|
|
mod = importlib.util.module_from_spec(spec)
|
||
|
|
spec.loader.exec_module(mod)
|
||
|
|
check_staleness = mod.check_staleness
|
||
|
|
fix_hashes = mod.fix_hashes
|
||
|
|
compute_file_hash = mod.compute_file_hash
|
||
|
|
|
||
|
|
|
||
|
|
def test_fresh_entry():
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
src = os.path.join(tmpdir, "source.py")
|
||
|
|
with open(src, "w") as f:
|
||
|
|
f.write("print('hello')")
|
||
|
|
h = compute_file_hash(src)
|
||
|
|
idx = os.path.join(tmpdir, "index.json")
|
||
|
|
with open(idx, "w") as f:
|
||
|
|
json.dump({"facts": [{"fact": "hello", "source_file": "source.py", "source_hash": h}]}, f)
|
||
|
|
results = check_staleness(idx, tmpdir)
|
||
|
|
assert results[0]["status"] == "fresh"
|
||
|
|
print("PASS: test_fresh_entry")
|
||
|
|
|
||
|
|
|
||
|
|
def test_stale_entry():
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
src = os.path.join(tmpdir, "source.py")
|
||
|
|
with open(src, "w") as f:
|
||
|
|
f.write("original content")
|
||
|
|
idx = os.path.join(tmpdir, "index.json")
|
||
|
|
with open(idx, "w") as f:
|
||
|
|
json.dump({"facts": [{"fact": "old", "source_file": "source.py", "source_hash": "sha256:wrong"}]}, f)
|
||
|
|
# Now change the source
|
||
|
|
with open(src, "w") as f:
|
||
|
|
f.write("modified content")
|
||
|
|
results = check_staleness(idx, tmpdir)
|
||
|
|
assert results[0]["status"] == "stale"
|
||
|
|
print("PASS: test_stale_entry")
|
||
|
|
|
||
|
|
|
||
|
|
def test_missing_source():
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
idx = os.path.join(tmpdir, "index.json")
|
||
|
|
with open(idx, "w") as f:
|
||
|
|
json.dump({"facts": [{"fact": "gone", "source_file": "nonexistent.py", "source_hash": "sha256:abc"}]}, f)
|
||
|
|
results = check_staleness(idx, tmpdir)
|
||
|
|
assert results[0]["status"] == "missing_source"
|
||
|
|
print("PASS: test_missing_source")
|
||
|
|
|
||
|
|
|
||
|
|
def test_no_hash():
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
src = os.path.join(tmpdir, "source.py")
|
||
|
|
with open(src, "w") as f:
|
||
|
|
f.write("content")
|
||
|
|
idx = os.path.join(tmpdir, "index.json")
|
||
|
|
with open(idx, "w") as f:
|
||
|
|
json.dump({"facts": [{"fact": "no hash", "source_file": "source.py"}]}, f)
|
||
|
|
results = check_staleness(idx, tmpdir)
|
||
|
|
assert results[0]["status"] == "no_hash"
|
||
|
|
assert results[0]["current_hash"].startswith("sha256:")
|
||
|
|
print("PASS: test_no_hash")
|
||
|
|
|
||
|
|
|
||
|
|
def test_no_source_field():
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
idx = os.path.join(tmpdir, "index.json")
|
||
|
|
with open(idx, "w") as f:
|
||
|
|
json.dump({"facts": [{"fact": "orphan"}]}, f)
|
||
|
|
results = check_staleness(idx, tmpdir)
|
||
|
|
assert results[0]["status"] == "no_source"
|
||
|
|
print("PASS: test_no_source_field")
|
||
|
|
|
||
|
|
|
||
|
|
def test_fix_hashes():
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
src = os.path.join(tmpdir, "source.py")
|
||
|
|
with open(src, "w") as f:
|
||
|
|
f.write("content for hashing")
|
||
|
|
idx = os.path.join(tmpdir, "index.json")
|
||
|
|
with open(idx, "w") as f:
|
||
|
|
json.dump({"facts": [{"fact": "needs hash", "source_file": "source.py"}]}, f)
|
||
|
|
fixed = fix_hashes(idx, tmpdir)
|
||
|
|
assert fixed == 1
|
||
|
|
# Verify hash was added
|
||
|
|
with open(idx) as f:
|
||
|
|
data = json.load(f)
|
||
|
|
assert data["facts"][0]["source_hash"].startswith("sha256:")
|
||
|
|
print("PASS: test_fix_hashes")
|
||
|
|
|
||
|
|
|
||
|
|
def test_empty_index():
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
idx = os.path.join(tmpdir, "index.json")
|
||
|
|
with open(idx, "w") as f:
|
||
|
|
json.dump({"facts": []}, f)
|
||
|
|
results = check_staleness(idx, tmpdir)
|
||
|
|
assert results == []
|
||
|
|
print("PASS: test_empty_index")
|
||
|
|
|
||
|
|
|
||
|
|
def test_compute_hash_nonexistent():
|
||
|
|
h = compute_file_hash("/nonexistent/path/file.py")
|
||
|
|
assert h is None
|
||
|
|
print("PASS: test_compute_hash_nonexistent")
|
||
|
|
|
||
|
|
|
||
|
|
def run_all():
|
||
|
|
test_fresh_entry()
|
||
|
|
test_stale_entry()
|
||
|
|
test_missing_source()
|
||
|
|
test_no_hash()
|
||
|
|
test_no_source_field()
|
||
|
|
test_fix_hashes()
|
||
|
|
test_empty_index()
|
||
|
|
test_compute_hash_nonexistent()
|
||
|
|
print("\nAll 8 tests passed!")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
run_all()
|