From e656610a80962db9477f849233f69c179df3ac2b Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Thu, 29 Oct 2026 19:36:41 +0000 Subject: [PATCH] fix(codegen): find generated classes by name, and derive emitted envelope bases from the bundle scripts/post_generate_fixes.py matched 29 exact class-signature strings such as "class ListCreativesResponse(AdcpVersionEnvelope, ProtocolEnvelope):". adcp#8547 makes core/protocol-envelope.json compose core/version-envelope.json and every task response compose only the protocol envelope, so the generated classes become "class X(ProtocolEnvelope):"; one pass then raised "ListCreativesResponse marker missing" and the status-alias, verify-brand-claim and compatibility-enum passes would have skipped silently (#1680). Every pass that needs a generated class now finds it by name (find_class_header / require_class_header) and keeps whatever bases the generator wrote; a class that is gone is a RuntimeError naming the pass. The classes this script emits itself (response arms, the constructible compatibility bases, the verify_brand_claim modules) take their envelope bases from response_envelope_bases(), which reads whether the generated ProtocolEnvelope already composes AdcpVersionEnvelope, and drop the version-envelope import when nothing uses it. The identity-match constant anchors on either envelope import. Measured both ways: the current 3.3.0-beta.1 tree regenerates byte-for-byte (make validate-generated), and against adcp#8547's schemas the post-generation passes run to completion and emit "class X(AdcpResponse, ProtocolEnvelope):" throughout, where they crashed before. Closes #1680 Co-Authored-By: Claude Fable 5.1 --- scripts/post_generate_fixes.py | 214 ++++++++++++------ tests/test_post_generate_envelope_matchers.py | 165 ++++++++++++++ 2 files changed, 307 insertions(+), 72 deletions(-) create mode 100644 tests/test_post_generate_envelope_matchers.py diff --git a/scripts/post_generate_fixes.py b/scripts/post_generate_fixes.py index d63fc3cbe..a19b820d6 100644 --- a/scripts/post_generate_fixes.py +++ b/scripts/post_generate_fixes.py @@ -242,6 +242,74 @@ def schema_keyword_raise(keyword: str, message: str, *, indent: str = " " * 8) - ) +def find_class_header(source: str, name: str) -> re.Match[str] | None: + """The ``class ():`` line in *source*, whatever its bases. + + A pass that needs a generated class must find it by NAME: the base-class + list is the generator's rendering of how the schema composes its + envelopes, and it changed when response schemas composed + ``core/protocol-envelope.json`` alone (adcp#8547, #1680). The match's + ``bases`` group carries whatever the generator wrote. + """ + return re.search(rf"^class {re.escape(name)}\((?P[^)]*)\):", source, re.MULTILINE) + + +def require_class_header(source: str, name: str, where: str) -> re.Match[str]: + """Like :func:`find_class_header`, failing loudly when the class is gone.""" + match = find_class_header(source, name) + if match is None: + raise RuntimeError(f"{where}: class {name} is not in the generated module") + return match + + +def protocol_envelope_composes_version_envelope() -> bool: + """Whether the generated ``ProtocolEnvelope`` already carries the version fields. + + ``core/protocol-envelope.json`` composes ``core/version-envelope.json`` + from adcp#8547 on, and the generator then renders ``class + ProtocolEnvelope(AdcpVersionEnvelope, ...)``. Read off the generated + module rather than the bundle so the answer is the one the generated + classes in this tree actually have. + """ + module = OUTPUT_DIR / "core" / "protocol_envelope.py" + if not module.exists(): + return False + header = find_class_header(module.read_text(), "ProtocolEnvelope") + return header is not None and "AdcpVersionEnvelope" in header.group("bases") + + +def response_envelope_bases(*, protocol: bool) -> str: + """The envelope bases a class this script emits should inherit. + + *protocol* names a task response (or an arm of one), which composes the + protocol envelope; a request-shaped class composes only the version + envelope. When ``ProtocolEnvelope`` already composes ``AdcpVersionEnvelope``, + a response inherits ``ProtocolEnvelope`` alone, like every generated + response of the same bundle. + """ + if not protocol: + return "AdcpVersionEnvelope" + if protocol_envelope_composes_version_envelope(): + return "ProtocolEnvelope" + return "AdcpVersionEnvelope, ProtocolEnvelope" + + +def normalize_emitted_envelope_bases(module_text: str) -> str: + """Rewrite the envelope bases of a hand-emitted module to the bundle's shape. + + The module literals this script writes spell the pre-#8547 composition, + ``(AdcpVersionEnvelope, ProtocolEnvelope)``. On a bundle whose protocol + envelope composes the version envelope that spelling is wrong, so the + bases are replaced and the then-unused version-envelope import dropped. + """ + bases = response_envelope_bases(protocol=True) + text = module_text.replace("(AdcpVersionEnvelope, ProtocolEnvelope)", f"({bases})") + if "AdcpVersionEnvelope" not in text.replace(_VERSION_ENVELOPE_IMPORT, ""): + text = text.replace(_VERSION_ENVELOPE_IMPORT, "") + text = re.sub(r"\n{3,}(from |import )", r"\n\n\1", text, count=1) + return text + + _VERSION_FILE = REPO_ROOT / "src" / "adcp" / "ADCP_VERSION" _BUNDLE_KEY = resolve_bundle_key(_VERSION_FILE.read_text().strip()) @@ -4356,6 +4424,24 @@ def fix_protocol_envelope_status_default() -> None: print(" core/protocol_envelope.py: defaulted status to completed") +def _envelope_import_anchor(source: str) -> str: + """The envelope import line a module-level constant is inserted after. + + The version-envelope import was the anchor; a response module on a bundle + whose protocol envelope composes the version envelope imports only + ``ProtocolEnvelope``, so either envelope import serves, and a module with + neither has no place for the constant (which is an error, not a skip). + """ + for pattern in ( + r"^from \.\.core\.version_envelope import AdcpVersionEnvelope\n", + r"^from \.\.core\.protocol_envelope import ProtocolEnvelope\n", + ): + match = re.search(pattern, source, re.MULTILINE) + if match is not None: + return match.group(0) + raise RuntimeError("trusted_match/identity_match_response.py imports no envelope class") + + def fix_trusted_match_runtime_validators() -> None: """Restore Trusted Match constraints that datamodel-codegen does not model. @@ -4428,7 +4514,7 @@ def _require_identity_match_dimensions(self) -> TmpProviderRegistration1: source = identity_match_response.read_text() if "_validate_tmpx_provider_ids" in source: changed = False - version_import = "from ..core.version_envelope import AdcpVersionEnvelope\n" + version_import = _envelope_import_anchor(source) if "_PROVIDER_ID_PATTERN =" not in source: source = source.replace( version_import, @@ -4467,7 +4553,7 @@ def _require_identity_match_dimensions(self) -> TmpProviderRegistration1: print(" trusted_match/identity_match_response.py pydantic import shape not found") return - version_import = "from ..core.version_envelope import AdcpVersionEnvelope\n" + version_import = _envelope_import_anchor(source) source = source.replace( version_import, version_import + "\n_PROVIDER_ID_PATTERN = re.compile(r'^[A-Za-z0-9_]{1,64}$')\n", @@ -4480,9 +4566,7 @@ def _require_identity_match_dimensions(self) -> TmpProviderRegistration1: print(" trusted_match/identity_match_response.py response class not found") return response_class = response_class_match.group(1) - source = ( - source.rstrip() - + f""" + source = source.rstrip() + f""" @model_validator(mode='after') def _validate_tmpx_provider_ids(self) -> {response_class}: @@ -4497,7 +4581,6 @@ def _validate_tmpx_provider_ids(self) -> {response_class}: raise ValueError('tmpx_providers keys must be valid provider_id values') return self """ - ) identity_match_response.write_text(source.rstrip() + "\n") print(" trusted_match/identity_match_response.py: added runtime validators") @@ -5332,9 +5415,7 @@ def restore_flattened_contract_field_types() -> None: if "@model_validator(mode='before')" not in source: source = ensure_pydantic_import(source, "model_validator") - source = ( - source.rstrip() - + """ + source = source.rstrip() + """ @model_validator(mode='before') @classmethod @@ -5349,7 +5430,6 @@ def _reject_seller_bound_manifest_fields(cls, data: Any) -> Any: ) return data """ - ) representation_target.write_text(source) print(" core/creative_representation.py: restored canonical format contract") @@ -5382,9 +5462,7 @@ def enforce_transformer_output_contract() -> None: if "class Transformer(" not in source: raise RuntimeError("transformer.py: Transformer class not found") source = ensure_pydantic_import(source, "model_validator") - target.write_text( - source.rstrip() - + """ + target.write_text(source.rstrip() + """ @model_validator(mode='after') def _require_output_format_declaration(self) -> Transformer: @@ -5399,8 +5477,7 @@ def _require_output_format_declaration(self) -> Transformer: 'one of output_capability_ids or deprecated output_format_ids is required' ) return self -""" - ) +""") print(" core/transformer.py: enforced output declaration requirement") @@ -5490,7 +5567,8 @@ def restore_constructible_response_bases() -> None: if first_arm_position < 0: raise RuntimeError(f"{relative_path}: unable to locate {first_arm.name}") arm_list = ",\n ".join(arm_names) - compatibility_base = f"""class {base_name}(ResponseArmDispatchMixin, AdcpVersionEnvelope, ProtocolEnvelope): + envelope_bases = response_envelope_bases(protocol=True) + compatibility_base = f"""class {base_name}(ResponseArmDispatchMixin, {envelope_bases}): \"\"\"Constructible compatibility base for generated response arms.\"\"\" @classmethod @@ -5931,7 +6009,7 @@ def emit_nested(self, preferred: str, schema: dict[str, Any]) -> str: self.nested_names.add(class_name) required = set(schema.get("required") or []) lines = [ - f"class {class_name}(AdcpVersionEnvelope):", + f"class {class_name}({response_envelope_bases(protocol=False)}):", " model_config = ConfigDict(extra='allow')", ] props = schema.get("properties") or {} @@ -6010,11 +6088,7 @@ def emit_response_class(self, class_name: str, arm: dict[str, Any]) -> str: # ``core/protocol-envelope.json`` carries the protocol envelope: # success arms, error arms and submitted arms alike. inherits_protocol_envelope = is_submitted or self.root_composes_protocol_envelope - bases = ( - "AdcpVersionEnvelope, ProtocolEnvelope" - if inherits_protocol_envelope - else "AdcpVersionEnvelope" - ) + bases = response_envelope_bases(protocol=inherits_protocol_envelope) if inherits_protocol_envelope: self.needs_protocol_envelope = True # Submitted arms once paired ``validate_default=True`` with @@ -6183,7 +6257,7 @@ def render(self, schema: dict[str, Any]) -> str: module = "\n".join(header) + "\n\n\n" + "\n\n\n".join(body) + "\n" if self.needs_no_explicit_null: module = _ensure_no_explicit_null_import(module) - return module + return normalize_emitted_envelope_bases(module) for relative, base in response_specs: target = OUTPUT_DIR / relative @@ -6261,9 +6335,11 @@ def fix_check_governance_status_alias() -> None: "from pydantic import AwareDatetime, ConfigDict, Field, model_validator", 1, ) - marker = "class CheckGovernanceResponse(AdcpVersionEnvelope):\n" - method = """class CheckGovernanceResponse(AdcpVersionEnvelope): - @model_validator(mode='before') + header = require_class_header( + source, "CheckGovernanceResponse", "governance/check_governance_response.py status alias" + ) + marker = header.group(0) + "\n" + method = marker + """ @model_validator(mode='before') @classmethod def _status_to_verdict(cls, data: Any) -> Any: if isinstance(data, dict) and 'verdict' not in data and 'status' in data: @@ -6272,9 +6348,6 @@ def _status_to_verdict(cls, data: Any) -> Any: return data """ - if marker not in source: - print(" governance/check_governance_response.py status alias marker not found") - return target.write_text(source.replace(marker, method, 1)) print(" governance/check_governance_response.py: mapped status to verdict") @@ -6300,9 +6373,13 @@ def fix_report_plan_outcome_status_alias() -> None: "from pydantic import ConfigDict, Field, model_validator", 1, ) - marker = "class ReportPlanOutcomeResponse(AdcpVersionEnvelope):\n" - method = """class ReportPlanOutcomeResponse(AdcpVersionEnvelope): - @model_validator(mode='before') + header = require_class_header( + source, + "ReportPlanOutcomeResponse", + "governance/report_plan_outcome_response.py status alias", + ) + marker = header.group(0) + "\n" + method = marker + """ @model_validator(mode='before') @classmethod def _status_to_outcome_state(cls, data: Any) -> Any: if isinstance(data, dict) and 'outcome_state' not in data and 'status' in data: @@ -6311,9 +6388,6 @@ def _status_to_outcome_state(cls, data: Any) -> Any: return data """ - if marker not in source: - print(" governance/report_plan_outcome_response.py status alias marker not found") - return target.write_text(source.replace(marker, method, 1)) print(" governance/report_plan_outcome_response.py: mapped status to outcome_state") @@ -6347,8 +6421,7 @@ def fix_verify_brand_claim_models() -> None: bulk_response = OUTPUT_DIR / "brand" / "verify_brand_claims_response.py" if request.exists() and "claim_type:" not in request.read_text(): - request.write_text( - """# generated by datamodel-codegen: + request.write_text(normalize_emitted_envelope_bases("""# generated by datamodel-codegen: # filename: brand/verify_brand_claim_request.json from __future__ import annotations @@ -6380,13 +6453,11 @@ class VerifyBrandClaimRequest(AdcpVersionEnvelope): dict[str, Any], Field(description='Claim payload. Shape varies by claim_type.'), ] -""" - ) +""")) print(" brand/verify_brand_claim_request.py: restored claim fields") if response.exists() and "VerifyBrandClaimSuccessResponse" not in response.read_text(): - response.write_text( - """# generated by datamodel-codegen: + response.write_text(normalize_emitted_envelope_bases("""# generated by datamodel-codegen: # filename: brand/verify_brand_claim_response.json from __future__ import annotations @@ -6445,8 +6516,7 @@ class VerifyBrandClaimErrorResponse(AdcpVersionEnvelope, ProtocolEnvelope): VerifyBrandClaimResponse = VerifyBrandClaimSuccessResponse | VerifyBrandClaimErrorResponse -""" - ) +""")) print(" brand/verify_brand_claim_response.py: restored response arms") response_schema = SCHEMA_DIR / "brand" / "verify-brand-claim-response.json" @@ -6455,8 +6525,7 @@ class VerifyBrandClaimErrorResponse(AdcpVersionEnvelope, ProtocolEnvelope): and response_schema.exists() and '"signed_response"' in response_schema.read_text() ): - response.write_text( - """# generated by datamodel-codegen: + response.write_text(normalize_emitted_envelope_bases("""# generated by datamodel-codegen: # filename: brand/verify_brand_claim_response.json from __future__ import annotations @@ -6607,8 +6676,7 @@ class VerifyBrandClaimErrorResponse(AdcpVersionEnvelope, ProtocolEnvelope): VerifyBrandClaimResponse = VerifyBrandClaimSuccessResponse | VerifyBrandClaimErrorResponse -""" - ) +""")) print(" brand/verify_brand_claim_response.py: restored signed response fields") bulk_response_schema = SCHEMA_DIR / "brand" / "verify-brand-claims-response.json" @@ -6618,7 +6686,7 @@ class VerifyBrandClaimErrorResponse(AdcpVersionEnvelope, ProtocolEnvelope): and '"signed_response"' in bulk_response_schema.read_text() ): bulk_response.write_text( - """# generated by datamodel-codegen: + normalize_emitted_envelope_bases("""# generated by datamodel-codegen: # filename: brand/verify_brand_claims_response.json from __future__ import annotations @@ -6801,7 +6869,7 @@ class VerifyBrandClaimsErrorResponse(AdcpVersionEnvelope, ProtocolEnvelope): VerifyBrandClaimsResponse = VerifyBrandClaimsResponseBulk | VerifyBrandClaimsErrorResponse -""" +""") ) print(" brand/verify_brand_claims_response.py: restored signed response fields") @@ -6821,9 +6889,12 @@ class VerifyBrandClaimsErrorResponse(AdcpVersionEnvelope, ProtocolEnvelope): "from ..core import ext as ext_1\n", 1, ) + bulk_header = require_class_header( + source, "VerifyBrandClaimsResponseBulk", "brand/verify_brand_claims_response.py" + ).group(0) source = source.replace( - "class VerifyBrandClaimsResponseBulk(AdcpVersionEnvelope, ProtocolEnvelope):\n pass\n", - """class VerifyBrandClaimsResponseBulk(AdcpVersionEnvelope, ProtocolEnvelope): + bulk_header + "\n pass\n", + bulk_header + """ model_config = ConfigDict( extra='allow', ) @@ -6982,7 +7053,10 @@ def _validate_format_reference_xor(self) -> Creative: Creatives = Creative Creatives1 = Creative """ - marker = "\n\nclass ListCreativesResponse(AdcpVersionEnvelope, ProtocolEnvelope):" + header = require_class_header( + source, "ListCreativesResponse", "creative/list_creatives_response.py merged XOR" + ) + marker = "\n\n" + header.group(0) if marker not in source: raise RuntimeError("ListCreativesResponse marker missing from merged creative output") target.write_text(source.replace(marker, merged_validator + marker, 1)) @@ -7021,9 +7095,13 @@ def _reject_legacy_format_ref(self) -> Creatives1: ) if "_reject_legacy_format_ref" not in source: source = source.replace( - "\n\nclass ListCreativesResponse(AdcpVersionEnvelope, ProtocolEnvelope):", + "\n\n" + + require_class_header( + source, "ListCreativesResponse", "creative/list_creatives_response.py XOR" + ).group(0), canonical_validator - + "\n\nclass ListCreativesResponse(AdcpVersionEnvelope, ProtocolEnvelope):", + + "\n\n" + + find_class_header(source, "ListCreativesResponse").group(0), # type: ignore[union-attr] 1, ) @@ -7199,9 +7277,12 @@ def enum_members(module_source: str, class_name: str) -> list[tuple[str, str]]: members = enum_members(product_fields.read_text(), "ProductResponseField") members.extend(enum_members(source, "Fields")) deduplicated = list(dict.fromkeys(members)) - marker = "\n\nclass GetProductsRequest(AdcpVersionEnvelope):" - if not deduplicated or marker not in source: + header = find_class_header(source, "GetProductsRequest") + if not deduplicated or header is None: return + marker = "\n\n" + header.group(0) + if marker not in source: + raise RuntimeError("media_buy/get_products_request.py: GetProductsRequest header moved") body = "\n".join(f" {name} = {value!r}" for name, value in deduplicated) compatibility_enum = ( "\n\nclass Field1(StrEnum):\n" @@ -7334,9 +7415,7 @@ def fix_legacy_purchase_accepted_losses() -> None: ) validator_marker = " def _accepted_losses_match_schema(" if validator_marker not in fixed and "field_validator" in fixed: - fixed = ( - fixed.rstrip() - + """ + fixed = fixed.rstrip() + """ @field_validator('selected_product_ids') @@ -7364,7 +7443,6 @@ def _accepted_losses_match_schema( raise ValueError('accepted_losses must include the required compatibility losses') return values """ - ) if fixed != source: target.write_text(fixed) print(" media_buy/legacy_purchase_continuation_input.py: narrowed accepted_losses") @@ -7615,9 +7693,7 @@ def _require_portable_bound(self) -> {class_name}: ) class_start = source.find("class MediaBuyChangeTerm(AdCPBaseModel):") if class_start >= 0 and "def _validate_constraint_action" not in source[class_start:]: - source = ( - source.rstrip() - + """ + source = source.rstrip() + """ @model_validator(mode='after') def _validate_constraint_action(self) -> MediaBuyChangeTerm: @@ -7637,9 +7713,7 @@ def _validate_constraint_action(self) -> MediaBuyChangeTerm: if action not in allowed.get(kind, set()): raise ValueError('constraint kind is incompatible with action') return self -""" - + "\n" - ) +""" + "\n" term_path.write_text(source) print(" media_buy/change_term.py: restored constraint/action compatibility") @@ -7652,9 +7726,7 @@ def _validate_constraint_action(self) -> MediaBuyChangeTerm: ) class_start = source.find("class CommercialTerms(AdCPBaseModel):") if class_start >= 0 and "def _validate_change_term_set" not in source[class_start:]: - source = ( - source.rstrip() - + """ + source = source.rstrip() + """ @model_validator(mode='after') def _validate_change_term_set(self) -> CommercialTerms: @@ -7703,9 +7775,7 @@ def _validate_change_term_set(self) -> CommercialTerms: ): raise ValueError('change-term earliest effective time exceeds latest time') return self -""" - + "\n" - ) +""" + "\n" terms_path.write_text(source) print(" media_buy/commercial_terms.py: restored change-term set invariants") diff --git a/tests/test_post_generate_envelope_matchers.py b/tests/test_post_generate_envelope_matchers.py new file mode 100644 index 000000000..d3b680a50 --- /dev/null +++ b/tests/test_post_generate_envelope_matchers.py @@ -0,0 +1,165 @@ +"""The post-generation passes find a generated class by name, not by its bases (#1680). + +adcp#8547 makes ``core/protocol-envelope.json`` compose +``core/version-envelope.json`` and every task response compose only the +protocol envelope, so the generated response classes change from +``class X(AdcpVersionEnvelope, ProtocolEnvelope):`` to ``class +X(ProtocolEnvelope):``. ``scripts/post_generate_fixes.py`` matched the full +class-signature string in 29 places: one raised ``ListCreativesResponse +marker missing`` on the new shape and the others would have skipped silently. +Every pass now locates its class by name and keeps whatever bases the +generator wrote, the classes this script emits itself derive their envelope +bases from the generated ``ProtocolEnvelope``, and a missing class is an +error. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from scripts import post_generate_fixes as fixes + +OLD = "class ProtocolEnvelope(AdCPBaseModel):\n pass\n" +NEW = "class ProtocolEnvelope(AdcpVersionEnvelope, AdCPBaseModel):\n pass\n" + + +@pytest.fixture +def generated(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + directory = tmp_path / "domains" + (directory / "core").mkdir(parents=True) + (directory / "governance").mkdir() + (directory / "creative").mkdir() + (directory / "media_buy").mkdir() + (directory / "brand").mkdir() + (directory / "trusted_match").mkdir() + monkeypatch.setattr(fixes, "OUTPUT_DIR", directory) + return directory + + +def _protocol_envelope(generated: Path, composition: str) -> None: + (generated / "core" / "protocol_envelope.py").write_text( + "from adcp.types.base import AdCPBaseModel\n\n\n" + composition + ) + + +def test_find_class_header_ignores_the_base_list() -> None: + for bases in ("AdcpVersionEnvelope, ProtocolEnvelope", "ProtocolEnvelope", "A, B, C"): + source = f"x = 1\n\n\nclass Thing({bases}):\n pass\n" + match = fixes.find_class_header(source, "Thing") + assert match is not None + assert match.group("bases") == bases + assert fixes.find_class_header("class Other(A):\n pass\n", "Thing") is None + with pytest.raises(RuntimeError, match="class Thing is not in the generated module"): + fixes.require_class_header("class Other(A):\n pass\n", "Thing", "somewhere") + + +@pytest.mark.parametrize( + "composition,protocol_bases", + [(OLD, "AdcpVersionEnvelope, ProtocolEnvelope"), (NEW, "ProtocolEnvelope")], + ids=["pre-8547", "post-8547"], +) +def test_emitted_envelope_bases_follow_the_generated_protocol_envelope( + generated: Path, composition: str, protocol_bases: str +) -> None: + _protocol_envelope(generated, composition) + assert fixes.response_envelope_bases(protocol=True) == protocol_bases + assert fixes.response_envelope_bases(protocol=False) == "AdcpVersionEnvelope" + module = ( + "from ..core.protocol_envelope import ProtocolEnvelope\n" + "from ..core.version_envelope import AdcpVersionEnvelope\n\n\n" + "class R(AdcpVersionEnvelope, ProtocolEnvelope):\n pass\n" + ) + normalized = fixes.normalize_emitted_envelope_bases(module) + assert f"class R({protocol_bases}):" in normalized + assert ("AdcpVersionEnvelope" in normalized) is (composition is OLD) + + +def test_envelope_bases_default_to_the_old_shape_without_a_generated_tree(generated: Path) -> None: + assert fixes.response_envelope_bases(protocol=True) == "AdcpVersionEnvelope, ProtocolEnvelope" + + +@pytest.mark.parametrize("bases", ["AdcpVersionEnvelope", "ProtocolEnvelope"]) +def test_status_alias_passes_find_their_class_whatever_its_bases( + generated: Path, bases: str +) -> None: + for module, cls, fix, alias in ( + ( + "check_governance_response", + "CheckGovernanceResponse", + fixes.fix_check_governance_status_alias, + "verdict", + ), + ( + "report_plan_outcome_response", + "ReportPlanOutcomeResponse", + fixes.fix_report_plan_outcome_status_alias, + "outcome_state", + ), + ): + target = generated / "governance" / f"{module}.py" + target.write_text( + "from typing import Any\n\n" + "from pydantic import AwareDatetime, ConfigDict, Field\n\n\n" + f"class {cls}({bases}):\n status: str | None = None\n" + ) + fix() + source = target.read_text() + assert f"class {cls}({bases}):" in source + assert f"data['{alias}'] = data['status']" in source + assert "model_validator" in source + + +def test_status_alias_pass_fails_loudly_when_the_class_is_gone(generated: Path) -> None: + target = generated / "governance" / "check_governance_response.py" + target.write_text( + "from pydantic import AwareDatetime, ConfigDict, Field\n\n\nclass Other(A):\n pass\n" + ) + with pytest.raises( + RuntimeError, match="CheckGovernanceResponse is not in the generated module" + ): + fixes.fix_check_governance_status_alias() + + +@pytest.mark.parametrize("bases", ["AdcpVersionEnvelope", "ProtocolEnvelope"]) +def test_get_products_compatibility_enum_anchors_on_the_class_name( + generated: Path, bases: str +) -> None: + (generated / "media_buy" / "product_fields.py").write_text( + "from adcp.types._str_enum import StrEnum\n\n\n" + "class ProductResponseField(StrEnum):\n product_id = 'product_id'\n" + ) + target = generated / "media_buy" / "get_products_request.py" + target.write_text( + "from adcp.types._str_enum import StrEnum\n\n\n" + "class Fields(StrEnum):\n brief = 'brief'\n\n\n" + f"class GetProductsRequest({bases}):\n fields: list[Fields] | None = None\n" + ) + fixes.restore_get_products_field_compatibility_enum() + source = target.read_text() + assert "class Field1(StrEnum):" in source + assert f"class GetProductsRequest({bases}):" in source + + +def test_identity_match_constant_anchors_on_either_envelope_import() -> None: + old = "from pydantic import Field\n\nfrom ..core.version_envelope import AdcpVersionEnvelope\n" + new = "from pydantic import Field\n\nfrom ..core.protocol_envelope import ProtocolEnvelope\n" + assert fixes._envelope_import_anchor(old).startswith("from ..core.version_envelope") + assert fixes._envelope_import_anchor(new).startswith("from ..core.protocol_envelope") + with pytest.raises(RuntimeError, match="imports no envelope class"): + fixes._envelope_import_anchor("from pydantic import Field\n") + + +def test_no_pass_matches_a_full_response_signature_literal() -> None: + """The remaining literal spellings are inside the modules this script emits itself.""" + source = Path(fixes.__file__).read_text() + offenders = [] + for number, line in enumerate(source.splitlines(), start=1): + stripped = line.strip() + if "AdcpVersionEnvelope, ProtocolEnvelope" not in stripped: + continue + # A marker is a quoted ``class`` header used to search generated text. + if stripped.startswith(('"', "'", 'f"', "marker", '+ "')) and "class " in stripped: + offenders.append(number) + assert offenders == [], f"exact class-signature markers remain at lines {offenders}"