Publish curated Unraid-AppOps-Gateway source from 4f99d8f6ea4fb8ae105b83570344b25fa05c31fd

backend/tests/test_mcp_rag_n8n_integrations.py

@@ -231,6 +231,59 @@ def test_n8n_uses_the_shared_grounded_knowledge_service(integration_client):
     )
 
 
+def test_forge_bootstrap_preflight_is_read_only_and_rejects_generic_apply(integration_client):
+    client, _engine = integration_client
+    from backend.app import mcp_server
+
+    profile = {
+        "api_version": "itworx.forge-bootstrap-preflight/v1",
+        "revision": "a" * 40,
+        "workspace_image_digest": "sha256:" + "b" * 64,
+        "frontdoor_image_digest": "sha256:" + "c" * 64,
+    }
+    response = client.post(
+        "/api/deployments/forge-bootstrap/preflight",
+        headers={**human_headers(client), "Idempotency-Key": "forge-api-preflight-0001"},
+        json=profile,
+    )
+    assert response.status_code == 200
+    result = response.json()
+    assert result["status"] == "preflight_failed"
+    assert result["preflight"]["status"] == "blocked"
+    assert "deployment_apply_policy" in result["preflight"]["blockers"]
+    assert "controller_trust" in result["preflight"]["blockers"]
+
+    refreshed = client.get(
+        f"/api/deployments/{result['id']}", headers=human_headers(client)
+    )
+    assert refreshed.status_code == 200
+    assert refreshed.json()["preflight"]["status"] == "blocked"
+    listed = client.get("/api/deployments", headers=human_headers(client))
+    assert any(item["id"] == result["id"] for item in listed.json()["items"])
+
+    generic_apply = client.post(
+        "/api/deployments",
+        headers={**human_headers(client), "Idempotency-Key": "forge-generic-apply-0001"},
+        json={"operation": "apply", "plan": profile},
+    )
+    assert generic_apply.status_code == 422
+
+    async def exercise_mcp():
+        async with Client(mcp_server.appops_mcp) as mcp_client:
+            preflight = await mcp_client.call_tool(
+                "deployment.preflight",
+                {"plan": profile, "idempotencyKey": "forge-mcp-preflight-0001"},
+            )
+            assert preflight.structured_content["preflight"]["status"] == "blocked"
+            apply = await mcp_client.call_tool(
+                "deployment.apply",
+                {"plan": profile, "idempotencyKey": "forge-mcp-apply-0001"},
+            )
+            assert apply.structured_content["status"] == "invalid"
+
+    asyncio.run(exercise_mcp())
+
+
 def test_local_rag_indexes_allowlisted_docs_and_redacts_secrets(tmp_path):
     from backend.app.services import knowledge_service