Nightly spec-drift: strict contract tests failing (since 2026-07-05)

[github-actions]

Nightly strict contract tests failed

The scheduled Spec Drift Detection run promoted additive drift to an error.
This means upstream OpenAPI/AsyncAPI changed and the SDK contract suite no longer matches.

To resolve: locally run uv run python scripts/sync_spec.py + uv run python scripts/generate.py, reconcile models/maps, and open a PR with Closes #<this issue>.

• Failing run: https://github.com/TexasCoding/kalshi-python-sdk/actions/runs/28735214139

Failing test output (last 200 lines)

        Plain ``"Foo"`` looks up ``components.schemas.Foo``.
        Dotted ``"Foo.bar.items"`` walks ``components.schemas.Foo.properties.bar.items``
        where each segment after the first navigates ``properties[segment]`` unless the
        segment is the literal ``"items"`` (in which case it descends into the array
        items schema). Used to address inline object schemas that the spec doesn't
        name at the top level — e.g., the per-entry shapes inside ``Batch*Orders*``
        response wrappers.
        """
        schemas = spec.get("components", {}).get("schemas", {})
        head, *rest = schema_name.split(".")
        node = schemas.get(head)
        if node is None:
>           pytest.fail(f"Schema '{head}' not found in OpenAPI spec")
E           Failed: Schema 'Announcement' not found in OpenAPI spec

tests/test_contracts.py:316: Failed
_______________ TestSpecDrift.test_required_drift[Announcement] ________________

self = <tests.test_contracts.TestSpecDrift object at 0x7f7e57facdd0>
entry = ContractEntry(sdk_model='kalshi.models.exchange.Announcement', spec_schema='Announcement', ignored_fields=frozenset(), notes='')

    @pytest.mark.parametrize(
        "entry",
        CONTRACT_MAP,
        ids=[e.sdk_model.rsplit(".", 1)[1] for e in CONTRACT_MAP],
    )
    def test_required_drift(self, entry: ContractEntry) -> None:
        """Fail when SDK fields are optional but spec marks them required.
    
        Hard-fail since #172. Was warn-only while the SDK kept ~134 fields
        Optional[T] | None against spec's required: true. The deviation is
        either resolved (drop None default) or recorded in EXCLUSIONS with
        kind='server_omits_despite_required' citing a demo+prod observation.
        """
>       spec_fields = _get_schema_fields(self.spec, entry.spec_schema)
                      ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

tests/test_contracts.py:783: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
tests/test_contracts.py:347: in _get_schema_fields
    schema = _resolve_schema(spec, schema_name)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

spec = {'openapi': '3.0.0', 'info': {'title': 'Kalshi Trade API Manual Endpoints', 'version': '3.23.0', 'description': 'Manua...endpoints was last validated: GetBalance, GetOrder(s), GetFills, GetPositions', 'tags': ['exchange'], ...}}, ...}, ...}
schema_name = 'Announcement'

    def _resolve_schema(spec: dict[str, Any], schema_name: str) -> dict[str, Any]:
        """Resolve a schema by name, supporting dotted-path syntax for inline schemas.
    
        Plain ``"Foo"`` looks up ``components.schemas.Foo``.
        Dotted ``"Foo.bar.items"`` walks ``components.schemas.Foo.properties.bar.items``
        where each segment after the first navigates ``properties[segment]`` unless the
        segment is the literal ``"items"`` (in which case it descends into the array
        items schema). Used to address inline object schemas that the spec doesn't
        name at the top level — e.g., the per-entry shapes inside ``Batch*Orders*``
        response wrappers.
        """
        schemas = spec.get("components", {}).get("schemas", {})
        head, *rest = schema_name.split(".")
        node = schemas.get(head)
        if node is None:
>           pytest.fail(f"Schema '{head}' not found in OpenAPI spec")
E           Failed: Schema 'Announcement' not found in OpenAPI spec

tests/test_contracts.py:316: Failed
______________________ TestSpecDrift.test_schema_coverage ______________________

self = <tests.test_contracts.TestSpecDrift object at 0x7f7e57e353d0>

    def test_schema_coverage(self) -> None:
        """Every mapped schema must resolve (supports dotted-path syntax)."""
        for entry in CONTRACT_MAP:
            # _resolve_schema fails the test internally if the path doesn't resolve.
>           _resolve_schema(self.spec, entry.spec_schema)

tests/test_contracts.py:796: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

spec = {'openapi': '3.0.0', 'info': {'title': 'Kalshi Trade API Manual Endpoints', 'version': '3.23.0', 'description': 'Manua...endpoints was last validated: GetBalance, GetOrder(s), GetFills, GetPositions', 'tags': ['exchange'], ...}}, ...}, ...}
schema_name = 'Announcement'

    def _resolve_schema(spec: dict[str, Any], schema_name: str) -> dict[str, Any]:
        """Resolve a schema by name, supporting dotted-path syntax for inline schemas.
    
        Plain ``"Foo"`` looks up ``components.schemas.Foo``.
        Dotted ``"Foo.bar.items"`` walks ``components.schemas.Foo.properties.bar.items``
        where each segment after the first navigates ``properties[segment]`` unless the
        segment is the literal ``"items"`` (in which case it descends into the array
        items schema). Used to address inline object schemas that the spec doesn't
        name at the top level — e.g., the per-entry shapes inside ``Batch*Orders*``
        response wrappers.
        """
        schemas = spec.get("components", {}).get("schemas", {})
        head, *rest = schema_name.split(".")
        node = schemas.get(head)
        if node is None:
>           pytest.fail(f"Schema '{head}' not found in OpenAPI spec")
E           Failed: Schema 'Announcement' not found in OpenAPI spec

tests/test_contracts.py:316: Failed
_______ TestRequestParamDrift.test_sync_params_match_spec[announcements] _______

self = <tests.test_contracts.TestRequestParamDrift object at 0x7f7e57f67740>
entry = MethodEndpointEntry(sdk_method='kalshi.resources.exchange.ExchangeResource.announcements', http_method='GET', path_template='/exchange/announcements', request_body_schema=None)

    def test_sync_params_match_spec(self, entry: MethodEndpointEntry) -> None:
>       self._assert_params_match(entry, async_=False)

tests/test_contracts.py:1121: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
tests/test_contracts.py:1148: in _assert_params_match
    spec_params_list = _resolve_path_params(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

spec = {'openapi': '3.0.0', 'info': {'title': 'Kalshi Trade API Manual Endpoints', 'version': '3.23.0', 'description': 'Manua...endpoints was last validated: GetBalance, GetOrder(s), GetFills, GetPositions', 'tags': ['exchange'], ...}}, ...}, ...}
path_template = '/exchange/announcements', http_method = 'GET'
max_ref_depth = 8

    def _resolve_path_params(
        spec: dict[str, Any],
        path_template: str,
        http_method: str,
        *,
        max_ref_depth: int = 8,
    ) -> list[dict[str, Any]]:
        """Return the merged list of parameter objects for a specific operation.
    
        Walks BOTH path-level ``parameters`` (shared across operations on that path)
        AND operation-level ``parameters`` (specific to the given HTTP method).
        Operation-level entries override path-level by the ``(name, in)`` key —
        OpenAPI permits the same parameter name in different locations (e.g., a
        path param and a query param both named ``ticker``), so dedup must consider
        both dimensions.
        Any ``$ref`` entries are resolved via ``_resolve_ref``.
    
        Raises ``KeyError`` if the path does not exist in the spec, or if the
        operation (HTTP method) does not exist on that path. Fail loud: a
        mismatched ``METHOD_ENDPOINT_MAP`` entry should surface at test time, not
        silently return partial data.
        """
        paths = spec.get("paths", {})
        if path_template not in paths:
>           raise KeyError(f"path {path_template!r} not found in spec")
E           KeyError: "path '/exchange/announcements' not found in spec"

tests/_contract_support.py:1736: KeyError
______ TestRequestParamDrift.test_async_params_match_spec[announcements] _______

self = <tests.test_contracts.TestRequestParamDrift object at 0x7f7e57e8fe90>
entry = MethodEndpointEntry(sdk_method='kalshi.resources.exchange.ExchangeResource.announcements', http_method='GET', path_template='/exchange/announcements', request_body_schema=None)

    def test_async_params_match_spec(self, entry: MethodEndpointEntry) -> None:
>       self._assert_params_match(entry, async_=True)

tests/test_contracts.py:1124: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
tests/test_contracts.py:1148: in _assert_params_match
    spec_params_list = _resolve_path_params(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

spec = {'openapi': '3.0.0', 'info': {'title': 'Kalshi Trade API Manual Endpoints', 'version': '3.23.0', 'description': 'Manua...endpoints was last validated: GetBalance, GetOrder(s), GetFills, GetPositions', 'tags': ['exchange'], ...}}, ...}, ...}
path_template = '/exchange/announcements', http_method = 'GET'
max_ref_depth = 8

    def _resolve_path_params(
        spec: dict[str, Any],
        path_template: str,
        http_method: str,
        *,
        max_ref_depth: int = 8,
    ) -> list[dict[str, Any]]:
        """Return the merged list of parameter objects for a specific operation.
    
        Walks BOTH path-level ``parameters`` (shared across operations on that path)
        AND operation-level ``parameters`` (specific to the given HTTP method).
        Operation-level entries override path-level by the ``(name, in)`` key —
        OpenAPI permits the same parameter name in different locations (e.g., a
        path param and a query param both named ``ticker``), so dedup must consider
        both dimensions.
        Any ``$ref`` entries are resolved via ``_resolve_ref``.
    
        Raises ``KeyError`` if the path does not exist in the spec, or if the
        operation (HTTP method) does not exist on that path. Fail loud: a
        mismatched ``METHOD_ENDPOINT_MAP`` entry should surface at test time, not
        silently return partial data.
        """
        paths = spec.get("paths", {})
        if path_template not in paths:
>           raise KeyError(f"path {path_template!r} not found in spec")
E           KeyError: "path '/exchange/announcements' not found in spec"

tests/_contract_support.py:1736: KeyError
=========================== short test summary info ============================
FAILED tests/test_contracts.py::TestSpecDrift::test_additive_drift[Announcement] - Failed: Schema 'Announcement' not found in OpenAPI spec
FAILED tests/test_contracts.py::TestSpecDrift::test_required_drift[Announcement] - Failed: Schema 'Announcement' not found in OpenAPI spec
FAILED tests/test_contracts.py::TestSpecDrift::test_schema_coverage - Failed: Schema 'Announcement' not found in OpenAPI spec
FAILED tests/test_contracts.py::TestRequestParamDrift::test_sync_params_match_spec[announcements] - KeyError: "path '/exchange/announcements' not found in spec"
FAILED tests/test_contracts.py::TestRequestParamDrift::test_async_params_match_spec[announcements] - KeyError: "path '/exchange/announcements' not found in spec"
================== 5 failed, 649 passed in 243.69s (0:04:03) ===================

[github-actions]

Still failing as of 2026-07-06 (run).

Latest failing output (last 200 lines)

        Plain ``"Foo"`` looks up ``components.schemas.Foo``.
        Dotted ``"Foo.bar.items"`` walks ``components.schemas.Foo.properties.bar.items``
        where each segment after the first navigates ``properties[segment]`` unless the
        segment is the literal ``"items"`` (in which case it descends into the array
        items schema). Used to address inline object schemas that the spec doesn't
        name at the top level — e.g., the per-entry shapes inside ``Batch*Orders*``
        response wrappers.
        """
        schemas = spec.get("components", {}).get("schemas", {})
        head, *rest = schema_name.split(".")
        node = schemas.get(head)
        if node is None:
