x

"""Integration tests for /api/outputs. Boots the real FastAPI app through the shared `client` fixture so every route exercises auth middleware + lifespan. Anthropic + the model registry are monkeypatched per-test for the LLM-driven endpoints. The agent_manager-spawning endpoint reuses the existing `stub_agent_loop` fixture so the real `launch_agent` runs (creating an in-memory session) without spawning the SDK. Layout mirrors `outputs.py`: - CRUD (/list, /create, /{id}, PUT, DELETE) + legacy migration - Workspace seed / read / write / delete - File serve (workspace + saved output, with token rewrite + _d payload injection) - Backend execute - auto-run (LLM-mocked) - auto-run-agent (stub_agent_loop + AgentConfig spy) - Auth control mirroring test_api_agents.test_protected_route_requires_auth """ from __future__ import annotations import base64 import json import os import sys import pytest # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _create_output(client, **overrides) -> dict: """POST /create with sensible defaults, return the persisted output dict.""" payload = { "name": "Test Output", "description": "test", "input_schema": { "type": "object", "properties": {"x": {"type": "integer"}}, "required": ["x"], }, "files": {"index.html": ""}, } payload.update(overrides) resp = client.post("/api/outputs/create", json=payload) assert resp.status_code == 200, resp.text return resp.json()["output"] # --------------------------------------------------------------------------- # CRUD + legacy migration # --------------------------------------------------------------------------- def test_list_empty_on_fresh_dir(client): resp = client.get("/api/outputs/list") assert resp.status_code == 200 assert resp.json() == {"outputs": []} def test_create_get_update_delete_round_trip(client): created = _create_output(client, name="Alpha") output_id = created["id"] assert created["name"] == "Alpha" listed = client.get("/api/outputs/list").json()["outputs"] assert any(o["id"] == output_id for o in listed) fetched = client.get(f"/api/outputs/{output_id}") assert fetched.status_code == 200 assert fetched.json()["name"] == "Alpha" upd = client.put( f"/api/outputs/{output_id}", json={ "name": "Beta", "auto_run_config": { "enabled": True, "prompt": "fetch X", "mode": "agent", "model": "sonnet", }, }, ) assert upd.status_code == 200 body = upd.json()["output"] assert body["name"] == "Beta" assert body["auto_run_config"]["enabled"] is True assert body["auto_run_config"]["prompt"] == "fetch X" deleted = client.delete(f"/api/outputs/{output_id}") assert deleted.status_code == 200 from backend.config.paths import OUTPUTS_DIR assert not os.path.exists(os.path.join(OUTPUTS_DIR, f"{output_id}.json")) gone = client.get(f"/api/outputs/{output_id}") assert gone.status_code == 404 def test_get_unknown_output_returns_404(client): resp = client.get("/api/outputs/does-not-exist") assert resp.status_code == 404 def test_create_migrates_legacy_frontend_backend_code(client): resp = client.post( "/api/outputs/create", json={ "name": "Legacy", "frontend_code": "old", "backend_code": "result = {}", }, ) assert resp.status_code == 200 output = resp.json()["output"] assert output["files"]["index.html"] == "old" assert output["files"]["backend.py"] == "result = {}" def test_update_unknown_output_returns_404(client): resp = client.put("/api/outputs/missing", json={"name": "x"}) assert resp.status_code == 404 def test_delete_unknown_output_returns_404(client): resp = client.delete("/api/outputs/missing") assert resp.status_code == 404 # --------------------------------------------------------------------------- # Workspace seed # --------------------------------------------------------------------------- def test_workspace_seed_with_explicit_files(client): from backend.config.paths import OUTPUTS_WORKSPACE_DIR workspace_id = "ws-explicit" resp = client.post( "/api/outputs/workspace/seed", json={ "workspace_id": workspace_id, "files": { "index.html": "seeded", "schema.json": '{"type":"object"}', }, "meta": {"name": "X", "description": "y"}, }, ) assert resp.status_code == 200 folder = os.path.join(OUTPUTS_WORKSPACE_DIR, workspace_id) assert os.path.isfile(os.path.join(folder, "index.html")) assert os.path.isfile(os.path.join(folder, "schema.json")) assert os.path.isfile(os.path.join(folder, "SKILL.md")) with open(os.path.join(folder, "meta.json")) as f: meta = json.load(f) assert meta["name"] == "X" def test_workspace_seed_empty_uses_default_template(client): from backend.apps.outputs.view_builder_templates import VIEW_TEMPLATE_FILES from backend.config.paths import OUTPUTS_WORKSPACE_DIR workspace_id = "ws-default" resp = client.post( "/api/outputs/workspace/seed", json={"workspace_id": workspace_id}, ) assert resp.status_code == 200 folder = os.path.join(OUTPUTS_WORKSPACE_DIR, workspace_id) for rel_path in VIEW_TEMPLATE_FILES: assert os.path.isfile(os.path.join(folder, rel_path)), rel_path assert os.path.isfile(os.path.join(folder, "SKILL.md")) def test_workspace_seed_drops_path_traversal_keys(client): """Keys that escape the workspace folder via `..` are silently skipped (continue branch in seed_workspace).""" from backend.config.paths import OUTPUTS_WORKSPACE_DIR workspace_id = "ws-traversal" resp = client.post( "/api/outputs/workspace/seed", json={ "workspace_id": workspace_id, "files": { "ok.txt": "kept", "../escape.html": "should-not-write", }, }, ) assert resp.status_code == 200 folder = os.path.join(OUTPUTS_WORKSPACE_DIR, workspace_id) assert os.path.isfile(os.path.join(folder, "ok.txt")) parent = os.path.dirname(os.path.normpath(folder)) assert not os.path.exists(os.path.join(parent, "escape.html")) # --------------------------------------------------------------------------- # Workspace read # --------------------------------------------------------------------------- def test_workspace_read_returns_files_and_meta(client): workspace_id = "ws-read" client.post( "/api/outputs/workspace/seed", json={ "workspace_id": workspace_id, "files": {"index.html": ""}, "meta": {"name": "Read me"}, }, ) resp = client.get(f"/api/outputs/workspace/{workspace_id}") assert resp.status_code == 200 body = resp.json() assert body["files"]["index.html"] == "" assert body["meta"] == {"name": "Read me"} assert body["path"].endswith(workspace_id) def test_workspace_read_returns_none_meta_for_bad_meta_json(client): """Garbage meta.json triggers the JSONDecodeError swallow branch in `read_workspace`, returning meta=None.""" from backend.config.paths import OUTPUTS_WORKSPACE_DIR workspace_id = "ws-bad-meta" folder = os.path.join(OUTPUTS_WORKSPACE_DIR, workspace_id) os.makedirs(folder, exist_ok=True) with open(os.path.join(folder, "meta.json"), "w") as f: f.write("{not valid json") resp = client.get(f"/api/outputs/workspace/{workspace_id}") assert resp.status_code == 200 assert resp.json()["meta"] is None def test_workspace_read_missing_returns_404(client): resp = client.get("/api/outputs/workspace/does-not-exist") assert resp.status_code == 404 # --------------------------------------------------------------------------- # Workspace file write / delete # --------------------------------------------------------------------------- def test_workspace_write_and_delete_file(client): from backend.config.paths import OUTPUTS_WORKSPACE_DIR workspace_id = "ws-write" client.post("/api/outputs/workspace/seed", json={"workspace_id": workspace_id}) write = client.put( f"/api/outputs/workspace/{workspace_id}/file/sub/dir/app.css", json={"content": "body { color: red; }"}, ) assert write.status_code == 200 assert write.json() == {"ok": True} full = os.path.join(OUTPUTS_WORKSPACE_DIR, workspace_id, "sub", "dir", "app.css") assert os.path.isfile(full) delete = client.delete(f"/api/outputs/workspace/{workspace_id}/file/sub/dir/app.css") assert delete.status_code == 200 assert not os.path.exists(full) # Empty parent dirs collapse up to the workspace root. assert not os.path.exists(os.path.join(OUTPUTS_WORKSPACE_DIR, workspace_id, "sub")) def test_workspace_write_traversal_rejected(client): workspace_id = "ws-write-trav" client.post("/api/outputs/workspace/seed", json={"workspace_id": workspace_id}) resp = client.put( f"/api/outputs/workspace/{workspace_id}/file/..%2Fescape.html", json={"content": "x"}, ) assert resp.status_code == 403 def test_workspace_write_missing_workspace_404(client): resp = client.put( "/api/outputs/workspace/missing/file/foo.txt", json={"content": "x"}, ) assert resp.status_code == 404 def test_workspace_delete_traversal_rejected(client): workspace_id = "ws-del-trav" client.post("/api/outputs/workspace/seed", json={"workspace_id": workspace_id}) resp = client.delete(f"/api/outputs/workspace/{workspace_id}/file/..%2Fescape.html") assert resp.status_code == 403 def test_workspace_delete_missing_workspace_404(client): resp = client.delete("/api/outputs/workspace/missing/file/foo.txt") assert resp.status_code == 404 def test_workspace_delete_missing_file_is_idempotent(client): """DELETE on an existing workspace but missing file still returns {"ok": True} (no-op branch).""" workspace_id = "ws-del-idem" client.post("/api/outputs/workspace/seed", json={"workspace_id": workspace_id}) resp = client.delete(f"/api/outputs/workspace/{workspace_id}/file/nope.txt") assert resp.status_code == 200 assert resp.json() == {"ok": True} # --------------------------------------------------------------------------- # Serve endpoints # --------------------------------------------------------------------------- def test_workspace_serve_non_html_is_raw(client): workspace_id = "ws-serve-css" client.post( "/api/outputs/workspace/seed", json={ "workspace_id": workspace_id, "files": {"app.css": "body { color: red; }"}, }, ) resp = client.get(f"/api/outputs/workspace/{workspace_id}/serve/app.css") assert resp.status_code == 200 assert resp.text == "body { color: red; }" assert resp.headers["content-type"].startswith("text/css") def test_workspace_serve_index_html_injects_default_globals(client, auth_token): workspace_id = "ws-serve-html" html = ( 'x' '' '' "" ) client.post( "/api/outputs/workspace/seed", json={"workspace_id": workspace_id, "files": {"index.html": html}}, ) resp = client.get(f"/api/outputs/workspace/{workspace_id}/serve/index.html") assert resp.status_code == 200 body = resp.text assert "window.OUTPUT_INPUT = {}" in body assert "window.OUTPUT_BACKEND_RESULT = null" in body # Relative got the token; absolute