f3e21c6b98
Accumulated uncommitted work: GHOST Web backend, WebGL/GPU web layer, web workflow engine and protocol, tooling, docs and task records. Large generated evidence, user-testing archives, build snapshots and imported test assets remain untracked. Co-Authored-By: Claude Code <noreply@anthropic.com>
233 lines
9.9 KiB
Python
233 lines
9.9 KiB
Python
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
|
|
from wbv2_step_cache import StepCache
|
|
|
|
|
|
class StepCacheTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name) / "project"
|
|
self.root.mkdir()
|
|
self.source = self.root / "source.cc"
|
|
self.source.write_bytes(b"original input")
|
|
self.log = self.root / "check.log"
|
|
self.log.write_bytes(b"verified")
|
|
self.artifact = self.root / "module.wasm"
|
|
self.artifact.write_bytes(b"compiled output")
|
|
self.case = self.root / "case.json"
|
|
self.case.write_text('{"passed":true}', encoding="utf-8")
|
|
self.cache = StepCache(self.root, "cache")
|
|
self.receipt = {
|
|
"exitCode": 0, "durationSeconds": 12.75, "logPath": "check.log",
|
|
"logSha256": self.digest(self.log), "pid": 1234,
|
|
"startedAt": "2026-09-09T00:00:00Z", "endedAt": "2026-09-09T00:00:12Z",
|
|
"argv": ["python3", "validate.py"], "cwd": str(self.root),
|
|
}
|
|
|
|
@staticmethod
|
|
def digest(path):
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
def key(self, cache=None, **changes):
|
|
arguments = {
|
|
"job": {"id": "MODULE-CHECK"}, "check_id": "case-one",
|
|
"argv": ["python3", "validate.py"], "cwd": ".", "env": {},
|
|
"unset_env": [], "input_hashes": {"source.cc": self.digest(self.source)},
|
|
}
|
|
arguments.update(changes)
|
|
return (cache or self.cache).key(**arguments)
|
|
|
|
def artifacts(self):
|
|
return [{"path": p.name, "sha256": self.digest(p)}
|
|
for p in (self.artifact, self.case)]
|
|
|
|
def store(self, **changes):
|
|
key = self.key()
|
|
self.cache.store(key, self.receipt | changes, self.artifacts())
|
|
return key
|
|
|
|
def test_reuses_success_without_rewriting_evidence(self):
|
|
before = self.log.stat().st_mtime_ns
|
|
key = self.store()
|
|
result = self.cache.load(key)
|
|
self.assertEqual(result["durationSeconds"], 12.75)
|
|
self.assertEqual(result["logPath"], self.receipt["logPath"])
|
|
self.assertEqual(result["startedAt"], self.receipt["startedAt"])
|
|
self.assertEqual(result["endedAt"], self.receipt["endedAt"])
|
|
self.assertEqual(result["pid"], 1234)
|
|
self.assertEqual(result["reusedFrom"]["cacheKey"], key)
|
|
self.assertEqual(self.log.stat().st_mtime_ns, before)
|
|
self.assertNotIn("reusedFrom", self.receipt)
|
|
|
|
def test_fresh_process_preserves_original_provenance(self):
|
|
key = self.store()
|
|
fresh = StepCache(self.root, "cache")
|
|
self.assertEqual(self.key(fresh), key)
|
|
result = fresh.load(key)
|
|
self.assertEqual(result["pid"], self.receipt["pid"])
|
|
self.assertEqual(result["argv"], self.receipt["argv"])
|
|
self.assertEqual(result["cwd"], self.receipt["cwd"])
|
|
self.assertEqual(result["reusedFrom"]["durationSeconds"], 12.75)
|
|
|
|
def test_input_change_invalidates_key_and_checkpoint(self):
|
|
key = self.store()
|
|
self.source.write_bytes(b"new input")
|
|
self.assertNotEqual(self.key(), key)
|
|
self.assertIsNone(self.cache.load(key))
|
|
|
|
def test_environment_changes_invalidate_key_without_storing_values(self):
|
|
key = self.key(env={"SECRET_TOKEN": "private-environment-value"})
|
|
other = self.key(env={"SECRET_TOKEN": "other-value"})
|
|
self.assertNotEqual(key, other)
|
|
self.assertNotEqual(key, self.key(env={"SECRET_TOKEN": "private-environment-value"},
|
|
unset_env=["THREADS"]))
|
|
self.cache.store(key, self.receipt, self.artifacts())
|
|
record = (self.root / "cache" / (key + ".json")).read_text()
|
|
self.assertNotIn("SECRET_TOKEN", record)
|
|
self.assertNotIn("private-environment-value", record)
|
|
|
|
def test_identity_command_and_directory_affect_key(self):
|
|
key = self.key()
|
|
(self.root / "subdir").mkdir()
|
|
for changes in ({"job": "another-job"}, {"check_id": "other-check"},
|
|
{"argv": ["python3", "different.py"]}, {"cwd": "subdir"}):
|
|
with self.subTest(changes=changes):
|
|
self.assertNotEqual(key, self.key(**changes))
|
|
|
|
def test_artifact_drift_invalidates_checkpoint(self):
|
|
key = self.store()
|
|
self.artifact.write_bytes(b"drifted output")
|
|
self.assertIsNone(self.cache.load(key))
|
|
|
|
def test_case_receipt_drift_invalidates_checkpoint(self):
|
|
key = self.store()
|
|
self.case.write_text('{"passed":false}', encoding="utf-8")
|
|
self.assertIsNone(self.cache.load(key))
|
|
|
|
def test_missing_artifact_invalidates_checkpoint(self):
|
|
key = self.store()
|
|
self.artifact.unlink()
|
|
self.assertIsNone(self.cache.load(key))
|
|
|
|
def test_log_drift_invalidates_checkpoint(self):
|
|
key = self.store()
|
|
self.log.write_bytes(b"changed log")
|
|
self.assertIsNone(self.cache.load(key))
|
|
|
|
def test_failed_or_interrupted_checks_are_never_stored(self):
|
|
for changes in ({"exitCode": 1}, {"exitCode": False}, {"exitCode": "0"},
|
|
{"timedOut": True}, {"idleTimedOut": True},
|
|
{"logLimitExceeded": True}, {"interrupted": True},
|
|
{"providerAborted": True}, {"signal": "SIGTERM"},
|
|
{"stopReason": "model_idle_timeout"}):
|
|
with self.subTest(changes=changes):
|
|
key = self.store(**changes)
|
|
self.assertIsNone(self.cache.load(key))
|
|
self.assertFalse((self.root / "cache" / (key + ".json")).exists())
|
|
|
|
def test_source_changes_during_execution_prevent_store(self):
|
|
key = self.key()
|
|
self.source.write_bytes(b"changed while checking")
|
|
self.cache.store(key, self.receipt, self.artifacts())
|
|
self.assertIsNone(self.cache.load(key))
|
|
|
|
def test_unpinned_store_is_rejected(self):
|
|
key = "a" * 64
|
|
self.cache.store(key, self.receipt, self.artifacts())
|
|
self.assertIsNone(self.cache.load(key))
|
|
with self.assertRaises(ValueError):
|
|
self.key(input_hashes={})
|
|
|
|
def test_missing_log_hash_or_raw_environment_prevents_store(self):
|
|
for changes in ({"logSha256": None}, {"durationSeconds": float("nan")},
|
|
{"env": {"SECRET": "private"}}):
|
|
with self.subTest(changes=changes):
|
|
self.assertIsNone(self.cache.load(self.store(**changes)))
|
|
|
|
def test_no_artifacts_prevents_store(self):
|
|
key = self.key()
|
|
self.cache.store(key, self.receipt, [])
|
|
self.assertIsNone(self.cache.load(key))
|
|
|
|
def test_corrupt_and_malformed_cache_records_are_misses(self):
|
|
key = self.store()
|
|
path = self.root / "cache" / (key + ".json")
|
|
good = json.loads(path.read_text())
|
|
for value in ("{broken", "null", "[]", "true", "{}",
|
|
json.dumps(good | {"key": "b" * 64}),
|
|
json.dumps(good | {"schemaVersion": True}),
|
|
json.dumps(good | {"artifacts": [None]}),
|
|
json.dumps(good | {"inputHashes": []}),
|
|
json.dumps(good | {"processReceipt": []})):
|
|
with self.subTest(value=value):
|
|
path.write_text(value)
|
|
self.assertIsNone(self.cache.load(key))
|
|
|
|
def test_traversal_paths_are_rejected(self):
|
|
with self.assertRaises(ValueError):
|
|
StepCache(self.root, "cache/../other")
|
|
with self.assertRaises(ValueError):
|
|
self.key(cwd="../project")
|
|
self.assertIsNone(self.cache.load("../key"))
|
|
self.assertIsNone(self.cache.load(self.store(logPath="cache/../check.log")))
|
|
|
|
def test_external_symlink_paths_are_rejected(self):
|
|
outside = Path(self.temp.name) / "outside"
|
|
outside.mkdir()
|
|
(outside / "output").write_bytes(self.artifact.read_bytes())
|
|
(self.root / "escape").symlink_to(outside, target_is_directory=True)
|
|
with self.assertRaises(ValueError):
|
|
StepCache(self.root, "escape/cache")
|
|
key = self.key()
|
|
artifacts = [{"path": "escape/output", "sha256": self.digest(self.artifact)}]
|
|
self.cache.store(key, self.receipt, artifacts)
|
|
self.assertIsNone(self.cache.load(key))
|
|
|
|
def test_replaced_artifact_symlink_is_rejected(self):
|
|
key = self.store()
|
|
outside = Path(self.temp.name) / "outside.wasm"
|
|
outside.write_bytes(self.artifact.read_bytes())
|
|
self.artifact.unlink()
|
|
self.artifact.symlink_to(outside)
|
|
self.assertIsNone(self.cache.load(key))
|
|
|
|
def test_internal_artifact_symlink_retarget_invalidates_checkpoint(self):
|
|
alias = self.root / "runtime.wasm"
|
|
alias.symlink_to(self.artifact)
|
|
key = self.key()
|
|
self.cache.store(key, self.receipt,
|
|
[{"path": alias.name, "sha256": self.digest(alias)}])
|
|
self.assertIsNotNone(self.cache.load(key))
|
|
other = self.root / "other.wasm"
|
|
other.write_bytes(b"changed compiled output")
|
|
alias.unlink()
|
|
alias.symlink_to(other)
|
|
self.assertIsNone(self.cache.load(key))
|
|
|
|
def test_input_symlink_escape_invalidates_checkpoint(self):
|
|
alias = self.root / "linked-source.cc"
|
|
alias.symlink_to(self.source)
|
|
key = self.key(input_hashes={alias.name: self.digest(alias)})
|
|
self.cache.store(key, self.receipt, self.artifacts())
|
|
self.assertIsNotNone(self.cache.load(key))
|
|
outside = Path(self.temp.name) / "outside.cc"
|
|
outside.write_bytes(self.source.read_bytes())
|
|
alias.unlink()
|
|
alias.symlink_to(outside)
|
|
self.assertIsNone(self.cache.load(key))
|
|
|
|
def test_symlink_loop_is_a_cache_miss(self):
|
|
key = self.store()
|
|
self.artifact.unlink()
|
|
self.artifact.symlink_to(self.artifact)
|
|
self.assertIsNone(self.cache.load(key))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|