fix: harden workflow config rpc target routing

This commit is contained in:
lda
2026-06-03 09:49:56 +07:00 Verified
parent 8a1d627f06
commit 9389dd6df6
14 changed files with 379 additions and 47 deletions
+3 -2
View File
@@ -94,8 +94,9 @@ implementation state.
`WorkflowServer` through fixed dotted methods. Remote CLI targeting remains `WorkflowServer` through fixed dotted methods. Remote CLI targeting remains
the next transport-facing slice. the next transport-facing slice.
- Completed: workflow config now distinguishes client targets from server - Completed: workflow config now distinguishes client targets from server
hosting config, and selected `wf` commands can target JSON-RPC HTTP with hosting config, selected `wf cap`/`wf run` commands can target JSON-RPC
explicit CLI overrides. HTTP with explicit CLI overrides, and local-only draft/artifact/deploy
commands fail fast for remote targets until they are wired.
5. **CLI/API alignment** 5. **CLI/API alignment**
- Let the CLI target either local process-backed stores/runtime or the future - Let the CLI target either local process-backed stores/runtime or the future
@@ -305,7 +305,7 @@ Recommended Pydantic shape:
```python ```python
class StdlibSourceConfig(BaseModel): class StdlibSourceConfig(BaseModel):
kind: Literal["stdlib"] kind: Literal["stdlib"]
id: str id: Literal["wf.std", "wf.recipes"]
class McpStdioTransportConfig(BaseModel): class McpStdioTransportConfig(BaseModel):
@@ -458,16 +458,20 @@ First slice implemented:
- neutral `wf_config` models and loader - neutral `wf_config` models and loader
- filesystem server store config - filesystem server store config
- stdlib source bootstrap config - stdlib source bootstrap config is parsed and fail-fast limited to currently
wired source ids (`wf.std`, `wf.recipes`)
- local and JSON-RPC client targets - local and JSON-RPC client targets
- `wf` root overrides for `--local`, `--url`, and `--timeout` - `wf` root overrides for `--local`, `--url`, and `--timeout`
- remote JSON-RPC client support for capability and run CLI commands - remote JSON-RPC client support for capability and run CLI commands
- `wf-rpc-server --config` support for server store and RPC HTTP transport - local-only CLI commands fail fast for `rpc_http` targets until they are wired
- `wf-rpc-server --config` support for server store and RPC HTTP transport,
including configured RPC path
Still future: Still future:
- store-backed mutable source registry - store-backed mutable source registry
- MCP/OpenAPI source config - MCP/OpenAPI source config
- arbitrary stdlib source aliases
- `/mcp` hosting from neutral server config - `/mcp` hosting from neutral server config
- remote draft/artifact/deployment CLI commands - remote draft/artifact/deployment CLI commands
- auth and SQL stores - auth and SQL stores
+3 -3
View File
@@ -5,7 +5,7 @@ from typing import Annotated, Literal
import typer import typer
from wf_cli.context import config_path_from_context, load_cli_context from wf_cli.context import load_local_cli_context_from_typer as load_cli_context
from wf_cli.formats import ListOutputFormat, emit_list_payload from wf_cli.formats import ListOutputFormat, emit_list_payload
from wf_cli.io import emit_json from wf_cli.io import emit_json
@@ -37,7 +37,7 @@ def list_artifacts(
] = ListOutputFormat.JSON, ] = ListOutputFormat.JSON,
) -> None: ) -> None:
"""List compact saved artifact summaries.""" """List compact saved artifact summaries."""
context = load_cli_context(config_path_from_context(ctx)) context = load_cli_context(ctx)
payload = asyncio.run( payload = asyncio.run(
context.handlers.list_artifacts( context.handlers.list_artifacts(
query=query, query=query,
@@ -62,7 +62,7 @@ def inspect_artifact(
version: Annotated[int, typer.Argument(min=1, help="Artifact version.")], version: Annotated[int, typer.Argument(min=1, help="Artifact version.")],
) -> None: ) -> None:
"""Inspect one saved artifact version.""" """Inspect one saved artifact version."""
context = load_cli_context(config_path_from_context(ctx)) context = load_cli_context(ctx)
emit_json( emit_json(
asyncio.run( asyncio.run(
context.handlers.inspect_artifact(artifact_id=artifact_id, version=version) context.handlers.inspect_artifact(artifact_id=artifact_id, version=version)
+6 -6
View File
@@ -6,7 +6,7 @@ from typing import Annotated
import typer import typer
from wf_cli.context import config_path_from_context, load_cli_context from wf_cli.context import load_local_cli_context_from_typer as load_cli_context
from wf_cli.formats import ListOutputFormat, emit_list_payload from wf_cli.formats import ListOutputFormat, emit_list_payload
from wf_cli.io import CliInputError, emit_json, parse_bindings, parse_json_input from wf_cli.io import CliInputError, emit_json, parse_bindings, parse_json_input
@@ -30,7 +30,7 @@ def validate_deployment(
] = False, ] = False,
) -> None: ) -> None:
"""Validate one saved workflow deployment.""" """Validate one saved workflow deployment."""
context = load_cli_context(config_path_from_context(ctx)) context = load_cli_context(ctx)
payload = asyncio.run( payload = asyncio.run(
context.handlers.validate_deployment( context.handlers.validate_deployment(
deployment_id=deployment_id, deployment_id=deployment_id,
@@ -48,7 +48,7 @@ def list_deployments(
] = ListOutputFormat.JSON, ] = ListOutputFormat.JSON,
) -> None: ) -> None:
"""List saved workflow deployments.""" """List saved workflow deployments."""
context = load_cli_context(config_path_from_context(ctx)) context = load_cli_context(ctx)
payload = asyncio.run(context.handlers.list_deployments()) payload = asyncio.run(context.handlers.list_deployments())
emit_list_payload( emit_list_payload(
payload, payload,
@@ -65,7 +65,7 @@ def inspect_deployment(
deployment_id: Annotated[str, typer.Argument(help="Deployment id.")], deployment_id: Annotated[str, typer.Argument(help="Deployment id.")],
) -> None: ) -> None:
"""Inspect one saved deployment.""" """Inspect one saved deployment."""
context = load_cli_context(config_path_from_context(ctx)) context = load_cli_context(ctx)
emit_json( emit_json(
asyncio.run(context.handlers.inspect_deployment(deployment_id=deployment_id)) asyncio.run(context.handlers.inspect_deployment(deployment_id=deployment_id))
) )
@@ -106,7 +106,7 @@ def save_deployment(
) )
except CliInputError as exc: except CliInputError as exc:
raise typer.BadParameter(str(exc)) from exc raise typer.BadParameter(str(exc)) from exc
context = load_cli_context(config_path_from_context(ctx)) context = load_cli_context(ctx)
emit_json(asyncio.run(context.handlers.save_deployment(payload))) emit_json(asyncio.run(context.handlers.save_deployment(payload)))
@@ -116,7 +116,7 @@ def delete_deployment(
deployment_id: Annotated[str, typer.Argument(help="Deployment id.")], deployment_id: Annotated[str, typer.Argument(help="Deployment id.")],
) -> None: ) -> None:
"""Delete one saved deployment.""" """Delete one saved deployment."""
context = load_cli_context(config_path_from_context(ctx)) context = load_cli_context(ctx)
emit_json( emit_json(
asyncio.run(context.handlers.delete_deployment(deployment_id=deployment_id)) asyncio.run(context.handlers.delete_deployment(deployment_id=deployment_id))
) )
+7 -7
View File
@@ -6,7 +6,7 @@ from typing import Annotated, Literal
import typer import typer
from wf_cli.context import config_path_from_context, load_cli_context from wf_cli.context import load_local_cli_context_from_typer as load_cli_context
from wf_cli.formats import ListOutputFormat, emit_list_payload from wf_cli.formats import ListOutputFormat, emit_list_payload
from wf_cli.io import CliInputError, emit_json, parse_bindings, parse_json_value from wf_cli.io import CliInputError, emit_json, parse_bindings, parse_json_value
@@ -25,7 +25,7 @@ def list_drafts(
] = ListOutputFormat.JSON, ] = ListOutputFormat.JSON,
) -> None: ) -> None:
"""List stored draft workspaces.""" """List stored draft workspaces."""
context = load_cli_context(config_path_from_context(ctx)) context = load_cli_context(ctx)
payload = asyncio.run(context.handlers.list_draft_workspaces()) payload = asyncio.run(context.handlers.list_draft_workspaces())
emit_list_payload( emit_list_payload(
payload, payload,
@@ -45,7 +45,7 @@ def inspect_draft(
] = False, ] = False,
) -> None: ) -> None:
"""Inspect one draft workspace.""" """Inspect one draft workspace."""
context = load_cli_context(config_path_from_context(ctx)) context = load_cli_context(ctx)
emit_json( emit_json(
asyncio.run( asyncio.run(
context.handlers.get_draft_workspace( context.handlers.get_draft_workspace(
@@ -69,7 +69,7 @@ def create_from_capability(
] = None, ] = None,
) -> None: ) -> None:
"""Bootstrap a draft workspace from inspect_capability wrapper hints.""" """Bootstrap a draft workspace from inspect_capability wrapper hints."""
context = load_cli_context(config_path_from_context(ctx)) context = load_cli_context(ctx)
emit_json( emit_json(
asyncio.run( asyncio.run(
context.handlers.create_draft_workspace_from_capability( context.handlers.create_draft_workspace_from_capability(
@@ -103,7 +103,7 @@ def patch_draft(
raise typer.BadParameter(str(exc)) from exc raise typer.BadParameter(str(exc)) from exc
if not isinstance(patch, list): if not isinstance(patch, list):
raise typer.BadParameter("draft patch input must be a JSON array") raise typer.BadParameter("draft patch input must be a JSON array")
context = load_cli_context(config_path_from_context(ctx)) context = load_cli_context(ctx)
emit_json( emit_json(
asyncio.run( asyncio.run(
context.handlers.patch_draft_workspace( context.handlers.patch_draft_workspace(
@@ -121,7 +121,7 @@ def validate_draft(
workspace_id: Annotated[str, typer.Argument(help="Draft workspace id.")], workspace_id: Annotated[str, typer.Argument(help="Draft workspace id.")],
) -> None: ) -> None:
"""Validate one stored draft workspace.""" """Validate one stored draft workspace."""
context = load_cli_context(config_path_from_context(ctx)) context = load_cli_context(ctx)
emit_json( emit_json(
asyncio.run( asyncio.run(
context.handlers.validate_draft_workspace(workspace_id=workspace_id) context.handlers.validate_draft_workspace(workspace_id=workspace_id)
@@ -156,7 +156,7 @@ def save_draft(
source_bindings = parse_bindings(binding or []) source_bindings = parse_bindings(binding or [])
except CliInputError as exc: except CliInputError as exc:
raise typer.BadParameter(str(exc)) from exc raise typer.BadParameter(str(exc)) from exc
context = load_cli_context(config_path_from_context(ctx)) context = load_cli_context(ctx)
if kind == "wrapper": if kind == "wrapper":
payload = asyncio.run( payload = asyncio.run(
context.handlers.create_wrapper_from_workspace( context.handlers.create_wrapper_from_workspace(
+109 -11
View File
@@ -2,19 +2,21 @@ from __future__ import annotations
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
import json
import typer import typer
from pydantic import ValidationError
from wf_api import WorkflowApi from wf_api import WorkflowApi
from wf_mcp.broker import build_service_from_config, load_broker_config
from wf_mcp.broker.service import WfMcpService
from wf_mcp.broker.service.workflow_operation_context import context_from_service
from wf_config import ( from wf_config import (
FilesystemStoreConfig, FilesystemStoreConfig,
LocalTargetConfig, LocalTargetConfig,
RpcHttpTargetConfig, RpcHttpTargetConfig,
load_workflow_config, load_workflow_config,
) )
from wf_mcp.broker import build_service_from_config, load_broker_config
from wf_mcp.broker.service import WfMcpService
from wf_mcp.broker.service.workflow_operation_context import context_from_service
from wf_server import build_local_static_workflow_server from wf_server import build_local_static_workflow_server
from wf_transport_rpc_http import RpcWorkflowApiClient from wf_transport_rpc_http import RpcWorkflowApiClient
@@ -28,6 +30,15 @@ class CliContext:
handlers: "WorkflowApi | RpcWorkflowApiClient" handlers: "WorkflowApi | RpcWorkflowApiClient"
@dataclass(frozen=True)
class LocalCliContext:
"""CLI context for commands that still require same-process WorkflowApi."""
config_path: Path
service: WfMcpService | None
handlers: WorkflowApi
def config_path_from_context(ctx: typer.Context) -> str: def config_path_from_context(ctx: typer.Context) -> str:
"""Return the root --config path captured by the Typer callback.""" """Return the root --config path captured by the Typer callback."""
obj = ctx.obj if isinstance(ctx.obj, dict) else {} obj = ctx.obj if isinstance(ctx.obj, dict) else {}
@@ -44,7 +55,24 @@ def load_cli_context(
) -> CliContext: ) -> CliContext:
"""Load config and build workflow-surface handlers for CLI commands.""" """Load config and build workflow-surface handlers for CLI commands."""
resolved_config_path = Path(config_path) resolved_config_path = Path(config_path)
if resolved_config_path.name == "wf_mcp.config.json": if force_local and rpc_url is not None:
raise ValueError("--local and --url are mutually exclusive")
if rpc_url is not None:
_validate_rpc_url(rpc_url)
return CliContext(
config_path=resolved_config_path,
service=None,
handlers=RpcWorkflowApiClient(
url=rpc_url,
timeout_seconds=_rpc_timeout_from_optional_config(
resolved_config_path,
override=rpc_timeout_seconds,
),
),
)
if _is_legacy_mcp_config(resolved_config_path):
config = load_broker_config(resolved_config_path) config = load_broker_config(resolved_config_path)
service = build_service_from_config(config) service = build_service_from_config(config)
return CliContext( return CliContext(
@@ -55,13 +83,6 @@ def load_cli_context(
config = load_workflow_config(resolved_config_path) config = load_workflow_config(resolved_config_path)
target = config.client.target target = config.client.target
if rpc_url is not None:
timeout = rpc_timeout_seconds if rpc_timeout_seconds is not None else 30.0
return CliContext(
config_path=resolved_config_path,
service=None,
handlers=RpcWorkflowApiClient(url=rpc_url, timeout_seconds=timeout),
)
if force_local or isinstance(target, LocalTargetConfig): if force_local or isinstance(target, LocalTargetConfig):
store = config.server.store store = config.server.store
if not isinstance(store, FilesystemStoreConfig): if not isinstance(store, FilesystemStoreConfig):
@@ -88,6 +109,32 @@ def load_cli_context(
raise ValueError(f"unsupported workflow target {target!r}") raise ValueError(f"unsupported workflow target {target!r}")
def load_local_cli_context(
config_path: str | Path,
*,
force_local: bool = False,
rpc_url: str | None = None,
rpc_timeout_seconds: float | None = None,
) -> LocalCliContext:
"""Load a local WorkflowApi context for commands not remote-enabled yet."""
context = load_cli_context(
config_path,
force_local=force_local,
rpc_url=rpc_url,
rpc_timeout_seconds=rpc_timeout_seconds,
)
if not isinstance(context.handlers, WorkflowApi):
raise ValueError(
"this CLI command is not available for rpc_http targets yet; "
"use --local or run a cap/run command"
)
return LocalCliContext(
config_path=context.config_path,
service=context.service,
handlers=context.handlers,
)
def force_local_from_context(ctx: typer.Context) -> bool: def force_local_from_context(ctx: typer.Context) -> bool:
obj = ctx.obj if isinstance(ctx.obj, dict) else {} obj = ctx.obj if isinstance(ctx.obj, dict) else {}
return bool(obj.get("force_local", False)) return bool(obj.get("force_local", False))
@@ -106,9 +153,60 @@ def rpc_timeout_from_context(ctx: typer.Context) -> float | None:
def load_cli_context_from_typer(ctx: typer.Context) -> CliContext: def load_cli_context_from_typer(ctx: typer.Context) -> CliContext:
try:
return load_cli_context( return load_cli_context(
config_path_from_context(ctx), config_path_from_context(ctx),
force_local=force_local_from_context(ctx), force_local=force_local_from_context(ctx),
rpc_url=rpc_url_from_context(ctx), rpc_url=rpc_url_from_context(ctx),
rpc_timeout_seconds=rpc_timeout_from_context(ctx), rpc_timeout_seconds=rpc_timeout_from_context(ctx),
) )
except ValueError as exc:
raise typer.BadParameter(str(exc)) from exc
def load_local_cli_context_from_typer(ctx: typer.Context) -> LocalCliContext:
try:
return load_local_cli_context(
config_path_from_context(ctx),
force_local=force_local_from_context(ctx),
rpc_url=rpc_url_from_context(ctx),
rpc_timeout_seconds=rpc_timeout_from_context(ctx),
)
except ValueError as exc:
raise typer.BadParameter(str(exc)) from exc
def _is_legacy_mcp_config(path: Path) -> bool:
"""Detect legacy broker config by content, not filename.
This keeps `wf_mcp.config.json` compatibility without making the filename a
load-bearing part of the neutral workflow config migration.
"""
data = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(data, dict):
return False
if any(key in data for key in ("version", "client", "server")):
return False
return any(key in data for key in ("store_root", "connections"))
def _rpc_timeout_from_optional_config(
path: Path,
*,
override: float | None,
) -> float:
if override is not None:
return override
try:
config = load_workflow_config(path)
except FileNotFoundError, json.JSONDecodeError, ValidationError:
return 30.0
target = config.client.target
if isinstance(target, RpcHttpTargetConfig):
return target.timeout_seconds
return 30.0
def _validate_rpc_url(url: str) -> None:
if not url.startswith(("http://", "https://")):
raise ValueError("rpc url must start with http:// or https://")
+9 -2
View File
@@ -18,9 +18,16 @@ class LocalTargetConfig(WorkflowConfigModel):
class RpcHttpTargetConfig(WorkflowConfigModel): class RpcHttpTargetConfig(WorkflowConfigModel):
kind: Literal["rpc_http"] kind: Literal["rpc_http"]
url: str url: str = Field(min_length=1)
timeout_seconds: float = Field(default=30.0, gt=0) timeout_seconds: float = Field(default=30.0, gt=0)
@field_validator("url")
@classmethod
def validate_url(cls, value: str) -> str:
if not value.startswith(("http://", "https://")):
raise ValueError("rpc_http target url must start with http:// or https://")
return value
TargetConfig = Annotated[ TargetConfig = Annotated[
LocalTargetConfig | RpcHttpTargetConfig, LocalTargetConfig | RpcHttpTargetConfig,
@@ -65,7 +72,7 @@ ServerTransportConfig = Annotated[
class StdlibSourceConfig(WorkflowConfigModel): class StdlibSourceConfig(WorkflowConfigModel):
kind: Literal["stdlib"] kind: Literal["stdlib"]
id: str = Field(min_length=1) id: Literal["wf.std", "wf.recipes"]
SourceConfig = Annotated[ SourceConfig = Annotated[
+4 -2
View File
@@ -25,16 +25,18 @@ from .models import (
) )
def create_rpc_app(server: WorkflowServer) -> jsonrpc.API: def create_rpc_app(server: WorkflowServer, *, rpc_path: str = "/rpc") -> jsonrpc.API:
"""Build a JSON-RPC HTTP app over an existing WorkflowServer. """Build a JSON-RPC HTTP app over an existing WorkflowServer.
Transport code owns only JSON-RPC envelope handling. Workflow semantics stay Transport code owns only JSON-RPC envelope handling. Workflow semantics stay
behind server.api, so this package remains swappable with WebSocket/MCP behind server.api, so this package remains swappable with WebSocket/MCP
transports later. transports later.
""" """
if not rpc_path.startswith("/"):
raise ValueError("rpc_path must start with '/'")
app = jsonrpc.API() app = jsonrpc.API()
entrypoint = jsonrpc.Entrypoint("/rpc") entrypoint = jsonrpc.Entrypoint(rpc_path)
@app.get("/healthz") @app.get("/healthz")
async def healthz() -> dict[str, str]: async def healthz() -> dict[str, str]:
+15 -3
View File
@@ -30,13 +30,24 @@ def serve(
"--store-root", "--store-root",
help="Override filesystem workflow store root.", help="Override filesystem workflow store root.",
), ),
host: str | None = typer.Option(None, "--host"), host: str | None = typer.Option(
port: int | None = typer.Option(None, "--port", min=1, max=65535), None,
"--host",
help="Override RPC bind host; defaults to config or 127.0.0.1.",
),
port: int | None = typer.Option(
None,
"--port",
min=1,
max=65535,
help="Override RPC bind port; defaults to config or 8765.",
),
) -> None: ) -> None:
"""Serve the local/static WorkflowApi over JSON-RPC HTTP.""" """Serve the local/static WorkflowApi over JSON-RPC HTTP."""
resolved_store_root = store_root resolved_store_root = store_root
resolved_host = host resolved_host = host
resolved_port = port resolved_port = port
resolved_rpc_path = "/rpc"
if config is not None: if config is not None:
workflow_config = load_workflow_config(config) workflow_config = load_workflow_config(config)
store = workflow_config.server.store store = workflow_config.server.store
@@ -56,13 +67,14 @@ def serve(
if rpc_transport is not None: if rpc_transport is not None:
resolved_host = host or rpc_transport.host resolved_host = host or rpc_transport.host
resolved_port = port or rpc_transport.port resolved_port = port or rpc_transport.port
resolved_rpc_path = rpc_transport.path
if resolved_store_root is None: if resolved_store_root is None:
raise typer.BadParameter( raise typer.BadParameter(
"--store-root is required when --config is not supplied" "--store-root is required when --config is not supplied"
) )
server = build_local_static_workflow_server(resolved_store_root) server = build_local_static_workflow_server(resolved_store_root)
rpc_app = create_rpc_app(server) rpc_app = create_rpc_app(server, rpc_path=resolved_rpc_path)
uvicorn.run( uvicorn.run(
rpc_app, rpc_app,
host=resolved_host or "127.0.0.1", host=resolved_host or "127.0.0.1",
+2
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import json import json
from pathlib import Path from pathlib import Path
from wf_api import WorkflowApi
from wf_cli.context import load_cli_context from wf_cli.context import load_cli_context
@@ -29,6 +30,7 @@ def test_load_cli_context_builds_service_and_handlers(tmp_path: Path) -> None:
context = load_cli_context(config_path) context = load_cli_context(config_path)
service = context.service service = context.service
assert service is not None assert service is not None
assert isinstance(context.handlers, WorkflowApi)
assert context.config_path == config_path assert context.config_path == config_path
assert service.connections.list_all()[0].id == "demo.personal" assert service.connections.list_all()[0].id == "demo.personal"
+106 -1
View File
@@ -7,7 +7,7 @@ from typer.testing import CliRunner
from wf_api.models import RawWorkflowPlan from wf_api.models import RawWorkflowPlan
from wf_cli.app import app from wf_cli.app import app
from wf_cli.context import load_cli_context from wf_cli.context import load_cli_context, load_local_cli_context
from wf_core import END from wf_core import END
from wf_server import build_local_static_workflow_server from wf_server import build_local_static_workflow_server
from wf_transport_rpc_http import RpcWorkflowApiClient, create_rpc_app from wf_transport_rpc_http import RpcWorkflowApiClient, create_rpc_app
@@ -66,6 +66,111 @@ def test_load_cli_context_local_override_beats_rpc_config(tmp_path) -> None:
assert context.config_path == config_path assert context.config_path == config_path
def test_load_cli_context_rejects_local_and_url_conflict(tmp_path) -> None:
config_path = tmp_path / "wf.json"
config_path.write_text('{"version": 1}', encoding="utf-8")
try:
load_cli_context(
config_path,
force_local=True,
rpc_url="http://127.0.0.1:8765/rpc",
)
except ValueError as exc:
message = str(exc)
else:
raise AssertionError("expected ValueError")
assert "--local and --url are mutually exclusive" in message
def test_load_cli_context_uses_workflow_shape_not_filename(tmp_path) -> None:
config_path = tmp_path / "wf_mcp.config.json"
config_path.write_text(
json.dumps(
{
"version": 1,
"client": {
"target": {
"kind": "rpc_http",
"url": "http://127.0.0.1:8765/rpc",
}
},
}
),
encoding="utf-8",
)
context = load_cli_context(config_path)
assert isinstance(context.handlers, RpcWorkflowApiClient)
def test_load_cli_context_uses_broker_shape_not_filename(tmp_path) -> None:
config_path = tmp_path / "wf.json"
config_path.write_text(
json.dumps({"store_root": ".wf_mcp_store", "connections": []}),
encoding="utf-8",
)
context = load_cli_context(config_path)
assert context.service is not None
assert not isinstance(context.handlers, RpcWorkflowApiClient)
def test_load_cli_context_url_override_reuses_config_timeout(tmp_path) -> None:
config_path = tmp_path / "wf.json"
config_path.write_text(
json.dumps(
{
"version": 1,
"client": {
"target": {
"kind": "rpc_http",
"url": "http://127.0.0.1:8765/rpc",
"timeout_seconds": 77,
}
},
}
),
encoding="utf-8",
)
context = load_cli_context(config_path, rpc_url="http://127.0.0.1:9999/rpc")
assert isinstance(context.handlers, RpcWorkflowApiClient)
assert context.handlers.url == "http://127.0.0.1:9999/rpc"
assert context.handlers.timeout_seconds == 77
def test_local_cli_context_rejects_rpc_target_for_local_only_commands(tmp_path) -> None:
config_path = tmp_path / "wf.json"
config_path.write_text(
json.dumps(
{
"version": 1,
"client": {
"target": {
"kind": "rpc_http",
"url": "http://127.0.0.1:8765/rpc",
}
},
}
),
encoding="utf-8",
)
try:
load_local_cli_context(config_path)
except ValueError as exc:
message = str(exc)
else:
raise AssertionError("expected ValueError")
assert "not available for rpc_http targets yet" in message
def _constant_plan() -> RawWorkflowPlan: def _constant_plan() -> RawWorkflowPlan:
return RawWorkflowPlan.model_validate( return RawWorkflowPlan.model_validate(
{ {
+20
View File
@@ -92,6 +92,26 @@ def test_workflow_config_rejects_unknown_target_kind() -> None:
) )
def test_workflow_config_rejects_invalid_rpc_http_url() -> None:
with pytest.raises(ValidationError, match="http:// or https://"):
WorkflowConfigFile.model_validate(
{
"version": 1,
"client": {"target": {"kind": "rpc_http", "url": "not-a-url"}},
}
)
def test_workflow_config_rejects_unwired_stdlib_source_id() -> None:
with pytest.raises(ValidationError):
WorkflowConfigFile.model_validate(
{
"version": 1,
"server": {"sources": [{"kind": "stdlib", "id": "custom.id"}]},
}
)
def test_load_workflow_config_resolves_filesystem_store_relative_to_config( def test_load_workflow_config_resolves_filesystem_store_relative_to_config(
tmp_path, tmp_path,
) -> None: ) -> None:
+24
View File
@@ -68,6 +68,30 @@ def test_rpc_unknown_method_returns_json_rpc_error(tmp_path) -> None:
asyncio.run(scenario()) 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: def test_rpc_draft_artifact_deployment_lifecycle(tmp_path) -> None:
async def scenario() -> None: async def scenario() -> None:
server = build_local_static_workflow_server(tmp_path / "store") server = build_local_static_workflow_server(tmp_path / "store")
+57
View File
@@ -42,3 +42,60 @@ def test_rpc_server_cli_accepts_config_file(tmp_path) -> None:
assert result.exit_code == 0 assert result.exit_code == 0
assert "--config" in result.output assert "--config" in result.output
def test_rpc_server_cli_uses_configured_store_and_transport(
monkeypatch, tmp_path
) -> None:
config_path = tmp_path / "wf.json"
config_path.write_text(
json.dumps(
{
"version": 1,
"server": {
"store": {"kind": "filesystem", "root": ".wf_store"},
"transports": [
{
"kind": "rpc_http",
"host": "127.0.0.2",
"port": 9999,
"path": "/workflow-rpc",
}
],
},
}
),
encoding="utf-8",
)
captured: dict[str, object] = {}
def fake_build_server(root):
captured["store_root"] = root
return object()
def fake_create_rpc_app(server, *, rpc_path="/rpc"):
captured["server"] = server
captured["rpc_path"] = rpc_path
return object()
def fake_uvicorn_run(app_obj, *, host, port, access_log):
captured["app"] = app_obj
captured["host"] = host
captured["port"] = port
captured["access_log"] = access_log
monkeypatch.setattr(
"wf_transport_rpc_http.cli.build_local_static_workflow_server",
fake_build_server,
)
monkeypatch.setattr("wf_transport_rpc_http.cli.create_rpc_app", fake_create_rpc_app)
monkeypatch.setattr("wf_transport_rpc_http.cli.uvicorn.run", fake_uvicorn_run)
result = CliRunner().invoke(app, ["--config", str(config_path)])
assert result.exit_code == 0, result.output
assert captured["store_root"] == (tmp_path / ".wf_store").resolve()
assert captured["rpc_path"] == "/workflow-rpc"
assert captured["host"] == "127.0.0.2"
assert captured["port"] == 9999
assert captured["access_log"] is False