"""Single-page UI tests: stages render, fragments refresh, and the full click-through lifecycle (run → approve → train → evaluate → promote) works.""" from pathlib import Path import pytest from fastapi.testclient import TestClient from forja_core.api import deps from forja_core.main import create_app FIXTURES = Path(__file__).parent.parent / "fixtures" @pytest.fixture(autouse=True) def patch_settings(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: import shutil # Agents are copied to tmp because the promotion test mutates the registry. agents_dir = tmp_path / "agents" shutil.copytree(FIXTURES / "agents", agents_dir) monkeypatch.setenv("DATA_DIR", str(tmp_path / "data")) monkeypatch.setenv("AGENTS_DIR", str(agents_dir)) monkeypatch.setenv("POLICIES_DIR", str(FIXTURES / "policies")) monkeypatch.setenv("DATASETS_DIR", str(FIXTURES / "datasets")) monkeypatch.setenv("MODELS_DIR", str(FIXTURES / "models")) monkeypatch.setenv("TRAINING_BACKEND", "mock") monkeypatch.setenv("LLM_PROVIDER", "mock") for fn in ( deps.get_settings, deps.get_registry, deps.get_policy_store, deps.get_dataset_store, deps.get_model_store, deps.get_training_backend, deps.get_orchestrator, deps.get_llm_provider, deps.get_guardrail_engine, ): fn.cache_clear() def test_index_renders_all_six_stages() -> None: r = TestClient(create_app()).get("/") assert r.status_code == 200 for anchor in ( 'id="define"', 'id="run"', 'id="approve"', 'id="train"', 'id="promote"', 'id="audit"', ): assert anchor in r.text # Define stage: agent with version/state chips and the policy validators. assert "test_agent" in r.text assert "v1 · active" in r.text assert "v2 · draft" in r.text assert "detect_pii" in r.text # Run stage: the one-click demo button. assert "Run demo incident" in r.text def test_run_invoke_completes_and_lands_in_history_fragment() -> None: client = TestClient(create_app()) # "mos" maps to a low-risk canonical mock response → completes without HITL. r = client.post("/run", data={"agent_name": "test_agent", "input": "mos degradation in pool"}) assert r.status_code == 200 assert "Completed" in r.text assert "validate_input" in r.text assert "finalize" in r.text h = client.get("/fragments/history") assert h.status_code == 200 assert "test_agent" in h.text def test_run_invoke_hitl_appears_in_approvals_fragment() -> None: client = TestClient(create_app()) # "hss" maps to a risk-5 action requiring approval → execution pauses in HITL. r = client.post("/run", data={"agent_name": "test_agent", "input": "hss capacity alarm"}) assert r.status_code == 200 assert "Pending approval" in r.text a = client.get("/fragments/approvals") assert a.status_code == 200 assert "test_agent" in a.text assert "isolate_replication_link" in a.text assert "Approve all" in a.text def test_run_invoke_unknown_agent_returns_error() -> None: client = TestClient(create_app()) r = client.post("/run", data={"agent_name": "missing", "input": "x"}) assert r.status_code == 400 assert "Execution error" in r.text @pytest.mark.parametrize( ("path", "target"), [ ("/agents", "/#define"), ("/policies", "/#define"), ("/run", "/#run"), ("/approvals", "/#approve"), ("/training", "/#train"), ("/promotions", "/#promote"), ("/history", "/#audit"), ("/armory", "/#audit"), ("/agents/test_agent", "/#define"), ], ) def test_legacy_routes_redirect_to_anchors(path: str, target: str) -> None: client = TestClient(create_app()) r = client.get(path, follow_redirects=False) assert r.status_code == 301 assert r.headers["location"] == target def test_full_lifecycle_click_through() -> None: """The exact sequence of requests the page buttons fire, end to end.""" client = TestClient(create_app()) # Stage 02 click: run the demo incident → pauses at the approval gate. r = client.post("/run", data={"agent_name": "test_agent", "input": "hss capacity alarm"}) assert "Pending approval" in r.text # Stage 03 click: approve all pending actions → run completes. pending = client.get("/api/executions").json() trace_id = next(e["trace_id"] for e in pending if e["status"] == "awaiting_approval") approved = client.post( f"/api/executions/{trace_id}/approve", json={"approved_action_ids": [], "comment": "Approved from UI"}, ).json() assert approved["status"] == "completed" assert client.get("/fragments/approvals").text.count("Pending approval") == 0 assert "test_agent" in client.get("/fragments/history").text # Stage 04 click: submit a training run for the draft candidate (v2). run = client.post( "/api/training/runs", json={ "dataset_name": "incident_sft", "model_name": "gpt4o_lora_base", "agent_name": "test_agent", "agent_version": "v2", }, ).json() assert run["status"] == "succeeded" frag = client.get("/fragments/training").text assert "Run evaluation" in frag # Stage 04 click: evaluate the candidate. report = client.post(f"/api/training/runs/{run['id']}/evaluate").json() assert report["passed"] is True frag = client.get("/fragments/training").text assert "Request promotion" in frag # Stage 04 click: request promotion → shows as pending in stages 04 and 05. promo = client.post( "/api/promotions", json={ "agent_name": "test_agent", "agent_version": "v2", "training_run_id": run["id"], "evaluation_report_id": report["id"], "requested_by": "web-ui", }, ) assert promo.status_code == 201 assert "Promotion pending" in client.get("/fragments/training").text assert "test_agent" in client.get("/fragments/promotions").text # Stage 05 click: approve the promotion → candidate flips to active. approve = client.post( f"/api/promotions/{promo.json()['id']}/approve", json={"approved_by": "web-ui", "comment": "ok"}, ) assert approve.status_code == 200 assert "✓ Promoted" in client.get("/fragments/training").text assert "v2 · active" in client.get("/fragments/agents").text agent = client.get("/api/agents/test_agent").json() assert agent["version"] == "v2" assert agent["state"] == "active"