>           pytest.fail(f"Schema '{head}' not found in OpenAPI spec")
E           Failed: Schema 'Announcement' not found in OpenAPI spec

tests/test_contracts.py:316: Failed
_______________ TestSpecDrift.test_required_drift[Announcement] ________________

self = <tests.test_contracts.TestSpecDrift object at 0x7f23b6784d70>
entry = ContractEntry(sdk_model='kalshi.models.exchange.Announcement', spec_schema='Announcement', ignored_fields=frozenset(), notes='')

    @pytest.mark.parametrize(
        "entry",
        CONTRACT_MAP,
        ids=[e.sdk_model.rsplit(".", 1)[1] for e in CONTRACT_MAP],
    )
    def test_required_drift(self, entry: ContractEntry) -> None:
        """Fail when SDK fields are optional but spec marks them required.
    
        Hard-fail since #172. Was warn-only while the SDK kept ~134 fields
        Optional[T] | None against spec's required: true. The deviation is
        either resolved (drop None default) or recorded in EXCLUSIONS with
        kind='server_omits_despite_required' citing a demo+prod observation.
        """
>       spec_fields = _get_schema_fields(self.spec, entry.spec_schema)
                      ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

tests/test_contracts.py:783: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
tests/test_contracts.py:347: in _get_schema_fields
    schema = _resolve_schema(spec, schema_name)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

