Spaces:
Sleeping
Sleeping
File size: 2,994 Bytes
e545bf5 6946637 e545bf5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 | """Deployment smoke test. Run against a live server:
python smoke_test.py http://127.0.0.1:7860 (or the Space URL)
Verifies: health, REST tool call, full MCP workflow, regression pin (>=26/27
links reconciled on Kincora), report download, agent 400-without-key.
Requires a local Kincora_Phase_2.inp or any valid .inp passed as argv[2].
"""
import asyncio, json, sys
import httpx
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
BASE = sys.argv[1].rstrip("/") if len(sys.argv) > 1 else "http://127.0.0.1:7860"
INP = sys.argv[2] if len(sys.argv) > 2 else "Kincora_Phase_2.inp"
async def main():
h = httpx.get(f"{BASE}/health", timeout=30).json()
assert h["status"] == "ok", h
print("health:", h["server"], h["version"], "| tools", h["tools"])
r = httpx.post(f"{BASE}/api/tool/list_sessions", json={}, timeout=30)
assert r.status_code == 200
print("REST surface: OK")
inp = open(INP, encoding="utf-8", errors="replace").read()
async with streamablehttp_client(f"{BASE}/mcp") as (read, write, _):
async with ClientSession(read, write) as s:
await s.initialize()
tools = await s.list_tools()
assert len(tools.tools) >= 16
up = json.loads((await s.call_tool("upload_model",
{"inp_content": inp, "filename": INP})).content[0].text)
sid = up["session_id"]
run = json.loads((await s.call_tool("run_simulation", {"session_id": sid})).content[0].text)
recon = run["rpt_reconciliation"]
print(f"MCP workflow: session {sid} | recon {recon.get('ok')}/{recon.get('links_checked')}")
if "Kincora" in INP:
assert recon.get("ok", 0) >= 26, "REGRESSION PIN FAILED"
# 1x1 red PNG — validates the attach_figure -> embedded-report path
tiny_png = ("iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR4"
"2mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg==")
fig = json.loads((await s.call_tool("attach_figure",
{"session_id": sid, "image_base64": tiny_png,
"caption": "Smoke-test figure", "section": "results"})).content[0].text)
assert fig["figure_id"] == "FIG-01", fig
print("attach_figure: OK", fig["figure_id"])
rep = json.loads((await s.call_tool("generate_report",
{"session_id": sid, "project_name": "Smoke Test"})).content[0].text)
dl = httpx.get(f"{BASE}{rep['files']['docx']}", timeout=60)
assert dl.status_code == 200 and len(dl.content) > 30000
print("report + download: OK", len(dl.content), "bytes")
await s.call_tool("close_session", {"session_id": sid})
r = httpx.post(f"{BASE}/api/agent", json={"question": "x", "provider": "anthropic"}, timeout=30)
print("agent (no key expected 400 unless secret set):", r.status_code)
print("\nSMOKE TEST: PASS")
asyncio.run(main())
|