135 lines
4.8 KiB
Python
135 lines
4.8 KiB
Python
import json
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from quant60.ledger import canonical_json
|
|
from quant60.release_reachability import LAYERS
|
|
|
|
|
|
PROJECT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(PROJECT))
|
|
|
|
from tools.build_release_reachability import ( # noqa: E402
|
|
DEFAULT_OUTPUT,
|
|
JOINQUANT_TARGET_PACKAGE_OBSERVED_AT,
|
|
JOINQUANT_TARGET_PACKAGE_RUN_ID,
|
|
JOINQUANT_TARGET_PACKAGE_RUN_PATH,
|
|
PLATFORM_ENTRYPOINTS,
|
|
RELEASE_ID,
|
|
build_manifest,
|
|
)
|
|
|
|
|
|
class BuildReleaseReachabilityTests(unittest.TestCase):
|
|
def test_checked_in_manifest_exactly_matches_generator(self):
|
|
expected = build_manifest(repository_root=PROJECT)
|
|
observed = json.loads(DEFAULT_OUTPUT.read_text(encoding="utf-8"))
|
|
self.assertEqual(observed, expected)
|
|
self.assertEqual(
|
|
DEFAULT_OUTPUT.read_bytes(),
|
|
(canonical_json(expected) + "\n").encode("utf-8"),
|
|
)
|
|
|
|
def test_only_execution_has_real_joinquant_observation(self):
|
|
manifest = build_manifest(repository_root=PROJECT)
|
|
self.assertEqual(manifest["release_id"], RELEASE_ID)
|
|
for layer in LAYERS:
|
|
claims = manifest["layers"][layer]
|
|
for declaration in PLATFORM_ENTRYPOINTS.values():
|
|
claim = claims[declaration["capability"]]
|
|
self.assertTrue(claim["status"])
|
|
self.assertEqual(len(claim["artifacts"]), 1)
|
|
artifact = claim["artifacts"][0]
|
|
self.assertIn("targetpackagev1", artifact["artifact_id"])
|
|
self.assertEqual(artifact["path"], declaration["path"])
|
|
self.assertEqual(artifact["platform"], declaration["platform"])
|
|
observed = claims["real_platform_observed"]
|
|
if layer != "execution":
|
|
self.assertEqual(
|
|
observed,
|
|
{"status": False, "artifacts": []},
|
|
)
|
|
continue
|
|
self.assertTrue(observed["status"])
|
|
self.assertEqual(len(observed["artifacts"]), 1)
|
|
artifact = observed["artifacts"][0]
|
|
self.assertEqual(artifact["kind"], "platform_run")
|
|
self.assertEqual(artifact["platform"], "joinquant")
|
|
self.assertEqual(
|
|
artifact["path"],
|
|
JOINQUANT_TARGET_PACKAGE_RUN_PATH,
|
|
)
|
|
self.assertEqual(
|
|
artifact["observed_at"],
|
|
JOINQUANT_TARGET_PACKAGE_OBSERVED_AT,
|
|
)
|
|
self.assertEqual(
|
|
artifact["run_id"],
|
|
JOINQUANT_TARGET_PACKAGE_RUN_ID,
|
|
)
|
|
self.assertFalse(
|
|
manifest["summary"]["all_layers_real_platform_observed"]
|
|
)
|
|
|
|
def test_tracked_joinquant_run_preserves_conservative_boundary(self):
|
|
run = json.loads(
|
|
(PROJECT / JOINQUANT_TARGET_PACKAGE_RUN_PATH).read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
self.assertEqual(run["run_id"], JOINQUANT_TARGET_PACKAGE_RUN_ID)
|
|
self.assertEqual(run["observed_at"], JOINQUANT_TARGET_PACKAGE_OBSERVED_AT)
|
|
self.assertEqual(
|
|
run["platform_evidence_class"],
|
|
"real_platform_runtime",
|
|
)
|
|
self.assertEqual(
|
|
run["input_evidence_class"],
|
|
"synthetic-research",
|
|
)
|
|
self.assertFalse(run["performance_claim"])
|
|
self.assertEqual(
|
|
run["target_package"]["evidence_class"],
|
|
"synthetic-research",
|
|
)
|
|
self.assertEqual(run["counts"], {
|
|
"orders": 3,
|
|
"fills": 3,
|
|
"skipped_orders": 0,
|
|
})
|
|
boundary = run["claim_boundary"]
|
|
self.assertEqual(
|
|
boundary["real_platform_observed_layers"],
|
|
["execution"],
|
|
)
|
|
self.assertFalse(boundary["upstream_layers_real_platform_observed"])
|
|
self.assertFalse(boundary["qmt_peer_observed"])
|
|
self.assertFalse(boundary["same_input_cross_engine_parity_observed"])
|
|
self.assertFalse(boundary["all_five_layers_real_platform_observed"])
|
|
self.assertFalse(boundary["g9_passed"])
|
|
self.assertFalse(boundary["baseline_60_passed"])
|
|
self.assertFalse(boundary["investment_value_claim"])
|
|
|
|
def test_check_mode_verifies_current_file(self):
|
|
completed = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(PROJECT / "tools/build_release_reachability.py"),
|
|
"--check",
|
|
],
|
|
cwd=PROJECT,
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
self.assertEqual(completed.returncode, 0, completed.stderr)
|
|
result = json.loads(completed.stdout)
|
|
self.assertTrue(result["ok"])
|
|
self.assertTrue(result["checked"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|