spec = {'openapi': '3.0.0', 'info': {'title': 'Kalshi Trade API Manual Endpoints', 'version': '3.23.0', 'description': 'Manua...endpoints was last validated: GetBalance, GetOrder(s), GetFills, GetPositions', 'tags': ['exchange'], ...}}, ...}, ...}
schema_name = 'Announcement'

    def _resolve_schema(spec: dict[str, Any], schema_name: str) -> dict[str, Any]:
        """Resolve a schema by name, supporting dotted-path syntax for inline schemas.
    
        Plain ``"Foo"`` looks up ``components.schemas.Foo``.
        Dotted ``"Foo.bar.items"`` walks ``components.schemas.Foo.properties.bar.items``
        where each segment after the first navigates ``properties[segment]`` unless the
        segment is the literal ``"items"`` (in which case it descends into the array
        items schema). Used to address inline object schemas that the spec doesn't
        name at the top level — e.g., the per-entry shapes inside ``Batch*Orders*``
        response wrappers.
        """
        schemas = spec.get("components", {}).get("schemas", {})
        head, *rest = schema_name.split(".")
        node = schemas.get(head)
        if node is None:
>           pytest.fail(f"Schema '{head}' not found in OpenAPI spec")
E           Failed: Schema 'Announcement' not found in OpenAPI spec

