diff --git a/src/vercel/_internal/sandbox/core.py b/src/vercel/_internal/sandbox/core.py index 3b36fa83..69603c3c 100644 --- a/src/vercel/_internal/sandbox/core.py +++ b/src/vercel/_internal/sandbox/core.py @@ -15,7 +15,7 @@ import posixpath import sys import tarfile -from collections.abc import AsyncGenerator, AsyncIterator, Generator +from collections.abc import AsyncGenerator, AsyncIterator, Generator, Mapping from contextlib import asynccontextmanager from datetime import timedelta from importlib.metadata import version as _pkg_version @@ -46,7 +46,6 @@ SandboxServerError, ) from vercel._internal.sandbox.models import ( - ApiNetworkPolicy, CommandFinishedResponse, CommandResponse, CreateSandboxRequest, @@ -325,12 +324,12 @@ async def update_network_policy( self, *, sandbox_id: str, - network_policy: ApiNetworkPolicy, + network_policy: Mapping[str, Any], ) -> SandboxResponse: data = await self._request_client.request_json( "POST", f"/v1/sandboxes/{sandbox_id}/network-policy", - body=JSONBody(network_policy.model_dump(by_alias=True, exclude_none=True)), + body=JSONBody(dict(network_policy)), ) return SandboxResponse.model_validate(data) diff --git a/src/vercel/_internal/sandbox/models.py b/src/vercel/_internal/sandbox/models.py index 0432b1ac..3d72ce67 100644 --- a/src/vercel/_internal/sandbox/models.py +++ b/src/vercel/_internal/sandbox/models.py @@ -7,9 +7,10 @@ from pydantic import ( AliasChoices, BaseModel, + BeforeValidator, ConfigDict, Field, - PrivateAttr, + PlainSerializer, StrictInt, StrictStr, TypeAdapter, @@ -90,124 +91,81 @@ class NetworkPolicyCustom(_CreateModel): NetworkPolicy: TypeAlias = Literal["allow-all", "deny-all"] | NetworkPolicyCustom +_NETWORK_POLICY_ADAPTER: TypeAdapter[NetworkPolicy] = TypeAdapter(NetworkPolicy) -class ApiNetworkInjectionRule(_CreateModel): - """Wire-format injection rule for a single domain.""" +def parse_network_policy(payload: Mapping[str, Any]) -> NetworkPolicy: + """Parse sandbox API network-policy JSON into the public NetworkPolicy type.""" + mode = payload.get("mode") + if mode in ("allow-all", "deny-all"): + return cast(Literal["allow-all", "deny-all"], mode) + if mode != "custom": + raise ValueError("networkPolicy.mode must be 'allow-all', 'deny-all', or 'custom'") - domain: str - headers: dict[str, str] | None = None - header_names: list[str] | None = Field( - default=None, - validation_alias=AliasChoices("header_names", "headerNames"), - serialization_alias="headerNames", - ) + allowed_domains = _get_optional_str_list(payload, "allowed_domains", "allowedDomains") + injection_rules = _get_optional_mapping_list(payload, "injection_rules", "injectionRules") + subnets = _subnets_from_payload(payload) - def to_redacted_headers(self) -> dict[str, str]: - if self.header_names: - redacted: dict[str, str] = {} - lower_to_name: dict[str, str] = {} - for name in self.header_names: - lower_name = name.lower() - previous_name = lower_to_name.get(lower_name) - if previous_name is not None and previous_name != name: - redacted.pop(previous_name, None) - lower_to_name[lower_name] = name - redacted[name] = _REDACTED_HEADER_VALUE - return redacted - return dict.fromkeys(self.headers or {}, _REDACTED_HEADER_VALUE) - - -class ApiNetworkPolicy(_CreateModel): - """Wire-format network policy returned by the Sandbox API.""" - - mode: Literal["allow-all", "deny-all", "custom"] - allowed_domains: list[str] | None = Field( - default=None, - validation_alias=AliasChoices("allowed_domains", "allowedDomains"), - serialization_alias="allowedDomains", - ) - injection_rules: list[ApiNetworkInjectionRule] | None = Field( - default=None, - validation_alias=AliasChoices("injection_rules", "injectionRules"), - serialization_alias="injectionRules", - ) - allowed_cidrs: list[str] | None = Field( - default=None, - validation_alias=AliasChoices("allowed_cidrs", "allowedCIDRs"), - serialization_alias="allowedCIDRs", - ) - denied_cidrs: list[str] | None = Field( - default=None, - validation_alias=AliasChoices("denied_cidrs", "deniedCIDRs"), - serialization_alias="deniedCIDRs", - ) + if not injection_rules: + return NetworkPolicyCustom(allow=allowed_domains or [], subnets=subnets) - @classmethod - def from_payload(cls, payload: Mapping[str, Any]) -> ApiNetworkPolicy: - return cls.model_validate(payload) + allow: dict[str, list[NetworkPolicyRule]] = {domain: [] for domain in allowed_domains or []} + for rule in injection_rules: + domain = rule.get("domain") + if not isinstance(domain, str): + raise TypeError("networkPolicy.injectionRules[].domain must be a string") - @classmethod - def from_network_policy(cls, network_policy: NetworkPolicy) -> ApiNetworkPolicy: - if isinstance(network_policy, str): - return cls(mode=network_policy) - - if isinstance(network_policy.allow, list): - allowed_cidrs = None - denied_cidrs = None - if network_policy.subnets is not None: - allowed_cidrs = network_policy.subnets.allow - denied_cidrs = network_policy.subnets.deny - return cls( - mode="custom", - allowed_domains=list(network_policy.allow), - allowed_cidrs=allowed_cidrs, - denied_cidrs=denied_cidrs, - ) + allow.setdefault(domain, []) + headers = _redacted_headers_from_rule(rule) + if not headers: + continue + allow[domain].append(NetworkPolicyRule(transform=[NetworkTransformer(headers=headers)])) - injection_rules: list[ApiNetworkInjectionRule] = [] - for domain, rules in network_policy.allow.items(): - headers = _merge_rule_headers(rules) - if not headers: - continue - injection_rules.append(ApiNetworkInjectionRule(domain=domain, headers=headers)) - - allowed_cidrs = None - denied_cidrs = None - if network_policy.subnets is not None: - allowed_cidrs = network_policy.subnets.allow - denied_cidrs = network_policy.subnets.deny - - return cls( - mode="custom", - allowed_domains=list(network_policy.allow.keys()), - injection_rules=injection_rules or None, - allowed_cidrs=allowed_cidrs, - denied_cidrs=denied_cidrs, - ) + return NetworkPolicyCustom(allow=allow, subnets=subnets) + + +def serialize_network_policy(network_policy: NetworkPolicy) -> dict[str, Any]: + """Serialize a public NetworkPolicy into sandbox API JSON.""" + network_policy = _NETWORK_POLICY_ADAPTER.validate_python(network_policy) + if isinstance(network_policy, str): + return {"mode": network_policy} + + if isinstance(network_policy.allow, list): + payload: dict[str, Any] = { + "mode": "custom", + "allowedDomains": list(network_policy.allow), + } + _set_subnet_payload(payload, network_policy.subnets) + return payload + + injection_rules: list[dict[str, Any]] = [] + for domain, rules in network_policy.allow.items(): + headers = _merge_rule_headers(rules) + if not headers: + continue + injection_rules.append({"domain": domain, "headers": headers}) + + payload = { + "mode": "custom", + "allowedDomains": list(network_policy.allow.keys()), + } + if injection_rules: + payload["injectionRules"] = injection_rules + _set_subnet_payload(payload, network_policy.subnets) + return payload - def to_network_policy(self) -> NetworkPolicy: - if self.mode in ("allow-all", "deny-all"): - return cast(Literal["allow-all", "deny-all"], self.mode) - - allowed_domains = list(self.allowed_domains or []) - injection_rules = list(self.injection_rules or []) - subnets = _subnets_from_api(self) - - if not injection_rules: - return NetworkPolicyCustom(allow=allowed_domains, subnets=subnets) - - allow: dict[str, list[NetworkPolicyRule]] = {domain: [] for domain in allowed_domains} - for rule in injection_rules: - allow.setdefault(rule.domain, []) - headers = rule.to_redacted_headers() - if not headers: - continue - allow[rule.domain].append( - NetworkPolicyRule(transform=[NetworkTransformer(headers=headers)]) - ) - return NetworkPolicyCustom(allow=allow, subnets=subnets) +def _coerce_network_policy_value(value: object) -> NetworkPolicy: + if isinstance(value, Mapping): + return parse_network_policy(value) + return _NETWORK_POLICY_ADAPTER.validate_python(value) + + +NetworkPolicyCodec: TypeAlias = Annotated[ + NetworkPolicy, + BeforeValidator(_coerce_network_policy_value), + PlainSerializer(serialize_network_policy, return_type=dict[str, Any]), +] def _merge_headers_case_insensitively( @@ -245,20 +203,86 @@ def _merge_rule_transform_headers(rule: NetworkPolicyRule) -> dict[str, str]: return merged -def _subnets_from_api(network_policy: ApiNetworkPolicy) -> NetworkPolicySubnets | None: - if network_policy.allowed_cidrs is None and network_policy.denied_cidrs is None: +def _redacted_headers_from_rule(rule: Mapping[str, Any]) -> dict[str, str]: + header_names = _get_optional_str_list(rule, "header_names", "headerNames") + if header_names: + redacted: dict[str, str] = {} + lower_to_name: dict[str, str] = {} + for name in header_names: + lower_name = name.lower() + previous_name = lower_to_name.get(lower_name) + if previous_name is not None and previous_name != name: + redacted.pop(previous_name, None) + lower_to_name[lower_name] = name + redacted[name] = _REDACTED_HEADER_VALUE + return redacted + + headers = _get_optional_str_mapping(rule, "headers") + return dict.fromkeys(headers, _REDACTED_HEADER_VALUE) + + +def _get_optional_str_list(payload: Mapping[str, Any], *keys: str) -> list[str] | None: + value = _get_alias_value(payload, *keys) + if value is None: + return None + if not isinstance(value, list) or not all(isinstance(item, str) for item in value): + joined = "/".join(keys) + raise TypeError(f"networkPolicy.{joined} must be a list of strings") + return list(value) + + +def _get_optional_mapping_list( + payload: Mapping[str, Any], *keys: str +) -> list[Mapping[str, Any]] | None: + value = _get_alias_value(payload, *keys) + if value is None: + return None + if not isinstance(value, list) or not all(isinstance(item, Mapping) for item in value): + joined = "/".join(keys) + raise TypeError(f"networkPolicy.{joined} must be a list of objects") + return [cast(Mapping[str, Any], item) for item in value] + + +def _get_optional_str_mapping(payload: Mapping[str, Any], *keys: str) -> dict[str, str]: + value = _get_alias_value(payload, *keys) + if value is None: + return {} + if not isinstance(value, Mapping) or not all( + isinstance(key, str) and isinstance(item, str) for key, item in value.items() + ): + joined = "/".join(keys) + raise TypeError(f"networkPolicy.{joined} must be a string map") + return dict(cast(Mapping[str, str], value)) + + +def _get_alias_value(payload: Mapping[str, Any], *keys: str) -> object | None: + for key in keys: + if key in payload: + return payload[key] + return None + + +def _subnets_from_payload(payload: Mapping[str, Any]) -> NetworkPolicySubnets | None: + allowed_cidrs = _get_optional_str_list(payload, "allowed_cidrs", "allowedCIDRs") + denied_cidrs = _get_optional_str_list(payload, "denied_cidrs", "deniedCIDRs") + if allowed_cidrs is None and denied_cidrs is None: return None return NetworkPolicySubnets( - allow=( - list(network_policy.allowed_cidrs) if network_policy.allowed_cidrs is not None else None - ), - deny=( - list(network_policy.denied_cidrs) if network_policy.denied_cidrs is not None else None - ), + allow=allowed_cidrs, + deny=denied_cidrs, ) +def _set_subnet_payload(payload: dict[str, Any], subnets: NetworkPolicySubnets | None) -> None: + if subnets is None: + return + if subnets.allow is not None: + payload["allowedCIDRs"] = subnets.allow + if subnets.deny is not None: + payload["deniedCIDRs"] = subnets.deny + + def _value_error(loc: tuple[str, ...], message: str, input_value: object) -> InitErrorDetails: return cast( InitErrorDetails, @@ -457,26 +481,13 @@ class CreateSandboxRequest(_CreateModel): timeout: int | timedelta | None = None resources: Resources | None = None runtime: StrictStr | None = None - network_policy: ( - ApiNetworkPolicy | NetworkPolicyCustom | Literal["allow-all", "deny-all"] | None - ) = Field( + network_policy: NetworkPolicyCodec | None = Field( default=None, serialization_alias="networkPolicy", ) interactive: bool | None = Field(default=None, serialization_alias="__interactive") env: dict[str, str] | None = None - @field_validator("network_policy", mode="before") - @classmethod - def _coerce_network_policy(cls, value: object) -> ApiNetworkPolicy | None: - if value is None: - return None - if isinstance(value, ApiNetworkPolicy): - return value - if isinstance(value, dict): - return ApiNetworkPolicy.model_validate(value) - return ApiNetworkPolicy.from_network_policy(cast(NetworkPolicy, value)) - @field_validator("timeout", mode="before") @classmethod def _coerce_timeout(cls, value: object) -> int | None: @@ -562,29 +573,16 @@ class Sandbox(BaseModel): cwd: str updated_at: int = Field(alias="updatedAt") interactive_port: int | None = Field(default=None, alias="interactivePort") - network_policy_data: ApiNetworkPolicy | None = Field(default=None, alias="networkPolicy") - _network_policy: NetworkPolicy | None = PrivateAttr(default=None) + network_policy: NetworkPolicyCodec | None = Field(default=None, alias="networkPolicy") - @field_validator("network_policy_data", mode="before") + @field_validator("network_policy", mode="before") @classmethod - def _parse_network_policy_data(cls, value: object) -> ApiNetworkPolicy | None: + def _parse_network_policy(cls, value: object) -> object: if value is None: return None - if isinstance(value, ApiNetworkPolicy): - return value - if isinstance(value, dict): - return ApiNetworkPolicy.from_payload(value) - raise TypeError("networkPolicy must be a mapping") - - def model_post_init(self, __context: object) -> None: - if self.network_policy_data is None: - self._network_policy = None - return - self._network_policy = self.network_policy_data.to_network_policy() - - @property - def network_policy(self) -> NetworkPolicy | None: - return self._network_policy + if isinstance(value, Mapping): + return parse_network_policy(value) + return value class SandboxRoute(BaseModel): diff --git a/src/vercel/sandbox/sandbox.py b/src/vercel/sandbox/sandbox.py index 7af5fe66..41b24a6b 100644 --- a/src/vercel/sandbox/sandbox.py +++ b/src/vercel/sandbox/sandbox.py @@ -13,7 +13,6 @@ from vercel._internal.sandbox.core import AsyncSandboxOpsClient, SyncSandboxOpsClient from vercel._internal.sandbox.errors import SandboxNotFoundError from vercel._internal.sandbox.models import ( - ApiNetworkPolicy, CommandResponse, GitSource, NetworkPolicy, @@ -29,6 +28,7 @@ WriteFile, parse_resources, parse_source, + serialize_network_policy, ) from vercel._internal.sandbox.pagination import SandboxListParams @@ -269,7 +269,7 @@ async def refresh(self) -> None: async def update_network_policy(self, network_policy: NetworkPolicy) -> NetworkPolicy: response = await self.client.update_network_policy( sandbox_id=self.sandbox.id, - network_policy=ApiNetworkPolicy.from_network_policy(network_policy), + network_policy=serialize_network_policy(network_policy), ) self.sandbox = response.sandbox updated_network_policy = self.sandbox.network_policy @@ -702,7 +702,7 @@ def update_network_policy(self, network_policy: NetworkPolicy) -> NetworkPolicy: response = iter_coroutine( self.client.update_network_policy( sandbox_id=self.sandbox.id, - network_policy=ApiNetworkPolicy.from_network_policy(network_policy), + network_policy=serialize_network_policy(network_policy), ) ) self.sandbox = response.sandbox diff --git a/tests/unit/test_sandbox_network_policy_conversion.py b/tests/unit/test_sandbox_network_policy_conversion.py index 1631daa6..4af34b06 100644 --- a/tests/unit/test_sandbox_network_policy_conversion.py +++ b/tests/unit/test_sandbox_network_policy_conversion.py @@ -8,10 +8,7 @@ import pytest from hypothesis import HealthCheck, given, settings, strategies as st -from vercel._internal.sandbox.models import ( - ApiNetworkInjectionRule, - ApiNetworkPolicy, -) +from vercel._internal.sandbox.models import parse_network_policy, serialize_network_policy from vercel.sandbox import ( NetworkPolicy, NetworkPolicyCustom, @@ -192,25 +189,25 @@ class TestNetworkPolicyModes: @pytest.mark.parametrize( ("policy", "api_payload"), [ - ("allow-all", ApiNetworkPolicy(mode="allow-all")), - ("deny-all", ApiNetworkPolicy(mode="deny-all")), + ("allow-all", {"mode": "allow-all"}), + ("deny-all", {"mode": "deny-all"}), ], ids=["allow_all", "deny_all"], ) def test_mode_strings_round_trip_exactly( - self, policy: NetworkPolicy, api_payload: ApiNetworkPolicy + self, policy: NetworkPolicy, api_payload: dict[str, str] ) -> None: - assert ApiNetworkPolicy.from_network_policy(policy) == api_payload - assert api_payload.to_network_policy() == policy + assert serialize_network_policy(policy) == api_payload + assert parse_network_policy(api_payload) == policy @given(st.sampled_from(["allow-all", "deny-all"])) @settings(max_examples=2, deadline=None, suppress_health_check=[HealthCheck.too_slow]) def test_mode_strings_round_trip_property_based(self, policy: NetworkPolicy) -> None: - assert ApiNetworkPolicy.from_network_policy(policy).to_network_policy() == policy + assert parse_network_policy(serialize_network_policy(policy)) == policy @pytest.mark.parametrize("policy", ["allow-all", "deny-all"]) - def test_api_network_policy_codec_methods_round_trip_modes(self, policy: NetworkPolicy) -> None: - assert ApiNetworkPolicy.from_network_policy(policy).to_network_policy() == policy + def test_internal_codec_round_trip_modes(self, policy: NetworkPolicy) -> None: + assert parse_network_policy(serialize_network_policy(policy)) == policy class TestNetworkPolicyExamples: @@ -222,15 +219,15 @@ def test_custom_list_form_converts_with_subnets(self) -> None: deny=["192.168.0.0/16"], ), ) - api_payload = ApiNetworkPolicy( - mode="custom", - allowed_domains=["example.com", "*.example.net"], - allowed_cidrs=["10.0.0.0/8"], - denied_cidrs=["192.168.0.0/16"], - ) + api_payload = { + "mode": "custom", + "allowedDomains": ["example.com", "*.example.net"], + "allowedCIDRs": ["10.0.0.0/8"], + "deniedCIDRs": ["192.168.0.0/16"], + } - assert ApiNetworkPolicy.from_network_policy(policy) == api_payload - assert api_payload.to_network_policy() == policy + assert serialize_network_policy(policy) == api_payload + assert parse_network_policy(api_payload) == policy def test_custom_record_form_converts_with_injection_rules(self) -> None: policy = NetworkPolicyCustom( @@ -240,29 +237,21 @@ def test_custom_record_form_converts_with_injection_rules(self) -> None: ] } ) - api_payload = ApiNetworkPolicy( - mode="custom", - allowed_domains=["example.com"], - injection_rules=[ - ApiNetworkInjectionRule( - domain="example.com", - headers={"X-API-Key": "value-for-x-api-key"}, - ) + api_payload = { + "mode": "custom", + "allowedDomains": ["example.com"], + "injectionRules": [ + {"domain": "example.com", "headers": {"X-API-Key": "value-for-x-api-key"}} ], - ) - api_response = ApiNetworkPolicy( - mode="custom", - allowed_domains=["example.com"], - injection_rules=[ - ApiNetworkInjectionRule( - domain="example.com", - header_names=["X-API-Key"], - ) - ], - ) + } + api_response = { + "mode": "custom", + "allowedDomains": ["example.com"], + "injectionRules": [{"domain": "example.com", "headerNames": ["X-API-Key"]}], + } - assert ApiNetworkPolicy.from_network_policy(policy) == api_payload - assert api_response.to_network_policy() == NetworkPolicyCustom( + assert serialize_network_policy(policy) == api_payload + assert parse_network_policy(api_response) == NetworkPolicyCustom( allow={ "example.com": [ NetworkPolicyRule( @@ -290,44 +279,40 @@ def test_mixed_record_form_merges_headers_per_domain(self) -> None: ], } ) - api_payload = ApiNetworkPolicy( - mode="custom", - allowed_domains=["*", "api.example.com", "docs.example.com"], - injection_rules=[ - ApiNetworkInjectionRule(domain="*", headers={"X-Wild": "value-for-x-wild"}), - ApiNetworkInjectionRule( - domain="api.example.com", - headers={ + api_payload = { + "mode": "custom", + "allowedDomains": ["*", "api.example.com", "docs.example.com"], + "injectionRules": [ + {"domain": "*", "headers": {"X-Wild": "value-for-x-wild"}}, + { + "domain": "api.example.com", + "headers": { "X-Api": "value-for-x-api", "X-Dupe": "value-for-x-dupe", }, - ), - ApiNetworkInjectionRule( - domain="docs.example.com", - headers={ + }, + { + "domain": "docs.example.com", + "headers": { "X-Dupe": "value-for-x-dupe", "X-Docs": "value-for-x-docs", }, - ), + }, ], - ) + } - assert ApiNetworkPolicy.from_network_policy(policy) == api_payload - assert ApiNetworkPolicy( - mode="custom", - allowed_domains=["*", "api.example.com", "docs.example.com"], - injection_rules=[ - ApiNetworkInjectionRule(domain="*", header_names=["X-Wild"]), - ApiNetworkInjectionRule( - domain="api.example.com", - header_names=["X-Api", "X-Dupe"], - ), - ApiNetworkInjectionRule( - domain="docs.example.com", - header_names=["X-Dupe", "X-Docs"], - ), - ], - ).to_network_policy() == NetworkPolicyCustom( + assert serialize_network_policy(policy) == api_payload + assert parse_network_policy( + { + "mode": "custom", + "allowedDomains": ["*", "api.example.com", "docs.example.com"], + "injectionRules": [ + {"domain": "*", "headerNames": ["X-Wild"]}, + {"domain": "api.example.com", "headerNames": ["X-Api", "X-Dupe"]}, + {"domain": "docs.example.com", "headerNames": ["X-Dupe", "X-Docs"]}, + ], + } + ) == NetworkPolicyCustom( allow={ "*": [ NetworkPolicyRule( @@ -360,14 +345,16 @@ class TestNetworkPolicyEdgeCases: def test_empty_rule_arrays_remain_valid_allowed_domains(self) -> None: policy = NetworkPolicyCustom(allow={"example.com": []}) - assert ApiNetworkPolicy.from_network_policy(policy) == ApiNetworkPolicy( - mode="custom", - allowed_domains=["example.com"], - ) - assert ApiNetworkPolicy( - mode="custom", - allowed_domains=["example.com"], - ).to_network_policy() == NetworkPolicyCustom(allow=["example.com"]) + assert serialize_network_policy(policy) == { + "mode": "custom", + "allowedDomains": ["example.com"], + } + assert parse_network_policy( + { + "mode": "custom", + "allowedDomains": ["example.com"], + } + ) == NetworkPolicyCustom(allow=["example.com"]) def test_domains_with_empty_transforms_do_not_emit_injection_rules(self) -> None: policy = NetworkPolicyCustom( @@ -379,16 +366,13 @@ def test_domains_with_empty_transforms_do_not_emit_injection_rules(self) -> None } ) - assert ApiNetworkPolicy.from_network_policy(policy) == ApiNetworkPolicy( - mode="custom", - allowed_domains=["example.com", "api.example.com"], - injection_rules=[ - ApiNetworkInjectionRule( - domain="api.example.com", - headers={"X-Api": "value-for-x-api"}, - ) + assert serialize_network_policy(policy) == { + "mode": "custom", + "allowedDomains": ["example.com", "api.example.com"], + "injectionRules": [ + {"domain": "api.example.com", "headers": {"X-Api": "value-for-x-api"}} ], - ) + } def test_multiple_transforms_for_same_domain_merge_headers(self) -> None: policy = NetworkPolicyCustom( @@ -408,20 +392,20 @@ def test_multiple_transforms_for_same_domain_merge_headers(self) -> None: } ) - assert ApiNetworkPolicy.from_network_policy(policy) == ApiNetworkPolicy( - mode="custom", - allowed_domains=["example.com"], - injection_rules=[ - ApiNetworkInjectionRule( - domain="example.com", - headers={ + assert serialize_network_policy(policy) == { + "mode": "custom", + "allowedDomains": ["example.com"], + "injectionRules": [ + { + "domain": "example.com", + "headers": { "X-First": "one", "X-Dupe": "second", "X-Second": "two", }, - ) + } ], - ) + } def test_multiple_transforms_merge_case_insensitive_header_names(self) -> None: policy = NetworkPolicyCustom( @@ -439,16 +423,16 @@ def test_multiple_transforms_merge_case_insensitive_header_names(self) -> None: } ) - assert ApiNetworkPolicy.from_network_policy(policy) == ApiNetworkPolicy( - mode="custom", - allowed_domains=["example.com"], - injection_rules=[ - ApiNetworkInjectionRule( - domain="example.com", - headers={"x-trace": "second", "X-Other": "other-value"}, - ) + assert serialize_network_policy(policy) == { + "mode": "custom", + "allowedDomains": ["example.com"], + "injectionRules": [ + { + "domain": "example.com", + "headers": {"x-trace": "second", "X-Other": "other-value"}, + } ], - ) + } def test_single_rule_preserves_distinct_case_variants_in_api_headers(self) -> None: policy = NetworkPolicyCustom( @@ -463,16 +447,16 @@ def test_single_rule_preserves_distinct_case_variants_in_api_headers(self) -> No } ) - assert ApiNetworkPolicy.from_network_policy(policy) == ApiNetworkPolicy( - mode="custom", - allowed_domains=["example.com"], - injection_rules=[ - ApiNetworkInjectionRule( - domain="example.com", - headers={"X-Trace": "first", "x-trace": "second"}, - ) + assert serialize_network_policy(policy) == { + "mode": "custom", + "allowedDomains": ["example.com"], + "injectionRules": [ + { + "domain": "example.com", + "headers": {"X-Trace": "first", "x-trace": "second"}, + } ], - ) + } def test_later_rules_replace_earlier_case_variants_for_same_header(self) -> None: policy = NetworkPolicyCustom( @@ -487,28 +471,27 @@ def test_later_rules_replace_earlier_case_variants_for_same_header(self) -> None } ) - assert ApiNetworkPolicy.from_network_policy(policy) == ApiNetworkPolicy( - mode="custom", - allowed_domains=["example.com"], - injection_rules=[ - ApiNetworkInjectionRule( - domain="example.com", - headers={"x-trace": "second", "X-Other": "other"}, - ) + assert serialize_network_policy(policy) == { + "mode": "custom", + "allowedDomains": ["example.com"], + "injectionRules": [ + { + "domain": "example.com", + "headers": {"x-trace": "second", "X-Other": "other"}, + } ], - ) + } def test_api_response_header_names_merge_case_insensitively(self) -> None: - assert ApiNetworkPolicy( - mode="custom", - allowed_domains=["example.com"], - injection_rules=[ - ApiNetworkInjectionRule( - domain="example.com", - header_names=["X-Trace", "x-trace", "X-Other"], - ) - ], - ).to_network_policy() == NetworkPolicyCustom( + assert parse_network_policy( + { + "mode": "custom", + "allowedDomains": ["example.com"], + "injectionRules": [ + {"domain": "example.com", "headerNames": ["X-Trace", "x-trace", "X-Other"]} + ], + } + ) == NetworkPolicyCustom( allow={ "example.com": [ NetworkPolicyRule( @@ -523,16 +506,15 @@ def test_api_response_header_names_merge_case_insensitively(self) -> None: ) def test_api_response_header_names_keep_last_case_variant(self) -> None: - assert ApiNetworkPolicy( - mode="custom", - allowed_domains=["example.com"], - injection_rules=[ - ApiNetworkInjectionRule( - domain="example.com", - header_names=["X-Trace", "x-trace", "X-TRACE"], - ) - ], - ).to_network_policy() == NetworkPolicyCustom( + assert parse_network_policy( + { + "mode": "custom", + "allowedDomains": ["example.com"], + "injectionRules": [ + {"domain": "example.com", "headerNames": ["X-Trace", "x-trace", "X-TRACE"]} + ], + } + ) == NetworkPolicyCustom( allow={ "example.com": [ NetworkPolicyRule( @@ -543,17 +525,19 @@ def test_api_response_header_names_keep_last_case_variant(self) -> None: ) def test_empty_api_header_names_fall_back_to_headers(self) -> None: - assert ApiNetworkPolicy( - mode="custom", - allowed_domains=["example.com"], - injection_rules=[ - ApiNetworkInjectionRule( - domain="example.com", - headers={"X-Trace": "trace-value"}, - header_names=[], - ) - ], - ).to_network_policy() == NetworkPolicyCustom( + assert parse_network_policy( + { + "mode": "custom", + "allowedDomains": ["example.com"], + "injectionRules": [ + { + "domain": "example.com", + "headers": {"X-Trace": "trace-value"}, + "headerNames": [], + } + ], + } + ) == NetworkPolicyCustom( allow={ "example.com": [ NetworkPolicyRule( @@ -564,16 +548,13 @@ def test_empty_api_header_names_fall_back_to_headers(self) -> None: ) def test_injection_rules_for_unknown_domains_surface_in_public_result(self) -> None: - assert ApiNetworkPolicy( - mode="custom", - allowed_domains=["example.com"], - injection_rules=[ - ApiNetworkInjectionRule( - domain="api.example.com", - header_names=["X-Trace"], - ) - ], - ).to_network_policy() == NetworkPolicyCustom( + assert parse_network_policy( + { + "mode": "custom", + "allowedDomains": ["example.com"], + "injectionRules": [{"domain": "api.example.com", "headerNames": ["X-Trace"]}], + } + ) == NetworkPolicyCustom( allow={ "example.com": [], "api.example.com": [ @@ -585,13 +566,15 @@ def test_injection_rules_for_unknown_domains_surface_in_public_result(self) -> N ) def test_missing_subnet_fields_do_not_create_empty_subnets(self) -> None: - assert ApiNetworkPolicy( - mode="custom", - allowed_domains=["example.com"], - ).to_network_policy() == NetworkPolicyCustom(allow=["example.com"]) + assert parse_network_policy( + { + "mode": "custom", + "allowedDomains": ["example.com"], + } + ) == NetworkPolicyCustom(allow=["example.com"]) def test_missing_optional_fields_do_not_raise_conversion_errors(self) -> None: - assert ApiNetworkPolicy(mode="custom").to_network_policy() == NetworkPolicyCustom(allow=[]) + assert parse_network_policy({"mode": "custom"}) == NetworkPolicyCustom(allow=[]) class TestNetworkPolicyGeneratedCases: @@ -600,16 +583,16 @@ class TestNetworkPolicyGeneratedCases: def test_generated_list_form_policies_round_trip_exactly( self, policy: NetworkPolicyCustom ) -> None: - assert _policy_semantics( - ApiNetworkPolicy.from_network_policy(policy).to_network_policy() - ) == _policy_semantics(policy) + assert _policy_semantics(parse_network_policy(serialize_network_policy(policy))) == ( + _policy_semantics(policy) + ) @given(_record_policy_strategy()) @settings(max_examples=25, deadline=None, suppress_health_check=[HealthCheck.too_slow]) def test_generated_record_form_policies_preserve_normalized_semantics( self, policy: NetworkPolicyCustom ) -> None: - round_tripped = ApiNetworkPolicy.from_network_policy(policy).to_network_policy() + round_tripped = parse_network_policy(serialize_network_policy(policy)) assert isinstance(round_tripped, NetworkPolicyCustom) assert _normalized_custom_policy_semantics( @@ -621,7 +604,7 @@ def test_generated_record_form_policies_preserve_normalized_semantics( def test_generated_record_form_policies_decode_to_redacted_headers( self, policy: NetworkPolicyCustom ) -> None: - round_tripped = ApiNetworkPolicy.from_network_policy(policy).to_network_policy() + round_tripped = parse_network_policy(serialize_network_policy(policy)) assert isinstance(round_tripped, NetworkPolicyCustom) assert _header_values(round_tripped) <= {""}