77 lines
3.1 KiB
Python
77 lines
3.1 KiB
Python
"""Execute the thesis's Python session, substituting only its transport setup."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
import re
|
|
from collections.abc import Coroutine
|
|
from pathlib import Path
|
|
from typing import Any, cast
|
|
|
|
import pytest
|
|
|
|
from wf_client import App, Deployment, Run
|
|
from wf_client.protocols import WorkflowClientPort
|
|
from wf_config import load_workflow_config
|
|
from wf_server.config import build_workflow_server_from_workflow_config
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
THESIS = ROOT / "docs/thesis/system-design-implementation.md"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_thesis_python_session_preserves_report_and_repair(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
"""Catch broken displayed bindings, client calls, and missing saved results.
|
|
|
|
Execute trusted repository Markdown, not submitted workflow/user text.
|
|
No operations or persistence are mocked: only the displayed HTTP connection
|
|
is replaced by the real API in process. This does not test the HTTP adapter.
|
|
"""
|
|
config = load_workflow_config(ROOT / "examples/report_workflow/wf.config.json")
|
|
config.server.store.root = tmp_path / "store"
|
|
server = build_workflow_server_from_workflow_config(config)
|
|
app = App._from_port(cast(WorkflowClientPort, server.api))
|
|
monkeypatch.chdir(ROOT)
|
|
|
|
manuscript = THESIS.read_text(encoding="utf-8")
|
|
chapter = manuscript.split("# Case Study: Deterministic Report Workflow\n", 1)[1]
|
|
chapter = chapter.split("\n# Evaluation\n", 1)[0]
|
|
blocks = re.findall(r"^```python\n(.*?)^```", chapter, re.MULTILINE | re.DOTALL)
|
|
assert blocks, "The case study must contain an executable Python session"
|
|
session = "\n".join(blocks)
|
|
connection = 'app = App.from_http_jsonrpc("http://127.0.0.1:8771/rpc")'
|
|
assert session.count(connection) == 1, (
|
|
"Review the documented transport substitution"
|
|
)
|
|
session = session.replace(connection, "app = supplied_app")
|
|
namespace: dict[str, Any] = {"supplied_app": app}
|
|
# The session does not import future annotations; do not inherit this file's
|
|
# compiler flags and accidentally change Pydantic's model construction.
|
|
code = compile(
|
|
session,
|
|
str(THESIS),
|
|
"exec",
|
|
flags=ast.PyCF_ALLOW_TOP_LEVEL_AWAIT,
|
|
dont_inherit=True,
|
|
)
|
|
await cast(Coroutine[Any, Any, None], eval(code, namespace))
|
|
|
|
run = namespace["run"]
|
|
assert isinstance(run, Run)
|
|
assert run.status == "completed"
|
|
deployment = namespace["deployment"]
|
|
assert isinstance(deployment, Deployment)
|
|
assert deployment.bindings["local.report"] == "local.report_runtime"
|
|
stored = await server.api.inspect_run(run_id=run.run_id)
|
|
output = stored["output"]
|
|
assert output is not None
|
|
assert output["report"]["title"] == "Weekly Project Update"
|
|
assert len(output["report"]["action_items"]) == 3
|
|
assert output["markdown"].startswith("# Weekly Project Update")
|
|
diagnostic = capsys.readouterr().out
|
|
assert "invalid_source_path" in diagnostic
|
|
assert "output[0].path" in diagnostic
|
|
assert namespace["trace"].frames
|