lda.chat/lda-wf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: migrate rpc tests to pytest asyncio

Browse Source
This commit is contained in:
lda
2026-06-04 18:54:42 +07:00 Unverified
parent f764b1717f
commit f9673a4ad0
5 changed files with 865 additions and 949 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+2
View File
@@ -28,12 +28,14 @@ dev = [
"basedpyright>=1.39.6", "basedpyright>=1.39.6",
"pytest>=8", "pytest>=8",
# "pytest-sugar>=1", # "pytest-sugar>=1",
"pytest-asyncio>=1.4.0",
"ruff>=0.15.15", "ruff>=0.15.15",
] ]
[tool.pytest.ini_options] [tool.pytest.ini_options]
addopts = "-p no:cacheprovider" addopts = "-p no:cacheprovider"
pythonpath = ["."] pythonpath = ["."]
asyncio_mode = "auto"
[tool.uv] [tool.uv]
package = true package = true
tests/wf_transport_rpc_http/test_app.py
+328 -350
View File
@@ -1,6 +1,5 @@
from __future__ import annotations from __future__ import annotations
import asyncio
from typing import Any from typing import Any
import httpx import httpx
@@ -22,311 +21,293 @@ async def _rpc(
return response.json() return response.json()
def test_rpc_health_and_capability_methods(tmp_path) -> None: async def test_rpc_health_and_capability_methods(tmp_path) -> None:
async def scenario() -> None: server = build_local_static_workflow_server(tmp_path / "store")
server = build_local_static_workflow_server(tmp_path / "store") app = create_rpc_app(server)
app = create_rpc_app(server) transport = httpx.ASGITransport(app=app)
transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(
async with httpx.AsyncClient( transport=transport, base_url="http://test"
transport=transport, base_url="http://test" ) as client:
) as client: health_response = await client.get("/healthz")
health_response = await client.get("/healthz") health = await _rpc(client, "workflow.health", {})
health = await _rpc(client, "workflow.health", {}) listed = await _rpc(
listed = await _rpc( client,
client, "workflow.capabilities.list",
"workflow.capabilities.list", {"source_id": "wf.std", "limit": 10},
{"source_id": "wf.std", "limit": 10},
)
inspected = await _rpc(
client,
"workflow.capabilities.inspect",
{"qualified_name": "wf.std.constant"},
)
assert health_response.status_code == 200
assert health_response.json()["status"] == "ok"
assert health["result"]["status"] == "ok"
assert listed["result"]["capabilities"]
assert inspected["result"]["name"] == "wf.std.constant"
asyncio.run(scenario())
def test_rpc_unknown_method_returns_json_rpc_error(tmp_path) -> None:
async def scenario() -> None:
server = build_local_static_workflow_server(tmp_path / "store")
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://test"
) as client:
payload = await _rpc(client, "workflow.nope", {})
assert payload["error"]["code"] == -32601
assert payload["error"]["message"] == "Method not found"
asyncio.run(scenario())
def test_rpc_app_mounts_configured_rpc_path(tmp_path) -> None:
async def scenario() -> None:
server = build_local_static_workflow_server(tmp_path / "store")
app = create_rpc_app(server, rpc_path="/workflow-rpc")
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://test"
) as client:
response = await client.post(
"/workflow-rpc",
json={
"jsonrpc": "2.0",
"id": "test",
"method": "workflow.health",
"params": {},
},
)
assert response.status_code == 200
assert response.json()["result"]["status"] == "ok"
asyncio.run(scenario())
def test_rpc_draft_artifact_deployment_lifecycle(tmp_path) -> None:
async def scenario() -> None:
server = build_local_static_workflow_server(tmp_path / "store")
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://test"
) as client:
draft_ws = await _rpc(
client,
"workflow.drafts.create_from_capability",
{
"workspace_id": "constant_ws",
"capability_name": "wf.std.constant",
"name": "constant_workflow",
"title": "Constant Workflow",
"input_map": {},
"output_map": {"value": "state.result"},
},
)
draft = {
"name": "rpc_constant",
"input_schema": {"type": "object", "properties": {}},
"state_schema": {
"type": "object",
"properties": {
"result": {"type": "string", "reducer": "wf.std.replace"}
},
},
"output_schema": {
"type": "object",
"properties": {"result": {"type": "string"}},
"required": ["result"],
},
"start": "constant",
"steps": {
"constant": {
"use": "wf.std.constant",
"input": [
{
"value": "hello over rpc",
"target": {"root": "local", "parts": ["value"]},
}
],
"output": [
{
"source": {"root": "local", "parts": ["value"]},
"target": {"root": "state", "parts": ["result"]},
}
],
}
},
"routes": {"constant": {"ok": "__end__"}},
"output": [
{
"path": {"root": "state", "parts": ["result"]},
"target": {"root": "local", "parts": ["result"]},
}
],
}
validate_draft = await _rpc(
client,
"workflow.drafts.validate",
{"draft": draft},
)
compiled_plan = validate_draft["result"]["compiled_plan"]
artifact = await _rpc(
client,
"workflow.artifacts.save",
{
"artifact": {
"id": "constant_rpc",
"version": 1,
"kind": "wrapper",
"title": "Constant RPC",
"input_schema": {"type": "object", "properties": {}},
"output_schema": {
"type": "object",
"properties": {"result": {"type": "string"}},
"required": ["result"],
},
"outcomes": ["ok"],
"required_capabilities": {},
"source_bindings": {"wf.std": "wf.std"},
"plan": compiled_plan,
},
},
)
deployment = await _rpc(
client,
"workflow.deployments.save",
{
"deployment": {
"id": "constant_rpc.default",
"artifact_id": "constant_rpc",
"artifact_version": 1,
"bindings": [
{"logical_source": "wf.std", "concrete_source": "wf.std"}
],
},
},
)
validate_deployment = await _rpc(
client,
"workflow.deployments.validate",
{"deployment_id": "constant_rpc.default"},
)
assert draft_ws["result"]["workspace_id"] == "constant_ws"
assert validate_draft["result"]["status"] == "valid"
assert artifact["result"]["artifact_id"] == "constant_rpc"
assert deployment["result"]["deployment_id"] == "constant_rpc.default"
assert validate_deployment["result"]["status"] == "runnable"
asyncio.run(scenario())
def test_rpc_artifact_and_deployment_catalog_methods(tmp_path) -> None:
async def scenario() -> None:
server = build_local_static_workflow_server(tmp_path / "store")
await server.api.create_artifact_from_plan(
artifact_id="rpc_lifecycle",
version=1,
title="RPC Lifecycle",
plan=_constant_plan(),
outcomes=["ok"],
source_bindings={"wf.std": "wf.std"},
) )
await server.api.save_deployment( inspected = await _rpc(
client,
"workflow.capabilities.inspect",
{"qualified_name": "wf.std.constant"},
)
assert health_response.status_code == 200
assert health_response.json()["status"] == "ok"
assert health["result"]["status"] == "ok"
assert listed["result"]["capabilities"]
assert inspected["result"]["name"] == "wf.std.constant"
async def test_rpc_unknown_method_returns_json_rpc_error(tmp_path) -> None:
server = build_local_static_workflow_server(tmp_path / "store")
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://test"
) as client:
payload = await _rpc(client, "workflow.nope", {})
assert payload["error"]["code"] == -32601
assert payload["error"]["message"] == "Method not found"
async def test_rpc_app_mounts_configured_rpc_path(tmp_path) -> None:
server = build_local_static_workflow_server(tmp_path / "store")
app = create_rpc_app(server, rpc_path="/workflow-rpc")
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://test"
) as client:
response = await client.post(
"/workflow-rpc",
json={
"jsonrpc": "2.0",
"id": "test",
"method": "workflow.health",
"params": {},
},
)
assert response.status_code == 200
assert response.json()["result"]["status"] == "ok"
async def test_rpc_draft_artifact_deployment_lifecycle(tmp_path) -> None:
server = build_local_static_workflow_server(tmp_path / "store")
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://test"
) as client:
draft_ws = await _rpc(
client,
"workflow.drafts.create_from_capability",
{ {
"id": "rpc_lifecycle.default", "workspace_id": "constant_ws",
"artifact_id": "rpc_lifecycle", "capability_name": "wf.std.constant",
"artifact_version": 1, "name": "constant_workflow",
"bindings": [{"logical_source": "wf.std", "concrete_source": "wf.std"}], "title": "Constant Workflow",
} "input_map": {},
"output_map": {"value": "state.result"},
},
) )
app = create_rpc_app(server) draft = {
transport = httpx.ASGITransport(app=app) "name": "rpc_constant",
async with httpx.AsyncClient( "input_schema": {"type": "object", "properties": {}},
transport=transport, base_url="http://test" "state_schema": {
) as client: "type": "object",
listed_artifacts = await _rpc(client, "workflow.artifacts.list", {}) "properties": {
inspected_artifact = await _rpc( "result": {"type": "string", "reducer": "wf.std.replace"}
client,
"workflow.artifacts.inspect",
{"artifact_id": "rpc_lifecycle", "version": 1},
)
listed_deployments = await _rpc(client, "workflow.deployments.list", {})
inspected_deployment = await _rpc(
client,
"workflow.deployments.inspect",
{"deployment_id": "rpc_lifecycle.default"},
)
deleted = await _rpc(
client,
"workflow.deployments.delete",
{"deployment_id": "rpc_lifecycle.default"},
)
assert listed_artifacts["result"]["nodes"]
assert inspected_artifact["result"]["id"] == "rpc_lifecycle"
assert listed_deployments["result"]["deployments"]
assert inspected_deployment["result"]["id"] == "rpc_lifecycle.default"
assert deleted["result"]["deployment_id"] == "rpc_lifecycle.default"
asyncio.run(scenario())
def test_rpc_draft_workspace_methods(tmp_path) -> None:
async def scenario() -> None:
server = build_local_static_workflow_server(tmp_path / "store")
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://test"
) as client:
created = await _rpc(
client,
"workflow.draft_workspaces.create_from_capability",
{
"workspace_id": "remote_ws",
"capability_name": "wf.std.constant",
"name": "remote_constant",
"title": "Remote Constant",
"input_map": {},
"output_map": {"value": "state.result"},
}, },
) },
listed = await _rpc(client, "workflow.draft_workspaces.list", {}) "output_schema": {
fetched = await _rpc( "type": "object",
client, "properties": {"result": {"type": "string"}},
"workflow.draft_workspaces.get", "required": ["result"],
{"workspace_id": "remote_ws"}, },
) "start": "constant",
validated = await _rpc( "steps": {
client, "constant": {
"workflow.draft_workspaces.validate", "use": "wf.std.constant",
{"workspace_id": "remote_ws"}, "input": [
) {
patched = await _rpc( "value": "hello over rpc",
client, "target": {"root": "local", "parts": ["value"]},
"workflow.draft_workspaces.patch", }
],
"output": [
{
"source": {"root": "local", "parts": ["value"]},
"target": {"root": "state", "parts": ["result"]},
}
],
}
},
"routes": {"constant": {"ok": "__end__"}},
"output": [
{ {
"workspace_id": "remote_ws", "path": {"root": "state", "parts": ["result"]},
"revision": created["result"]["revision"], "target": {"root": "local", "parts": ["result"]},
"patch": [ }
{"op": "replace", "path": "/name", "value": "remote_renamed"} ],
}
validate_draft = await _rpc(
client,
"workflow.drafts.validate",
{"draft": draft},
)
compiled_plan = validate_draft["result"]["compiled_plan"]
artifact = await _rpc(
client,
"workflow.artifacts.save",
{
"artifact": {
"id": "constant_rpc",
"version": 1,
"kind": "wrapper",
"title": "Constant RPC",
"input_schema": {"type": "object", "properties": {}},
"output_schema": {
"type": "object",
"properties": {"result": {"type": "string"}},
"required": ["result"],
},
"outcomes": ["ok"],
"required_capabilities": {},
"source_bindings": {"wf.std": "wf.std"},
"plan": compiled_plan,
},
},
)
deployment = await _rpc(
client,
"workflow.deployments.save",
{
"deployment": {
"id": "constant_rpc.default",
"artifact_id": "constant_rpc",
"artifact_version": 1,
"bindings": [
{"logical_source": "wf.std", "concrete_source": "wf.std"}
], ],
}, },
) },
artifact = await _rpc( )
client, validate_deployment = await _rpc(
"workflow.draft_workspaces.create_artifact", client,
{ "workflow.deployments.validate",
"workspace_id": "remote_ws", {"deployment_id": "constant_rpc.default"},
"artifact_id": "remote_artifact", )
"version": 1,
"title": "Remote Artifact",
"outcomes": ["ok"],
"kind": "workflow",
"source_bindings": {"wf.std": "wf.std"},
},
)
assert created["result"]["workspace_id"] == "remote_ws" assert draft_ws["result"]["workspace_id"] == "constant_ws"
assert listed["result"]["workspaces"] assert validate_draft["result"]["status"] == "valid"
assert fetched["result"]["workspace_id"] == "remote_ws" assert artifact["result"]["artifact_id"] == "constant_rpc"
assert validated["result"]["status"] in {"valid", "invalid"} assert deployment["result"]["deployment_id"] == "constant_rpc.default"
assert patched["result"]["revision"] == created["result"]["revision"] + 1 assert validate_deployment["result"]["status"] == "runnable"
assert artifact["result"]["artifact_id"] == "remote_artifact"
asyncio.run(scenario())
async def test_rpc_artifact_and_deployment_catalog_methods(tmp_path) -> None:
server = build_local_static_workflow_server(tmp_path / "store")
await server.api.create_artifact_from_plan(
artifact_id="rpc_lifecycle",
version=1,
title="RPC Lifecycle",
plan=_constant_plan(),
outcomes=["ok"],
source_bindings={"wf.std": "wf.std"},
)
await server.api.save_deployment(
{
"id": "rpc_lifecycle.default",
"artifact_id": "rpc_lifecycle",
"artifact_version": 1,
"bindings": [{"logical_source": "wf.std", "concrete_source": "wf.std"}],
}
)
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://test"
) as client:
listed_artifacts = await _rpc(client, "workflow.artifacts.list", {})
inspected_artifact = await _rpc(
client,
"workflow.artifacts.inspect",
{"artifact_id": "rpc_lifecycle", "version": 1},
)
listed_deployments = await _rpc(client, "workflow.deployments.list", {})
inspected_deployment = await _rpc(
client,
"workflow.deployments.inspect",
{"deployment_id": "rpc_lifecycle.default"},
)
deleted = await _rpc(
client,
"workflow.deployments.delete",
{"deployment_id": "rpc_lifecycle.default"},
)
assert listed_artifacts["result"]["nodes"]
assert inspected_artifact["result"]["id"] == "rpc_lifecycle"
assert listed_deployments["result"]["deployments"]
assert inspected_deployment["result"]["id"] == "rpc_lifecycle.default"
assert deleted["result"]["deployment_id"] == "rpc_lifecycle.default"
async def test_rpc_draft_workspace_methods(tmp_path) -> None:
server = build_local_static_workflow_server(tmp_path / "store")
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://test"
) as client:
created = await _rpc(
client,
"workflow.draft_workspaces.create_from_capability",
{
"workspace_id": "remote_ws",
"capability_name": "wf.std.constant",
"name": "remote_constant",
"title": "Remote Constant",
"input_map": {},
"output_map": {"value": "state.result"},
},
)
listed = await _rpc(client, "workflow.draft_workspaces.list", {})
fetched = await _rpc(
client,
"workflow.draft_workspaces.get",
{"workspace_id": "remote_ws"},
)
validated = await _rpc(
client,
"workflow.draft_workspaces.validate",
{"workspace_id": "remote_ws"},
)
patched = await _rpc(
client,
"workflow.draft_workspaces.patch",
{
"workspace_id": "remote_ws",
"revision": created["result"]["revision"],
"patch": [
{"op": "replace", "path": "/name", "value": "remote_renamed"}
],
},
)
artifact = await _rpc(
client,
"workflow.draft_workspaces.create_artifact",
{
"workspace_id": "remote_ws",
"artifact_id": "remote_artifact",
"version": 1,
"title": "Remote Artifact",
"outcomes": ["ok"],
"kind": "workflow",
"source_bindings": {"wf.std": "wf.std"},
},
)
assert created["result"]["workspace_id"] == "remote_ws"
assert listed["result"]["workspaces"]
assert fetched["result"]["workspace_id"] == "remote_ws"
assert validated["result"]["status"] in {"valid", "invalid"}
assert patched["result"]["revision"] == created["result"]["revision"] + 1
assert artifact["result"]["artifact_id"] == "remote_artifact"
def _constant_plan() -> RawWorkflowPlan: def _constant_plan() -> RawWorkflowPlan:
@@ -377,60 +358,57 @@ def _constant_plan() -> RawWorkflowPlan:
) )
def test_rpc_runs_deployment_and_reads_bounded_trace(tmp_path) -> None: async def test_rpc_runs_deployment_and_reads_bounded_trace(tmp_path) -> None:
async def scenario() -> None: server = build_local_static_workflow_server(tmp_path / "store")
server = build_local_static_workflow_server(tmp_path / "store") await server.api.create_artifact_from_plan(
await server.api.create_artifact_from_plan( artifact_id="rpc_constant",
artifact_id="rpc_constant", version=1,
version=1, title="RPC Constant",
title="RPC Constant", plan=_constant_plan(),
plan=_constant_plan(), outcomes=["ok"],
outcomes=["ok"], source_bindings={"wf.std": "wf.std"},
source_bindings={"wf.std": "wf.std"}, )
) await server.api.save_deployment(
await server.api.save_deployment( {
"id": "rpc_constant.default",
"artifact_id": "rpc_constant",
"artifact_version": 1,
"bindings": [{"logical_source": "wf.std", "concrete_source": "wf.std"}],
}
)
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://test"
) as client:
run = await _rpc(
client,
"workflow.runs.start",
{ {
"id": "rpc_constant.default", "deployment_id": "rpc_constant.default",
"artifact_id": "rpc_constant", "workflow_input": {},
"artifact_version": 1, "trace_range": {"start": 0, "limit": 1},
"bindings": [{"logical_source": "wf.std", "concrete_source": "wf.std"}], },
} )
inspected = await _rpc(
client,
"workflow.runs.inspect",
{"run_id": run["result"]["run_id"]},
)
trace = await _rpc(
client,
"workflow.runs.trace",
{
"run_id": run["result"]["run_id"],
"trace_range": {"start": 0, "limit": 1},
},
) )
app = create_rpc_app(server) assert run["result"]["status"] == "completed"
transport = httpx.ASGITransport(app=app) assert run["result"]["output"]["result"] == "hello over rpc"
async with httpx.AsyncClient( assert "trace" not in inspected["result"]
transport=transport, base_url="http://test" assert inspected["result"]["trace_count"] >= 1
) as client: assert trace["result"]["trace_start"] == 0
run = await _rpc( assert trace["result"]["trace_limit"] == 1
client, assert len(trace["result"]["trace"]) == 1
"workflow.runs.start",
{
"deployment_id": "rpc_constant.default",
"workflow_input": {},
"trace_range": {"start": 0, "limit": 1},
},
)
inspected = await _rpc(
client,
"workflow.runs.inspect",
{"run_id": run["result"]["run_id"]},
)
trace = await _rpc(
client,
"workflow.runs.trace",
{
"run_id": run["result"]["run_id"],
"trace_range": {"start": 0, "limit": 1},
},
)
assert run["result"]["status"] == "completed"
assert run["result"]["output"]["result"] == "hello over rpc"
assert "trace" not in inspected["result"]
assert inspected["result"]["trace_count"] >= 1
assert trace["result"]["trace_start"] == 0
assert trace["result"]["trace_limit"] == 1
assert len(trace["result"]["trace"]) == 1
asyncio.run(scenario())
tests/wf_transport_rpc_http/test_client.py
+239 -265
View File
@@ -1,7 +1,5 @@
from __future__ import annotations from __future__ import annotations
import asyncio
import httpx import httpx
from wf_api.models import RawWorkflowPlan, TraceRange from wf_api.models import RawWorkflowPlan, TraceRange
@@ -58,282 +56,258 @@ def _constant_plan() -> RawWorkflowPlan:
) )
def test_rpc_workflow_client_lists_and_inspects_capabilities(tmp_path) -> None: async def test_rpc_workflow_client_lists_and_inspects_capabilities(tmp_path) -> None:
async def scenario() -> None: server = build_local_static_workflow_server(tmp_path / "store")
server = build_local_static_workflow_server(tmp_path / "store") app = create_rpc_app(server)
app = create_rpc_app(server) transport = httpx.ASGITransport(app=app)
transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(
async with httpx.AsyncClient( transport=transport,
transport=transport, base_url="http://test",
base_url="http://test", ) as http_client:
) as http_client: client = RpcWorkflowApiClient(
client = RpcWorkflowApiClient( url="http://test/rpc",
url="http://test/rpc", timeout_seconds=5,
timeout_seconds=5, http_client=http_client,
http_client=http_client,
)
listed = await client.list_capabilities(source_id="wf.std", limit=5)
inspected = await client.inspect_capability(
qualified_name="wf.std.constant"
)
assert listed["capabilities"]
assert inspected["name"] == "wf.std.constant"
asyncio.run(scenario())
def test_rpc_workflow_client_lists_and_inspects_sources(tmp_path) -> None:
async def scenario() -> None:
server = build_local_static_workflow_server(tmp_path / "store")
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport,
base_url="http://test",
) as http_client:
client = RpcWorkflowApiClient(
url="http://test/rpc",
timeout_seconds=5,
http_client=http_client,
)
listed = await client.list_sources(limit=10)
inspected = await client.inspect_source(source_id="wf.std")
source_ids = {source["id"] for source in listed["sources"]}
assert "wf.std" in source_ids
assert inspected["id"] == "wf.std"
asyncio.run(scenario())
def test_rpc_workflow_client_reads_admin_state(tmp_path) -> None:
async def scenario() -> None:
server = build_local_static_workflow_server(tmp_path / "store")
server.events.record_workflow_event(
"workflow_test_event",
capability_id="workflow.demo.v1",
payload={"ok": True},
) )
app = create_rpc_app(server) listed = await client.list_capabilities(source_id="wf.std", limit=5)
transport = httpx.ASGITransport(app=app) inspected = await client.inspect_capability(
async with httpx.AsyncClient( qualified_name="wf.std.constant"
transport=transport,
base_url="http://test",
) as http_client:
client = RpcWorkflowApiClient(
url="http://test/rpc",
timeout_seconds=5,
http_client=http_client,
)
connections = await client.list_connections()
statuses = await client.get_connection_statuses()
events = await client.list_events()
assert connections == {"connections": [], "total": 0}
assert statuses == {"statuses": [], "total": 0}
assert events["total"] == 1
assert events["events"][0]["kind"] == "workflow_test_event"
asyncio.run(scenario())
def test_rpc_workflow_client_runs_and_reads_trace(tmp_path) -> None:
async def scenario() -> None:
server = build_local_static_workflow_server(tmp_path / "store")
await server.api.create_artifact_from_plan(
artifact_id="client_constant",
version=1,
title="Client Constant",
plan=_constant_plan(),
outcomes=["ok"],
source_bindings={"wf.std": "wf.std"},
) )
await server.api.save_deployment(
{ assert listed["capabilities"]
"id": "client_constant.default", assert inspected["name"] == "wf.std.constant"
"artifact_id": "client_constant",
"artifact_version": 1,
"bindings": [{"logical_source": "wf.std", "concrete_source": "wf.std"}], async def test_rpc_workflow_client_lists_and_inspects_sources(tmp_path) -> None:
} server = build_local_static_workflow_server(tmp_path / "store")
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport,
base_url="http://test",
) as http_client:
client = RpcWorkflowApiClient(
url="http://test/rpc",
timeout_seconds=5,
http_client=http_client,
) )
app = create_rpc_app(server) listed = await client.list_sources(limit=10)
transport = httpx.ASGITransport(app=app) inspected = await client.inspect_source(source_id="wf.std")
async with httpx.AsyncClient(
transport=transport,
base_url="http://test",
) as http_client:
client = RpcWorkflowApiClient(
url="http://test/rpc",
timeout_seconds=5,
http_client=http_client,
)
run = await client.run_deployment(
deployment_id="client_constant.default",
workflow_input={},
trace_range=TraceRange(start=0, limit=1),
)
inspected = await client.inspect_run(run_id=run["run_id"])
trace = await client.read_run_trace(
run_id=run["run_id"],
trace_range=TraceRange(start=0, limit=1),
)
assert run["status"] == "completed" source_ids = {source["id"] for source in listed["sources"]}
assert run["output"]["result"] == "hello from rpc client" assert "wf.std" in source_ids
assert inspected["trace_count"] >= 1 assert inspected["id"] == "wf.std"
assert len(trace["trace"]) == 1
asyncio.run(scenario())
def test_rpc_workflow_client_raises_for_rpc_error(tmp_path) -> None: async def test_rpc_workflow_client_reads_admin_state(tmp_path) -> None:
async def scenario() -> None: server = build_local_static_workflow_server(tmp_path / "store")
server = build_local_static_workflow_server(tmp_path / "store") server.events.record_workflow_event(
app = create_rpc_app(server) "workflow_test_event",
transport = httpx.ASGITransport(app=app) capability_id="workflow.demo.v1",
async with httpx.AsyncClient( payload={"ok": True},
transport=transport, )
base_url="http://test", app = create_rpc_app(server)
) as http_client: transport = httpx.ASGITransport(app=app)
client = RpcWorkflowApiClient( async with httpx.AsyncClient(
url="http://test/rpc", transport=transport,
timeout_seconds=5, base_url="http://test",
http_client=http_client, ) as http_client:
) client = RpcWorkflowApiClient(
try: url="http://test/rpc",
await client.inspect_capability(qualified_name="missing.capability") timeout_seconds=5,
except RuntimeError as exc: http_client=http_client,
message = str(exc)
else:
raise AssertionError("expected RuntimeError")
assert "Workflow operation failed" in message
assert "missing.capability" in message
asyncio.run(scenario())
def test_rpc_workflow_client_lists_and_inspects_artifacts(tmp_path) -> None:
async def scenario() -> None:
server = build_local_static_workflow_server(tmp_path / "store")
await server.api.create_artifact_from_plan(
artifact_id="client_art",
version=1,
title="Client Art",
plan=_constant_plan(),
outcomes=["ok"],
source_bindings={"wf.std": "wf.std"},
) )
app = create_rpc_app(server) connections = await client.list_connections()
transport = httpx.ASGITransport(app=app) statuses = await client.get_connection_statuses()
async with httpx.AsyncClient( events = await client.list_events()
transport=transport, base_url="http://test"
) as http_client:
client = RpcWorkflowApiClient(
url="http://test/rpc", timeout_seconds=5, http_client=http_client
)
listed = await client.list_artifacts()
inspected = await client.inspect_artifact(
artifact_id="client_art", version=1
)
assert listed["nodes"] assert connections == {"connections": [], "total": 0}
assert inspected["id"] == "client_art" assert statuses == {"statuses": [], "total": 0}
assert events["total"] == 1
asyncio.run(scenario()) assert events["events"][0]["kind"] == "workflow_test_event"
def test_rpc_workflow_client_lists_inspects_validates_and_deletes_deployments( async def test_rpc_workflow_client_runs_and_reads_trace(tmp_path) -> None:
server = build_local_static_workflow_server(tmp_path / "store")
await server.api.create_artifact_from_plan(
artifact_id="client_constant",
version=1,
title="Client Constant",
plan=_constant_plan(),
outcomes=["ok"],
source_bindings={"wf.std": "wf.std"},
)
await server.api.save_deployment(
{
"id": "client_constant.default",
"artifact_id": "client_constant",
"artifact_version": 1,
"bindings": [{"logical_source": "wf.std", "concrete_source": "wf.std"}],
}
)
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport,
base_url="http://test",
) as http_client:
client = RpcWorkflowApiClient(
url="http://test/rpc",
timeout_seconds=5,
http_client=http_client,
)
run = await client.run_deployment(
deployment_id="client_constant.default",
workflow_input={},
trace_range=TraceRange(start=0, limit=1),
)
inspected = await client.inspect_run(run_id=run["run_id"])
trace = await client.read_run_trace(
run_id=run["run_id"],
trace_range=TraceRange(start=0, limit=1),
)
assert run["status"] == "completed"
assert run["output"]["result"] == "hello from rpc client"
assert inspected["trace_count"] >= 1
assert len(trace["trace"]) == 1
async def test_rpc_workflow_client_raises_for_rpc_error(tmp_path) -> None:
server = build_local_static_workflow_server(tmp_path / "store")
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport,
base_url="http://test",
) as http_client:
client = RpcWorkflowApiClient(
url="http://test/rpc",
timeout_seconds=5,
http_client=http_client,
)
try:
await client.inspect_capability(qualified_name="missing.capability")
except RuntimeError as exc:
message = str(exc)
else:
raise AssertionError("expected RuntimeError")
assert "Workflow operation failed" in message
assert "missing.capability" in message
async def test_rpc_workflow_client_lists_and_inspects_artifacts(tmp_path) -> None:
server = build_local_static_workflow_server(tmp_path / "store")
await server.api.create_artifact_from_plan(
artifact_id="client_art",
version=1,
title="Client Art",
plan=_constant_plan(),
outcomes=["ok"],
source_bindings={"wf.std": "wf.std"},
)
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://test"
) as http_client:
client = RpcWorkflowApiClient(
url="http://test/rpc", timeout_seconds=5, http_client=http_client
)
listed = await client.list_artifacts()
inspected = await client.inspect_artifact(
artifact_id="client_art", version=1
)
assert listed["nodes"]
assert inspected["id"] == "client_art"
async def test_rpc_workflow_client_lists_inspects_validates_and_deletes_deployments(
tmp_path, tmp_path,
) -> None: ) -> None:
async def scenario() -> None: server = build_local_static_workflow_server(tmp_path / "store")
server = build_local_static_workflow_server(tmp_path / "store") await server.api.create_artifact_from_plan(
await server.api.create_artifact_from_plan( artifact_id="client_deploy_art",
artifact_id="client_deploy_art", version=1,
title="Client Deploy Art",
plan=_constant_plan(),
outcomes=["ok"],
source_bindings={"wf.std": "wf.std"},
)
await server.api.save_deployment(
{
"id": "client_deploy_art.default",
"artifact_id": "client_deploy_art",
"artifact_version": 1,
"bindings": [{"logical_source": "wf.std", "concrete_source": "wf.std"}],
}
)
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://test"
) as http_client:
client = RpcWorkflowApiClient(
url="http://test/rpc", timeout_seconds=5, http_client=http_client
)
listed = await client.list_deployments()
inspected = await client.inspect_deployment(
deployment_id="client_deploy_art.default"
)
validated = await client.validate_deployment(
deployment_id="client_deploy_art.default"
)
deleted = await client.delete_deployment(
deployment_id="client_deploy_art.default"
)
assert listed["deployments"]
assert inspected["id"] == "client_deploy_art.default"
assert validated["status"] == "runnable"
assert deleted["deployment_id"] == "client_deploy_art.default"
async def test_rpc_workflow_client_draft_workspace_lifecycle(tmp_path) -> None:
server = build_local_static_workflow_server(tmp_path / "store")
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://test"
) as http_client:
client = RpcWorkflowApiClient(
url="http://test/rpc", timeout_seconds=5, http_client=http_client
)
created = await client.create_draft_workspace_from_capability(
workspace_id="client_ws",
capability_name="wf.std.constant",
name="client_constant",
title="Client Constant",
input_map={},
output_map={"value": "state.result"},
)
listed = await client.list_draft_workspaces()
fetched = await client.get_draft_workspace(workspace_id="client_ws")
validated = await client.validate_draft_workspace(workspace_id="client_ws")
patched = await client.patch_draft_workspace(
workspace_id="client_ws",
revision=created["revision"],
patch=[{"op": "replace", "path": "/name", "value": "client_renamed"}],
)
artifact = await client.create_artifact_from_workspace(
workspace_id="client_ws",
artifact_id="client_ws_art",
version=1, version=1,
title="Client Deploy Art", title="Client WS Art",
plan=_constant_plan(), outcomes=("ok",),
outcomes=["ok"], kind="workflow",
source_bindings={"wf.std": "wf.std"}, source_bindings={"wf.std": "wf.std"},
) )
await server.api.save_deployment(
{
"id": "client_deploy_art.default",
"artifact_id": "client_deploy_art",
"artifact_version": 1,
"bindings": [{"logical_source": "wf.std", "concrete_source": "wf.std"}],
}
)
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://test"
) as http_client:
client = RpcWorkflowApiClient(
url="http://test/rpc", timeout_seconds=5, http_client=http_client
)
listed = await client.list_deployments()
inspected = await client.inspect_deployment(
deployment_id="client_deploy_art.default"
)
validated = await client.validate_deployment(
deployment_id="client_deploy_art.default"
)
deleted = await client.delete_deployment(
deployment_id="client_deploy_art.default"
)
assert listed["deployments"] assert created["workspace_id"] == "client_ws"
assert inspected["id"] == "client_deploy_art.default" assert listed["workspaces"]
assert validated["status"] == "runnable" assert fetched["workspace_id"] == "client_ws"
assert deleted["deployment_id"] == "client_deploy_art.default" assert validated["status"] in {"valid", "invalid"}
assert patched["revision"] == created["revision"] + 1
asyncio.run(scenario()) assert artifact["artifact_id"] == "client_ws_art"
def test_rpc_workflow_client_draft_workspace_lifecycle(tmp_path) -> None:
async def scenario() -> None:
server = build_local_static_workflow_server(tmp_path / "store")
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://test"
) as http_client:
client = RpcWorkflowApiClient(
url="http://test/rpc", timeout_seconds=5, http_client=http_client
)
created = await client.create_draft_workspace_from_capability(
workspace_id="client_ws",
capability_name="wf.std.constant",
name="client_constant",
title="Client Constant",
input_map={},
output_map={"value": "state.result"},
)
listed = await client.list_draft_workspaces()
fetched = await client.get_draft_workspace(workspace_id="client_ws")
validated = await client.validate_draft_workspace(workspace_id="client_ws")
patched = await client.patch_draft_workspace(
workspace_id="client_ws",
revision=created["revision"],
patch=[{"op": "replace", "path": "/name", "value": "client_renamed"}],
)
artifact = await client.create_artifact_from_workspace(
workspace_id="client_ws",
artifact_id="client_ws_art",
version=1,
title="Client WS Art",
outcomes=("ok",),
kind="workflow",
source_bindings={"wf.std": "wf.std"},
)
assert created["workspace_id"] == "client_ws"
assert listed["workspaces"]
assert fetched["workspace_id"] == "client_ws"
assert validated["status"] in {"valid", "invalid"}
assert patched["revision"] == created["revision"] + 1
assert artifact["artifact_id"] == "client_ws_art"
asyncio.run(scenario())
tests/wf_transport_rpc_http/test_source_registry_rpc.py
+282 -334
View File
@@ -1,6 +1,5 @@
from __future__ import annotations from __future__ import annotations
import asyncio
from dataclasses import dataclass, replace from dataclasses import dataclass, replace
from typing import Any from typing import Any
@@ -100,390 +99,339 @@ def _server_with_mutation_provider(tmp_path: Any) -> Any:
# --- read-only tests (unchanged) --- # --- read-only tests (unchanged) ---
def test_rpc_source_registry_list_unavailable_on_local_static(tmp_path) -> None: async def test_rpc_source_registry_list_unavailable_on_local_static(tmp_path) -> None:
async def scenario() -> None: server = build_local_static_workflow_server(tmp_path / "store")
server = build_local_static_workflow_server(tmp_path / "store") app = create_rpc_app(server)
app = create_rpc_app(server) transport = httpx.ASGITransport(app=app)
transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(
async with httpx.AsyncClient( transport=transport, base_url="http://test"
transport=transport, base_url="http://test" ) as client:
) as client: payload = await _rpc(
payload = await _rpc( client, "workflow.admin.source_registry.list", {"limit": 10}
client, "workflow.admin.source_registry.list", {"limit": 10}
)
assert "error" in payload
assert payload["error"]["data"]["code"] == "source_registry_unavailable"
asyncio.run(scenario())
def test_rpc_source_registry_inspect_unavailable_on_local_static(tmp_path) -> None:
async def scenario() -> None:
server = build_local_static_workflow_server(tmp_path / "store")
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://test"
) as client:
payload = await _rpc(
client,
"workflow.admin.source_registry.inspect",
{"source_id": "github.work"},
)
assert "error" in payload
assert payload["error"]["data"]["code"] == "source_registry_unavailable"
asyncio.run(scenario())
def test_rpc_source_registry_methods_return_registry_payloads(tmp_path) -> None:
async def scenario() -> None:
server = replace(
build_local_static_workflow_server(tmp_path / "store"),
source_registry_admin=WorkflowSourceRegistryApi(
provider=FakeRegistryProvider()
),
) )
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://test"
) as client:
listed = await _rpc(
client, "workflow.admin.source_registry.list", {"limit": 10}
)
inspected = await _rpc(
client,
"workflow.admin.source_registry.inspect",
{"source_id": "github.work"},
)
assert listed["result"]["entries"][0]["id"] == "github.work" assert "error" in payload
assert listed["result"]["entries"][0]["shadowed_by_config"] is True assert payload["error"]["data"]["code"] == "source_registry_unavailable"
assert inspected["result"]["entry"]["transport"]["kind"] == "stdio"
assert inspected["result"]["shadowed_by_config"] is True
asyncio.run(scenario())
async def test_rpc_source_registry_inspect_unavailable_on_local_static(tmp_path) -> None:
server = build_local_static_workflow_server(tmp_path / "store")
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://test"
) as client:
payload = await _rpc(
client,
"workflow.admin.source_registry.inspect",
{"source_id": "github.work"},
)
assert "error" in payload
assert payload["error"]["data"]["code"] == "source_registry_unavailable"
async def test_rpc_source_registry_methods_return_registry_payloads(tmp_path) -> None:
server = replace(
build_local_static_workflow_server(tmp_path / "store"),
source_registry_admin=WorkflowSourceRegistryApi(
provider=FakeRegistryProvider()
),
)
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://test"
) as client:
listed = await _rpc(
client, "workflow.admin.source_registry.list", {"limit": 10}
)
inspected = await _rpc(
client,
"workflow.admin.source_registry.inspect",
{"source_id": "github.work"},
)
assert listed["result"]["entries"][0]["id"] == "github.work"
assert listed["result"]["entries"][0]["shadowed_by_config"] is True
assert inspected["result"]["entry"]["transport"]["kind"] == "stdio"
assert inspected["result"]["shadowed_by_config"] is True
# --- mutation unavailable tests --- # --- mutation unavailable tests ---
def test_rpc_source_registry_add_unavailable_on_local_static(tmp_path) -> None: async def test_rpc_source_registry_add_unavailable_on_local_static(tmp_path) -> None:
async def scenario() -> None: server = build_local_static_workflow_server(tmp_path / "store")
server = build_local_static_workflow_server(tmp_path / "store") app = create_rpc_app(server)
app = create_rpc_app(server) transport = httpx.ASGITransport(app=app)
transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(
async with httpx.AsyncClient( transport=transport, base_url="http://test"
transport=transport, base_url="http://test" ) as client:
) as client: payload = await _rpc(
payload = await _rpc( client,
client, "workflow.admin.source_registry.add",
"workflow.admin.source_registry.add", {"entry": {"id": "new.source", "kind": "mcp"}},
{"entry": {"id": "new.source", "kind": "mcp"}}, )
)
assert "error" in payload assert "error" in payload
assert payload["error"]["data"]["code"] == "source_registry_unavailable" assert payload["error"]["data"]["code"] == "source_registry_unavailable"
asyncio.run(scenario())
def test_rpc_source_registry_update_unavailable_on_local_static(tmp_path) -> None: async def test_rpc_source_registry_update_unavailable_on_local_static(tmp_path) -> None:
async def scenario() -> None: server = build_local_static_workflow_server(tmp_path / "store")
server = build_local_static_workflow_server(tmp_path / "store") app = create_rpc_app(server)
app = create_rpc_app(server) transport = httpx.ASGITransport(app=app)
transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(
async with httpx.AsyncClient( transport=transport, base_url="http://test"
transport=transport, base_url="http://test" ) as client:
) as client: payload = await _rpc(
payload = await _rpc( client,
client, "workflow.admin.source_registry.update",
"workflow.admin.source_registry.update", {"source_id": "github.work", "patch": {"enabled": False}},
{"source_id": "github.work", "patch": {"enabled": False}}, )
)
assert "error" in payload assert "error" in payload
assert payload["error"]["data"]["code"] == "source_registry_unavailable" assert payload["error"]["data"]["code"] == "source_registry_unavailable"
asyncio.run(scenario())
def test_rpc_source_registry_enable_unavailable_on_local_static(tmp_path) -> None: async def test_rpc_source_registry_enable_unavailable_on_local_static(tmp_path) -> None:
async def scenario() -> None: server = build_local_static_workflow_server(tmp_path / "store")
server = build_local_static_workflow_server(tmp_path / "store") app = create_rpc_app(server)
app = create_rpc_app(server) transport = httpx.ASGITransport(app=app)
transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(
async with httpx.AsyncClient( transport=transport, base_url="http://test"
transport=transport, base_url="http://test" ) as client:
) as client: payload = await _rpc(
payload = await _rpc( client,
client, "workflow.admin.source_registry.enable",
"workflow.admin.source_registry.enable", {"source_id": "github.work"},
{"source_id": "github.work"}, )
)
assert "error" in payload assert "error" in payload
assert payload["error"]["data"]["code"] == "source_registry_unavailable" assert payload["error"]["data"]["code"] == "source_registry_unavailable"
asyncio.run(scenario())
def test_rpc_source_registry_disable_unavailable_on_local_static(tmp_path) -> None: async def test_rpc_source_registry_disable_unavailable_on_local_static(tmp_path) -> None:
async def scenario() -> None: server = build_local_static_workflow_server(tmp_path / "store")
server = build_local_static_workflow_server(tmp_path / "store") app = create_rpc_app(server)
app = create_rpc_app(server) transport = httpx.ASGITransport(app=app)
transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(
async with httpx.AsyncClient( transport=transport, base_url="http://test"
transport=transport, base_url="http://test" ) as client:
) as client: payload = await _rpc(
payload = await _rpc( client,
client, "workflow.admin.source_registry.disable",
"workflow.admin.source_registry.disable", {"source_id": "github.work"},
{"source_id": "github.work"}, )
)
assert "error" in payload assert "error" in payload
assert payload["error"]["data"]["code"] == "source_registry_unavailable" assert payload["error"]["data"]["code"] == "source_registry_unavailable"
asyncio.run(scenario())
def test_rpc_source_registry_remove_unavailable_on_local_static(tmp_path) -> None: async def test_rpc_source_registry_remove_unavailable_on_local_static(tmp_path) -> None:
async def scenario() -> None: server = build_local_static_workflow_server(tmp_path / "store")
server = build_local_static_workflow_server(tmp_path / "store") app = create_rpc_app(server)
app = create_rpc_app(server) transport = httpx.ASGITransport(app=app)
transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(
async with httpx.AsyncClient( transport=transport, base_url="http://test"
transport=transport, base_url="http://test" ) as client:
) as client: payload = await _rpc(
payload = await _rpc( client,
client, "workflow.admin.source_registry.remove",
"workflow.admin.source_registry.remove", {"source_id": "github.work"},
{"source_id": "github.work"}, )
)
assert "error" in payload assert "error" in payload
assert payload["error"]["data"]["code"] == "source_registry_unavailable" assert payload["error"]["data"]["code"] == "source_registry_unavailable"
asyncio.run(scenario())
# --- mutation success tests --- # --- mutation success tests ---
def test_rpc_source_registry_add_returns_entry(tmp_path) -> None: async def test_rpc_source_registry_add_returns_entry(tmp_path) -> None:
async def scenario() -> None: server = _server_with_mutation_provider(tmp_path)
server = _server_with_mutation_provider(tmp_path) app = create_rpc_app(server)
app = create_rpc_app(server) transport = httpx.ASGITransport(app=app)
transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(
async with httpx.AsyncClient( transport=transport, base_url="http://test"
transport=transport, base_url="http://test" ) as client:
) as client: payload = await _rpc(
payload = await _rpc( client,
client, "workflow.admin.source_registry.add",
"workflow.admin.source_registry.add", {"entry": {"id": "new.mcp", "kind": "mcp", "enabled": True}},
{"entry": {"id": "new.mcp", "kind": "mcp", "enabled": True}},
)
assert "result" in payload
assert payload["result"]["entry"]["id"] == "new.mcp"
assert payload["result"]["entry"]["kind"] == "mcp"
asyncio.run(scenario())
def test_rpc_source_registry_update_returns_entry(tmp_path) -> None:
async def scenario() -> None:
server = _server_with_mutation_provider(tmp_path)
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://test"
) as client:
payload = await _rpc(
client,
"workflow.admin.source_registry.update",
{"source_id": "github.work", "patch": {"enabled": False}},
)
assert "result" in payload
assert payload["result"]["entry"]["id"] == "github.work"
assert payload["result"]["entry"]["enabled"] is False
asyncio.run(scenario())
def test_rpc_source_registry_enable_returns_entry(tmp_path) -> None:
async def scenario() -> None:
mutation = FakeMutationProvider()
mutation.entries["github.work"]["enabled"] = False
server = replace(
build_local_static_workflow_server(tmp_path / "store"),
source_registry_admin=WorkflowSourceRegistryApi(
provider=FakeRegistryProvider(),
mutation_provider=mutation,
),
) )
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://test"
) as client:
payload = await _rpc(
client,
"workflow.admin.source_registry.enable",
{"source_id": "github.work"},
)
assert "result" in payload assert "result" in payload
assert payload["result"]["entry"]["enabled"] is True assert payload["result"]["entry"]["id"] == "new.mcp"
assert payload["result"]["entry"]["kind"] == "mcp"
asyncio.run(scenario())
def test_rpc_source_registry_disable_returns_entry(tmp_path) -> None: async def test_rpc_source_registry_update_returns_entry(tmp_path) -> None:
async def scenario() -> None: server = _server_with_mutation_provider(tmp_path)
server = _server_with_mutation_provider(tmp_path) app = create_rpc_app(server)
app = create_rpc_app(server) transport = httpx.ASGITransport(app=app)
transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(
async with httpx.AsyncClient( transport=transport, base_url="http://test"
transport=transport, base_url="http://test" ) as client:
) as client: payload = await _rpc(
payload = await _rpc( client,
client, "workflow.admin.source_registry.update",
"workflow.admin.source_registry.disable", {"source_id": "github.work", "patch": {"enabled": False}},
{"source_id": "github.work"}, )
)
assert "result" in payload assert "result" in payload
assert payload["result"]["entry"]["enabled"] is False assert payload["result"]["entry"]["id"] == "github.work"
assert payload["result"]["entry"]["enabled"] is False
asyncio.run(scenario())
def test_rpc_source_registry_remove_returns_removed(tmp_path) -> None: async def test_rpc_source_registry_enable_returns_entry(tmp_path) -> None:
async def scenario() -> None: mutation = FakeMutationProvider()
server = _server_with_mutation_provider(tmp_path) mutation.entries["github.work"]["enabled"] = False
app = create_rpc_app(server) server = replace(
transport = httpx.ASGITransport(app=app) build_local_static_workflow_server(tmp_path / "store"),
async with httpx.AsyncClient( source_registry_admin=WorkflowSourceRegistryApi(
transport=transport, base_url="http://test" provider=FakeRegistryProvider(),
) as client: mutation_provider=mutation,
payload = await _rpc( ),
client, )
"workflow.admin.source_registry.remove", app = create_rpc_app(server)
{"source_id": "github.work"}, transport = httpx.ASGITransport(app=app)
) async with httpx.AsyncClient(
transport=transport, base_url="http://test"
) as client:
payload = await _rpc(
client,
"workflow.admin.source_registry.enable",
{"source_id": "github.work"},
)
assert "result" in payload assert "result" in payload
assert payload["result"]["removed"] is True assert payload["result"]["entry"]["enabled"] is True
assert payload["result"]["source_id"] == "github.work"
asyncio.run(scenario())
async def test_rpc_source_registry_disable_returns_entry(tmp_path) -> None:
server = _server_with_mutation_provider(tmp_path)
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://test"
) as client:
payload = await _rpc(
client,
"workflow.admin.source_registry.disable",
{"source_id": "github.work"},
)
assert "result" in payload
assert payload["result"]["entry"]["enabled"] is False
async def test_rpc_source_registry_remove_returns_removed(tmp_path) -> None:
server = _server_with_mutation_provider(tmp_path)
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport, base_url="http://test"
) as client:
payload = await _rpc(
client,
"workflow.admin.source_registry.remove",
{"source_id": "github.work"},
)
assert "result" in payload
assert payload["result"]["removed"] is True
assert payload["result"]["source_id"] == "github.work"
# --- mutation error tests --- # --- mutation error tests ---
def test_rpc_source_registry_add_missing_entry_raises_error(tmp_path) -> None: async def test_rpc_source_registry_add_missing_entry_raises_error(tmp_path) -> None:
async def scenario() -> None: server = _server_with_mutation_provider(tmp_path)
server = _server_with_mutation_provider(tmp_path) app = create_rpc_app(server)
app = create_rpc_app(server) transport = httpx.ASGITransport(app=app)
transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(
async with httpx.AsyncClient( transport=transport, base_url="http://test"
transport=transport, base_url="http://test" ) as client:
) as client: payload = await _rpc(
payload = await _rpc( client,
client, "workflow.admin.source_registry.add",
"workflow.admin.source_registry.add", {"entry": {}},
{"entry": {}}, )
)
assert "error" in payload assert "error" in payload
asyncio.run(scenario())
def test_rpc_source_registry_update_missing_source_raises_error(tmp_path) -> None: async def test_rpc_source_registry_update_missing_source_raises_error(tmp_path) -> None:
async def scenario() -> None: server = _server_with_mutation_provider(tmp_path)
server = _server_with_mutation_provider(tmp_path) app = create_rpc_app(server)
app = create_rpc_app(server) transport = httpx.ASGITransport(app=app)
transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(
async with httpx.AsyncClient( transport=transport, base_url="http://test"
transport=transport, base_url="http://test" ) as client:
) as client: payload = await _rpc(
payload = await _rpc( client,
client, "workflow.admin.source_registry.update",
"workflow.admin.source_registry.update", {"source_id": "nonexistent", "patch": {"enabled": False}},
{"source_id": "nonexistent", "patch": {"enabled": False}}, )
)
assert "error" in payload assert "error" in payload
asyncio.run(scenario())
def test_rpc_source_registry_remove_missing_source_raises_error(tmp_path) -> None: async def test_rpc_source_registry_remove_missing_source_raises_error(tmp_path) -> None:
async def scenario() -> None: server = _server_with_mutation_provider(tmp_path)
server = _server_with_mutation_provider(tmp_path) app = create_rpc_app(server)
app = create_rpc_app(server) transport = httpx.ASGITransport(app=app)
transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(
async with httpx.AsyncClient( transport=transport, base_url="http://test"
transport=transport, base_url="http://test" ) as client:
) as client: payload = await _rpc(
payload = await _rpc( client,
client, "workflow.admin.source_registry.remove",
"workflow.admin.source_registry.remove", {"source_id": "nonexistent"},
{"source_id": "nonexistent"}, )
)
assert "error" in payload assert "error" in payload
asyncio.run(scenario())
# --- client method tests --- # --- client method tests ---
def test_rpc_client_source_registry_calls_correct_methods(tmp_path) -> None: async def test_rpc_client_source_registry_calls_correct_methods(tmp_path) -> None:
async def scenario() -> None: server = build_local_static_workflow_server(tmp_path / "store")
server = build_local_static_workflow_server(tmp_path / "store") app = create_rpc_app(server)
app = create_rpc_app(server) transport = httpx.ASGITransport(app=app)
transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(
async with httpx.AsyncClient( transport=transport, base_url="http://test"
transport=transport, base_url="http://test" ) as http_client:
) as http_client: client = RpcWorkflowApiClient(
client = RpcWorkflowApiClient( url="http://test/rpc",
url="http://test/rpc", timeout_seconds=5,
timeout_seconds=5, http_client=http_client,
http_client=http_client, )
) try:
try: await client.list_registry_entries(limit=5)
await client.list_registry_entries(limit=5) except RuntimeError as exc:
except RuntimeError as exc: list_error = str(exc)
list_error = str(exc) else:
else: list_error = None
list_error = None
try: try:
await client.inspect_registry_entry(source_id="x") await client.inspect_registry_entry(source_id="x")
except RuntimeError as exc: except RuntimeError as exc:
inspect_error = str(exc) inspect_error = str(exc)
else: else:
inspect_error = None inspect_error = None
assert list_error is not None assert list_error is not None
assert "source registry admin reads are not available" in list_error assert "source registry admin reads are not available" in list_error
assert inspect_error is not None assert inspect_error is not None
assert "source registry admin reads are not available" in inspect_error assert "source registry admin reads are not available" in inspect_error
asyncio.run(scenario())
def test_rpc_client_source_registry_mutation_methods_exist() -> None: async def test_rpc_client_source_registry_mutation_methods_exist() -> None:
from wf_transport_rpc_http.client import RpcWorkflowApiClient from wf_transport_rpc_http.client import RpcWorkflowApiClient
client = RpcWorkflowApiClient.__new__(RpcWorkflowApiClient) client = RpcWorkflowApiClient.__new__(RpcWorkflowApiClient)
@@ -495,31 +443,31 @@ def test_rpc_client_source_registry_mutation_methods_exist() -> None:
client._call = fake_call # type: ignore[assignment] client._call = fake_call # type: ignore[assignment]
asyncio.run(client.add_registry_entry(entry={"id": "x", "kind": "mcp"})) await client.add_registry_entry(entry={"id": "x", "kind": "mcp"})
assert calls[-1] == ( assert calls[-1] == (
"workflow.admin.source_registry.add", "workflow.admin.source_registry.add",
{"entry": {"id": "x", "kind": "mcp"}}, {"entry": {"id": "x", "kind": "mcp"}},
) )
asyncio.run(client.update_registry_entry(source_id="s", patch={"enabled": False})) await client.update_registry_entry(source_id="s", patch={"enabled": False})
assert calls[-1] == ( assert calls[-1] == (
"workflow.admin.source_registry.update", "workflow.admin.source_registry.update",
{"source_id": "s", "patch": {"enabled": False}}, {"source_id": "s", "patch": {"enabled": False}},
) )
asyncio.run(client.enable_registry_entry(source_id="s")) await client.enable_registry_entry(source_id="s")
assert calls[-1] == ( assert calls[-1] == (
"workflow.admin.source_registry.enable", "workflow.admin.source_registry.enable",
{"source_id": "s"}, {"source_id": "s"},
) )
asyncio.run(client.disable_registry_entry(source_id="s")) await client.disable_registry_entry(source_id="s")
assert calls[-1] == ( assert calls[-1] == (
"workflow.admin.source_registry.disable", "workflow.admin.source_registry.disable",
{"source_id": "s"}, {"source_id": "s"},
) )
asyncio.run(client.remove_registry_entry(source_id="s")) await client.remove_registry_entry(source_id="s")
assert calls[-1] == ( assert calls[-1] == (
"workflow.admin.source_registry.remove", "workflow.admin.source_registry.remove",
{"source_id": "s"}, {"source_id": "s"},
uv.lock
Generated
+14
View File
@@ -660,6 +660,7 @@ dependencies = [
dev = [ dev = [
{ name = "basedpyright" }, { name = "basedpyright" },
{ name = "pytest" }, { name = "pytest" },
{ name = "pytest-asyncio" },
{ name = "ruff" }, { name = "ruff" },
] ]
@@ -681,6 +682,7 @@ requires-dist = [
dev = [ dev = [
{ name = "basedpyright", specifier = ">=1.39.6" }, { name = "basedpyright", specifier = ">=1.39.6" },
{ name = "pytest", specifier = ">=8" }, { name = "pytest", specifier = ">=8" },
{ name = "pytest-asyncio", specifier = ">=1.4.0" },
{ name = "ruff", specifier = ">=0.15.15" }, { name = "ruff", specifier = ">=0.15.15" },
] ]
@@ -1065,6 +1067,18 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/d4/24/a372aaf5c9b7208e7112038812994107bc65a84cd00e0354a88c2c77a617/pytest-9.0.3-py3-none-any.whl", hash = "sha256:2c5efc453d45394fdd706ade797c0a81091eccd1d6e4bccfcd476e2b8e0ab5d9", size = 375249, upload-time = "2026-04-07T17:16:16.13Z" }, { url = "https://files.pythonhosted.org/packages/d4/24/a372aaf5c9b7208e7112038812994107bc65a84cd00e0354a88c2c77a617/pytest-9.0.3-py3-none-any.whl", hash = "sha256:2c5efc453d45394fdd706ade797c0a81091eccd1d6e4bccfcd476e2b8e0ab5d9", size = 375249, upload-time = "2026-04-07T17:16:16.13Z" },
] ]
[[package]]
name = "pytest-asyncio"
version = "1.4.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "pytest" },
]
sdist = { url = "https://files.pythonhosted.org/packages/43/7c/d36d04db312ecf4298932ef77e6e4a9e8ad017906e24e34f0b0c361a2473/pytest_asyncio-1.4.0.tar.gz", hash = "sha256:c6c0d2259945122819f171a32ecea2c349ead889ee28176caaf492143424be42", size = 58514, upload-time = "2026-05-26T09:56:04.083Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/03/e2/08a497ef684b88559c9cc5f4ad53a37e7b99e727094a86d6ea32536d5d3c/pytest_asyncio-1.4.0-py3-none-any.whl", hash = "sha256:933ca923a23075a87fb7070c0ec272a6848489824d887c85c812670932835aa1", size = 16930, upload-time = "2026-05-26T09:56:02.576Z" },
]
[[package]] [[package]]
name = "python-dotenv" name = "python-dotenv"
version = "1.2.2" version = "1.2.2"