Add Core Hub operator CLI wrappers

tests/test_operator_cli.py

@@ -0,0 +1,130 @@
+import json
+
+from core_hub.operator_cli import build_readiness_summary, collect_bootstrap_status
+
+OPERATOR_HEADERS = {"Authorization": "Bearer operator-token"}
+
+
+def _seed_ops_hub(client):
+    hub = client.post(
+        "/api/v2/hubs",
+        headers=OPERATOR_HEADERS,
+        json={
+            "slug": "ops-hub",
+            "name": "Ops Hub",
+            "domain": "ops.coulomb.social",
+            "hubKind": "domain",
+            "hubFamily": "vsm",
+            "vsmFunction": "OPS",
+            "vsmSystem": "1",
+        },
+    ).json()
+    manifest = client.post(
+        "/api/v2/hub-capability-manifests",
+        headers=OPERATOR_HEADERS,
+        json={"hubId": hub["id"], "manifestVersion": "1.0"},
+    ).json()
+    manifest = client.post(
+        f"/api/v2/hub-capability-manifests/{manifest['id']}/activate",
+        headers=OPERATOR_HEADERS,
+    ).json()
+    consumer = client.post(
+        "/api/v2/api-consumers",
+        headers=OPERATOR_HEADERS,
+        json={"slug": "ops-hub", "name": "ops-hub", "hubCapabilityManifestId": manifest["id"]},
+    ).json()
+    key = client.post(
+        f"/api/v2/api-consumers/{consumer['id']}/api-keys",
+        headers=OPERATOR_HEADERS,
+        json={"scopes": "framework:read hub:ops-hub:read hub:ops-hub:write"},
+    ).json()["fullKey"]
+    runtime_headers = {"Authorization": f"Bearer {key}"}
+    widget = client.post(
+        "/api/v2/widgets",
+        headers=runtime_headers,
+        json={
+            "hubId": hub["id"],
+            "name": "Gitea Registry Readiness",
+            "widgetType": "ops-readiness-gate",
+            "capabilityRef": "ops:readiness:gitea-registry",
+            "viewContext": "ops-hub/readiness/gitea-registry",
+            "policyScope": "ops-registry",
+        },
+    ).json()
+    client.post(
+        "/api/v2/interaction-events",
+        headers=runtime_headers,
+        json={
+            "widgetId": widget["id"],
+            "eventType": "ops-endpoint-verified",
+            "viewContext": "railiance-apps/workplans/RAIL-AP-WP-0001",
+            "metadata": {"expectedStatus": 401},
+        },
+    )
+    return key
+
+
+def test_bootstrap_status_summarizes_ops_hub_without_secret_leak(client):
+    full_key = _seed_ops_hub(client)
+
+    report = collect_bootstrap_status(client, "operator-token")
+
+    assert report["ok"] is True
+    assert report["hubs"]["opsHubCandidates"][0]["slug"] == "ops-hub"
+    assert report["manifests"]["active"] == 1
+    assert report["apiConsumers"]["opsHubCandidates"][0]["keyPrefix"].startswith("ch_")
+    assert report["widgets"]["opsReadinessGates"] == 1
+    assert report["interactionEvents"]["opsEndpointVerified"] == 1
+
+    rendered = json.dumps(report, sort_keys=True)
+    assert full_key not in rendered
+    assert "operator-token" not in rendered
+
+
+def test_readiness_summary_reports_cutover_ready_from_non_secret_evidence(tmp_path):
+    smoke = tmp_path / "smoke.json"
+    smoke.write_text(
+        json.dumps(
+            {
+                "ok": True,
+                "runId": "smoke-1",
+                "checks": [{"ok": True}],
+                "bootstrap": {"event": {"id": "event-1"}},
+            }
+        ),
+        encoding="utf-8",
+    )
+    migration = tmp_path / "migration.json"
+    migration.write_text(
+        json.dumps(
+            {
+                "ok": True,
+                "dryRun": False,
+                "source": "inter-hub-haskell",
+                "bundleSha256": "abc",
+                "counts": {},
+            }
+        ),
+        encoding="utf-8",
+    )
+    activity = tmp_path / "activity.json"
+    activity.write_text(json.dumps([{"status": "posted", "verified": True}]), encoding="utf-8")
+
+    report = build_readiness_summary(
+        deployed_smoke_report=smoke,
+        migration_report=migration,
+        activity_core_report=activity,
+    )
+
+    assert report["readyForCutover"] is True
+    assert "legacy_inter_hub_reference" in report["openGates"]
+
+
+def test_readiness_summary_keeps_dry_run_import_open(tmp_path):
+    migration = tmp_path / "migration.json"
+    migration.write_text(json.dumps({"ok": True, "dryRun": True}), encoding="utf-8")
+
+    report = build_readiness_summary(migration_report=migration)
+
+    assert report["readyForCutover"] is False
+    assert "staging_import" in report["openGates"]