fix: validate rpc target urls with pydantic
This commit is contained in:
@@ -98,7 +98,7 @@ def load_cli_context(
|
||||
config_path=resolved_config_path,
|
||||
service=None,
|
||||
handlers=RpcWorkflowApiClient(
|
||||
url=target.url,
|
||||
url=str(target.url),
|
||||
timeout_seconds=(
|
||||
rpc_timeout_seconds
|
||||
if rpc_timeout_seconds is not None
|
||||
|
||||
@@ -3,7 +3,7 @@ from __future__ import annotations
|
||||
from pathlib import Path
|
||||
from typing import Annotated, Literal
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
|
||||
from pydantic import AnyHttpUrl, BaseModel, ConfigDict, Field, field_validator, model_validator
|
||||
|
||||
|
||||
class WorkflowConfigModel(BaseModel):
|
||||
@@ -18,16 +18,9 @@ class LocalTargetConfig(WorkflowConfigModel):
|
||||
|
||||
class RpcHttpTargetConfig(WorkflowConfigModel):
|
||||
kind: Literal["rpc_http"]
|
||||
url: str = Field(min_length=1)
|
||||
url: AnyHttpUrl
|
||||
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[
|
||||
LocalTargetConfig | RpcHttpTargetConfig,
|
||||
|
||||
@@ -60,7 +60,7 @@ def test_workflow_config_parses_rpc_http_target_and_transport() -> None:
|
||||
)
|
||||
|
||||
assert isinstance(config.client.target, RpcHttpTargetConfig)
|
||||
assert config.client.target.url == "http://127.0.0.1:8765/rpc"
|
||||
assert str(config.client.target.url) == "http://127.0.0.1:8765/rpc"
|
||||
assert config.client.target.timeout_seconds == 12
|
||||
assert isinstance(config.server.transports[0], RpcHttpTransportConfig)
|
||||
assert config.server.transports[0].host == "0.0.0.0"
|
||||
@@ -93,7 +93,7 @@ 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://"):
|
||||
with pytest.raises(ValidationError):
|
||||
WorkflowConfigFile.model_validate(
|
||||
{
|
||||
"version": 1,
|
||||
|
||||
Reference in New Issue
Block a user