test: add contract tests for intelligence, ops, machine registry, yiacad
- test_intelligence_tui_contract.py - test_log_ops_contract.py - test_machine_registry_contract.py - test_runtime_ai_gateway_contract.py - test_yiacad_native_surface_contract.py - test_yiacad_uiux_tui_contract.py - test_zeroclaw_n8n_workflow_contract.py Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
L'électron rare
@@ -0,0 +1,231 @@
|
||||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
SCRIPT = REPO_ROOT / "tools" / "cockpit" / "intelligence_tui.sh"
|
||||
LEGACY_SCRIPT = REPO_ROOT / "tools" / "cockpit" / "intelligence_program_tui.sh"
|
||||
|
||||
|
||||
class IntelligenceTuiContractTests(unittest.TestCase):
|
||||
def run_script(self, *args: str) -> dict:
|
||||
proc = subprocess.run(
|
||||
["bash", str(SCRIPT), *args],
|
||||
cwd=str(REPO_ROOT),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
self.assertTrue(proc.stdout.strip(), proc.stderr)
|
||||
return json.loads(proc.stdout)
|
||||
|
||||
def run_script_from_cwd(self, cwd: Path, *args: str) -> dict:
|
||||
proc = subprocess.run(
|
||||
["bash", str(SCRIPT), *args],
|
||||
cwd=str(cwd),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
self.assertTrue(proc.stdout.strip(), proc.stderr)
|
||||
return json.loads(proc.stdout)
|
||||
|
||||
def run_text_script(self, *args: str) -> str:
|
||||
proc = subprocess.run(
|
||||
["bash", str(SCRIPT), *args],
|
||||
cwd=str(REPO_ROOT),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
self.assertTrue(proc.stdout.strip(), proc.stderr)
|
||||
return proc.stdout
|
||||
|
||||
def test_status_reports_canonical_sources_and_extension_repos(self) -> None:
|
||||
payload = self.run_script("--action", "status", "--json")
|
||||
self.assertEqual(payload["contract_version"], "cockpit-v1")
|
||||
self.assertEqual(payload["component"], "intelligence_tui")
|
||||
self.assertEqual(payload["action"], "status")
|
||||
self.assertIn("canonical_sources", payload)
|
||||
self.assertGreaterEqual(len(payload["canonical_sources"]), 4)
|
||||
self.assertTrue(payload["spec_doc"].endswith("specs/agentic_intelligence_integration_spec.md"))
|
||||
self.assertTrue(payload["plan_doc"].endswith("docs/plans/22_plan_integration_intelligence_agentique.md"))
|
||||
self.assertTrue(payload["todo_doc"].endswith("docs/plans/22_todo_integration_intelligence_agentique.md"))
|
||||
self.assertTrue(payload["web_spec_doc"].endswith("specs/yiacad_git_eda_platform_spec.md"))
|
||||
self.assertTrue(payload["web_plan_doc"].endswith("docs/plans/23_plan_yiacad_git_eda_platform.md"))
|
||||
self.assertTrue(payload["web_todo_doc"].endswith("docs/plans/23_todo_yiacad_git_eda_platform.md"))
|
||||
labels = {source["label"] for source in payload["canonical_sources"]}
|
||||
self.assertIn("Plan 23", labels)
|
||||
self.assertIn("Web spec", labels)
|
||||
repo_names = {repo["name"] for repo in payload["extension_repos"]}
|
||||
self.assertTrue({"kill-life-studio", "kill-life-mesh", "kill-life-operator"}.issubset(repo_names))
|
||||
|
||||
def test_memory_writes_latest_artifacts(self) -> None:
|
||||
payload = self.run_script("--action", "memory", "--json")
|
||||
memory_artifacts = payload["memory_artifacts"]
|
||||
latest_json = Path(memory_artifacts["json"])
|
||||
latest_md = Path(memory_artifacts["markdown"])
|
||||
self.assertTrue(latest_json.exists())
|
||||
self.assertTrue(latest_md.exists())
|
||||
persisted = json.loads(latest_json.read_text(encoding="utf-8"))
|
||||
self.assertEqual(persisted["component"], "intelligence_tui")
|
||||
self.assertIn("intelligence_views", persisted)
|
||||
self.assertEqual(
|
||||
set(persisted["intelligence_views"].keys()),
|
||||
{"scorecard", "comparison", "recommendations"},
|
||||
)
|
||||
|
||||
def test_status_is_repo_root_stable_outside_repo_cwd(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
payload = self.run_script_from_cwd(Path(tmp_dir), "--action", "status", "--json")
|
||||
self.assertTrue(payload["spec_doc"].startswith(str(REPO_ROOT)))
|
||||
self.assertTrue(payload["plan_doc"].startswith(str(REPO_ROOT)))
|
||||
self.assertTrue(payload["todo_doc"].startswith(str(REPO_ROOT)))
|
||||
self.assertTrue(payload["audit_doc"].startswith(str(REPO_ROOT)))
|
||||
self.assertTrue(payload["feature_map_doc"].startswith(str(REPO_ROOT)))
|
||||
self.assertTrue(payload["research_doc"].startswith(str(REPO_ROOT)))
|
||||
self.assertTrue(payload["web_spec_doc"].startswith(str(REPO_ROOT)))
|
||||
self.assertTrue(payload["web_plan_doc"].startswith(str(REPO_ROOT)))
|
||||
self.assertTrue(payload["web_todo_doc"].startswith(str(REPO_ROOT)))
|
||||
self.assertTrue(
|
||||
payload["intelligence_views"]["scorecard"]["json"].startswith(str(REPO_ROOT))
|
||||
)
|
||||
|
||||
def test_next_actions_is_exposed(self) -> None:
|
||||
payload = self.run_script("--action", "next-actions", "--json")
|
||||
self.assertEqual(payload["action"], "next-actions")
|
||||
self.assertIn("next_steps", payload)
|
||||
self.assertGreaterEqual(len(payload["next_steps"]), 1)
|
||||
|
||||
def test_scorecard_persists_lane_maturity_and_fragmentation(self) -> None:
|
||||
payload = self.run_script("--action", "scorecard", "--json")
|
||||
self.assertEqual(payload["action"], "scorecard")
|
||||
self.assertIn("fragmentation_score", payload)
|
||||
self.assertIn("overall_maturity_score", payload)
|
||||
self.assertIn("lane_maturity", payload)
|
||||
self.assertGreaterEqual(len(payload["lane_maturity"]), 4)
|
||||
lane_names = {lane["lane"] for lane in payload["lane_maturity"]}
|
||||
self.assertIn("Program-Governance", lane_names)
|
||||
self.assertIn("Contracts", lane_names)
|
||||
self.assertIn("Runtime-Gateway", lane_names)
|
||||
self.assertIn("Web-Git-EDA", lane_names)
|
||||
scorecard_json = Path(payload["artifacts"][1])
|
||||
scorecard_md = Path(payload["artifacts"][2])
|
||||
self.assertTrue(scorecard_json.exists())
|
||||
self.assertTrue(scorecard_md.exists())
|
||||
|
||||
def test_comparison_covers_five_target_repos(self) -> None:
|
||||
payload = self.run_script("--action", "comparison", "--json")
|
||||
self.assertEqual(payload["action"], "comparison")
|
||||
repo_names = {repo["name"] for repo in payload["repos"]}
|
||||
self.assertEqual(
|
||||
repo_names,
|
||||
{
|
||||
"Kill_LIFE",
|
||||
"ai-agentic-embedded-base",
|
||||
"kill-life-studio",
|
||||
"kill-life-mesh",
|
||||
"kill-life-operator",
|
||||
},
|
||||
)
|
||||
studio = next(repo for repo in payload["repos"] if repo["name"] == "kill-life-studio")
|
||||
self.assertEqual(studio["governance_signal"], "consumer")
|
||||
self.assertIn("governance", studio["enabled_capabilities"])
|
||||
|
||||
def test_recommendations_exposes_prioritized_queue(self) -> None:
|
||||
payload = self.run_script("--action", "recommendations", "--json")
|
||||
self.assertEqual(payload["action"], "recommendations")
|
||||
queue = payload["queue"]
|
||||
self.assertGreaterEqual(len(queue), 5)
|
||||
self.assertEqual(queue[0]["id"], "AI-RQ-101")
|
||||
priorities = [item["priority"] for item in queue]
|
||||
self.assertEqual(priorities, sorted(priorities, key=lambda value: {"P0": 0, "P1": 1, "P2": 2}[value]))
|
||||
recommendation_json = Path(payload["artifacts"][1])
|
||||
recommendation_md = Path(payload["artifacts"][2])
|
||||
self.assertTrue(recommendation_json.exists())
|
||||
self.assertTrue(recommendation_md.exists())
|
||||
|
||||
def test_research_uses_current_official_table(self) -> None:
|
||||
payload = self.run_script("--action", "research", "--json")
|
||||
self.assertEqual(payload["action"], "research")
|
||||
self.assertTrue(payload["research_doc"].startswith(str(REPO_ROOT)))
|
||||
self.assertGreaterEqual(len(payload["research"]), 1)
|
||||
first_row = payload["research"][0]
|
||||
self.assertTrue(
|
||||
any(
|
||||
key in first_row
|
||||
for key in (
|
||||
"Source officielle",
|
||||
"Pattern / source primaire",
|
||||
"Projet / source",
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
def test_summary_short_json_is_compact_and_stable(self) -> None:
|
||||
payload = self.run_script("--action", "summary-short", "--json")
|
||||
self.assertEqual(payload["contract_version"], "summary-short/v1")
|
||||
self.assertEqual(payload["component"], "intelligence_tui")
|
||||
self.assertEqual(payload["action"], "summary-short")
|
||||
self.assertEqual(payload["owner_repo"], "Kill_LIFE")
|
||||
self.assertEqual(payload["owner_agent"], "PM-Mesh")
|
||||
self.assertEqual(payload["owner_subagent"], "Plan-Orchestrator")
|
||||
self.assertIn(payload["status"], {"ready", "degraded", "blocked"})
|
||||
self.assertIsInstance(payload["write_set"], list)
|
||||
self.assertGreaterEqual(len(payload["write_set"]), 1)
|
||||
self.assertIsInstance(payload["evidence"], list)
|
||||
self.assertGreaterEqual(len(payload["evidence"]), 1)
|
||||
self.assertTrue(payload["summary_short"])
|
||||
self.assertLessEqual(len(payload["summary_short"]), 320)
|
||||
self.assertIsInstance(payload["open_task_count"], int)
|
||||
self.assertIsInstance(payload["intelligence_open_todo_count"], int)
|
||||
self.assertIsInstance(payload["global_open_task_count"], int)
|
||||
self.assertLessEqual(len(payload["next_steps"]), 3)
|
||||
self.assertIn("goal", payload)
|
||||
self.assertIn("state", payload)
|
||||
self.assertIn("blockers", payload)
|
||||
self.assertIn("next", payload)
|
||||
self.assertIn("owner", payload)
|
||||
self.assertIn("first_next_step", payload)
|
||||
self.assertIn("first_priority_lane", payload)
|
||||
self.assertNotIn("canonical_sources", payload)
|
||||
self.assertNotIn("owners", payload)
|
||||
self.assertNotIn("research", payload)
|
||||
self.assertNotIn("extension_repos", payload)
|
||||
|
||||
def test_summary_short_text_is_extension_friendly(self) -> None:
|
||||
output = self.run_text_script("--action", "summary-short")
|
||||
lines = [line.strip() for line in output.splitlines() if line.strip()]
|
||||
self.assertGreaterEqual(len(lines), 6)
|
||||
self.assertTrue(all("=" in line for line in lines))
|
||||
self.assertEqual(lines[0], "contract_version=summary-short/v1")
|
||||
self.assertIn("component=intelligence_tui", lines)
|
||||
self.assertIn("action=summary-short", lines)
|
||||
self.assertTrue(any(line.startswith("owner_repo=Kill_LIFE") for line in lines))
|
||||
self.assertTrue(any(line.startswith("summary_short=") for line in lines))
|
||||
self.assertTrue(any(line.startswith("status=") for line in lines))
|
||||
self.assertTrue(any(line.startswith("open_task_count=") for line in lines))
|
||||
self.assertTrue(any(line.startswith("first_next_step=") for line in lines))
|
||||
self.assertTrue(any(line.startswith("evidence=") for line in lines))
|
||||
|
||||
def test_legacy_wrapper_still_works(self) -> None:
|
||||
proc = subprocess.run(
|
||||
["bash", str(LEGACY_SCRIPT), "--action", "summary-short", "--json"],
|
||||
cwd=str(REPO_ROOT),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
payload = json.loads(proc.stdout)
|
||||
self.assertEqual(payload["component"], "intelligence_tui")
|
||||
self.assertEqual(payload["action"], "summary-short")
|
||||
self.assertEqual(payload["contract_version"], "summary-short/v1")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,103 @@
|
||||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
SCRIPT = REPO_ROOT / "tools" / "cockpit" / "log_ops.sh"
|
||||
|
||||
|
||||
class LogOpsContractTests(unittest.TestCase):
|
||||
def run_script(self, target_dirs: list[Path], *args: str) -> dict:
|
||||
env = os.environ.copy()
|
||||
env["LOG_OPS_TARGET_DIRS"] = ":".join(str(path) for path in target_dirs)
|
||||
proc = subprocess.run(
|
||||
["bash", str(SCRIPT), *args],
|
||||
cwd=str(REPO_ROOT),
|
||||
env=env,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
self.assertTrue(proc.stdout.strip(), proc.stderr)
|
||||
return json.loads(proc.stdout)
|
||||
|
||||
def test_summary_is_ok_when_no_logs_are_present(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
artifacts = root / "artifacts"
|
||||
logs = root / "logs"
|
||||
scratch = root / "tmp"
|
||||
for directory in (artifacts, logs, scratch):
|
||||
directory.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
payload = self.run_script([artifacts, logs, scratch], "--action", "summary", "--json")
|
||||
self.assertEqual(payload["contract_version"], "cockpit-v1")
|
||||
self.assertEqual(payload["component"], "log_ops")
|
||||
self.assertEqual(payload["status"], "done")
|
||||
self.assertEqual(payload["contract_status"], "ok")
|
||||
self.assertEqual(payload["count"], 0)
|
||||
self.assertEqual(payload["stale"], 0)
|
||||
|
||||
def test_purge_removes_stale_logs_from_overridden_targets(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
artifacts = root / "artifacts"
|
||||
logs = root / "logs"
|
||||
scratch = root / "tmp"
|
||||
for directory in (artifacts, logs, scratch):
|
||||
directory.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
stale_file = logs / "stale.log"
|
||||
stale_file.write_text("old log\n", encoding="utf-8")
|
||||
old_epoch = stale_file.stat().st_mtime - (3 * 24 * 60 * 60)
|
||||
os.utime(stale_file, (old_epoch, old_epoch))
|
||||
|
||||
summary = self.run_script(
|
||||
[artifacts, logs, scratch],
|
||||
"--action",
|
||||
"summary",
|
||||
"--retention-days",
|
||||
"1",
|
||||
"--json",
|
||||
)
|
||||
self.assertEqual(summary["status"], "degraded")
|
||||
self.assertEqual(summary["stale"], 1)
|
||||
|
||||
purge = self.run_script(
|
||||
[artifacts, logs, scratch],
|
||||
"--action",
|
||||
"purge",
|
||||
"--retention-days",
|
||||
"1",
|
||||
"--apply",
|
||||
"--json",
|
||||
)
|
||||
self.assertEqual(purge["status"], "done")
|
||||
self.assertEqual(purge["purged_count"], 1)
|
||||
self.assertFalse(stale_file.exists())
|
||||
|
||||
def test_summary_json_escapes_paths_with_quotes(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
artifacts = root / "artifacts"
|
||||
logs = root / "logs"
|
||||
scratch = root / "tmp"
|
||||
for directory in (artifacts, logs, scratch):
|
||||
directory.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
quoted_file = logs / 'quoted"log.log'
|
||||
quoted_file.write_text("quoted path\n", encoding="utf-8")
|
||||
|
||||
payload = self.run_script([artifacts, logs, scratch], "--action", "summary", "--json")
|
||||
self.assertEqual(payload["status"], "done")
|
||||
self.assertIn(str(quoted_file), payload["files"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,45 @@
|
||||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import subprocess
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
SCRIPT = REPO_ROOT / "tools" / "cockpit" / "machine_registry.sh"
|
||||
|
||||
|
||||
class MachineRegistryContractTests(unittest.TestCase):
|
||||
def run_script(self, *args: str) -> dict:
|
||||
proc = subprocess.run(
|
||||
["bash", str(SCRIPT), *args],
|
||||
cwd=str(REPO_ROOT),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
self.assertTrue(proc.stdout.strip(), proc.stderr)
|
||||
return json.loads(proc.stdout)
|
||||
|
||||
def test_summary_uses_cockpit_contract(self) -> None:
|
||||
payload = self.run_script("--action", "summary", "--json")
|
||||
self.assertEqual(payload["contract_version"], "cockpit-v1")
|
||||
self.assertEqual(payload["component"], "machine_registry")
|
||||
self.assertEqual(payload["contract_status"], "ok")
|
||||
self.assertEqual(payload["status"], "ok")
|
||||
self.assertEqual(payload["action"], "summary")
|
||||
self.assertIn("tower", payload["priority_order"])
|
||||
self.assertIn("root-reserve", payload["reserve_targets"])
|
||||
self.assertGreaterEqual(payload["target_count"], 1)
|
||||
self.assertTrue(payload["artifacts"])
|
||||
|
||||
def test_show_returns_single_machine(self) -> None:
|
||||
payload = self.run_script("--action", "show", "--machine", "tower", "--json")
|
||||
self.assertEqual(payload["action"], "show")
|
||||
self.assertEqual(len(payload["targets"]), 1)
|
||||
self.assertEqual(payload["targets"][0]["id"], "tower")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,157 @@
|
||||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
SCRIPT = REPO_ROOT / "tools" / "cockpit" / "runtime_ai_gateway.sh"
|
||||
|
||||
|
||||
class RuntimeAiGatewayContractTests(unittest.TestCase):
|
||||
def test_status_aggregates_sources_into_runtime_mcp_ia_contract(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
root = Path(tmp_dir)
|
||||
intelligence = root / "intelligence.json"
|
||||
mesh = root / "mesh.json"
|
||||
mascarade = root / "mascarade.json"
|
||||
|
||||
intelligence.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"contract_version": "summary-short/v1",
|
||||
"status": "degraded",
|
||||
"summary_short": "IA lane degraded with open governance items.",
|
||||
"evidence": ["artifacts/cockpit/intelligence_program/latest.json"],
|
||||
"open_task_count": 3,
|
||||
"next_steps": ["Do intelligence thing"],
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
mesh.write_text(
|
||||
"\n".join(
|
||||
[
|
||||
"INFO 2026-03-21 mesh preflight",
|
||||
json.dumps(
|
||||
{
|
||||
"mesh_status": "ready",
|
||||
"load_profile": "tower-first",
|
||||
"host_order": ["clems", "kxkm"],
|
||||
}
|
||||
),
|
||||
]
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
mascarade.write_text(
|
||||
json.dumps({"status": "ok", "provider": "ollama", "model": "qwen"}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
proc = subprocess.run(
|
||||
[
|
||||
"bash",
|
||||
str(SCRIPT),
|
||||
"--action",
|
||||
"status",
|
||||
"--json",
|
||||
"--intelligence-report",
|
||||
str(intelligence),
|
||||
"--mesh-report",
|
||||
str(mesh),
|
||||
"--mascarade-report",
|
||||
str(mascarade),
|
||||
],
|
||||
cwd=str(REPO_ROOT),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
|
||||
payload = json.loads(proc.stdout)
|
||||
self.assertEqual(payload["contract_version"], "runtime-mcp-ia-gateway/v1")
|
||||
self.assertEqual(payload["component"], "runtime-mcp-ia-gateway")
|
||||
self.assertEqual(payload["owner_repo"], "Kill_LIFE")
|
||||
self.assertEqual(payload["owner_agent"], "Runtime-Companion")
|
||||
self.assertEqual(payload["owner_subagent"], "MCP-Health")
|
||||
self.assertEqual(payload["status"], "degraded")
|
||||
self.assertTrue(payload["summary_short"])
|
||||
self.assertLessEqual(len(payload["summary_short"]), 320)
|
||||
self.assertIn("surfaces", payload)
|
||||
self.assertEqual(payload["surfaces"]["ia"]["status"], "degraded")
|
||||
self.assertEqual(payload["surfaces"]["ia"]["open_task_count"], 3)
|
||||
self.assertEqual(payload["surfaces"]["mcp"]["status"], "ready")
|
||||
self.assertEqual(payload["surfaces"]["mcp"]["host_order"], ["clems", "kxkm"])
|
||||
self.assertEqual(payload["surfaces"]["runtime"]["status"], "ready")
|
||||
self.assertIn("firmware_cad", payload["surfaces"])
|
||||
self.assertIn(payload["surfaces"]["firmware_cad"]["status"], {"ready", "degraded", "blocked"})
|
||||
self.assertIn("firmware_cad", payload["summary_short_artifacts"])
|
||||
firmware_cad_summary_path = REPO_ROOT / payload["summary_short_artifacts"]["firmware_cad"]["json"]
|
||||
self.assertTrue(firmware_cad_summary_path.exists())
|
||||
self.assertEqual(payload["sources"]["mascarade"]["provider"], "ollama")
|
||||
self.assertGreaterEqual(len(payload["evidence"]), 1)
|
||||
self.assertIn("Do intelligence thing", payload["next_steps"])
|
||||
|
||||
def test_refresh_degrades_fast_when_mesh_probe_times_out(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
root = Path(tmp_dir)
|
||||
intelligence_script = root / "intelligence.sh"
|
||||
mesh_script = root / "mesh.sh"
|
||||
mascarade_script = root / "mascarade.sh"
|
||||
|
||||
intelligence_script.write_text(
|
||||
"#!/usr/bin/env bash\n"
|
||||
"printf '%s\\n' '{\"contract_version\":\"summary-short/v1\",\"status\":\"ready\",\"summary_short\":\"ok\",\"evidence\":[\"a\"],\"open_task_count\":0,\"next_steps\":[]}'\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
mesh_script.write_text(
|
||||
"#!/usr/bin/env bash\n"
|
||||
"sleep 2\n"
|
||||
"printf '%s\\n' '{\"mesh_status\":\"ready\",\"load_profile\":\"tower-first\",\"host_order\":[\"clems\"]}'\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
mascarade_script.write_text(
|
||||
"#!/usr/bin/env bash\n"
|
||||
"printf '%s\\n' '{\"status\":\"ok\",\"provider\":\"ollama\",\"model\":\"qwen\"}'\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
env = os.environ.copy()
|
||||
env.update(
|
||||
{
|
||||
"RUNTIME_GATEWAY_INTELLIGENCE_SCRIPT": str(intelligence_script),
|
||||
"RUNTIME_GATEWAY_MESH_SCRIPT": str(mesh_script),
|
||||
"RUNTIME_GATEWAY_MASCARADE_SCRIPT": str(mascarade_script),
|
||||
"RUNTIME_GATEWAY_INTELLIGENCE_TIMEOUT_SEC": "5",
|
||||
"RUNTIME_GATEWAY_MESH_TIMEOUT_SEC": "1",
|
||||
"RUNTIME_GATEWAY_MASCARADE_TIMEOUT_SEC": "5",
|
||||
}
|
||||
)
|
||||
|
||||
proc = subprocess.run(
|
||||
["bash", str(SCRIPT), "--action", "status", "--json", "--refresh"],
|
||||
cwd=str(REPO_ROOT),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
env=env,
|
||||
)
|
||||
|
||||
payload = json.loads(proc.stdout)
|
||||
self.assertEqual(payload["surfaces"]["runtime"]["status"], "ready")
|
||||
self.assertEqual(payload["surfaces"]["ia"]["status"], "ready")
|
||||
self.assertEqual(payload["surfaces"]["mcp"]["status"], "degraded")
|
||||
self.assertTrue(payload["sources"]["mesh"]["refresh_timeout"])
|
||||
self.assertIn("mesh-refresh-timeout", payload["degraded_reasons"])
|
||||
self.assertTrue(
|
||||
any("timed out" in step.lower() for step in payload["next_steps"])
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,175 @@
|
||||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
KICAD_MANAGER_CONTROL = (
|
||||
REPO_ROOT
|
||||
/ ".runtime-home"
|
||||
/ "cad-ai-native-forks"
|
||||
/ "kicad-ki"
|
||||
/ "kicad"
|
||||
/ "tools"
|
||||
/ "kicad_manager_control.cpp"
|
||||
)
|
||||
PCB_TOOLBAR = (
|
||||
REPO_ROOT
|
||||
/ ".runtime-home"
|
||||
/ "cad-ai-native-forks"
|
||||
/ "kicad-ki"
|
||||
/ "pcbnew"
|
||||
/ "toolbars_pcb_editor.cpp"
|
||||
)
|
||||
SCH_TOOLBAR = (
|
||||
REPO_ROOT
|
||||
/ ".runtime-home"
|
||||
/ "cad-ai-native-forks"
|
||||
/ "kicad-ki"
|
||||
/ "eeschema"
|
||||
/ "toolbars_sch_editor.cpp"
|
||||
)
|
||||
BOARD_EDITOR_CONTROL = (
|
||||
REPO_ROOT
|
||||
/ ".runtime-home"
|
||||
/ "cad-ai-native-forks"
|
||||
/ "kicad-ki"
|
||||
/ "pcbnew"
|
||||
/ "tools"
|
||||
/ "board_editor_control.cpp"
|
||||
)
|
||||
SCH_EDITOR_CONTROL = (
|
||||
REPO_ROOT
|
||||
/ ".runtime-home"
|
||||
/ "cad-ai-native-forks"
|
||||
/ "kicad-ki"
|
||||
/ "eeschema"
|
||||
/ "tools"
|
||||
/ "sch_editor_control.cpp"
|
||||
)
|
||||
KICAD_PLUGIN = (
|
||||
REPO_ROOT
|
||||
/ ".runtime-home"
|
||||
/ "cad-ai-native-forks"
|
||||
/ "kicad-ki"
|
||||
/ "scripting"
|
||||
/ "plugins"
|
||||
/ "yiacad_kicad_plugin"
|
||||
/ "yiacad_action.py"
|
||||
)
|
||||
FREECAD_WORKBENCH = (
|
||||
REPO_ROOT
|
||||
/ ".runtime-home"
|
||||
/ "cad-ai-native-forks"
|
||||
/ "freecad-ki"
|
||||
/ "src"
|
||||
/ "Mod"
|
||||
/ "YiACADWorkbench"
|
||||
/ "yiacad_freecad_gui.py"
|
||||
)
|
||||
FREECAD_MAIN_WINDOW = (
|
||||
REPO_ROOT
|
||||
/ ".runtime-home"
|
||||
/ "cad-ai-native-forks"
|
||||
/ "freecad-ki"
|
||||
/ "src"
|
||||
/ "Gui"
|
||||
/ "MainWindow.cpp"
|
||||
)
|
||||
REQUIRED_NATIVE_SURFACES = (
|
||||
KICAD_MANAGER_CONTROL,
|
||||
PCB_TOOLBAR,
|
||||
SCH_TOOLBAR,
|
||||
BOARD_EDITOR_CONTROL,
|
||||
SCH_EDITOR_CONTROL,
|
||||
KICAD_PLUGIN,
|
||||
FREECAD_WORKBENCH,
|
||||
FREECAD_MAIN_WINDOW,
|
||||
)
|
||||
|
||||
|
||||
class YiacadNativeSurfaceContractTests(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls) -> None:
|
||||
missing = [str(path) for path in REQUIRED_NATIVE_SURFACES if not path.exists()]
|
||||
if missing:
|
||||
raise unittest.SkipTest(
|
||||
"Native CAD forks are not installed under .runtime-home: " + ", ".join(missing)
|
||||
)
|
||||
|
||||
def read_text(self, path: Path) -> str:
|
||||
self.assertTrue(path.exists(), path)
|
||||
return path.read_text(encoding="utf-8")
|
||||
|
||||
def assert_contains_all(self, path: Path, *needles: str) -> str:
|
||||
content = self.read_text(path)
|
||||
for needle in needles:
|
||||
self.assertIn(needle, content, f"Missing {needle!r} in {path}")
|
||||
return content
|
||||
|
||||
def test_kicad_manager_exposes_full_yiacad_native_action_set(self) -> None:
|
||||
self.assert_contains_all(
|
||||
KICAD_MANAGER_CONTROL,
|
||||
"ShowYiacadStatus",
|
||||
"ShowYiacadErcDrc",
|
||||
"ShowYiacadBomReview",
|
||||
"ShowYiacadEcadMcadSync",
|
||||
"yiacad_backend_client.py",
|
||||
)
|
||||
|
||||
def test_kicad_shell_toolbars_keep_yiacad_review_group(self) -> None:
|
||||
self.assert_contains_all(PCB_TOOLBAR, '_( "YiACAD Review" )')
|
||||
self.assert_contains_all(SCH_TOOLBAR, '_( "YiACAD Review" )')
|
||||
|
||||
def test_kicad_control_layers_share_the_service_first_bridge(self) -> None:
|
||||
for path in (BOARD_EDITOR_CONTROL, SCH_EDITOR_CONTROL):
|
||||
self.assert_contains_all(
|
||||
path,
|
||||
"yiacad_backend_client.py",
|
||||
"PYTHON_EXECUTABLE",
|
||||
'python3',
|
||||
"ShowYiacadStatus",
|
||||
"ShowYiacadErcDrc",
|
||||
"ShowYiacadBomReview",
|
||||
"ShowYiacadEcadMcadSync",
|
||||
)
|
||||
|
||||
def test_kicad_plugin_keeps_palette_review_center_and_persistence(self) -> None:
|
||||
content = self.assert_contains_all(
|
||||
KICAD_PLUGIN,
|
||||
"PALETTE_ACTIONS = [",
|
||||
"latest_review_session.json",
|
||||
"review_history.json",
|
||||
"YiACAD Command Palette",
|
||||
'label="Command Palette"',
|
||||
'label="Review Center"',
|
||||
)
|
||||
self.assertEqual(content.count("def load_review_history() -> dict:"), 1)
|
||||
|
||||
def test_freecad_workbench_keeps_palette_review_center_and_persistence(self) -> None:
|
||||
self.assert_contains_all(
|
||||
FREECAD_WORKBENCH,
|
||||
'DOCK_TITLE = "YiACAD Inspector"',
|
||||
"PALETTE_ACTIONS = [",
|
||||
"latest_review_session.json",
|
||||
"review_history.json",
|
||||
'QtGui.QLabel("Command Palette")',
|
||||
'QtGui.QLabel("Review Center")',
|
||||
'"YiACAD_Inspector"',
|
||||
)
|
||||
|
||||
def test_freecad_shell_anchor_remains_available_for_native_convergence(self) -> None:
|
||||
self.assert_contains_all(
|
||||
FREECAD_MAIN_WINDOW,
|
||||
'yiacadShellDockObjectName = "Std_YiACADShellView"',
|
||||
'QDockWidget::tr("YiACAD Shell")',
|
||||
'"<b>Surface:</b> freecad-shell"',
|
||||
'"<b>Status:</b> degraded"',
|
||||
'"<b>Artifacts:</b> output contract, review logs, workbench evidence"',
|
||||
'"<b>Next steps:</b> attach normalized result cards, route artifacts, then converge with the persistent inspector contract."',
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,60 @@
|
||||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import subprocess
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
SCRIPT = REPO_ROOT / "tools" / "cockpit" / "yiacad_uiux_tui.sh"
|
||||
PROOF_SCRIPT = REPO_ROOT / "tools" / "cockpit" / "yiacad_backend_proof.sh"
|
||||
|
||||
|
||||
class YiacadUiuxTuiContractTests(unittest.TestCase):
|
||||
def run_json(self, *args: str) -> dict:
|
||||
proc = subprocess.run(
|
||||
["bash", str(SCRIPT), *args, "--json"],
|
||||
cwd=str(REPO_ROOT),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
self.assertTrue(proc.stdout.strip(), proc.stderr)
|
||||
return json.loads(proc.stdout)
|
||||
|
||||
def test_status_json_is_parseable_and_stable(self) -> None:
|
||||
payload = self.run_json("--action", "status")
|
||||
self.assertEqual(payload["component"], "yiacad")
|
||||
self.assertEqual(payload["surface"], "tui")
|
||||
self.assertEqual(payload["action"], "status.surface")
|
||||
self.assertIn(payload["status"], {"done", "degraded", "blocked"})
|
||||
self.assertGreaterEqual(len(payload["artifacts"]), 1)
|
||||
self.assertGreaterEqual(len(payload["next_steps"]), 1)
|
||||
|
||||
def test_logs_summary_json_remains_parseable(self) -> None:
|
||||
payload = self.run_json("--action", "logs-summary")
|
||||
self.assertEqual(payload["status"], "done")
|
||||
self.assertIn("uiux_tui_files", payload)
|
||||
self.assertIn("backend_service_files", payload)
|
||||
|
||||
def test_backend_proof_confirms_service_first_transport(self) -> None:
|
||||
proc = subprocess.run(
|
||||
["bash", str(PROOF_SCRIPT), "--action", "run", "--json"],
|
||||
cwd=str(REPO_ROOT),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
self.assertTrue(proc.stdout.strip(), proc.stderr)
|
||||
payload = json.loads(proc.stdout)
|
||||
self.assertEqual(payload["component"], "yiacad-backend-proof")
|
||||
self.assertEqual(payload["status"], "done")
|
||||
self.assertEqual(payload["transport"], "local-facade")
|
||||
self.assertTrue(payload["contract_ok"])
|
||||
self.assertEqual(payload["kicad_transport_status"], "done")
|
||||
self.assertEqual(payload["freecad_transport_status"], "done")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,39 @@
|
||||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
WORKFLOW_FILE = REPO_ROOT / "tools" / "ai" / "integrations" / "n8n" / "kill_life_smoke_workflow.json"
|
||||
|
||||
|
||||
class ZeroclawN8nWorkflowContractTests(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls) -> None:
|
||||
cls.payload = json.loads(WORKFLOW_FILE.read_text(encoding="utf-8"))
|
||||
|
||||
def test_tracked_workflow_keeps_stable_identity(self) -> None:
|
||||
self.assertEqual(self.payload["id"], "kill-life-n8n-smoke")
|
||||
self.assertEqual(self.payload["name"], "Kill LIFE n8n smoke")
|
||||
|
||||
def test_smoke_workflow_uses_an_activatable_trigger(self) -> None:
|
||||
trigger_nodes = [node for node in self.payload["nodes"] if node["type"] == "n8n-nodes-base.scheduleTrigger"]
|
||||
self.assertEqual(len(trigger_nodes), 1)
|
||||
trigger = trigger_nodes[0]
|
||||
intervals = trigger["parameters"]["rule"]["interval"]
|
||||
self.assertEqual(intervals[0]["field"], "cronExpression")
|
||||
self.assertTrue(intervals[0]["expression"].strip())
|
||||
node_types = {node["type"] for node in self.payload["nodes"]}
|
||||
self.assertNotIn("n8n-nodes-base.manualTrigger", node_types)
|
||||
|
||||
def test_trigger_connects_to_the_noop_smoke_node(self) -> None:
|
||||
connections = self.payload["connections"]
|
||||
self.assertIn("Yearly Trigger", connections)
|
||||
first_link = connections["Yearly Trigger"]["main"][0][0]
|
||||
self.assertEqual(first_link["node"], "No Op")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user