boat/quant-os
Public Access
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

feat: preserve Quant OS target-package vertical slice

Browse Source
This commit is contained in:
zebzhang
2026-07-30 21:27:44 +08:00
parent 00efeb7ec2
commit 919c64c679
42 changed files with 15454 additions and 404 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_release_reachability.py
+405
View File
@@ -0,0 +1,405 @@
import copy
import hashlib
import json
import tempfile
import unittest
from pathlib import Path
from quant60.ledger import canonical_json
from quant60.release_reachability import (
CAPABILITIES,
LAYERS,
ReleaseReachabilityError,
build_evidence_artifact,
build_release_reachability,
load_verified_release_reachability,
release_reachability_sha256,
seal_release_reachability,
validate_release_reachability,
write_release_reachability,
)
GENERATED_AT = "2026-07-26T12:00:00Z"
class ReleaseReachabilityTests(unittest.TestCase):
def setUp(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
self.root = Path(temporary.name)
for directory in ("src", "tests", "tools", "platforms", "evidence"):
(self.root / directory).mkdir()
for layer in LAYERS:
(self.root / "src" / f"{layer}.py").write_text(
f"LAYER = {layer!r}\n",
encoding="utf-8",
)
(self.root / "tests" / f"test_{layer}.py").write_text(
f"def test_{layer}():\n assert True\n",
encoding="utf-8",
)
(self.root / "tools" / f"{layer}_entry.py").write_text(
f"from src import {layer}\n",
encoding="utf-8",
)
(self.root / "platforms" / "joinquant_entry.py").write_text(
"def execute():\n return 'submitted'\n",
encoding="utf-8",
)
(self.root / "platforms" / "qmt_entry.py").write_text(
"def execute():\n return 'submitted'\n",
encoding="utf-8",
)
(self.root / "evidence" / "jq-execution.json").write_text(
canonical_json(
{
"platform": "joinquant",
"layer": "execution",
"release_id": "quant60-baseline-v1",
"result": "passed",
}
)
+ "\n",
encoding="utf-8",
)
@staticmethod
def _false_claim():
return {"status": False, "artifacts": []}
def _artifact(
self,
*,
artifact_id,
kind,
path,
platform,
observed_at=None,
run_id=None,
):
return build_evidence_artifact(
artifact_id=artifact_id,
kind=kind,
path=path,
platform=platform,
repository_root=self.root,
observed_at=observed_at,
run_id=run_id,
)
def _layers(self):
layers = {}
for layer in LAYERS:
layers[layer] = {
"implemented": {
"status": True,
"artifacts": [
self._artifact(
artifact_id=f"{layer}-source",
kind="source",
path=f"src/{layer}.py",
platform="repository",
)
],
},
"unit_tested": {
"status": True,
"artifacts": [
self._artifact(
artifact_id=f"{layer}-test",
kind="test",
path=f"tests/test_{layer}.py",
platform="repository",
)
],
},
"local_entrypoint_reachable": {
"status": True,
"artifacts": [
self._artifact(
artifact_id=f"{layer}-local-entry",
kind="entrypoint",
path=f"tools/{layer}_entry.py",
platform="local",
)
],
},
"jq_entrypoint_reachable": self._false_claim(),
"qmt_entrypoint_reachable": self._false_claim(),
"real_platform_observed": self._false_claim(),
}
layers["execution"]["jq_entrypoint_reachable"] = {
"status": True,
"artifacts": [
self._artifact(
artifact_id="execution-jq-entry",
kind="entrypoint",
path="platforms/joinquant_entry.py",
platform="joinquant",
)
],
}
layers["execution"]["real_platform_observed"] = {
"status": True,
"artifacts": [
self._artifact(
artifact_id="execution-jq-run",
kind="platform_run",
path="evidence/jq-execution.json",
platform="joinquant",
observed_at="2026-07-26T11:30:00Z",
run_id="jq-backtest-20260726-01",
)
],
}
return layers
def _manifest(self):
return build_release_reachability(
release_id="quant60-baseline-v1",
generated_at=GENERATED_AT,
layers=self._layers(),
repository_root=self.root,
)
def test_build_write_load_preserves_exact_six_claims_per_layer(self):
manifest = self._manifest()
self.assertEqual(
manifest["manifest_sha256"],
release_reachability_sha256(manifest),
)
for layer in LAYERS:
self.assertEqual(
set(manifest["layers"][layer]),
set(CAPABILITIES),
)
self.assertTrue(manifest["summary"]["all_layers_implemented"])
self.assertTrue(manifest["summary"]["all_layers_unit_tested"])
self.assertTrue(
manifest["summary"]["all_layers_local_entrypoint_reachable"]
)
self.assertFalse(
manifest["summary"]["all_layers_jq_entrypoint_reachable"]
)
self.assertFalse(
manifest["summary"]["all_layers_qmt_entrypoint_reachable"]
)
self.assertFalse(
manifest["summary"]["all_layers_real_platform_observed"]
)
path = write_release_reachability(
manifest,
self.root / "reachability.json",
repository_root=self.root,
)
loaded = load_verified_release_reachability(
path,
repository_root=self.root,
expected_release_id="quant60-baseline-v1",
expected_manifest_sha256=manifest["manifest_sha256"],
)
self.assertEqual(loaded, manifest)
self.assertEqual(
path.read_bytes(),
(canonical_json(manifest) + "\n").encode("utf-8"),
)
def test_manifest_and_artifact_tampering_both_fail_closed(self):
manifest = self._manifest()
status_tamper = copy.deepcopy(manifest)
status_tamper["layers"]["alpha"]["implemented"]["status"] = False
with self.assertRaisesRegex(
ReleaseReachabilityError,
"SHA-256 mismatch",
):
validate_release_reachability(
status_tamper,
repository_root=self.root,
)
path = write_release_reachability(
manifest,
self.root / "reachability.json",
repository_root=self.root,
)
(self.root / "src" / "alpha.py").write_text(
"LAYER = 'tampered'\n",
encoding="utf-8",
)
with self.assertRaisesRegex(
ReleaseReachabilityError,
"artifact hash mismatch",
):
load_verified_release_reachability(
path,
repository_root=self.root,
)
def test_source_hash_cannot_masquerade_as_real_platform_observation(self):
layers = self._layers()
source = self.root / "src" / "execution.py"
layers["execution"]["real_platform_observed"] = {
"status": True,
"artifacts": [
{
"artifact_id": "fake-platform-run",
"kind": "platform_run",
"path": "src/execution.py",
"sha256": hashlib.sha256(source.read_bytes()).hexdigest(),
"platform": "joinquant",
"observed_at": "2026-07-26T11:30:00Z",
"run_id": "fake-run",
}
],
}
with self.assertRaisesRegex(
ReleaseReachabilityError,
"real-platform evidence must live under",
):
build_release_reachability(
release_id="quant60-baseline-v1",
generated_at=GENERATED_AT,
layers=layers,
repository_root=self.root,
)
def test_true_status_requires_typed_evidence_and_false_forbids_it(self):
true_without_evidence = self._layers()
true_without_evidence["risk"]["unit_tested"] = {
"status": True,
"artifacts": [],
}
with self.assertRaisesRegex(
ReleaseReachabilityError,
"iff at least one evidence",
):
build_release_reachability(
release_id="quant60-baseline-v1",
generated_at=GENERATED_AT,
layers=true_without_evidence,
repository_root=self.root,
)
wrong_evidence_type = self._layers()
wrong_evidence_type["risk"]["unit_tested"]["artifacts"][0].update(
{"kind": "source", "platform": "repository"}
)
with self.assertRaisesRegex(
ReleaseReachabilityError,
"requires kind='test'",
):
build_release_reachability(
release_id="quant60-baseline-v1",
generated_at=GENERATED_AT,
layers=wrong_evidence_type,
repository_root=self.root,
)
def test_real_observation_requires_corresponding_platform_entrypoint(self):
layers = self._layers()
layers["execution"]["jq_entrypoint_reachable"] = self._false_claim()
with self.assertRaisesRegex(
ReleaseReachabilityError,
"requires jq_entrypoint_reachable",
):
build_release_reachability(
release_id="quant60-baseline-v1",
generated_at=GENERATED_AT,
layers=layers,
repository_root=self.root,
)
def test_path_hash_id_and_clock_validation_are_strict(self):
cases = []
absolute = self._layers()
absolute["alpha"]["implemented"]["artifacts"][0]["path"] = str(
self.root / "src" / "alpha.py"
)
cases.append((absolute, "repository-relative path"))
traversal = self._layers()
traversal["alpha"]["implemented"]["artifacts"][0]["path"] = (
"src/../tests/test_alpha.py"
)
cases.append((traversal, "normalized repository-relative path"))
placeholder = self._layers()
placeholder["alpha"]["implemented"]["artifacts"][0]["sha256"] = (
"0" * 64
)
cases.append((placeholder, "all-zero placeholder"))
duplicate = self._layers()
duplicate["alpha"]["implemented"]["artifacts"][0]["artifact_id"] = (
duplicate["universe"]["implemented"]["artifacts"][0]["artifact_id"]
)
cases.append((duplicate, "reused across claims"))
future_run = self._layers()
future_run["execution"]["real_platform_observed"]["artifacts"][0][
"observed_at"
] = "2026-07-26T12:00:01Z"
cases.append((future_run, "cannot be after generated_at"))
for layers, message in cases:
with self.subTest(message=message):
with self.assertRaisesRegex(
ReleaseReachabilityError,
message,
):
build_release_reachability(
release_id="quant60-baseline-v1",
generated_at=GENERATED_AT,
layers=layers,
repository_root=self.root,
)
def test_strict_loader_rejects_duplicate_json_keys(self):
path = self.root / "duplicate.json"
path.write_text(
'{"schema_version":"1.0","schema_version":"1.0"}',
encoding="utf-8",
)
with self.assertRaisesRegex(
ReleaseReachabilityError,
"duplicate JSON key",
):
load_verified_release_reachability(
path,
repository_root=self.root,
)
def test_resealed_derived_summary_tamper_is_rejected(self):
manifest = self._manifest()
manifest["summary"]["all_layers_qmt_entrypoint_reachable"] = True
manifest["manifest_sha256"] = release_reachability_sha256(manifest)
with self.assertRaisesRegex(
ReleaseReachabilityError,
"summary is not the exact derived",
):
validate_release_reachability(
manifest,
repository_root=self.root,
)
def test_seal_rejects_prepopulated_derived_fields(self):
body = {
"schema_version": "1.0",
"artifact_type": "quant60-release-reachability",
"release_id": "quant60-baseline-v1",
"generated_at": GENERATED_AT,
"layers": self._layers(),
"summary": {},
}
with self.assertRaisesRegex(
ReleaseReachabilityError,
"derived summary",
):
seal_release_reachability(body, repository_root=self.root)
if __name__ == "__main__":
unittest.main()