diff --git a/.chronus/changes/python-fix-typeddict-shared-clid-2026-6-2-0-0-0.md b/.chronus/changes/python-fix-typeddict-shared-clid-2026-6-2-0-0-0.md new file mode 100644 index 00000000000..5dd91200351 --- /dev/null +++ b/.chronus/changes/python-fix-typeddict-shared-clid-2026-6-2-0-0-0.md @@ -0,0 +1,9 @@ +--- +changeKind: fix +packages: + - "@typespec/http-client-python" +--- + +Fix dangling `_types.X` references when template-instantiated models (e.g. `ResourceUpdateModel`) share a `crossLanguageDefinitionId`. The TypedDict deduplication now pairs each model with its own copy by name, so distinct models such as `CacheUpdate` and `VolumeUpdate` are all rendered in `types.py`. + +Also stop emitting unused TypedDicts for response-only models in `types.py`. Output-only models already render as classes in `models/` and are referenced via `_models.X`, so their TypedDict copies (e.g. `GetResponse`) were dead code. The set of TypedDicts (and discriminated-base union aliases) rendered in `types.py` is now the transitive closure of the request-body input models over their base classes, discriminated subtypes and property types. Input body overloads (including spread bodies whose usage lacks the `Input` flag) are still emitted, and any output-only model reachable from an input model — such as a discriminated subtype or an ARM `SystemData` property — is kept so no forward reference is left undefined. This fixes a `NameError` at import time when an output-only union alias (e.g. `Dinosaur = Union[TRex]`) referenced an excluded subtype, and a pyright `reportUndefinedVariable` error when an input model referenced an excluded property type (e.g. `SystemData`). diff --git a/packages/http-client-python/generator/pygen/codegen/models/model_type.py b/packages/http-client-python/generator/pygen/codegen/models/model_type.py index c4f83fc6c18..49fb22a547b 100644 --- a/packages/http-client-python/generator/pygen/codegen/models/model_type.py +++ b/packages/http-client-python/generator/pygen/codegen/models/model_type.py @@ -85,6 +85,10 @@ def __init__( def is_usage_output(self) -> bool: return bool(self.usage & UsageFlags.Output.value) + @property + def is_usage_input(self) -> bool: + return bool(self.usage & UsageFlags.Input.value) + @property def is_used_in_operations_via_types(self) -> bool: """Whether this model would be imported from types.py (not models) in operations.""" diff --git a/packages/http-client-python/generator/pygen/codegen/serializers/types_serializer.py b/packages/http-client-python/generator/pygen/codegen/serializers/types_serializer.py index 5ee26ebf70c..30e1cb802e4 100644 --- a/packages/http-client-python/generator/pygen/codegen/serializers/types_serializer.py +++ b/packages/http-client-python/generator/pygen/codegen/serializers/types_serializer.py @@ -5,7 +5,8 @@ # -------------------------------------------------------------------------- import keyword import re -from typing import Optional +from functools import cached_property +from typing import Any, Optional from ..models import ModelType, CodeModel from ..models.enum_type import EnumType from ..models.imports import FileImport, ImportType @@ -71,28 +72,114 @@ def declare_literal_enum(self, enum: EnumType) -> str: values = [enum.get_declaration(v.value) for v in enum.values] return f"{enum.name} = Literal[{', '.join(values)}]" + @staticmethod + def _renders_as_input_typeddict(m: "ModelType") -> bool: + """Whether a non-json model is a *seed* for the types.py input surface. + + TypedDicts (and discriminated-base union aliases) in ``types.py`` describe request-body + (*input*) shapes. Output-only models already render as classes in ``models/`` and are + referenced via ``_models.*``, so a response-only model that nothing input references would be + dead code. A model seeds the input surface when it is: + + * a typeddict copy (``base == "typeddict"``) — these only exist as input body overloads + (their ``usage`` may carry ``Spread``/``Json`` rather than ``Input``), or + * ``is_typed_dict_only`` — includes every model in full ``typeddict`` mode (responses too), + and input-only anonymous bodies, or + * used as input (``is_usage_input``) — e.g. a model shared between request and response. + + The full set rendered in types.py is the transitive closure of these seeds over base + classes, discriminated subtypes and property types (see :attr:`_types_file_model_names`), so + an output-only model that *is* referenced by an input model (e.g. ARM ``SystemData`` on + ``Resource``) is still rendered. + """ + return m.base == "typeddict" or m.is_typed_dict_only or m.is_usage_input + + @staticmethod + def _iter_referenced_models(base_type: Any): + """Yield ModelType instances directly referenced by a type, recursing through containers. + + Handles list/dict ``element_type``, constant/enum ``value_type`` and combined ``types``. + A referenced ModelType is yielded but not descended into — the closure walk descends into a + model's own properties/parents/subtypes when that model is itself visited. + """ + stack = [base_type] + seen: set[int] = set() + while stack: + t = stack.pop() + if t is None or id(t) in seen: + continue + seen.add(id(t)) + if isinstance(t, ModelType): + yield t + continue + for attr in ("element_type", "value_type"): + child = getattr(t, attr, None) + if child is not None: + stack.append(child) + for child in getattr(t, "types", []) or []: + stack.append(child) + + @cached_property + def _types_file_model_names(self) -> set[str]: + """Names of every model that must be rendered in types.py. + + Starts from the input seeds (:meth:`_renders_as_input_typeddict`) and takes the transitive + closure over base classes, discriminated subtypes and property types. Keyed on model + ``name`` (dpg models and their typeddict copies share a name and render as one TypedDict), so + the result is stable regardless of which copy a reference points at. + + Cached: the closure is a full walk over the model graph and is read several times per file + (via :attr:`typeddict_models` and :attr:`discriminated_base_models`). ``self._models`` is set + once at construction and never mutated, so memoizing on the instance is safe. + """ + needed: set[str] = set() + stack = [m for m in self._models if m.base != "json" and self._renders_as_input_typeddict(m)] + while stack: + m = stack.pop() + if m.base == "json" or m.name in needed: + continue + needed.add(m.name) + stack.extend(m.parents) + stack.extend(m.discriminated_subtypes.values()) + for prop in m.properties: + stack.extend(self._iter_referenced_models(prop.type)) + return needed + @property def typeddict_models(self) -> list[ModelType]: """Models that should be rendered as TypedDicts (excluding discriminated bases which become unions). - When both a dpg model and its typeddict copy exist (same crossLanguageDefinitionId), + When both a dpg model and its typeddict copy exist for the same model, prefer the dpg model (it already renders as a TypedDict in types.py) and skip the copy. + + The pairing is keyed on the model ``name`` (the copy is a shallow copy of the source, so it + shares the source's name). ``crossLanguageDefinitionId`` cannot be used here: template + instantiated models such as ``ResourceUpdateModel`` all share the + template's cross-language id, so keying on it would wrongly collapse distinct models + (e.g. ``CacheUpdate`` and ``VolumeUpdate``) into one and drop the rest from types.py. + + Only models in the input-surface closure (:attr:`_types_file_model_names`) are rendered, so + response-only models (e.g. ``GetResponse``) are dropped while models reachable from an input + model — including output-only ones such as a discriminated subtype or an ARM ``SystemData`` + property — are kept, ensuring no forward reference is left undefined. """ - candidates = [m for m in self._models if m.base != "json" and not m.discriminated_subtypes] - seen_ids: dict[str, "ModelType"] = {} + needed = self._types_file_model_names + candidates = [ + m for m in self._models if m.base != "json" and not m.discriminated_subtypes and m.name in needed + ] + seen_names: dict[str, "ModelType"] = {} result: list["ModelType"] = [] for m in candidates: - clid = m.yaml_data.get("crossLanguageDefinitionId") - if clid and clid in seen_ids: + name = m.name + if name in seen_names: # Prefer the dpg model over the typeddict copy - if m.base == "dpg" and seen_ids[clid].base == "typeddict": + if m.base == "dpg" and seen_names[name].base == "typeddict": # Replace the typeddict copy with the dpg model - result = [r if r is not seen_ids[clid] else m for r in result] - seen_ids[clid] = m + result = [r if r is not seen_names[name] else m for r in result] + seen_names[name] = m # Otherwise skip this duplicate continue - if clid: - seen_ids[clid] = m + seen_names[name] = m result.append(m) return result @@ -100,10 +187,16 @@ def typeddict_models(self) -> list[ModelType]: def discriminated_base_models(self) -> list[ModelType]: """Discriminated base models that become Union type aliases in types.py. + Only bases in the input-surface closure (:attr:`_types_file_model_names`) are emitted: an + output-only ``Dinosaur = Union[TRex]`` alias is dead code and would reference subtype + TypedDicts that are themselves (correctly) omitted from types.py, causing a ``NameError`` at + import time. + Topologically sorted so that nested discriminated bases (e.g. Shark) are defined before their parents (e.g. Fish = Union[Salmon, Shark]). """ - bases = [m for m in self._models if m.base != "json" and m.discriminated_subtypes] + needed = self._types_file_model_names + bases = [m for m in self._models if m.base != "json" and m.discriminated_subtypes and m.name in needed] base_names = {m.name for m in bases} sorted_bases: list[ModelType] = [] visited: set[str] = set() diff --git a/packages/http-client-python/generator/pygen/preprocess/__init__.py b/packages/http-client-python/generator/pygen/preprocess/__init__.py index 6961d890a2d..605ced4b4ed 100644 --- a/packages/http-client-python/generator/pygen/preprocess/__init__.py +++ b/packages/http-client-python/generator/pygen/preprocess/__init__.py @@ -307,8 +307,18 @@ def is_tsp(self) -> bool: return self.options.get("tsp_file", False) @staticmethod - def _find_existing_typeddict(code_model: dict[str, Any], cross_lang_id: Optional[str]) -> Optional[dict[str, Any]]: - """Find an existing typeddict copy with the given crossLanguageDefinitionId.""" + def _find_existing_typeddict( + code_model: dict[str, Any], + cross_lang_id: Optional[str], + name: Optional[str] = None, + ) -> Optional[dict[str, Any]]: + """Find an existing typeddict copy for the given model. + + Matches on both ``crossLanguageDefinitionId`` and ``name``. The name is required because + template-instantiated models (e.g. ``ResourceUpdateModel``) all share + the template's cross-language id, so matching on the id alone would reuse one model's copy + (e.g. ``CacheUpdate``) for a different model (e.g. ``VolumeUpdate``). + """ if not cross_lang_id: return None return next( @@ -318,10 +328,50 @@ def _find_existing_typeddict(code_model: dict[str, Any], cross_lang_id: Optional if t.get("type") == "model" and t.get("base") == "typeddict" and t.get("crossLanguageDefinitionId") == cross_lang_id + and (name is None or t.get("name") == name) ), None, ) + @staticmethod + def _find_spread_original(code_model: dict[str, Any], json_model: dict[str, Any]) -> Optional[dict[str, Any]]: + """Recover the dpg model that a spread (json) body was cloned from. + + When a spread body type is also used elsewhere, the emitter clones it, renames the clone to + ``Request`` and sets ``base = "json"`` while keeping the original + ``crossLanguageDefinitionId``. To reference the real model's TypedDict we look the clone up + by that id. + + Distinct template-instantiated models share a single ``crossLanguageDefinitionId``, so an id + match alone is ambiguous. We only reuse an original when the choice is unambiguous: + + * a dpg candidate whose ``name`` equals the json model's name (the body was not renamed), or + * exactly one dpg candidate carries the id. + + Otherwise we return ``None`` so the caller falls back to the json model itself, avoiding a + reference to the wrong model's TypedDict. + """ + cross_lang_id = json_model.get("crossLanguageDefinitionId") + if not cross_lang_id: + return None + candidates = [ + t + for t in code_model["types"] + if t.get("type") == "model" + and t.get("base") == "dpg" + and t.get("crossLanguageDefinitionId") == cross_lang_id + and t is not json_model + ] + if not candidates: + return None + name = json_model.get("name") + for candidate in candidates: + if candidate.get("name") == name: + return candidate + if len(candidates) == 1: + return candidates[0] + return None + @staticmethod def _insert_typeddict_overload( code_model: dict[str, Any], @@ -376,25 +426,14 @@ def add_body_param_type( # Add typeddict overload for non-spread dpg models if self.options["models-mode"] == "dpg" and is_dpg_model: cross_lang_id = model_type.get("crossLanguageDefinitionId") - existing_td = self._find_existing_typeddict(code_model, cross_lang_id) + existing_td = self._find_existing_typeddict(code_model, cross_lang_id, model_type.get("name")) self._insert_typeddict_overload(code_model, body_parameter, model_type, origin_type, existing_td) # For spread bodies (json base), add a typeddict overload that references # the original model. This replaces the JSON single-body overload. if is_json_model: cross_lang_id = model_type.get("crossLanguageDefinitionId") - original = None - if cross_lang_id: - original = next( - ( - t - for t in code_model["types"] - if t.get("type") == "model" - and t.get("crossLanguageDefinitionId") == cross_lang_id - and t is not model_type - ), - None, - ) + original = self._find_spread_original(code_model, model_type) if is_typeddict_only and original: # In typeddict-only mode, the original dpg model already renders @@ -407,7 +446,7 @@ def add_body_param_type( body_parameter["type"]["types"].insert(1, td_list_or_dict) else: source = original or model_type - existing_td = self._find_existing_typeddict(code_model, cross_lang_id) + existing_td = self._find_existing_typeddict(code_model, cross_lang_id, source.get("name")) self._insert_typeddict_overload(code_model, body_parameter, source, origin_type, existing_td) if len(body_parameter["type"]["types"]) == 1: @@ -420,7 +459,6 @@ def add_body_param_type( code_model["types"].append(body_parameter["type"]) - def pad_reserved_words(self, name: str, pad_type: PadType, yaml_type: dict[str, Any]) -> str: # we want to pad hidden variables as well if not name: diff --git a/packages/http-client-python/tests/unit/test_typeddict.py b/packages/http-client-python/tests/unit/test_typeddict.py index 3fb20d7cc96..c9da8aa5066 100644 --- a/packages/http-client-python/tests/unit/test_typeddict.py +++ b/packages/http-client-python/tests/unit/test_typeddict.py @@ -10,6 +10,8 @@ from pygen.codegen.models import CodeModel, JSONModelType, DPGModelType from pygen.codegen.models.model_type import TypedDictModelType +from pygen.codegen.models.property import Property +from pygen.codegen.models.list_type import ListType from pygen.codegen.serializers.types_serializer import TypesSerializer from pygen.codegen.serializers.unions_serializer import UnionsSerializer @@ -117,6 +119,227 @@ def test_models_mode_dpg_typeddict_models_included(): assert len(ts.typeddict_models) == 1 +def _make_model_with_clid(code_model, name, clid, model_cls): + """Create a model carrying an explicit crossLanguageDefinitionId.""" + return model_cls( + yaml_data={ + "name": name, + "type": "model", + "snakeCaseName": name.lower(), + "usage": 2, + "crossLanguageDefinitionId": clid, + }, + code_model=code_model, + properties=[], + ) + + +def test_typeddict_models_shared_cross_language_id_not_collapsed(): + """Distinct models that share a template crossLanguageDefinitionId must all render. + + Template-instantiated models such as ``ResourceUpdateModel`` and + ``ResourceUpdateModel`` are named ``CacheUpdate`` / ``VolumeUpdate`` + but share the template's cross-language id. The dpg/copy deduplication must key on the model name so + these distinct models are not collapsed into one (which previously dropped ``CacheUpdate`` from + types.py, leaving a dangling ``_types.CacheUpdate`` reference in operations). + """ + code_model = _make_code_model(models_mode="dpg") + clid = "Azure.ResourceManager.Foundations.ResourceUpdateModel" + cache_dpg = _make_model_with_clid(code_model, "CacheUpdate", clid, DPGModelType) + cache_td = _make_model_with_clid(code_model, "CacheUpdate", clid, TypedDictModelType) + volume_dpg = _make_model_with_clid(code_model, "VolumeUpdate", clid, DPGModelType) + volume_td = _make_model_with_clid(code_model, "VolumeUpdate", clid, TypedDictModelType) + + env = _make_env() + ts = TypesSerializer( + code_model=code_model, + env=env, + models=[cache_dpg, cache_td, volume_dpg, volume_td], + ) + rendered = ts.typeddict_models + # Both distinct models render exactly once, and the dpg copy is preferred over the typeddict copy. + assert sorted(m.name for m in rendered) == ["CacheUpdate", "VolumeUpdate"] + assert all(m.base == "dpg" for m in rendered) + + +def _make_model_with_usage(code_model, name, usage, model_cls): + """Create a model with an explicit usage bitmask (2=Input, 4=Output, 6=both).""" + return model_cls( + yaml_data={ + "name": name, + "type": "model", + "snakeCaseName": name.lower(), + "usage": usage, + }, + code_model=code_model, + properties=[], + ) + + +def test_dpg_output_only_model_excluded_from_typeddict_models(): + """In dpg mode, response-only models must not render as TypedDicts in types.py. + + Anonymous response bodies (e.g. ``GetResponse``) render as classes in ``models/`` and are only + referenced via ``_models.*``; a TypedDict for them in types.py is never referenced via + ``_types.*`` and is dead code. Only input models belong in types.py. + """ + code_model = _make_code_model(models_mode="dpg") + send_request = _make_model_with_usage(code_model, "SendRequest", 2, DPGModelType) # Input + get_response = _make_model_with_usage(code_model, "GetResponse", 4, DPGModelType) # Output-only + + env = _make_env() + ts = TypesSerializer(code_model=code_model, env=env, models=[send_request, get_response]) + rendered = [m.name for m in ts.typeddict_models] + assert rendered == ["SendRequest"] + assert "GetResponse" not in rendered + + +def test_dpg_input_output_model_included_in_typeddict_models(): + """A dpg model used as both input and output still renders in types.py (it is an input).""" + code_model = _make_code_model(models_mode="dpg") + cat = _make_model_with_usage(code_model, "Cat", 6, DPGModelType) # Input | Output + + env = _make_env() + ts = TypesSerializer(code_model=code_model, env=env, models=[cat]) + assert [m.name for m in ts.typeddict_models] == ["Cat"] + + +def test_typeddict_mode_output_only_model_included(): + """In full typeddict mode, every model (incl. output-only) is consumed as a TypedDict.""" + code_model = _make_code_model(models_mode="typeddict") + get_response = _make_model_with_usage(code_model, "GetResponse", 4, TypedDictModelType) # Output-only + + env = _make_env() + ts = TypesSerializer(code_model=code_model, env=env, models=[get_response]) + # is_typed_dict_only is True for all models in typeddict mode, so it is kept despite being output-only. + assert [m.name for m in ts.typeddict_models] == ["GetResponse"] + + +def test_dpg_typeddict_copy_without_input_usage_still_kept(): + """A spread-body typeddict copy must be kept even when its usage lacks the Input bit. + + Spread request bodies (e.g. ``SendRequest``) are marked ``Json | Spread`` by tcgc but NOT + ``Input``. The referenced type is the ``base == "typeddict"`` copy, so operations import + ``_types.SendRequest``. Filtering these out would leave a dangling reference, so typeddict + copies are always kept regardless of their usage flags. + """ + code_model = _make_code_model(models_mode="dpg") + # usage 320 == Json(256) | Spread(64), no Input(2) bit — mirrors real tcgc output. + send_copy = _make_model_with_usage(code_model, "SendRequest", 320, TypedDictModelType) + + env = _make_env() + ts = TypesSerializer(code_model=code_model, env=env, models=[send_copy]) + assert [m.name for m in ts.typeddict_models] == ["SendRequest"] + assert not send_copy.is_usage_input # the copy is kept despite lacking the Input flag + + +def _make_discriminated_base(code_model, name, usage, subtypes, model_cls=DPGModelType): + """Create a discriminated base model carrying a discriminator_value->subtype mapping.""" + return model_cls( + yaml_data={ + "name": name, + "type": "model", + "snakeCaseName": name.lower(), + "usage": usage, + }, + code_model=code_model, + properties=[], + discriminated_subtypes={s.name.lower(): s for s in subtypes}, + ) + + +def test_dpg_output_only_discriminated_base_excluded(): + """An output-only discriminated base (e.g. ``Dinosaur = Union[TRex]``) must not render. + + Regression for the ``NameError: name 'TRex' is not defined`` import crash: the output-only base + was emitted as a union alias in types.py while its subtype ``TRex`` was correctly excluded, + leaving the alias referencing an undefined name. + """ + code_model = _make_code_model(models_mode="dpg") + trex = _make_model_with_usage(code_model, "TRex", 4, DPGModelType) # Output-only subtype + dinosaur = _make_discriminated_base(code_model, "Dinosaur", 4, [trex]) # Output-only base + + env = _make_env() + ts = TypesSerializer(code_model=code_model, env=env, models=[dinosaur, trex]) + assert [m.name for m in ts.discriminated_base_models] == [] + # The subtype is not force-included because no rendered base references it. + assert [m.name for m in ts.typeddict_models] == [] + + +def test_dpg_input_discriminated_base_included_with_subtypes(): + """An input discriminated base renders as a union and its subtypes render as TypedDicts.""" + code_model = _make_code_model(models_mode="dpg") + eagle = _make_model_with_usage(code_model, "Eagle", 2, DPGModelType) # Input + goose = _make_model_with_usage(code_model, "Goose", 2, DPGModelType) # Input + bird = _make_discriminated_base(code_model, "Bird", 2, [eagle, goose]) # Input base + + env = _make_env() + ts = TypesSerializer(code_model=code_model, env=env, models=[bird, eagle, goose]) + assert [m.name for m in ts.discriminated_base_models] == ["Bird"] + assert sorted(m.name for m in ts.typeddict_models) == ["Eagle", "Goose"] + + +def test_dpg_input_base_forces_output_only_subtype_into_types(): + """A subtype referenced by a rendered input base must render even if the subtype is output-only. + + ``Bird = Union[Eagle]`` requires ``Eagle`` to be defined in types.py; if ``Eagle``'s own usage + flags would exclude it, it must still be force-included so the union alias does not reference an + undefined name. + """ + code_model = _make_code_model(models_mode="dpg") + eagle = _make_model_with_usage(code_model, "Eagle", 4, DPGModelType) # Output-only subtype + bird = _make_discriminated_base(code_model, "Bird", 2, [eagle]) # Input base references it + + env = _make_env() + ts = TypesSerializer(code_model=code_model, env=env, models=[bird, eagle]) + assert [m.name for m in ts.discriminated_base_models] == ["Bird"] + assert [m.name for m in ts.typeddict_models] == ["Eagle"] + + +def _make_property(code_model, name, prop_type): + """Create a Property named ``name`` whose type is ``prop_type``.""" + return Property( + yaml_data={"wireName": name, "clientName": name, "optional": True}, + code_model=code_model, + type=prop_type, + ) + + +def test_dpg_input_model_pulls_in_output_only_property_type(): + """An output-only model referenced only as an input model's property type must still render. + + Regression for ``NameError: name 'SystemData' is not defined``: ARM ``Resource`` (input) has a + read-only ``systemData: "SystemData"`` property. ``SystemData`` is output-only, so it would be + dropped from types.py, leaving the forward reference dangling. The reachability closure must + keep it. + """ + code_model = _make_code_model(models_mode="dpg") + system_data = _make_model_with_usage(code_model, "SystemData", 4, DPGModelType) # Output-only + resource = _make_model_with_usage(code_model, "Resource", 2, DPGModelType) # Input + resource.properties = [_make_property(code_model, "systemData", system_data)] + # An unrelated output-only response model that nothing input references must stay excluded. + get_response = _make_model_with_usage(code_model, "GetResponse", 4, DPGModelType) + + env = _make_env() + ts = TypesSerializer(code_model=code_model, env=env, models=[resource, system_data, get_response]) + rendered = sorted(m.name for m in ts.typeddict_models) + assert rendered == ["Resource", "SystemData"] + assert "GetResponse" not in rendered + + +def test_dpg_input_model_pulls_in_output_only_type_inside_container(): + """Reachability must descend into list/dict element types, not just direct property references.""" + code_model = _make_code_model(models_mode="dpg") + item = _make_model_with_usage(code_model, "Item", 4, DPGModelType) # Output-only element + container = _make_model_with_usage(code_model, "Container", 2, DPGModelType) # Input + list_type = ListType(yaml_data={"type": "list"}, code_model=code_model, element_type=item) + container.properties = [_make_property(code_model, "items", list_type)] + + env = _make_env() + ts = TypesSerializer(code_model=code_model, env=env, models=[container, item]) + assert sorted(m.name for m in ts.typeddict_models) == ["Container", "Item"] + + # ---------- models-mode=typeddict ---------- diff --git a/packages/http-client-python/tests/unit/test_typeddict_overloads.py b/packages/http-client-python/tests/unit/test_typeddict_overloads.py index e98a0af65ec..0789877a7dc 100644 --- a/packages/http-client-python/tests/unit/test_typeddict_overloads.py +++ b/packages/http-client-python/tests/unit/test_typeddict_overloads.py @@ -100,3 +100,191 @@ def test_dpg_mode_still_emits_multiple_overloads(): # combined type has multiple members and overloads are generated. assert body_parameter["type"]["type"] == "combined" assert len(yaml_data["overloads"]) >= 2 + + +def _dpg_body_parameter(name: str, cross_lang_id: str) -> dict: + """A JSON dpg-model body parameter for the given model name and cross-language id.""" + model_type = { + "type": "model", + "base": "dpg", + "name": name, + "crossLanguageDefinitionId": cross_lang_id, + "properties": [], + } + return { + "wireName": "body", + "clientName": "body", + "location": "body", + "optional": False, + "implementation": "Method", + "contentTypes": ["application/json"], + "type": model_type, + } + + +def test_dpg_mode_shared_cross_language_id_creates_distinct_typeddict_copies(): + """Distinct models sharing a template cross-language id get their own typeddict copy. + + Template-instantiated models (e.g. ``ResourceUpdateModel`` and + ``ResourceUpdateModel``) share the template's crossLanguageDefinitionId but have + different names. ``_find_existing_typeddict`` must match on name as well, otherwise the second + model reuses the first model's typeddict copy and its operation references the wrong type. + """ + plugin = _plugin("dpg") + clid = "Azure.ResourceManager.Foundations.ResourceUpdateModel" + cache_body = _dpg_body_parameter("CacheUpdate", clid) + volume_body = _dpg_body_parameter("VolumeUpdate", clid) + code_model = {"types": [cache_body["type"], volume_body["type"]]} + + plugin.add_body_param_type(code_model, cache_body) + plugin.add_body_param_type(code_model, volume_body) + + typeddict_copies = [t for t in code_model["types"] if t.get("type") == "model" and t.get("base") == "typeddict"] + # Two distinct copies, one per model name. + assert sorted(t["name"] for t in typeddict_copies) == [ + "CacheUpdate", + "VolumeUpdate", + ] + # Each operation's typeddict overload references its OWN model, not a shared/wrong one. + cache_td = cache_body["type"]["types"][1] + volume_td = volume_body["type"]["types"][1] + assert cache_td["name"] == "CacheUpdate" + assert volume_td["name"] == "VolumeUpdate" + assert cache_td is not volume_td + + +def test_dpg_mode_same_model_reused_across_operations_shares_one_copy(): + """Regression guard for the intended deduplication: one model used in two bodies shares a single copy.""" + plugin = _plugin("dpg") + clid = "Contoso.Widget" + first_body = _dpg_body_parameter("Widget", clid) + second_body = _dpg_body_parameter("Widget", clid) + code_model = {"types": [first_body["type"]]} + + plugin.add_body_param_type(code_model, first_body) + plugin.add_body_param_type(code_model, second_body) + + typeddict_copies = [t for t in code_model["types"] if t.get("type") == "model" and t.get("base") == "typeddict"] + assert [t["name"] for t in typeddict_copies] == ["Widget"] + # Both operations reference the very same copy object. + assert first_body["type"]["types"][1] is second_body["type"]["types"][1] + + +def _dpg_list_body_parameter(name: str, cross_lang_id: str) -> dict: + """A JSON list body whose element is a dpg model with the given name and cross-language id.""" + element_type = { + "type": "model", + "base": "dpg", + "name": name, + "crossLanguageDefinitionId": cross_lang_id, + "properties": [], + } + return { + "wireName": "body", + "clientName": "body", + "location": "body", + "optional": False, + "implementation": "Method", + "contentTypes": ["application/json"], + "type": {"type": "list", "elementType": element_type}, + } + + +def test_dpg_list_body_shared_cross_language_id_creates_distinct_element_copies(): + """List-body element models that share a template clid must each get their own typeddict copy.""" + plugin = _plugin("dpg") + clid = "Azure.ResourceManager.Foundations.ResourceUpdateModel" + cache_body = _dpg_list_body_parameter("CacheUpdate", clid) + volume_body = _dpg_list_body_parameter("VolumeUpdate", clid) + code_model = {"types": [cache_body["type"]["elementType"], volume_body["type"]["elementType"]]} + + plugin.add_body_param_type(code_model, cache_body) + plugin.add_body_param_type(code_model, volume_body) + + typeddict_copies = [t for t in code_model["types"] if t.get("type") == "model" and t.get("base") == "typeddict"] + assert sorted(t["name"] for t in typeddict_copies) == ["CacheUpdate", "VolumeUpdate"] + # The inserted list overload's element references the matching model, not a shared/wrong one. + assert cache_body["type"]["types"][1]["elementType"]["name"] == "CacheUpdate" + assert volume_body["type"]["types"][1]["elementType"]["name"] == "VolumeUpdate" + + +def _json_spread_body_parameter(name: str, cross_lang_id: str) -> dict: + """A spread (base='json') body parameter, as the emitter emits for spread request bodies.""" + model_type = { + "type": "model", + "base": "json", + "name": name, + "crossLanguageDefinitionId": cross_lang_id, + "properties": [], + } + return { + "wireName": "body", + "clientName": "body", + "location": "body", + "optional": False, + "implementation": "Method", + "contentTypes": ["application/json"], + "type": model_type, + } + + +def test_spread_body_reuses_single_underlying_model(): + """A spread body with exactly one same-clid original reuses that original's TypedDict.""" + plugin = _plugin("dpg") + clid = "Contoso.Widget" + widget = _dpg_body_parameter("Widget", clid)["type"] + # Emitter renames a shared spread body clone to ``Request`` but keeps the clid. + create_body = _json_spread_body_parameter("CreateRequest", clid) + code_model = {"types": [widget, create_body["type"]]} + + plugin.add_body_param_type(code_model, create_body) + + inserted = create_body["type"]["types"][1] + # Unambiguous single original -> reference the real model's TypedDict. + assert inserted["name"] == "Widget" + assert inserted["base"] == "typeddict" + + +def test_spread_body_matches_original_by_name_not_first_clid_sibling(): + """When siblings share a clid, a non-renamed spread body must match the SAME-named original.""" + plugin = _plugin("dpg") + clid = "Azure.ResourceManager.Foundations.ResourceUpdateModel" + gadget = _dpg_body_parameter("Gadget", clid)["type"] + widget = _dpg_body_parameter("Widget", clid)["type"] + widget_spread = _json_spread_body_parameter("Widget", clid) + # ``Gadget`` is listed first on purpose: a clid-only lookup would wrongly pick it. + code_model = {"types": [gadget, widget, widget_spread["type"]]} + + plugin.add_body_param_type(code_model, widget_spread) + + inserted = widget_spread["type"]["types"][1] + assert inserted["name"] == "Widget" + + +def test_spread_body_shared_clid_renamed_clones_avoid_cross_reference(): + """Ambiguous same-clid siblings: renamed spread clones must NOT reference the wrong model. + + Both spread bodies are renamed clones (``Request``) of distinct template instantiations + that share the template clid. Since neither renamed name matches an original and multiple + originals share the clid, the choice is ambiguous. Rather than pick the wrong original (the + pre-fix bug produced a dangling ``_types.CacheUpdate`` in the volume operation), each body falls + back to its own uniquely-named TypedDict. + """ + plugin = _plugin("dpg") + clid = "Azure.ResourceManager.Foundations.ResourceUpdateModel" + cache = _dpg_body_parameter("CacheUpdate", clid)["type"] + volume = _dpg_body_parameter("VolumeUpdate", clid)["type"] + cache_body = _json_spread_body_parameter("CacheUpdateRequest", clid) + volume_body = _json_spread_body_parameter("VolumeUpdateRequest", clid) + code_model = {"types": [cache, volume, cache_body["type"], volume_body["type"]]} + + plugin.add_body_param_type(code_model, cache_body) + plugin.add_body_param_type(code_model, volume_body) + + cache_td = cache_body["type"]["types"][1] + volume_td = volume_body["type"]["types"][1] + # No cross-reference: the volume operation must not reference CacheUpdate. + assert cache_td["name"] == "CacheUpdateRequest" + assert volume_td["name"] == "VolumeUpdateRequest" + assert volume_td["name"] != "CacheUpdate" + assert cache_td is not volume_td