tests/test_contracts.py:316: Failed
______________________ TestSpecDrift.test_schema_coverage ______________________

self = <tests.test_contracts.TestSpecDrift object at 0x7f23b660d370>

    def test_schema_coverage(self) -> None:
        """Every mapped schema must resolve (supports dotted-path syntax)."""
        for entry in CONTRACT_MAP:
            # _resolve_schema fails the test internally if the path doesn't resolve.
>           _resolve_schema(self.spec, entry.spec_schema)

tests/test_contracts.py:796: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

spec = {'openapi': '3.0.0', 'info': {'title': 'Kalshi Trade API Manual Endpoints', 'version': '3.23.0', 'description': 'Manua...endpoints was last validated: GetBalance, GetOrder(s), GetFills, GetPositions', 'tags': ['exchange'], ...}}, ...}, ...}
schema_name = 'Announcement'

    def _resolve_schema(spec: dict[str, Any], schema_name: str) -> dict[str, Any]:
        """Resolve a schema by name, supporting dotted-path syntax for inline schemas.
    
        Plain ``"Foo"`` looks up ``components.schemas.Foo``.
        Dotted ``"Foo.bar.items"`` walks ``components.schemas.Foo.properties.bar.items``
        where each segment after the first navigates ``properties[segment]`` unless the
        segment is the literal ``"items"`` (in which case it descends into the array
        items schema). Used to address inline object schemas that the spec doesn't
        name at the top level — e.g., the per-entry shapes inside ``Batch*Orders*``
        response wrappers.
        """
        schemas = spec.get("components", {}).get("schemas", {})
        head, *rest = schema_name.split(".")
        node = schemas.get(head)
        if node is None:
>           pytest.fail(f"Schema '{head}' not found in OpenAPI spec")
E           Failed: Schema 'Announcement' not found in OpenAPI spec

tests/test_contracts.py:316: Failed
_______ TestRequestParamDrift.test_sync_params_match_spec[announcements] _______

self = <tests.test_contracts.TestRequestParamDrift object at 0x7f23b674f680>
entry = MethodEndpointEntry(sdk_method='kalshi.resources.exchange.ExchangeResource.announcements', http_method='GET', path_template='/exchange/announcements', request_body_schema=None)

    def test_sync_params_match_spec(self, entry: MethodEndpointEntry) -> None:
