376 lines
22 KiB
Python
376 lines
22 KiB
Python
"""Supervisor regression tests; synthetic receipts are not Blender coverage."""
|
|
import contextlib
|
|
import copy
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from wbv2_autopilot import Autopilot, CONTROL, Lease, POLICY, REQUIRED, ProviderUnavailable, inventory_proof, manifest_proof, queue_context, sha, validate_command, write_json
|
|
from wbv2_timing import WorkerTrace, display_times, minute_label, timing_report
|
|
|
|
|
|
class SupervisorTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temporary.cleanup)
|
|
self.root = Path(self.temporary.name)
|
|
self.policy = json.loads((Path(__file__).with_name("wbv2_autopilot_policy.json")).read_text())
|
|
self.policy.update(heartbeatSeconds=0.01, retryDelaySeconds=0, autoCommit=False, autoPush=False)
|
|
self.task = "WBV2-P42-001"
|
|
self.parent = "tests/golden/WBV2-P42-000/manifest.json"
|
|
self.file("docs/EXECUTION_QUEUE.md", f"| 当前任务 | `{self.task}` |\n| parent manifest | `{self.parent}` |\n")
|
|
self.file(f"docs/tasks/{self.task}.md", "one bounded source workflow\n")
|
|
self.file("source.cpp", "int real_source;\n")
|
|
self.native_source = "blender-5.2.0/source/blender/web_engine/test_fixture.cpp"
|
|
self.file(self.native_source, "int supervisor_test_fixture;\n")
|
|
self.file("engine.wasm", "test fixture, not production WASM\n")
|
|
write_json(self.root / POLICY, self.policy)
|
|
write_json(self.root / self.parent, {"task": "WBV2-P42-000", "status": "done", "nextTask": self.task})
|
|
write_json(self.root / "tests/golden/WBV2/dynamic-inventory.json", {"units": [{"id": "operator:TEST"}], "coverage": {"unverifiedUnits": 1}})
|
|
write_json(self.root / "web/app/src/capabilities/blender-feature-route-catalog.json", {"entries": [{"id": "operator:TEST", "route": "WEB_MAIN"}]})
|
|
self.runner = Autopilot(self.root, self.policy)
|
|
self.context = queue_context(self.root)
|
|
self.job = {"id": "SOURCE-JOB", "ownedPaths": ["source.cpp"], "title": "test source", "queueTask": self.task}
|
|
|
|
def file(self, path, content):
|
|
p = self.root / path
|
|
p.parent.mkdir(parents=True, exist_ok=True)
|
|
p.write_text(content)
|
|
return p
|
|
|
|
def ready_result(self, run_dir):
|
|
kinds = ["positive", "negative", "undo_redo", "save_reopen", "determinism"]
|
|
cases = [{"id": kind, "kind": kind, "checkId": "native", "receipt": str((run_dir / "cases" / f"{kind}.json").relative_to(self.root)), "unitIds": ["operator:TEST"]} for kind in kinds]
|
|
result = {"status": "ready", "summary": "synthetic supervisor fixture", "changedFiles": ["source.cpp", "check.py"], "sourceUnits": ["operator:TEST"], "dependencies": ["source.cpp", "check.py"], "artifacts": ["engine.wasm"], "checks": [{"id": "native", "kinds": sorted(REQUIRED), "argv": ["python3", "check.py"], "cwd": "."}], "cases": cases}
|
|
result.update(workflow="Synthetic supervisor workflow", routeIds=["operator:TEST"], sourceReferences=[self.native_source])
|
|
self.file("check.py", """import hashlib, json, os
|
|
from pathlib import Path
|
|
root = Path.cwd()
|
|
counter = root / 'executions.txt'
|
|
counter.write_text(str(int(counter.read_text()) + 1) if counter.exists() else '1')
|
|
out = Path(os.environ['WBV2_RECEIPT_DIR'])
|
|
out.mkdir(parents=True, exist_ok=True)
|
|
a, b = 'a' * 64, 'b' * 64
|
|
observations = {'positive': {'before': a, 'after': b}, 'negative': {'before': a, 'after': a}, 'undo_redo': {'before': a, 'after': b, 'undo': a, 'redo': b}, 'save_reopen': {'saved': b, 'reopened': b}, 'determinism': {'first': b, 'second': b}}
|
|
for kind, values in observations.items():
|
|
receipt = {'status': 'PASS', 'runtime': 'CHROMIUM_WEBENGINE', 'caseId': kind, 'kind': kind, 'unitIds': ['operator:TEST'], 'observations': values, 'artifacts': [{'path': 'engine.wasm', 'sha256': hashlib.sha256((root / 'engine.wasm').read_bytes()).hexdigest()}]}
|
|
(out / (kind + '.json')).write_text(json.dumps(receipt))
|
|
""")
|
|
return result
|
|
|
|
def worker_fixture(self, prompt, run_dir):
|
|
return self.ready_result(run_dir)
|
|
|
|
def execute_fixture(self):
|
|
with patch.object(self.runner, "worker", side_effect=self.worker_fixture), contextlib.redirect_stdout(io.StringIO()):
|
|
return self.runner.attempt(self.context, self.job)
|
|
|
|
def test_parent_pointer_mismatch_is_rejected(self):
|
|
write_json(self.root / self.parent, {"status": "done", "nextTask": "WBV2-P99-001"})
|
|
with self.assertRaisesRegex(ValueError, "does not authorize"):
|
|
queue_context(self.root)
|
|
|
|
def test_route_complete_or_deferred_is_not_runtime_acceptance(self):
|
|
p = self.root / f"tests/golden/{self.task}/manifest.json"
|
|
write_json(p, {"task": self.task, "status": "done", "nextTask": None, "implementationComplete": True, "verificationDeferred": True})
|
|
self.assertFalse(manifest_proof(self.root, self.task)["ok"])
|
|
|
|
def test_terminal_manifest_checks_artifact_bytes(self):
|
|
p = self.root / f"tests/golden/{self.task}/manifest.json"
|
|
write_json(p, {"task": self.task, "status": "done", "acceptanceComplete": True, "nextTask": None, "artifacts": {"source": {"path": "source.cpp", "sha256": sha(self.root / "source.cpp")}}})
|
|
self.assertTrue(manifest_proof(self.root, self.task)["ok"])
|
|
self.file("source.cpp", "changed")
|
|
self.assertFalse(manifest_proof(self.root, self.task)["ok"])
|
|
|
|
def test_policy_cannot_disable_required_checks(self):
|
|
self.policy["requiredChecks"] = ["typecheck"]
|
|
with self.assertRaisesRegex(ValueError, "mandatory"):
|
|
Autopilot(self.root, self.policy)
|
|
|
|
def test_zero_summary_does_not_hide_unverified_units_or_drift(self):
|
|
p = self.root / "tests/golden/WBV2/dynamic-inventory.json"
|
|
write_json(p, {"units": [{"id": "operator:TEST", "coverageStatus": "UNVERIFIED"}], "coverage": {"unverifiedUnits": 0}})
|
|
self.assertFalse(inventory_proof(self.root)["ok"])
|
|
write_json(p, {"units": [{"id": "operator:TEST", "coverageStatus": "VERIFIED", "coverage": {}}], "coverage": {"unverifiedUnits": 0}})
|
|
self.assertEqual(1, inventory_proof(self.root)["unitsWithInvalidReferences"])
|
|
|
|
def test_lock_excludes_concurrent_owner_and_reuses_inode(self):
|
|
p = self.root / "owner.lock"
|
|
with Lease(p):
|
|
inode = p.stat().st_ino
|
|
with self.assertRaisesRegex(RuntimeError, "owns"):
|
|
with Lease(p):
|
|
pass
|
|
with Lease(p):
|
|
self.assertEqual(inode, p.stat().st_ino)
|
|
|
|
def test_command_rejects_shell_inline_code_other_browsers_and_escape(self):
|
|
for argv, cwd in [(["bash", "-c", "true"], "."), (["node", "-e", "process.exit(0)"], "."), (["npm", "run", "test", "--", "--project=webkit"], "."), (["git", "reset", "--hard"], "."), (["python3", "test.py"], "../outside")]:
|
|
with self.subTest(argv=argv), self.assertRaises(ValueError):
|
|
validate_command(self.root, argv, cwd)
|
|
|
|
def test_symlink_escape_is_rejected(self):
|
|
(self.root / "external").symlink_to("/tmp", target_is_directory=True)
|
|
with self.assertRaises(ValueError):
|
|
validate_command(self.root, ["npm", "run", "test"], "external")
|
|
|
|
def test_agent_ready_without_behavior_matrix_is_not_success(self):
|
|
with patch.object(self.runner, "worker", return_value={"status": "ready", "summary": "done", "sourceUnits": ["operator:TEST"], "checks": []}), contextlib.redirect_stdout(io.StringIO()):
|
|
self.assertFalse(self.runner.attempt(self.context, self.job))
|
|
self.assertEqual("retry_pending", self.runner.state["jobs"]["SOURCE-JOB"]["state"])
|
|
|
|
def test_fresh_runtime_receipts_and_hashes_are_required(self):
|
|
run_dir = self.runner.directory / "runs/test"
|
|
result = self.ready_result(run_dir)
|
|
self.file(result["cases"][0]["receipt"], "{}")
|
|
with self.assertRaisesRegex(ValueError, "fresh"):
|
|
self.runner.validate_result(result, run_dir)
|
|
|
|
def test_real_process_checks_create_receipts_and_cache_invalidates(self):
|
|
self.assertTrue(self.execute_fixture())
|
|
record = self.runner.state["jobs"]["SOURCE-JOB"]
|
|
self.assertTrue(self.runner.receipt_valid(record))
|
|
self.assertEqual("1", (self.root / "executions.txt").read_text())
|
|
self.file("source.cpp", "changed native source")
|
|
self.assertFalse(self.runner.receipt_valid(record))
|
|
|
|
def test_identical_checks_are_run_once(self):
|
|
def worker(prompt, run_dir):
|
|
result = self.ready_result(run_dir)
|
|
duplicate = copy.deepcopy(result["checks"][0])
|
|
duplicate["id"] = "same-command"
|
|
result["checks"].append(duplicate)
|
|
return result
|
|
with patch.object(self.runner, "worker", side_effect=worker), contextlib.redirect_stdout(io.StringIO()):
|
|
self.assertTrue(self.runner.attempt(self.context, self.job))
|
|
self.assertEqual("1", (self.root / "executions.txt").read_text())
|
|
|
|
def test_source_mutation_during_validation_is_not_accepted(self):
|
|
def worker(prompt, run_dir):
|
|
result = self.ready_result(run_dir)
|
|
with (self.root / "check.py").open("a") as stream:
|
|
stream.write("\n(root / 'source.cpp').write_text('drift')\n")
|
|
return result
|
|
with patch.object(self.runner, "worker", side_effect=worker), contextlib.redirect_stdout(io.StringIO()):
|
|
self.assertFalse(self.runner.attempt(self.context, self.job))
|
|
self.assertEqual("needs_reconcile", self.runner.state["state"])
|
|
|
|
def test_protected_queue_edit_requires_reconciliation(self):
|
|
def worker(prompt, run_dir):
|
|
self.file("docs/EXECUTION_QUEUE.md", "changed")
|
|
return self.ready_result(run_dir)
|
|
with patch.object(self.runner, "worker", side_effect=worker), contextlib.redirect_stdout(io.StringIO()):
|
|
self.assertFalse(self.runner.attempt(self.context, self.job))
|
|
self.assertEqual("needs_reconcile", self.runner.state["state"])
|
|
|
|
def test_failed_check_never_marks_job_verified_and_retries_are_bounded(self):
|
|
def worker(prompt, run_dir):
|
|
result = self.ready_result(run_dir)
|
|
self.file("check.py", "raise SystemExit(7)\n")
|
|
return result
|
|
with patch.object(self.runner, "worker", side_effect=worker), contextlib.redirect_stdout(io.StringIO()):
|
|
for _ in range(3):
|
|
self.assertFalse(self.runner.attempt(self.context, self.job))
|
|
self.assertEqual("blocked", self.runner.state["state"])
|
|
self.assertEqual(3, self.runner.state["jobs"]["SOURCE-JOB"]["attempts"])
|
|
|
|
def test_negative_case_with_mutation_fails(self):
|
|
self.assertTrue(self.execute_fixture())
|
|
record = self.runner.state["jobs"]["SOURCE-JOB"]
|
|
run_dir = self.root / record["runDirectory"]
|
|
result = self.ready_result(run_dir)
|
|
p = run_dir / "cases/negative.json"
|
|
receipt = json.loads(p.read_text())
|
|
receipt["observations"]["after"] = "b" * 64
|
|
write_json(p, receipt)
|
|
with self.assertRaisesRegex(ValueError, "behavior assertion"):
|
|
self.runner.verify_cases(result)
|
|
|
|
def test_worker_result_does_not_override_nonzero_process_exit(self):
|
|
run_dir = self.runner.directory / "runs/failed-worker"
|
|
run_dir.mkdir(parents=True)
|
|
def process(*args, **kwargs):
|
|
self.file(str((run_dir / "worker.jsonl").relative_to(self.root)), '{"type":"turn.completed","usage":{"input_tokens":10,"output_tokens":2}}\n')
|
|
write_json(run_dir / "worker-result.json", {"status": "ready"})
|
|
return {"exitCode": 9, "interrupted": False, "timedOut": False, "logLimit": False}
|
|
with patch.object(self.runner, "run_process", side_effect=process), self.assertRaisesRegex(ValueError, "exit=9"):
|
|
self.runner.worker("test", run_dir)
|
|
self.assertEqual(12, self.runner.run_tokens)
|
|
|
|
def test_unknown_previous_writer_is_never_replayed(self):
|
|
self.runner.state["state"] = "implementing"
|
|
with patch.object(self.runner, "worker") as worker:
|
|
self.assertEqual(3, self.runner.run(max_cycles=1))
|
|
worker.assert_not_called()
|
|
self.assertEqual("needs_reconcile", self.runner.state["state"])
|
|
|
|
def test_timeout_kills_descendants_before_late_publication(self):
|
|
child = self.file("late.py", "import time\nfrom pathlib import Path\ntime.sleep(1)\nPath('late-result').write_text('wrong')\n")
|
|
parent = self.file("hang.py", f"import subprocess, sys, time\nsubprocess.Popen([sys.executable, {str(child)!r}])\ntime.sleep(30)\n")
|
|
self.runner.checkpoint("validating")
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
result = self.runner.run_process([sys.executable, str(parent)], self.root, self.runner.directory / "timeout.log", 0.1)
|
|
self.assertTrue(result["timedOut"])
|
|
time.sleep(1.1)
|
|
self.assertFalse((self.root / "late-result").exists())
|
|
|
|
def test_dirty_user_file_cannot_be_included_in_automatic_commit(self):
|
|
self.policy["autoCommit"] = True
|
|
write_json(self.runner.directory / "baseline.json", {"head": "none", "dirtyPaths": ["source.cpp"]})
|
|
with self.assertRaisesRegex(ValueError, "pre-existing edits"):
|
|
self.runner.publish(self.context, ["source.cpp"])
|
|
|
|
def test_commit_excludes_unrelated_staged_changes(self):
|
|
def git(*args):
|
|
return subprocess.check_output(["git", *args], cwd=self.root, stderr=subprocess.DEVNULL, text=True).strip()
|
|
git("init", "-b", "main")
|
|
git("config", "user.name", "Supervisor Test")
|
|
git("config", "user.email", "supervisor@example.invalid")
|
|
self.file("user.txt", "original")
|
|
git("add", ".")
|
|
git("commit", "-m", "fixture")
|
|
self.file("user.txt", "user staged edit")
|
|
git("add", "user.txt")
|
|
self.runner.remember_baseline()
|
|
self.file("source.cpp", "verified implementation")
|
|
self.policy["autoCommit"] = True
|
|
self.assertTrue(self.runner.publish(self.context, ["source.cpp"]))
|
|
self.assertEqual("source.cpp", git("diff-tree", "--no-commit-id", "--name-only", "-r", "HEAD"))
|
|
self.assertEqual("user.txt", git("diff", "--cached", "--name-only"))
|
|
self.assertEqual("user staged edit", (self.root / "user.txt").read_text())
|
|
|
|
def test_daemon_lease_handoff_keeps_single_owner(self):
|
|
lock = self.root / "daemon.lock"
|
|
module_dir = str(Path(__file__).parent)
|
|
with Lease(lock) as lease:
|
|
fd = lease.stream.fileno()
|
|
code = f"from pathlib import Path\nimport time\nfrom wbv2_autopilot import Lease\nwith Lease(Path({str(lock)!r}), {fd}):\n print('owned', flush=True)\n time.sleep(0.4)\n"
|
|
child = subprocess.Popen([sys.executable, "-c", code], stdout=subprocess.PIPE, text=True, pass_fds=(fd,), env={**os.environ, "PYTHONPATH": module_dir})
|
|
lease.transferred = True
|
|
try:
|
|
self.assertEqual("owned", child.stdout.readline().strip())
|
|
with self.assertRaises(RuntimeError):
|
|
with Lease(lock):
|
|
pass
|
|
self.assertEqual(0, child.wait(timeout=3))
|
|
finally:
|
|
if child.poll() is None:
|
|
child.kill()
|
|
child.wait()
|
|
child.stdout.close()
|
|
with Lease(lock):
|
|
pass
|
|
|
|
def test_legacy_preflight_is_read_only_and_rejects_incomplete_terminal(self):
|
|
repo = Path(__file__).resolve().parents[1]
|
|
for name in ("run-wbv2-scheduler.mjs", "wbv2-failure-ledger.mjs"):
|
|
self.file("tools/web/" + name, (repo / "tools/web" / name).read_text())
|
|
self.file("tools/web/check-task-context.mjs", "process.exit(0);\n")
|
|
write_json(self.root / "tests/golden/WBV2/task-index.json", {"activeTask": None, "entries": {self.task: {}}})
|
|
write_json(self.root / f"tests/golden/{self.task}/manifest.json", {"task": self.task, "status": "in_progress", "nextTask": None})
|
|
checkpoint = self.file("tests/golden/WBV2/continuous-scheduler.json", '{"state":"sentinel"}\n')
|
|
before = checkpoint.read_bytes()
|
|
result = subprocess.run(["node", "tools/web/run-wbv2-scheduler.mjs", "--preflight-only"], cwd=self.root, capture_output=True, text=True, timeout=10)
|
|
self.assertEqual(2, result.returncode, result.stderr)
|
|
self.assertIn("implementation_entrypoint_missing", result.stdout)
|
|
self.assertEqual(before, checkpoint.read_bytes())
|
|
self.assertFalse((self.root / "tests/golden/WBV2/continuous-scheduler.lock").exists())
|
|
|
|
def test_time_labels_are_beijing_minutes(self):
|
|
self.assertEqual("2026-09-08 10:10", minute_label("2026-09-08T02:10:59.999Z"))
|
|
self.assertEqual({"updatedAt": "2026-09-08 10:10", "durationSeconds": 0.04}, display_times({"updatedAt": "2026-09-08T02:10:59+00:00", "durationSeconds": 0.04}))
|
|
|
|
def test_quick_command_has_no_one_second_polling_floor(self):
|
|
script = self.file("quick.py", "import time\ntime.sleep(0.03)\n")
|
|
self.policy.update(heartbeatSeconds=5, monitorPollSeconds=0.25)
|
|
with patch.object(self.runner, "checkpoint", wraps=self.runner.checkpoint) as checkpoints, contextlib.redirect_stdout(io.StringIO()):
|
|
result = self.runner.run_process([sys.executable, str(script)], self.root, self.runner.directory / "quick.log", 5)
|
|
self.assertLess(result["durationSeconds"], 0.9)
|
|
self.assertEqual(2, checkpoints.call_count)
|
|
|
|
def model_script(self, source):
|
|
script = self.file("model_fixture.py", source)
|
|
self.policy.update(monitorPollSeconds=0.02, modelStartTimeoutSeconds=0.15, modelIdleTimeoutSeconds=0.15)
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
return self.runner.run_process([sys.executable, str(script)], self.root, self.runner.directory / "model.log", 5, monitor_worker=True, phase="model_request")
|
|
|
|
def test_provider_error_before_actions_releases_process_early(self):
|
|
result = self.model_script("import time\nprint('{\"type\":\"turn.started\"}', flush=True)\nprint('{\"type\":\"error\",\"message\":\"Concurrency limit exceeded\"}', flush=True)\ntime.sleep(10)\n")
|
|
self.assertEqual("provider_unavailable", result["providerAbort"])
|
|
self.assertTrue(result["workerTrace"]["safeToRetry"])
|
|
self.assertLess(result["durationSeconds"], 1)
|
|
|
|
def test_model_start_without_progress_is_bounded(self):
|
|
result = self.model_script("import time\nprint('{\"type\":\"turn.started\"}', flush=True)\ntime.sleep(10)\n")
|
|
self.assertEqual("model_start_timeout", result["providerAbort"])
|
|
self.assertTrue(result["workerTrace"]["safeToRetry"])
|
|
|
|
def test_stalled_model_after_actions_requires_reconciliation(self):
|
|
result = self.model_script("import time\nprint('{\"type\":\"turn.started\"}', flush=True)\nprint('{\"type\":\"item.completed\",\"item\":{\"id\":\"edit\",\"type\":\"file_change\"}}', flush=True)\ntime.sleep(10)\n")
|
|
self.assertEqual("model_idle_timeout", result["providerAbort"])
|
|
self.assertFalse(result["workerTrace"]["safeToRetry"])
|
|
|
|
def test_active_tool_is_not_mistaken_for_model_idle(self):
|
|
result = self.model_script("import time\nprint('{\"type\":\"turn.started\"}', flush=True)\nprint('{\"type\":\"item.started\",\"item\":{\"id\":\"build\",\"type\":\"command_execution\",\"command\":\"cmake --build build\"}}', flush=True)\ntime.sleep(0.25)\nprint('{\"type\":\"item.completed\",\"item\":{\"id\":\"build\",\"type\":\"command_execution\",\"exit_code\":0}}', flush=True)\n")
|
|
self.assertIsNone(result["providerAbort"])
|
|
self.assertEqual(0, result["exitCode"])
|
|
self.assertEqual("build", result["workerTrace"]["items"][0]["phase"])
|
|
|
|
def test_provider_failures_have_separate_bounded_budget(self):
|
|
with patch.object(self.runner, "worker", side_effect=ProviderUnavailable("rate limit")), contextlib.redirect_stdout(io.StringIO()):
|
|
for failure in range(1, 5):
|
|
self.assertFalse(self.runner.attempt(self.context, self.job))
|
|
record = self.runner.state["jobs"][self.job["id"]]
|
|
self.assertEqual(0, record["attempts"])
|
|
self.assertEqual(failure, record["providerFailures"])
|
|
self.assertEqual("provider_blocked", self.runner.state["state"])
|
|
|
|
def test_provider_failure_with_source_changes_is_not_retried(self):
|
|
def changed(*args):
|
|
self.file("source.cpp", "unexpected change")
|
|
raise ProviderUnavailable("overload")
|
|
with patch.object(self.runner, "worker", side_effect=changed), contextlib.redirect_stdout(io.StringIO()):
|
|
self.assertFalse(self.runner.attempt(self.context, self.job))
|
|
self.assertEqual("needs_reconcile", self.runner.state["state"])
|
|
|
|
def test_jsonl_partial_reads_and_nested_tool_timing(self):
|
|
trace = WorkerTrace()
|
|
p = self.file("events.jsonl", '{"type":"turn.star')
|
|
trace.read(p, 0.1)
|
|
self.assertFalse(trace.safe_to_retry)
|
|
with p.open("a") as stream:
|
|
stream.write('ted"}\n')
|
|
trace.read(p, 0.2)
|
|
self.assertTrue(trace.safe_to_retry)
|
|
trace.feed({"type": "item.started", "item": {"id": "read", "type": "command_execution", "command": "rg pattern source.cpp"}}, 1)
|
|
trace.feed({"type": "item.completed", "item": {"id": "read", "type": "command_execution", "exit_code": 0}}, 1.4)
|
|
self.assertAlmostEqual(0.4, trace.items[0]["durationSeconds"])
|
|
self.assertEqual("reading", trace.items[0]["phase"])
|
|
event = {"event": "process.finished", "argv": ["codex", "exec"], "log": "model.log", "startedAt": "2026-09-08T02:10:00+00:00", "finishedAt": "2026-09-08T02:10:02+00:00", "durationSeconds": 2, "workerTrace": trace.snapshot(2)}
|
|
self.file(CONTROL + "/events.jsonl", json.dumps(event) + "\n")
|
|
report = timing_report(self.runner.directory)
|
|
self.assertAlmostEqual(2 / 60, report["totalMeasuredMinutes"], places=4)
|
|
self.assertEqual("2026-09-08 10:10", report["steps"][0]["start"])
|
|
|
|
def test_business_routes_and_native_source_are_mandatory(self):
|
|
run_dir = self.runner.directory / "runs/contract"
|
|
result = self.ready_result(run_dir)
|
|
result["routeIds"] = []
|
|
with self.assertRaisesRegex(ValueError, "business workflow"):
|
|
self.runner.validate_result(result, run_dir)
|
|
result["routeIds"] = ["operator:TEST"]
|
|
result["sourceReferences"] = ["source.cpp"]
|
|
with self.assertRaisesRegex(ValueError, "native Blender source"):
|
|
self.runner.validate_result(result, run_dir)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|