Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions src/adcp/server/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -178,6 +178,7 @@ async def get_products(params, context=None):
register_test_controller,
)
from adcp.server.tmp import TmpHandler
from adcp.validation.envelope import UnsupportedVersionError, resolve_requested_adcp_version

__all__ = [
# Base classes
Expand Down Expand Up @@ -267,6 +268,7 @@ async def get_products(params, context=None):
# DX helpers
"AccountError",
"STANDARD_ERROR_CODES",
"UnsupportedVersionError",
"adcp_error",
"adcp_server",
"ADCPServerBuilder",
Expand All @@ -275,6 +277,7 @@ async def get_products(params, context=None):
"is_terminal_status",
"resolve_account",
"resolve_account_into_context",
"resolve_requested_adcp_version",
"valid_actions_for_status",
# Response builders
"activate_signal_response",
Expand Down
31 changes: 31 additions & 0 deletions src/adcp/validation/envelope.py
Original file line number Diff line number Diff line change
Expand Up @@ -129,3 +129,34 @@ def detect_wire_version(
return max(candidates, key=lambda v: int(v.split(".")[1].split("-")[0]))

return None


def resolve_requested_adcp_version(
payload: Any,
*,
supported: tuple[str, ...] = SUPPORTED_WIRE_VERSIONS,
default: str = DEFAULT_UNNEGOTIATED_ADCP_VERSION,
) -> str:
"""Return the AdCP release this request should be served as.

This is the public, adopter-facing version of the server dispatcher's
envelope-field resolution contract:

* explicit ``adcp_version`` wins and is normalized to release precision;
* legacy ``adcp_major_version`` maps to that major's base minor when
available, preserving pre-3.1 response-envelope semantics;
* no version signal resolves to ``default`` (currently ``"3.0"``).

The helper is intentionally payload-only. It does not run the dispatcher's
tool-specific legacy shape probes for adapter-routed versions such as 2.5.

Unsupported explicit claims, or an unnegotiated request whose default is
not in ``supported``, raise :class:`UnsupportedVersionError`, just like
:func:`detect_wire_version`.
"""
resolved = detect_wire_version(payload, supported=supported)
if resolved is not None:
return resolved
if default not in supported:
raise UnsupportedVersionError(default, supported)
return default
34 changes: 33 additions & 1 deletion tests/test_validation_envelope.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,11 @@

import pytest

from adcp.validation.envelope import UnsupportedVersionError, detect_wire_version
from adcp.validation.envelope import (
UnsupportedVersionError,
detect_wire_version,
resolve_requested_adcp_version,
)

# A canned supported set keeps the test independent of COMPATIBLE_ADCP_VERSIONS
# drift over time. Pinning the set inside the test also documents what
Expand Down Expand Up @@ -61,6 +65,34 @@ def test_neither_field_returns_none_fallback_to_sdk_pin() -> None:
assert detect_wire_version({"other_field": "x"}, supported=_SUPPORTED) is None


def test_resolve_requested_adcp_version_defaults_unnegotiated_to_30() -> None:
assert resolve_requested_adcp_version({}, supported=_SUPPORTED) == "3.0"


def test_resolve_requested_adcp_version_preserves_explicit_31() -> None:
payload = {"adcp_version": "3.1.0", "adcp_major_version": 3}
assert resolve_requested_adcp_version(payload, supported=_SUPPORTED) == "3.1"


def test_resolve_requested_adcp_version_public_server_import() -> None:
from adcp.server import (
UnsupportedVersionError as ServerUnsupportedVersionError,
)
from adcp.server import (
resolve_requested_adcp_version as server_resolve,
)

assert server_resolve({"adcp_major_version": 3}, supported=_SUPPORTED) == "3.0"
assert ServerUnsupportedVersionError is UnsupportedVersionError


def test_resolve_requested_adcp_version_rejects_unsupported_default() -> None:
with pytest.raises(UnsupportedVersionError) as exc_info:
resolve_requested_adcp_version({}, supported=("3.1",))
assert exc_info.value.wire_value == "3.0"
assert exc_info.value.supported == ("3.1",)


def test_non_dict_payload_returns_none() -> None:
"""Non-dict payloads can't carry the envelope — caller skips."""
assert detect_wire_version("not_a_dict", supported=_SUPPORTED) is None
Expand Down
Loading