>       self._assert_params_match(entry, async_=False)

tests/test_contracts.py:1121: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
tests/test_contracts.py:1148: in _assert_params_match
    spec_params_list = _resolve_path_params(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

spec = {'openapi': '3.0.0', 'info': {'title': 'Kalshi Trade API Manual Endpoints', 'version': '3.23.0', 'description': 'Manua...endpoints was last validated: GetBalance, GetOrder(s), GetFills, GetPositions', 'tags': ['exchange'], ...}}, ...}, ...}
path_template = '/exchange/announcements', http_method = 'GET'
max_ref_depth = 8

    def _resolve_path_params(
        spec: dict[str, Any],
        path_template: str,
        http_method: str,
        *,
        max_ref_depth: int = 8,
    ) -> list[dict[str, Any]]:
        """Return the merged list of parameter objects for a specific operation.
    
        Walks BOTH path-level ``parameters`` (shared across operations on that path)
        AND operation-level ``parameters`` (specific to the given HTTP method).
        Operation-level entries override path-level by the ``(name, in)`` key —
        OpenAPI permits the same parameter name in different locations (e.g., a
        path param and a query param both named ``ticker``), so dedup must consider
        both dimensions.
        Any ``$ref`` entries are resolved via ``_resolve_ref``.
    
        Raises ``KeyError`` if the path does not exist in the spec, or if the
        operation (HTTP method) does not exist on that path. Fail loud: a
        mismatched ``METHOD_ENDPOINT_MAP`` entry should surface at test time, not
        silently return partial data.
        """
        paths = spec.get("paths", {})
        if path_template not in paths:
>           raise KeyError(f"path {path_template!r} not found in spec")
E           KeyError: "path '/exchange/announcements' not found in spec"

tests/_contract_support.py:1736: KeyError
______ TestRequestParamDrift.test_async_params_match_spec[announcements] _______

self = <tests.test_contracts.TestRequestParamDrift object at 0x7f23b64efdd0>
entry = MethodEndpointEntry(sdk_method='kalshi.resources.exchange.ExchangeResource.announcements', http_method='GET', path_template='/exchange/announcements', request_body_schema=None)

    def test_async_params_match_spec(self, entry: MethodEndpointEntry) -> None:
>       self._assert_params_match(entry, async_=True)

tests/test_contracts.py:1124: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
tests/test_contracts.py:1148: in _assert_params_match
    spec_params_list = _resolve_path_params(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

spec = {'openapi': '3.0.0', 'info': {'title': 'Kalshi Trade API Manual Endpoints', 'version': '3.23.0', 'description': 'Manua...endpoints was last validated: GetBalance, GetOrder(s), GetFills, GetPositions', 'tags': ['exchange'], ...}}, ...}, ...}
path_template = '/exchange/announcements', http_method = 'GET'
max_ref_depth = 8

    def _resolve_path_params(
        spec: dict[str, Any],
        path_template: str,
        http_method: str,
        *,
        max_ref_depth: int = 8,
    ) -> list[dict[str, Any]]:
        """Return the merged list of parameter objects for a specific operation.
    
        Walks BOTH path-level ``parameters`` (shared across operations on that path)
        AND operation-level ``parameters`` (specific to the given HTTP method).
        Operation-level entries override path-level by the ``(name, in)`` key —
        OpenAPI permits the same parameter name in different locations (e.g., a
        path param and a query param both named ``ticker``), so dedup must consider
        both dimensions.
        Any ``$ref`` entries are resolved via ``_resolve_ref``.
    
        Raises ``KeyError`` if the path does not exist in the spec, or if the
        operation (HTTP method) does not exist on that path. Fail loud: a
        mismatched ``METHOD_ENDPOINT_MAP`` entry should surface at test time, not
        silently return partial data.
        """
        paths = spec.get("paths", {})
        if path_template not in paths:
>           raise KeyError(f"path {path_template!r} not found in spec")
E           KeyError: "path '/exchange/announcements' not found in spec"

