diff --git a/src/adcp/client.py b/src/adcp/client.py index dec5f9825..9ac23b874 100644 --- a/src/adcp/client.py +++ b/src/adcp/client.py @@ -377,10 +377,13 @@ def __init__( JSON schemas. Defaults (matching the TS port): requests in ``warn`` mode (drift logged but not blocked — partial payloads in error-path tests still work) and responses - in ``strict`` mode (agent drift fails the task). The - response mode flips to ``warn`` when any of ``ADCP_ENV`` - / ``PYTHON_ENV`` / ``ENV`` / ``ENVIRONMENT`` is set to - ``production`` / ``prod``. Storyboards and compliance + in ``strict`` mode (agent drift fails the task). + ``ADCP_VALIDATION_MODE=strict|warn|off`` overrides both + sides at call time (matches the TS port); ``ADCP_ENV`` + set to ``production`` / ``prod`` flips only the response + default to ``warn``. Generic ``ENV`` / ``ENVIRONMENT`` / + ``PYTHON_ENV`` are deliberately ignored — they collide + with unrelated tooling. Storyboards and compliance runners that want hard-stop enforcement everywhere pass ``validation=ValidationHookConfig(requests="strict", responses="strict")``; high-throughput callers can set diff --git a/src/adcp/server/mcp_tools.py b/src/adcp/server/mcp_tools.py index 963a69429..8816c215a 100644 --- a/src/adcp/server/mcp_tools.py +++ b/src/adcp/server/mcp_tools.py @@ -1173,6 +1173,49 @@ def _resolve(node: Any, seen: frozenset[str]) -> Any: return result +def _model_to_json_schema( + model_type: Any, *, allow_root_union: bool = False +) -> dict[str, Any] | None: + """Generate a flat JSON Schema for a Pydantic model or union. + + * Plain ``BaseModel`` subclasses use ``model_json_schema()``. + * Union / Optional types use ``TypeAdapter`` so discriminated unions + and aliases (``CreateMediaBuyResponse = ...Response1 | ...Response2``) + generate as ``anyOf``. + * ``$ref`` nodes are inlined (see :func:`_inline_refs`) so MCP + clients that don't resolve references see the full surface. + + When ``allow_root_union`` is ``False`` (the default — used for input + schemas), schemas with a root-level ``anyOf`` / ``$ref`` return + ``None`` so the caller falls back to a hand-crafted shape. + Input schemas need ``type: "object"`` at the root so MCP clients can + render the form. Output schemas can validly be a discriminated + union, so ``allow_root_union=True`` keeps the ``anyOf``. + + Returns ``None`` on any failure — callers fall back to skipping. + """ + try: + from pydantic import TypeAdapter + + if isinstance(model_type, type) and hasattr(model_type, "model_json_schema"): + schema = model_type.model_json_schema() + else: + adapter = TypeAdapter(model_type) + schema = adapter.json_schema() + except Exception: + return None + + schema.pop("title", None) + + if not allow_root_union and ("anyOf" in schema or "$ref" in schema): + return None + + try: + return _inline_refs(schema) + except Exception: + return None + + def _generate_pydantic_schemas() -> dict[str, dict[str, Any]]: """Generate JSON schemas from Pydantic request models. @@ -1188,8 +1231,6 @@ def _generate_pydantic_schemas() -> dict[str, dict[str, Any]]: against that regression by asserting every tool has an entry here. """ try: - from pydantic import TypeAdapter - from adcp.types import ( AcquireRightsRequest, ActivateSignalRequest, @@ -1331,49 +1372,212 @@ def _generate_pydantic_schemas() -> dict[str, dict[str, Any]]: schemas: dict[str, dict[str, Any]] = {} for tool_name, request_type in _tool_to_request.items(): - try: - # Handle union types (e.g. PreviewCreativeRequest, ComplyTestControllerRequest) - if isinstance(request_type, type) and hasattr(request_type, "model_json_schema"): - schema = request_type.model_json_schema() - else: - # Union types need TypeAdapter - adapter = TypeAdapter(request_type) - schema = adapter.json_schema() + # Input schemas must be flat ``type: "object"`` — root-level + # ``anyOf`` / ``$ref`` schemas are skipped so the hand-crafted + # stub stays in place. + schema = _model_to_json_schema(request_type, allow_root_union=False) + if schema is None: + logger.debug( + "Pydantic input-schema generation skipped for %s, using hand-crafted schema", + tool_name, + ) + continue + schemas[tool_name] = schema - schema.pop("title", None) + return schemas - # Union types produce anyOf with $ref at root — these can't - # be represented as flat MCP schemas. Keep hand-crafted. - if "anyOf" in schema or "$ref" in schema: - continue - # Inline every $ref into its $defs body so MCP clients that - # don't resolve JSON-Schema references (a surprisingly large - # slice of the ecosystem) still see the full tool surface. - # Spec-wise the schema is equivalent — just flat. - schema = _inline_refs(schema) +def _generate_pydantic_output_schemas() -> dict[str, dict[str, Any]]: + """Generate JSON schemas from Pydantic response models. - schemas[tool_name] = schema - except Exception: + Mirror of :func:`_generate_pydantic_schemas` for the response side. + Each AdCP tool has a corresponding ``Response`` type — for plain + success responses this is a single ``BaseModel`` subclass; for tools + that distinguish success / error / pending / rejected on the wire + (``CreateMediaBuyResponse``, ``AcquireRightsResponse``, etc.) it's a + union alias. + + Output schemas advertise the structured-content shape on + ``tools/list`` (matches the TS port) so MCP clients can validate + ``structuredContent`` without a separate spec lookup. + + Unlike input schemas, root-level ``anyOf`` is allowed — discriminated + response unions are valid JSON Schema and clients that consume + ``outputSchema`` already handle them. + """ + try: + from adcp.types import ( + AcquireRightsResponse, + ActivateSignalResponse, + BuildCreativeResponse, + CalibrateContentResponse, + CheckGovernanceResponse, + ComplyTestControllerResponse, + ContextMatchResponse, + CreateCollectionListResponse, + CreateContentStandardsResponse, + CreateMediaBuyResponse, + CreatePropertyListResponse, + DeleteCollectionListResponse, + DeletePropertyListResponse, + GetAccountFinancialsResponse, + GetAdcpCapabilitiesResponse, + GetBrandIdentityResponse, + GetCollectionListResponse, + GetContentStandardsResponse, + GetCreativeDeliveryResponse, + GetCreativeFeaturesResponse, + GetMediaBuyArtifactsResponse, + GetMediaBuyDeliveryResponse, + GetMediaBuysResponse, + GetPlanAuditLogsResponse, + GetProductsResponse, + GetPropertyListResponse, + GetRightsResponse, + GetSignalsResponse, + IdentityMatchResponse, + ListAccountsResponse, + ListCollectionListsResponse, + ListContentStandardsResponse, + ListCreativeFormatsResponse, + ListCreativesResponse, + ListPropertyListsResponse, + LogEventResponse, + PreviewCreativeResponse, + ProvidePerformanceFeedbackResponse, + ReportPlanOutcomeResponse, + ReportUsageResponse, + SiGetOfferingResponse, + SiInitiateSessionResponse, + SiSendMessageResponse, + SiTerminateSessionResponse, + SyncAccountsResponse, + SyncAudiencesResponse, + SyncCatalogsResponse, + SyncCreativesResponse, + SyncEventSourcesResponse, + SyncGovernanceResponse, + SyncPlansResponse, + UpdateCollectionListResponse, + UpdateContentStandardsResponse, + UpdateMediaBuyResponse, + UpdatePropertyListResponse, + UpdateRightsResponse, + ValidateContentDeliveryResponse, + ) + except ImportError: + return {} + + _tool_to_response: dict[str, Any] = { + # Catalog + "get_products": GetProductsResponse, + "list_creative_formats": ListCreativeFormatsResponse, + # Creative + "sync_creatives": SyncCreativesResponse, + "list_creatives": ListCreativesResponse, + "build_creative": BuildCreativeResponse, + "preview_creative": PreviewCreativeResponse, + "get_creative_delivery": GetCreativeDeliveryResponse, + # Media Buy + "create_media_buy": CreateMediaBuyResponse, + "update_media_buy": UpdateMediaBuyResponse, + "get_media_buy_delivery": GetMediaBuyDeliveryResponse, + "get_media_buys": GetMediaBuysResponse, + # Signals + "get_signals": GetSignalsResponse, + "activate_signal": ActivateSignalResponse, + # Account + "list_accounts": ListAccountsResponse, + "sync_accounts": SyncAccountsResponse, + "get_account_financials": GetAccountFinancialsResponse, + "report_usage": ReportUsageResponse, + # Events & Catalogs + "log_event": LogEventResponse, + "sync_event_sources": SyncEventSourcesResponse, + "sync_audiences": SyncAudiencesResponse, + "sync_catalogs": SyncCatalogsResponse, + "sync_governance": SyncGovernanceResponse, + # Feedback + "provide_performance_feedback": ProvidePerformanceFeedbackResponse, + # Protocol Discovery + "get_adcp_capabilities": GetAdcpCapabilitiesResponse, + # Compliance + "comply_test_controller": ComplyTestControllerResponse, + # Content Standards + "create_content_standards": CreateContentStandardsResponse, + "get_content_standards": GetContentStandardsResponse, + "list_content_standards": ListContentStandardsResponse, + "update_content_standards": UpdateContentStandardsResponse, + "calibrate_content": CalibrateContentResponse, + "validate_content_delivery": ValidateContentDeliveryResponse, + "get_media_buy_artifacts": GetMediaBuyArtifactsResponse, + # Governance + "get_creative_features": GetCreativeFeaturesResponse, + "sync_plans": SyncPlansResponse, + "check_governance": CheckGovernanceResponse, + "report_plan_outcome": ReportPlanOutcomeResponse, + "get_plan_audit_logs": GetPlanAuditLogsResponse, + # Property Lists + "create_property_list": CreatePropertyListResponse, + "get_property_list": GetPropertyListResponse, + "list_property_lists": ListPropertyListsResponse, + "update_property_list": UpdatePropertyListResponse, + "delete_property_list": DeletePropertyListResponse, + # Collection Lists + "create_collection_list": CreateCollectionListResponse, + "get_collection_list": GetCollectionListResponse, + "list_collection_lists": ListCollectionListsResponse, + "update_collection_list": UpdateCollectionListResponse, + "delete_collection_list": DeleteCollectionListResponse, + # Sponsored Intelligence + "si_get_offering": SiGetOfferingResponse, + "si_initiate_session": SiInitiateSessionResponse, + "si_send_message": SiSendMessageResponse, + "si_terminate_session": SiTerminateSessionResponse, + # Brand + "get_brand_identity": GetBrandIdentityResponse, + "get_rights": GetRightsResponse, + "acquire_rights": AcquireRightsResponse, + "update_rights": UpdateRightsResponse, + # TMP + "context_match": ContextMatchResponse, + "identity_match": IdentityMatchResponse, + } + + schemas: dict[str, dict[str, Any]] = {} + for tool_name, response_type in _tool_to_response.items(): + schema = _model_to_json_schema(response_type, allow_root_union=True) + if schema is None: logger.debug( - "Pydantic schema generation failed for %s, using hand-crafted schema", + "Pydantic output-schema generation failed for %s", tool_name, - exc_info=True, ) + continue + schemas[tool_name] = schema return schemas # Generate schemas once at import time _PYDANTIC_SCHEMAS = _generate_pydantic_schemas() +_PYDANTIC_OUTPUT_SCHEMAS = _generate_pydantic_output_schemas() def _apply_pydantic_schemas() -> None: - """Replace hand-crafted inputSchemas with Pydantic-generated ones.""" + """Apply Pydantic-generated input + output schemas to tool definitions. + + * ``inputSchema``: replaced when a Pydantic-generated schema is + available (handles drift between hand-crafted stubs and the spec). + * ``outputSchema``: added so ``tools/list`` advertises the structured + response shape — matches the TS port and lets MCP clients validate + ``structuredContent`` without a separate spec lookup. + """ for tool_def in ADCP_TOOL_DEFINITIONS: name = tool_def["name"] if name in _PYDANTIC_SCHEMAS: tool_def["inputSchema"] = _PYDANTIC_SCHEMAS[name] + if name in _PYDANTIC_OUTPUT_SCHEMAS: + tool_def["outputSchema"] = _PYDANTIC_OUTPUT_SCHEMAS[name] _apply_pydantic_schemas() diff --git a/src/adcp/server/serve.py b/src/adcp/server/serve.py index 14114ad1f..62cedf903 100644 --- a/src/adcp/server/serve.py +++ b/src/adcp/server/serve.py @@ -1310,6 +1310,7 @@ def _register_handler_tools( continue description = tool_def.get("description", "") input_schema = tool_def.get("inputSchema", {"type": "object", "properties": {}}) + output_schema = tool_def.get("outputSchema") caller = create_tool_caller(handler, tool_name, validation=validation) _register_tool( mcp, @@ -1319,6 +1320,7 @@ def _register_handler_tools( caller, context_factory=context_factory, middleware=middleware_tuple, + output_schema=output_schema, ) registered.append(tool_name) @@ -1339,6 +1341,7 @@ def _register_tool( *, context_factory: ContextFactory | None = None, middleware: tuple[SkillMiddleware, ...] = (), + output_schema: dict[str, Any] | None = None, ) -> None: """Register a single ADCP tool on a FastMCP server. @@ -1449,9 +1452,18 @@ def model_dump_one_level(self) -> dict[str, Any]: result.update(self.model_extra) return result + # Advertise the spec response schema on ``tools/list`` when one is + # available. FastMCP serializes ``Tool.output_schema`` (which reads + # ``fn_metadata.output_schema``) into the ``outputSchema`` field of + # the ``tools/list`` response — matches the TS port. Falls back to + # the auto-derived shape from the ``fn`` return annotation when no + # spec schema is mapped (e.g. handler-only custom tools). + effective_output_schema = ( + output_schema if output_schema is not None else tool.fn_metadata.output_schema + ) tool.fn_metadata = FuncMetadata( arg_model=_AdcpArgs, - output_schema=tool.fn_metadata.output_schema, + output_schema=effective_output_schema, output_model=tool.fn_metadata.output_model, wrap_output=False, ) diff --git a/src/adcp/validation/client_hooks.py b/src/adcp/validation/client_hooks.py index 24ca64e37..8c1d0af9a 100644 --- a/src/adcp/validation/client_hooks.py +++ b/src/adcp/validation/client_hooks.py @@ -38,9 +38,20 @@ class ValidationHookConfig: makes the SDK a compliance harness: drift from an agent fails the task on the first call, not the Nth storyboard run. - Only ``ADCP_ENV`` is consulted — generic ``ENV`` / ``ENVIRONMENT`` - would collide with unrelated tooling (rails, postgres, 12-factor) - and silently flip the SDK's default. + Resolution order for both sides at call time: + + 1. Explicit value on this config (``requests=`` / ``responses=``). + 2. ``ADCP_VALIDATION_MODE`` env var (``strict`` / ``warn`` / ``off``) + — applies to both sides unless overridden by an explicit value. + Matches the TS port (adcontextprotocol/adcp-client). + 3. ``ADCP_ENV=prod|production`` flips the response default to + ``warn``; requests fall back to the type default. + 4. Defaults: ``requests="warn"``, ``responses="strict"``. + + Only ``ADCP_ENV`` and ``ADCP_VALIDATION_MODE`` are consulted — + generic ``ENV`` / ``ENVIRONMENT`` would collide with unrelated + tooling (rails, postgres, 12-factor) and silently flip the SDK's + default. """ requests: ValidationMode | None = None @@ -59,6 +70,28 @@ class DebugLogEntry(TypedDict, total=False): issues: list[dict[str, Any]] +_VALID_MODES: frozenset[str] = frozenset({"strict", "warn", "off"}) + + +def _env_validation_mode() -> ValidationMode | None: + """Read ``ADCP_VALIDATION_MODE`` at call time. + + Returns ``None`` when the env var is unset, empty, or holds a value + that isn't one of the three valid modes. Unrecognized values are + ignored rather than raising — keeps the SDK robust against typos in + deploy environments where misreads would silently change validation + posture (better to fall back to the documented defaults than blow + up on the next request). + """ + val = os.environ.get("ADCP_VALIDATION_MODE") + if not val: + return None + normalized = val.strip().lower() + if normalized in _VALID_MODES: + return normalized # type: ignore[return-value] + return None + + def _default_response_mode() -> ValidationMode: """Response default: ``strict`` unless ``ADCP_ENV`` declares production. @@ -74,11 +107,25 @@ def _default_response_mode() -> ValidationMode: def resolve_validation_modes( config: ValidationHookConfig | None = None, ) -> tuple[ValidationMode, ValidationMode]: - """Return the effective ``(requests, responses)`` modes.""" - req: ValidationMode = (config.requests if config is not None else None) or "warn" - resp: ValidationMode = ( - config.responses if config is not None else None - ) or _default_response_mode() + """Return the effective ``(requests, responses)`` modes. + + Resolution order (per side): + + 1. Explicit ``config.requests`` / ``config.responses`` (when set). + 2. ``ADCP_VALIDATION_MODE`` env var — applies to both sides. + 3. ``ADCP_ENV=prod|production`` flips the response default to + ``warn``; requests fall back to ``warn`` (the type default). + 4. Hard defaults: ``requests="warn"``, ``responses="strict"``. + + Read at call time (not import time) so tests that mutate env vars + via ``patch.dict`` work without a module-level reset hook. + """ + explicit_req = config.requests if config is not None else None + explicit_resp = config.responses if config is not None else None + env_mode = _env_validation_mode() + + req: ValidationMode = explicit_req or env_mode or "warn" + resp: ValidationMode = explicit_resp or env_mode or _default_response_mode() return req, resp diff --git a/tests/test_tools_list_output_schema.py b/tests/test_tools_list_output_schema.py new file mode 100644 index 000000000..676f0c822 --- /dev/null +++ b/tests/test_tools_list_output_schema.py @@ -0,0 +1,178 @@ +"""``tools/list`` advertises ``outputSchema`` for every spec-mapped tool. + +JS parity: the TS SDK ships ``outputSchema`` alongside ``inputSchema`` on +``tools/list`` so MCP clients can validate ``structuredContent`` without +a separate spec lookup. Python previously shipped only ``inputSchema``. + +Two layers covered here: + +* :data:`ADCP_TOOL_DEFINITIONS` (the in-memory inventory) carries an + ``outputSchema`` for every Pydantic-mapped tool. +* The wire-level ``tools/list`` JSON-RPC response surfaces the same + schema on the FastMCP transport. Locks in the integration so a + regression on either side breaks loudly. +""" + +from __future__ import annotations + +import json +from typing import Any + +import httpx +import pytest +from asgi_lifespan import LifespanManager + +from adcp.server import ADCPHandler, create_mcp_server +from adcp.server.mcp_tools import ADCP_TOOL_DEFINITIONS + +# Sample of tools to lock in: one read-only catalog tool, one mutating +# tool with a discriminated union response, one discovery tool. If any +# of these regress to no outputSchema, the SDK has dropped JS parity. +_SAMPLE_TOOLS: tuple[str, ...] = ( + "get_products", + "create_media_buy", + "get_adcp_capabilities", +) + + +# ---------------------------------------------------------------------- +# In-memory inventory +# ---------------------------------------------------------------------- + + +def test_adcp_tool_definitions_has_output_schema_for_sample_tools() -> None: + by_name = {t["name"]: t for t in ADCP_TOOL_DEFINITIONS} + for tool_name in _SAMPLE_TOOLS: + tool_def = by_name[tool_name] + assert "outputSchema" in tool_def, ( + f"{tool_name} is missing outputSchema in ADCP_TOOL_DEFINITIONS — " + "JS parity regression" + ) + schema = tool_def["outputSchema"] + assert isinstance(schema, dict) + assert schema, f"{tool_name} outputSchema is empty" + + +def test_get_products_output_schema_is_object_shape() -> None: + """``get_products`` returns a single response model — its outputSchema + should be a flat ``type: object`` shape, not a discriminated union. + Catches regressions where the response generator silently falls back.""" + by_name = {t["name"]: t for t in ADCP_TOOL_DEFINITIONS} + schema = by_name["get_products"]["outputSchema"] + + assert schema.get("type") == "object" + assert "properties" in schema + # Spec: ``products`` is the load-bearing field. + assert "products" in schema["properties"] + + +def test_create_media_buy_output_schema_includes_response_union() -> None: + """``create_media_buy`` returns a 3-arm discriminated union (success / + error / pending). The advertised outputSchema must surface that — + JS parity ships ``anyOf`` for these.""" + by_name = {t["name"]: t for t in ADCP_TOOL_DEFINITIONS} + schema = by_name["create_media_buy"]["outputSchema"] + + assert "anyOf" in schema, ( + "create_media_buy outputSchema lost its discriminated union — " + "MCP clients can't tell success from error from pending without it" + ) + assert isinstance(schema["anyOf"], list) + assert len(schema["anyOf"]) >= 2 + + +# ---------------------------------------------------------------------- +# Wire-level: tools/list response +# ---------------------------------------------------------------------- + + +class _StubHandler(ADCPHandler): + """Minimal handler — we only need ``tools/list``, not actual tool calls.""" + + +@pytest.fixture +async def mcp_client() -> Any: + handler = _StubHandler() + mcp = create_mcp_server(handler, name="test-output-schema", advertise_all=True) + mcp.settings.stateless_http = True + mcp.settings.json_response = True + mcp.settings.transport_security.allowed_hosts = ["localhost", "127.0.0.1"] + app = mcp.streamable_http_app() + + async with LifespanManager(app): + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://localhost", + follow_redirects=True, + ) as client: + yield client + + +async def _initialize_session(client: httpx.AsyncClient) -> None: + body = { + "jsonrpc": "2.0", + "id": 0, + "method": "initialize", + "params": { + "protocolVersion": "2025-06-18", + "capabilities": {}, + "clientInfo": {"name": "test-client", "version": "1.0"}, + }, + } + headers = { + "content-type": "application/json", + "accept": "application/json, text/event-stream", + } + resp = await client.post("/mcp/", json=body, headers=headers) + assert resp.status_code == 200, resp.text + + +async def _list_tools(client: httpx.AsyncClient) -> dict[str, Any]: + body = {"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}} + headers = { + "content-type": "application/json", + "accept": "application/json, text/event-stream", + } + resp = await client.post("/mcp/", json=body, headers=headers) + assert resp.status_code == 200, resp.text + return _parse_event_stream(resp.text) + + +def _parse_event_stream(body: str) -> dict[str, Any]: + for line in body.splitlines(): + line = line.strip() + if line.startswith("data: "): + return json.loads(line.removeprefix("data: ")) + return json.loads(body) if body.strip() else {} + + +@pytest.mark.asyncio +async def test_tools_list_response_includes_output_schema(mcp_client: Any) -> None: + await _initialize_session(mcp_client) + payload = await _list_tools(mcp_client) + + assert "result" in payload, payload + tools = {t["name"]: t for t in payload["result"]["tools"]} + + for tool_name in _SAMPLE_TOOLS: + assert tool_name in tools, f"{tool_name} not advertised" + tool = tools[tool_name] + assert "outputSchema" in tool, ( + f"tools/list did not advertise outputSchema for {tool_name} — " + "JS parity regression on the wire" + ) + assert tool["outputSchema"], f"{tool_name} outputSchema is empty on the wire" + + +@pytest.mark.asyncio +async def test_tools_list_input_and_output_schemas_are_distinct(mcp_client: Any) -> None: + """Sanity: the request and response schemas describe different + shapes. Catches a regression where outputSchema is mistakenly set + to the inputSchema.""" + await _initialize_session(mcp_client) + payload = await _list_tools(mcp_client) + + tools = {t["name"]: t for t in payload["result"]["tools"]} + tool = tools["get_products"] + + assert tool["inputSchema"] != tool["outputSchema"] diff --git a/tests/test_validation_modes.py b/tests/test_validation_modes.py new file mode 100644 index 000000000..b9956a319 --- /dev/null +++ b/tests/test_validation_modes.py @@ -0,0 +1,155 @@ +"""Tests for ``resolve_validation_modes`` env-var resolution. + +JS parity: the TS SDK supports ``ADCP_VALIDATION_MODE=strict|warn|off`` to +override defaults for both sides at call time. ``ADCP_ENV=prod|production`` +narrowly flips only the response default to ``warn``. + +Resolution order (locked in here so a regression breaks loudly): + +1. Explicit ``requests=`` / ``responses=`` on ``ValidationHookConfig``. +2. ``ADCP_VALIDATION_MODE`` env var (applies to both sides unless + overridden). +3. ``ADCP_ENV=prod|production`` flip on the response side. +4. Hard defaults: ``requests="warn"``, ``responses="strict"``. +""" + +from __future__ import annotations + +from unittest.mock import patch + +import pytest + +from adcp.validation.client_hooks import ( + ValidationHookConfig, + resolve_validation_modes, +) + + +def test_defaults_when_no_env_or_config() -> None: + """``warn``/``strict`` is the documented default surface.""" + with patch.dict("os.environ", {}, clear=True): + req, resp = resolve_validation_modes() + + assert req == "warn" + assert resp == "strict" + + +def test_adcp_env_production_flips_only_response() -> None: + """Lock in the existing narrow behavior — request side stays at the + type default. Tests must mirror the spec, not the implementation.""" + with patch.dict("os.environ", {"ADCP_ENV": "production"}, clear=True): + req, resp = resolve_validation_modes() + + assert req == "warn" + assert resp == "warn" + + +def test_adcp_env_prod_alias_flips_response() -> None: + with patch.dict("os.environ", {"ADCP_ENV": "prod"}, clear=True): + _, resp = resolve_validation_modes() + + assert resp == "warn" + + +@pytest.mark.parametrize("mode", ["strict", "warn", "off"]) +def test_adcp_validation_mode_applies_to_both_sides(mode: str) -> None: + """Single env var sets both sides — matches the TS port.""" + with patch.dict("os.environ", {"ADCP_VALIDATION_MODE": mode}, clear=True): + req, resp = resolve_validation_modes() + + assert req == mode + assert resp == mode + + +def test_adcp_validation_mode_uppercase_is_normalized() -> None: + """Operators tend to TYPE_LIKE_THIS in shell exports; accept it.""" + with patch.dict("os.environ", {"ADCP_VALIDATION_MODE": "STRICT"}, clear=True): + req, resp = resolve_validation_modes() + + assert req == "strict" + assert resp == "strict" + + +def test_adcp_validation_mode_with_whitespace_is_normalized() -> None: + with patch.dict("os.environ", {"ADCP_VALIDATION_MODE": " warn "}, clear=True): + req, resp = resolve_validation_modes() + + assert req == "warn" + assert resp == "warn" + + +def test_invalid_validation_mode_falls_back_to_defaults() -> None: + """Typos in deploy env (``WARNING`` for ``warn``) fall back to defaults + rather than breaking the SDK on the next request.""" + with patch.dict("os.environ", {"ADCP_VALIDATION_MODE": "WARNING"}, clear=True): + req, resp = resolve_validation_modes() + + assert req == "warn" + assert resp == "strict" + + +def test_empty_validation_mode_is_treated_as_unset() -> None: + with patch.dict("os.environ", {"ADCP_VALIDATION_MODE": ""}, clear=True): + req, resp = resolve_validation_modes() + + assert req == "warn" + assert resp == "strict" + + +def test_validation_mode_takes_precedence_over_adcp_env() -> None: + """When both env vars are set, ``ADCP_VALIDATION_MODE`` wins. The + TS port treats ``ADCP_VALIDATION_MODE`` as the explicit override + and ``ADCP_ENV`` as the deploy-environment flip — explicit beats + implicit.""" + env = {"ADCP_VALIDATION_MODE": "strict", "ADCP_ENV": "production"} + with patch.dict("os.environ", env, clear=True): + req, resp = resolve_validation_modes() + + assert req == "strict" + assert resp == "strict" + + +def test_validation_mode_off_overrides_strict_default() -> None: + """High-throughput callers can disable validation entirely via env.""" + with patch.dict("os.environ", {"ADCP_VALIDATION_MODE": "off"}, clear=True): + req, resp = resolve_validation_modes() + + assert req == "off" + assert resp == "off" + + +def test_explicit_config_overrides_validation_mode_env() -> None: + """Explicit ``ValidationHookConfig`` is the highest precedence — + storyboards and compliance runners that pass strict on both sides + must not be silently downgraded by an env var.""" + config = ValidationHookConfig(requests="strict", responses="strict") + with patch.dict("os.environ", {"ADCP_VALIDATION_MODE": "off"}, clear=True): + req, resp = resolve_validation_modes(config) + + assert req == "strict" + assert resp == "strict" + + +def test_explicit_config_per_side_falls_through_to_env() -> None: + """Setting only one side explicitly leaves the other to the env-var + chain. Proves the config fields are independent.""" + config = ValidationHookConfig(requests="off") + with patch.dict("os.environ", {"ADCP_VALIDATION_MODE": "strict"}, clear=True): + req, resp = resolve_validation_modes(config) + + assert req == "off" # explicit wins + assert resp == "strict" # env-var chain + + +def test_resolution_is_evaluated_at_call_time_not_import_time() -> None: + """Tests that mutate env vars must see the new value on the next + call. Module-level caching of the resolved modes would break + ``patch.dict`` and force a reset hook.""" + with patch.dict("os.environ", {"ADCP_VALIDATION_MODE": "strict"}, clear=True): + first = resolve_validation_modes() + + with patch.dict("os.environ", {"ADCP_VALIDATION_MODE": "off"}, clear=True): + second = resolve_validation_modes() + + assert first == ("strict", "strict") + assert second == ("off", "off")