102 lines
6.0 KiB
Python
102 lines
6.0 KiB
Python
import contextlib
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
import record_runtime_issue as recorder
|
|
import prepare_feature_capability_batch as planner
|
|
import preflight_current_module_batch as preflight
|
|
import run_module_batch_pipeline as pipeline
|
|
|
|
|
|
class ProductRecoveryTests(unittest.TestCase):
|
|
def test_product_fault_registration_and_planning_preserve_business_identity(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
current = root / "docs/current"
|
|
current.mkdir(parents=True)
|
|
paths = {name: current / filename for name, filename in {
|
|
"REGISTRY": "feature-capability-registry.json", "ISSUES": "runtime-issue-ledger.json",
|
|
"STATE": "task-state.json", "OUTPUT": "module-batch-manifest.json",
|
|
"PIPELINE": "module-batch-pipeline.json", "PROGRESS": "scheduler-progress.json"}.items()}
|
|
paths["REGISTRY"].write_text('{"records": []}')
|
|
paths["STATE"].write_text('{"active_task": "task"}')
|
|
(current / "product-business-backlog.json").write_text(json.dumps({"records": [
|
|
{"id": "product:task-panel", "name": "Task panel", "source": "web/src/project-shell.tsx", "sourceLine": 1}]}))
|
|
with patch.multiple(recorder, ROOT=root, ISSUES=paths["ISSUES"]), \
|
|
patch.object(sys, "argv", ["record", "product:task-panel", "--error", "panels=0"]), \
|
|
contextlib.redirect_stdout(io.StringIO()):
|
|
self.assertEqual(recorder.main(), 0)
|
|
issue = json.loads(paths["ISSUES"].read_text())["issues"][0]
|
|
self.assertEqual(issue["capabilityId"], "product:task-panel")
|
|
self.assertEqual(issue["module"], "Web")
|
|
self.assertEqual(issue["cmd"], "Task panel")
|
|
before = paths["ISSUES"].read_bytes()
|
|
with patch.multiple(recorder, ROOT=root, ISSUES=paths["ISSUES"]), \
|
|
patch.object(sys, "argv", ["record", "product:task-panel", "--error", "panels=0", "--check-only"]), \
|
|
contextlib.redirect_stdout(io.StringIO()):
|
|
self.assertEqual(recorder.main(), 0)
|
|
self.assertEqual(paths["ISSUES"].read_bytes(), before)
|
|
with patch.multiple(planner, ROOT=root, **paths), \
|
|
patch.object(planner, "capability_source_fingerprint", return_value="stable"), \
|
|
patch.object(planner, "implementation_ready_ids", return_value=set()), \
|
|
patch.object(planner, "cooling_ids", return_value=set()), \
|
|
contextlib.redirect_stdout(io.StringIO()):
|
|
self.assertEqual(planner.main(["--runtime-only"]), 0)
|
|
batch = json.loads(paths["OUTPUT"].read_text())
|
|
self.assertEqual([row[0] for row in batch["records"]], ["product:task-panel"])
|
|
self.assertEqual(batch["selected"]["family"], "runtime-issue")
|
|
self.assertTrue(preflight.record_belongs_to_batch(batch, "Web", "product:task-panel"))
|
|
self.assertFalse(preflight.record_belongs_to_batch(batch, "Part", "product:task-panel"))
|
|
batch["selected"]["family"] = "desktop-command"
|
|
self.assertFalse(preflight.record_belongs_to_batch(batch, "Web", "product:task-panel"))
|
|
|
|
def test_unknown_product_is_rejected_without_creating_an_issue(self):
|
|
with tempfile.TemporaryDirectory() as directory, patch.object(recorder, "ROOT", Path(directory)), \
|
|
patch.object(sys, "argv", ["record", "product:invented", "--error", "missing"]):
|
|
with self.assertRaisesRegex(SystemExit, "unknown capability"):
|
|
recorder.main()
|
|
|
|
def test_runtime_repair_runs_real_contract_without_static_marker_or_receipt(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
(root / "docs/business-acceptance").mkdir(parents=True)
|
|
(root / "cases.py").write_text("pass\n")
|
|
contract = "docs/business-acceptance/Web.json"
|
|
(root / contract).write_text(json.dumps({"command": ["python3", "cases.py"], "coveredIds": ["product:task-panel"]}))
|
|
batch = {"batchId": "runtime", "strategy": "feature-first-deferred-validation",
|
|
"selected": {"module": "Web", "tier": "feature-workflow", "family": "runtime-issue"},
|
|
"records": [["product:task-panel"]]}
|
|
packet = {"batchId": "runtime", "outerPipelineAvailable": True, "acceptanceContract": contract}
|
|
paths = {name: root / (name + ".json") for name in ("BATCH_PATH", "PACKET_PATH", "OUTPUT_PATH")}
|
|
paths["BATCH_PATH"].write_text(json.dumps(batch))
|
|
paths["PACKET_PATH"].write_text(json.dumps(packet))
|
|
commands = []
|
|
def run(name, command, environment, timeout):
|
|
commands.append(command)
|
|
self.assertEqual(name, "business-acceptance")
|
|
self.assertIn("--require-issue-closure", command)
|
|
return {"name": name, "exitCode": 0, "seconds": 0.01}
|
|
with patch.multiple(pipeline, ROOT=root, **paths), \
|
|
patch.object(pipeline, "run_step", side_effect=run), \
|
|
patch.object(pipeline, "record_implementation_batch") as receipt, \
|
|
patch.object(pipeline, "static_readiness") as readiness, \
|
|
patch.object(pipeline.subprocess, "run"), contextlib.redirect_stdout(io.StringIO()):
|
|
self.assertEqual(pipeline.main(), 0)
|
|
self.assertEqual(len(commands), 1)
|
|
receipt.assert_not_called()
|
|
readiness.assert_not_called()
|
|
self.assertEqual(json.loads(paths["OUTPUT_PATH"].read_text())["status"], "pass")
|
|
(root / contract).write_text(json.dumps({"command": ["python3", "cases.py"], "coveredIds": []}))
|
|
self.assertEqual(pipeline.main(), 78)
|
|
self.assertEqual(len(commands), 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|