tests/_contract_support.py:1736: KeyError
=========================== short test summary info ============================
FAILED tests/test_contracts.py::TestSpecDrift::test_additive_drift[Announcement] - Failed: Schema 'Announcement' not found in OpenAPI spec
FAILED tests/test_contracts.py::TestSpecDrift::test_required_drift[Announcement] - Failed: Schema 'Announcement' not found in OpenAPI spec
FAILED tests/test_contracts.py::TestSpecDrift::test_schema_coverage - Failed: Schema 'Announcement' not found in OpenAPI spec
FAILED tests/test_contracts.py::TestRequestParamDrift::test_sync_params_match_spec[announcements] - KeyError: "path '/exchange/announcements' not found in spec"
FAILED tests/test_contracts.py::TestRequestParamDrift::test_async_params_match_spec[announcements] - KeyError: "path '/exchange/announcements' not found in spec"
================== 5 failed, 649 passed in 240.56s (0:04:00) ===================

[github-actions]

Still failing as of 2026-07-07 (run).

Latest failing output (last 200 lines)

        operation (HTTP method) does not exist on that path. Fail loud: a
        mismatched ``METHOD_ENDPOINT_MAP`` entry should surface at test time, not
        silently return partial data.
        """
        paths = spec.get("paths", {})
        if path_template not in paths:
>           raise KeyError(f"path {path_template!r} not found in spec")
E           KeyError: "path '/exchange/announcements' not found in spec"

tests/_contract_support.py:1736: KeyError
_________ TestRequestParamDrift.test_async_params_match_spec[balance] __________

self = <tests.test_contracts.TestRequestParamDrift object at 0x7efddd9fe2d0>
entry = MethodEndpointEntry(sdk_method='kalshi.resources.portfolio.PortfolioResource.balance', http_method='GET', path_template='/portfolio/balance', request_body_schema=None)

    def test_async_params_match_spec(self, entry: MethodEndpointEntry) -> None:
>       self._assert_params_match(entry, async_=True)

tests/test_contracts.py:1124: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.test_contracts.TestRequestParamDrift object at 0x7efddd9fe2d0>
entry = MethodEndpointEntry(sdk_method='kalshi.resources.portfolio.PortfolioResource.balance', http_method='GET', path_template='/portfolio/balance', request_body_schema=None)
async_ = True

    def _assert_params_match(
        self,
        entry: MethodEndpointEntry,
        *,
        async_: bool,
    ) -> None:
        sdk_fqn = entry.sdk_method
        if async_:
            parts = sdk_fqn.split(".")
            parts[-2] = f"Async{parts[-2]}"
            sdk_fqn = ".".join(parts)
            # Assert sibling exists
            module_name = ".".join(parts[:-2])
            cls_name = parts[-2]
            module = importlib.import_module(module_name)
            assert hasattr(module, cls_name), f"Missing async sibling {cls_name} in {module_name}"
    
        sdk_params = _signature_params(sdk_fqn)
        # #253: ``extra_headers`` is an SDK-meta kwarg (per-request header overrides)
        # threaded onto every public resource method. It is never part of the
        # OpenAPI spec, so exclude it from the drift comparison globally.
        sdk_params.discard("extra_headers")
        spec_params_list = _resolve_path_params(
            self.spec,
            entry.path_template,
            entry.http_method,
        )
        spec_params = {p["name"] for p in spec_params_list if p.get("in") in ("query", "path")}
        # Spec path params like ``{order_id}`` should appear in the path
        # template placeholders too; union them in so the test catches
        # cases where a spec operation declares a path param that the
        # SDK method signature doesn't accept.
        template_params = _path_params_from_template(entry.path_template)
        spec_params |= template_params
    
        # EXCLUSIONS is indexed by the SYNC method fqn; async tests reuse
        # the same allowlist entries.
        lookup_fqn = entry.sdk_method
    
        # ADD drift: spec has it, SDK missing
        missing = spec_params - sdk_params
        missing_unallowed = {p for p in missing if (lookup_fqn, p) not in EXCLUSIONS}
        # REMOVE drift: SDK has it, spec doesn't
        extra = sdk_params - spec_params
        extra_unallowed = {p for p in extra if (lookup_fqn, p) not in EXCLUSIONS}
    
        errors: list[str] = []
        if missing_unallowed:
            errors.append(f"[ADD drift] spec has params SDK missing: {sorted(missing_unallowed)}")
        if extra_unallowed:
            errors.append(f"[REMOVE drift] SDK has params spec doesn't: {sorted(extra_unallowed)}")
        if errors:
>           pytest.fail(
                f"{sdk_fqn} <-> {entry.http_method} {entry.path_template}\n"
                + "\n".join(errors)
                + f"\n(Allowlist via EXCLUSIONS[({lookup_fqn!r}, '...')])"
            )
E           Failed: kalshi.resources.portfolio.AsyncPortfolioResource.balance <-> GET /portfolio/balance
E           [ADD drift] spec has params SDK missing: ['exchange_index']
E           (Allowlist via EXCLUSIONS[('kalshi.resources.portfolio.PortfolioResource.balance', '...')])

tests/test_contracts.py:1178: Failed
___ TestRequestBodyDrift.test_body_properties_match_spec[transfer_position] ____

self = <tests.test_contracts.TestRequestBodyDrift object at 0x7efdddafd1c0>
entry = MethodEndpointEntry(sdk_method='kalshi.resources.subaccounts.SubaccountsResource.transfer_position', http_method='POST...lio/subaccounts/positions/transfer', request_body_schema='#/components/schemas/ApplySubaccountPositionTransferRequest')

    def test_body_properties_match_spec(
        self,
        entry: MethodEndpointEntry,
    ) -> None:
        assert entry.request_body_schema is not None
        model_fqn = BODY_MODEL_MAP.get(entry.request_body_schema)
        assert model_fqn is not None, (
            f"No request model registered in BODY_MODEL_MAP for "
            f"{entry.request_body_schema!r}. Add the mapping."
        )
        model_cls = _get_model_class_from_fqn(model_fqn)
    
        schema = _resolve_request_body_schema(
            self.spec,
            entry.path_template,
            entry.http_method,
        )
        assert schema is not None, (
            f"Spec has no body schema for {entry.http_method} {entry.path_template}"
        )
    
        errors: list[str] = []
        _check_model_drift(
            model_cls=model_cls,
            model_fqn=model_fqn,
            spec_schema=schema,
            errors=errors,
        )
    
        # Issue #52: also drift-check nested BaseModel fields (one level deep).
        for field_name, nested_cls, nested_schema in _iter_nested_body_models(
            model_cls,
            schema,
            self.spec,
        ):
            nested_fqn = f"{nested_cls.__module__}.{nested_cls.__qualname__}"
            _check_model_drift(
                model_cls=nested_cls,
                model_fqn=nested_fqn,
                spec_schema=nested_schema,
                errors=errors,
                context=f"{model_fqn}.{field_name}",
            )
    
        if errors:
>           pytest.fail(f"{model_fqn} <-> {entry.request_body_schema}\n" + "\n".join(errors))
E           Failed: kalshi.models.subaccounts.ApplySubaccountPositionTransferRequest <-> #/components/schemas/ApplySubaccountPositionTransferRequest
E           [ADD drift] spec has properties SDK model missing: ['price']
E           [REMOVE drift] SDK model emits wire names spec doesn't: ['price_cents']

tests/test_contracts.py:1478: Failed
__________ TestPerpsScmSpecDrift.test_response_drift[ObligationEntry] __________

self = <tests.test_contracts.TestPerpsScmSpecDrift object at 0x7efddda39280>
entry = ContractEntry(sdk_model='kalshi.perps.klear.models.margin.ObligationEntry', spec_schema='ObligationEntry', ignored_fields=frozenset(), notes='Folds spec allOf[ObligationInfo, inline] into one model')

    def test_response_drift(self, entry: ContractEntry | None) -> None:
        if entry is None:
            pytest.skip("no perps SCM response models registered yet")
>       _assert_perps_response_drift(entry, self.spec)

tests/test_contracts.py:1964: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

entry = ContractEntry(sdk_model='kalshi.perps.klear.models.margin.ObligationEntry', spec_schema='ObligationEntry', ignored_fields=frozenset(), notes='Folds spec allOf[ObligationInfo, inline] into one model')
spec = {'openapi': '3.0.0', 'info': {'title': 'Kalshi Self-Clearing Member API', 'version': '0.0.1', 'description': 'Klear AP...}, {'url': 'https://demo-api.kalshi.co/klear-api/v1', 'description': 'Demo'}], 'security': [{'kalshiBearer': []}], ...}

    def _assert_perps_response_drift(entry: ContractEntry, spec: dict[str, Any]) -> None:
        """Response-side drift assertion for one perps contract entry."""
        spec_fields = _get_schema_fields(spec, entry.spec_schema)
        model_class = _get_sdk_model_class(entry.sdk_model)
        additive, required_issues = _classify_drift(
            entry, spec, spec_fields, model_class, exclusions=PERPS_EXCLUSIONS
        )
        problems = additive + required_issues
        if problems:
>           pytest.fail(
                f"Perps spec drift in {entry.sdk_model}:\n"
                + "\n".join(f"  - {p}" for p in problems)
            )
E           Failed: Perps spec drift in kalshi.perps.klear.models.margin.ObligationEntry:
E             - Spec field 'asset_class' has no SDK mapping

tests/test_contracts.py:1839: Failed
=========================== short test summary info ============================
FAILED tests/test_contracts.py::TestSpecDrift::test_additive_drift[Announcement] - Failed: Schema 'Announcement' not found in OpenAPI spec
FAILED tests/test_contracts.py::TestSpecDrift::test_additive_drift[SubaccountTransfer] - Failed: Additive drift in kalshi.models.subaccounts.SubaccountTransfer:
  - Spec field 'price' has no SDK mapping
FAILED tests/test_contracts.py::TestSpecDrift::test_additive_drift[SubaccountNettingConfig] - Failed: Additive drift in kalshi.models.subaccounts.SubaccountNettingConfig:
  - Spec field 'exchange_index' has no SDK mapping
FAILED tests/test_contracts.py::TestSpecDrift::test_additive_drift[ApplySubaccountPositionTransferRequest] - Failed: Additive drift in kalshi.models.subaccounts.ApplySubaccountPositionTransferRequest:
  - Spec field 'price' has no SDK mapping
FAILED tests/test_contracts.py::TestSpecDrift::test_required_drift[Announcement] - Failed: Schema 'Announcement' not found in OpenAPI spec
FAILED tests/test_contracts.py::TestSpecDrift::test_schema_coverage - Failed: Schema 'Announcement' not found in OpenAPI spec
FAILED tests/test_contracts.py::TestRequestParamDrift::test_sync_params_match_spec[announcements] - KeyError: "path '/exchange/announcements' not found in spec"
FAILED tests/test_contracts.py::TestRequestParamDrift::test_sync_params_match_spec[balance] - Failed: kalshi.resources.portfolio.PortfolioResource.balance <-> GET /portfolio/balance
[ADD drift] spec has params SDK missing: ['exchange_index']
(Allowlist via EXCLUSIONS[('kalshi.resources.portfolio.PortfolioResource.balance', '...')])
FAILED tests/test_contracts.py::TestRequestParamDrift::test_async_params_match_spec[announcements] - KeyError: "path '/exchange/announcements' not found in spec"
FAILED tests/test_contracts.py::TestRequestParamDrift::test_async_params_match_spec[balance] - Failed: kalshi.resources.portfolio.AsyncPortfolioResource.balance <-> GET /portfolio/balance
[ADD drift] spec has params SDK missing: ['exchange_index']
(Allowlist via EXCLUSIONS[('kalshi.resources.portfolio.PortfolioResource.balance', '...')])
FAILED tests/test_contracts.py::TestRequestBodyDrift::test_body_properties_match_spec[transfer_position] - Failed: kalshi.models.subaccounts.ApplySubaccountPositionTransferRequest <-> #/components/schemas/ApplySubaccountPositionTransferRequest
[ADD drift] spec has properties SDK model missing: ['price']
[REMOVE drift] SDK model emits wire names spec doesn't: ['price_cents']
FAILED tests/test_contracts.py::TestPerpsScmSpecDrift::test_response_drift[ObligationEntry] - Failed: Perps spec drift in kalshi.perps.klear.models.margin.ObligationEntry:
  - Spec field 'asset_class' has no SDK mapping
================== 12 failed, 642 passed in 236.06s (0:03:56) ==================