Spaces:
Running
Running
| """End-to-end local smoke check used before publishing the Space.""" | |
| from __future__ import annotations | |
| import asyncio | |
| import importlib.util | |
| import sys | |
| import tempfile | |
| from pathlib import Path | |
| ROOT = Path(__file__).resolve().parents[1] | |
| sys.path.insert(0, str(ROOT)) | |
| from examples.demo_tools import TOOLS # noqa: E402 | |
| from trace2mcp import ( # noqa: E402 | |
| ToolRegistry, | |
| compile_trace, | |
| execute_workflow, | |
| generate_project, | |
| load_trace, | |
| ) | |
| def main() -> None: | |
| workflow = compile_trace( | |
| load_trace(ROOT / "examples" / "sample_trace.json"), name="company_brief" | |
| ) | |
| result = asyncio.run(execute_workflow(workflow, ToolRegistry(TOOLS))) | |
| assert result.metrics.nodes_executed == 6 | |
| assert result.metrics.layer_count == 4 | |
| assert result.metrics.max_parallelism == 3 | |
| assert result.output["format"] == "markdown" | |
| with tempfile.TemporaryDirectory(prefix="trace2mcp-smoke-") as temp: | |
| generated = generate_project(workflow, temp) | |
| sys.path.insert(0, temp) | |
| spec = importlib.util.spec_from_file_location("trace2mcp_smoke_server", generated["server"]) | |
| assert spec is not None and spec.loader is not None | |
| module = importlib.util.module_from_spec(spec) | |
| spec.loader.exec_module(module) | |
| assert len(module.WORKFLOW.nodes) == 6 | |
| sys.path.remove(temp) | |
| import app | |
| status, dashboard, compiled_json = app.compile_from_text(app.SAMPLE_TEXT) | |
| assert "Compilation complete" in status | |
| assert "Trace2MCP package compiled" in status | |
| assert "dashboard" in dashboard | |
| assert '"nodes"' in compiled_json | |
| print("Smoke OK: runtime, MCP SDK v2 scaffold, and Gradio Space loaded.") | |
| if __name__ == "__main__": | |
| main() | |