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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 64 additions & 0 deletions scripts/post_generate_fixes.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
3. Fixes BrandManifest forward references
4. Adds deprecated=True to fields marked deprecated in JSON schema
5. Unwraps specified RootModel unions to plain Union type aliases (#155)
6. Widens canceled: Literal[True] = True on request types to | None = None (#641)
"""

from __future__ import annotations
Expand Down Expand Up @@ -1053,6 +1054,68 @@ def _ensure_sequence_import(content: str) -> str:
return "from collections.abc import Sequence\n\n" + content


# Matches the four request-type 'canceled: Literal[True] = True' emissions.
# datamodel-codegen emits '= True' directly from "const": true boolean
# schema properties — it is NOT produced by inject_literal_discriminator_defaults()
# (which already skips bool-valued Literals). Each match rewrites only the
# annotation and the default; the Field description and the rest of the class
# are untouched. The regex is inherently idempotent: 'Literal[True] | None,'
# does not match 'Literal[True],' so a second pass is a no-op.
_CANCELED_FIELD_RE = re.compile(
r"( canceled: Annotated\[\n )"
r"Literal\[True\]"
r"(,\n Field\(\n description='Cancel[^']*'\n \),\n \])"
r" = True"
)


def fix_canceled_literal_defaults() -> None:
"""Widen ``canceled: Literal[True] = True`` on request types to ``Literal[True] | None = None``.

Issue #641: the generated ``= True`` default silently cancels media buys /
packages when a buyer omits the field from an update request. Changing to
``Literal[True] | None = None`` preserves wire semantics (the field still
only accepts ``true`` when present) while making omission non-destructive.

Response-side ``canceled: bool | None = False`` fields (status indicators
like ``core/package.py``) are out of scope — their default is already safe.

Root cause: ``datamodel-codegen`` emits ``= True`` from the schema's
``"const": true`` boolean property. This function corrects that misfire for
the four request-type emissions listed below.
"""
targets = [
OUTPUT_DIR / "media_buy/update_media_buy_request.py",
OUTPUT_DIR / "media_buy/package_update.py",
OUTPUT_DIR / "bundled/media_buy/update_media_buy_request.py",
]

total_fixed = 0
for py_file in targets:
if not py_file.exists():
print(f" {py_file.relative_to(OUTPUT_DIR)}: not found (skipping)")
continue

source = py_file.read_text()
new_source, count = _CANCELED_FIELD_RE.subn(
r"\1Literal[True] | None\2 = None",
source,
)

if count == 0:
print(f" {py_file.relative_to(OUTPUT_DIR)}: no destructive canceled defaults found")
continue

py_file.write_text(new_source)
total_fixed += count
print(f" {py_file.relative_to(OUTPUT_DIR)}: fixed {count} canceled field(s)")

if total_fixed > 0:
print(f" ✓ Widened {total_fixed} canceled Literal[True] default(s) to None")
else:
print(" No canceled field defaults needed fixing")


def main():
"""Apply all post-generation fixes."""
print("Applying post-generation fixes...")
Expand All @@ -1071,6 +1134,7 @@ def main():
restore_format_category_deprecation_shim()
inject_literal_discriminator_defaults()
widen_extension_point_lists_to_sequence()
fix_canceled_literal_defaults()

print("\n✓ Post-generation fixes complete\n")

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -4775,11 +4775,11 @@ class Package(AdCPBaseModel):
Field(description='Pause/resume specific package (true = paused, false = active)'),
] = None
canceled: Annotated[
Literal[True],
Literal[True] | None,
Field(
description='Cancel this specific package. Cancellation is irreversible — canceled packages stop delivery and cannot be reactivated. Sellers MAY reject with NOT_CANCELLABLE.'
),
] = True
] = None
cancellation_reason: Annotated[
str | None, Field(description='Reason for canceling this package.', max_length=500)
] = None
Expand Down Expand Up @@ -7575,11 +7575,11 @@ class UpdateMediaBuyRequest(AdCPBaseModel):
Field(description='Pause/resume the entire media buy (true = paused, false = active)'),
] = None
canceled: Annotated[
Literal[True],
Literal[True] | None,
Field(
description='Cancel the entire media buy. Cancellation is irreversible — canceled media buys cannot be reactivated. Sellers MAY reject with NOT_CANCELLABLE if the media buy cannot be canceled in its current state.'
),
] = True
] = None
cancellation_reason: Annotated[
str | None,
Field(
Expand Down
4 changes: 2 additions & 2 deletions src/adcp/types/generated_poc/media_buy/package_update.py
Original file line number Diff line number Diff line change
Expand Up @@ -97,11 +97,11 @@ class PackageUpdate(AdCPBaseModel):
Field(description='Pause/resume specific package (true = paused, false = active)'),
] = None
canceled: Annotated[
Literal[True],
Literal[True] | None,
Field(
description='Cancel this specific package. Cancellation is irreversible — canceled packages stop delivery and cannot be reactivated. Sellers MAY reject with NOT_CANCELLABLE.'
),
] = True
] = None
cancellation_reason: Annotated[
str | None, Field(description='Reason for canceling this package.', max_length=500)
] = None
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -50,11 +50,11 @@ class UpdateMediaBuyRequest(AdCPBaseModel):
Field(description='Pause/resume the entire media buy (true = paused, false = active)'),
] = None
canceled: Annotated[
Literal[True],
Literal[True] | None,
Field(
description='Cancel the entire media buy. Cancellation is irreversible — canceled media buys cannot be reactivated. Sellers MAY reject with NOT_CANCELLABLE if the media buy cannot be canceled in its current state.'
),
] = True
] = None
cancellation_reason: Annotated[
str | None,
Field(
Expand Down
146 changes: 146 additions & 0 deletions tests/test_canceled_literal_default.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,146 @@
"""Tests for issue #641 — canceled: Literal[True] = True destructive default.

UpdateMediaBuyRequest and PackageUpdate previously had ``canceled: Literal[True] = True``,
meaning any call that omitted the field silently triggered irreversible cancellation.
The fix widens the type to ``Literal[True] | None = None`` so omission is non-destructive.
"""

from __future__ import annotations

import pytest


class TestUpdateMediaBuyRequestCanceledDefault:
"""UpdateMediaBuyRequest.canceled must default to None (non-destructive)."""

_ACCOUNT_KWARGS = {
"account": {"account_id": "acc_test_001"},
"media_buy_id": "mbuy-123",
"idempotency_key": "a" * 16,
}

def test_canceled_defaults_to_none_when_omitted(self) -> None:
from adcp.types.generated_poc.media_buy.update_media_buy_request import (
UpdateMediaBuyRequest,
)

req = UpdateMediaBuyRequest(**self._ACCOUNT_KWARGS)
assert req.canceled is None

def test_canceled_true_still_accepted(self) -> None:
from adcp.types.generated_poc.media_buy.update_media_buy_request import (
UpdateMediaBuyRequest,
)

req = UpdateMediaBuyRequest(**self._ACCOUNT_KWARGS, canceled=True)
assert req.canceled is True

def test_canceled_none_excluded_from_wire_payload(self) -> None:
"""When canceled is None, model_dump(exclude_none=True) omits it — no cancellation on wire."""
from adcp.types.generated_poc.media_buy.update_media_buy_request import (
UpdateMediaBuyRequest,
)

req = UpdateMediaBuyRequest(**self._ACCOUNT_KWARGS)
payload = req.model_dump(mode="json", exclude_none=True)
assert "canceled" not in payload

def test_canceled_true_present_in_wire_payload(self) -> None:
"""Explicit canceled=True must appear in the wire payload to trigger cancellation."""
from adcp.types.generated_poc.media_buy.update_media_buy_request import (
UpdateMediaBuyRequest,
)

req = UpdateMediaBuyRequest(**self._ACCOUNT_KWARGS, canceled=True)
payload = req.model_dump(mode="json", exclude_none=True)
assert payload["canceled"] is True

def test_canceled_false_rejected(self) -> None:
"""Literal[True] still rejects False — the field is a one-way commit signal."""
from pydantic import ValidationError

from adcp.types.generated_poc.media_buy.update_media_buy_request import (
UpdateMediaBuyRequest,
)

with pytest.raises(ValidationError):
UpdateMediaBuyRequest(**self._ACCOUNT_KWARGS, canceled=False) # type: ignore[arg-type]


class TestPackageUpdateCanceledDefault:
"""PackageUpdate.canceled must default to None (non-destructive)."""

def test_canceled_defaults_to_none_when_omitted(self) -> None:
from adcp.types.generated_poc.media_buy.package_update import PackageUpdate

pkg = PackageUpdate(package_id="pkg-1")
assert pkg.canceled is None

def test_canceled_true_still_accepted(self) -> None:
from adcp.types.generated_poc.media_buy.package_update import PackageUpdate

pkg = PackageUpdate(package_id="pkg-1", canceled=True)
assert pkg.canceled is True

def test_canceled_none_excluded_from_wire_payload(self) -> None:
from adcp.types.generated_poc.media_buy.package_update import PackageUpdate

pkg = PackageUpdate(package_id="pkg-1")
payload = pkg.model_dump(mode="json", exclude_none=True)
assert "canceled" not in payload

def test_canceled_true_present_in_wire_payload(self) -> None:
from adcp.types.generated_poc.media_buy.package_update import PackageUpdate

pkg = PackageUpdate(package_id="pkg-1", canceled=True)
payload = pkg.model_dump(mode="json", exclude_none=True)
assert payload["canceled"] is True


class TestFixIdempotency:
"""The post-gen fix must be idempotent — running it twice produces the same output."""

def test_fix_is_idempotent_on_synthesized_source(self) -> None:
from scripts.post_generate_fixes import _CANCELED_FIELD_RE

already_fixed = (
" canceled: Annotated[\n"
" Literal[True] | None,\n"
" Field(\n"
" description='Cancel this specific package. Cancellation is irreversible"
" — canceled packages stop delivery and cannot be reactivated."
" Sellers MAY reject with NOT_CANCELLABLE.'\n"
" ),\n"
" ] = None\n"
)

result, count = _CANCELED_FIELD_RE.subn(
r"\1Literal[True] | None\2 = None",
already_fixed,
)
assert count == 0, "fix must not re-apply to already-widened source"
assert result == already_fixed

def test_fix_matches_destructive_source(self) -> None:
from scripts.post_generate_fixes import _CANCELED_FIELD_RE

destructive = (
" canceled: Annotated[\n"
" Literal[True],\n"
" Field(\n"
" description='Cancel this specific package. Cancellation is irreversible"
" — canceled packages stop delivery and cannot be reactivated."
" Sellers MAY reject with NOT_CANCELLABLE.'\n"
" ),\n"
" ] = True\n"
)

result, count = _CANCELED_FIELD_RE.subn(
r"\1Literal[True] | None\2 = None",
destructive,
)
assert count == 1
assert "Literal[True] | None," in result
assert "] = None" in result
assert "Literal[True]," not in result
assert "] = True" not in result
Loading