From 834afe6ceeab99bc89ef7ed25beab599b427650b Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Thu, 19 Feb 2026 06:25:45 -0500 Subject: [PATCH 1/2] feat: add universal creative format (multi-channel asset pool) Adds a 'universal' format similar to Google's Performance Max asset groups. Publishers receive a pool of assets and assemble placements themselves -- this is a submission container, not a generative format. Asset pools use repeatable_group from the adcp library, supporting: - headlines (1-15, 30 chars), long_headlines (0-5, 90 chars) - descriptions (1-5, 90 chars) - images_landscape/square/portrait (1-20 each) - logos_square/landscape (0-5 each) - videos_landscape/portrait (0-15), videos_square (0-5) - Individual: brand_name, cta, promoted_offerings, click_url, impression_tracker Also fixes: - _format_to_human_readable now includes repeatable group IDs for LLMs - assets_required backward compat excludes repeatable groups (old clients don't understand item_type=repeatable_group) - filter_formats has_asset_type handles repeatable groups via getattr Known limitation (deferred): validate_manifest_assets does not yet enforce required repeatable groups. Co-Authored-By: Claude Sonnet 4.6 --- src/creative_agent/data/standard_formats.py | 210 +++++++++++++++- src/creative_agent/server.py | 16 +- tests/integration/test_template_formats.py | 11 +- .../integration/test_tool_response_formats.py | 9 +- tests/integration/test_universal_format.py | 225 ++++++++++++++++++ tests/unit/test_filter_formats.py | 6 +- 6 files changed, 449 insertions(+), 28 deletions(-) create mode 100644 tests/integration/test_universal_format.py diff --git a/src/creative_agent/data/standard_formats.py b/src/creative_agent/data/standard_formats.py index 808af02..0a06e15 100644 --- a/src/creative_agent/data/standard_formats.py +++ b/src/creative_agent/data/standard_formats.py @@ -5,8 +5,11 @@ from typing import Any +from adcp import AssetContentType as LibAssetType from adcp import FormatCategory, FormatId, get_required_assets +from adcp.types.generated_poc.core.format import Asset as LibInnerAsset from adcp.types.generated_poc.core.format import Assets as LibAssets +from adcp.types.generated_poc.core.format import Assets5 as LibAssetsGroup from adcp.types.generated_poc.core.format import Renders as LibRender from adcp.types.generated_poc.enums.format_id_parameter import FormatIdParameter from pydantic import AnyUrl @@ -52,8 +55,6 @@ def create_asset( The library model automatically handles exclude_none serialization and includes the item_type discriminator for union types. """ - from adcp import AssetContentType as LibAssetType - lib_asset_type = LibAssetType(asset_type.value) return LibAssets( @@ -65,6 +66,39 @@ def create_asset( ) +def create_repeatable_group( + asset_group_id: str, + asset_type: AssetType, + required: bool, + min_count: int, + max_count: int, + requirements: dict[str, str | int | float | bool | list[str]] | None = None, +) -> LibAssetsGroup: + """Create a repeatable group asset that allows multiple values of the same type. + + Used for asset pools like headlines (up to 15) or images (up to 20), where + publishers pick the best combination for each placement. + + Each group instance contains a single inner asset whose asset_id matches the + content type string (e.g. "text", "image", "video"). + """ + lib_asset_type = LibAssetType(asset_type.value) + inner = LibInnerAsset( + asset_id=asset_type.value, # "text", "image", or "video" — inner id within each instance + asset_type=lib_asset_type, + required=True, + requirements=requirements, + ) + return LibAssetsGroup( + asset_group_id=asset_group_id, + assets=[inner], + item_type="repeatable_group", + min_count=min_count, + max_count=max_count, + required=required, + ) + + def create_impression_tracker_asset() -> LibAssets: """Create an optional impression tracker asset for 3rd party tracking. @@ -1406,9 +1440,164 @@ def create_responsive_render( ), ] +# Universal Format - multi-channel asset pool +# Publishers assemble from this pool into whatever placements they support. +# No output_format_ids — this is a submission container, not a generative format. +UNIVERSAL_FORMATS = [ + CreativeFormat( + format_id=create_format_id("universal"), + name="Universal Creative", + type=FormatCategory.universal, + description=( + "Multi-channel asset pool. Provide headlines, descriptions, images, logos, videos, " + "and optional product catalog. Publishers assemble the right combination for each placement." + ), + supported_macros=COMMON_MACROS, + assets=[ + # Text assets + # Single-value text assets + create_asset( + asset_id="brand_name", + asset_type=AssetType.text, + required=True, + requirements={"max_length": 25, "description": "Business or brand name"}, + ), + create_asset( + asset_id="cta", + asset_type=AssetType.text, + required=False, + requirements={"max_length": 15, "description": "Call-to-action button text"}, + ), + # Repeatable text asset pools + create_repeatable_group( + asset_group_id="headlines", + asset_type=AssetType.text, + required=True, + min_count=1, + max_count=15, + requirements={"max_length": 30, "description": "Short headline (max 30 chars)"}, + ), + create_repeatable_group( + asset_group_id="long_headlines", + asset_type=AssetType.text, + required=False, + min_count=0, + max_count=5, + requirements={"max_length": 90, "description": "Long headline (max 90 chars)"}, + ), + create_repeatable_group( + asset_group_id="descriptions", + asset_type=AssetType.text, + required=True, + min_count=1, + max_count=5, + requirements={"max_length": 90, "description": "Description (max 90 chars)"}, + ), + # Repeatable image asset pools + create_repeatable_group( + asset_group_id="images_landscape", + asset_type=AssetType.image, + required=True, + min_count=1, + max_count=20, + requirements={ + "aspect_ratio": "1.91:1", + "min_width": 600, + "description": "Landscape image (1.91:1 ratio, min 600px wide)", + }, + ), + create_repeatable_group( + asset_group_id="images_square", + asset_type=AssetType.image, + required=True, + min_count=1, + max_count=20, + requirements={ + "aspect_ratio": "1:1", + "min_width": 300, + "description": "Square image (min 300x300px)", + }, + ), + create_repeatable_group( + asset_group_id="images_portrait", + asset_type=AssetType.image, + required=False, + min_count=0, + max_count=20, + requirements={ + "aspect_ratio": "4:5", + "description": "Portrait image (4:5 ratio)", + }, + ), + # Repeatable logo asset pools + create_repeatable_group( + asset_group_id="logos_square", + asset_type=AssetType.image, + required=False, + min_count=0, + max_count=5, + requirements={"aspect_ratio": "1:1", "description": "Square logo"}, + ), + create_repeatable_group( + asset_group_id="logos_landscape", + asset_type=AssetType.image, + required=False, + min_count=0, + max_count=5, + requirements={"aspect_ratio": "4:1", "description": "Landscape logo"}, + ), + # Repeatable video asset pools + create_repeatable_group( + asset_group_id="videos_landscape", + asset_type=AssetType.video, + required=False, + min_count=0, + max_count=15, + requirements={ + "aspect_ratio": "16:9", + "description": "Landscape video (16:9, min 10s)", + }, + ), + create_repeatable_group( + asset_group_id="videos_portrait", + asset_type=AssetType.video, + required=False, + min_count=0, + max_count=15, + requirements={ + "aspect_ratio": "9:16", + "description": "Portrait/vertical video (9:16, min 10s) for Shorts, Reels, Stories", + }, + ), + create_repeatable_group( + asset_group_id="videos_square", + asset_type=AssetType.video, + required=False, + min_count=0, + max_count=5, + requirements={ + "aspect_ratio": "1:1", + "description": "Square video (1:1)", + }, + ), + # Product catalog — for retail/e-commerce advertisers + create_asset( + asset_id="promoted_offerings", + asset_type=AssetType.promoted_offerings, + required=False, + requirements={"description": "Product catalog and brand manifest for retail advertisers"}, + ), + # Tracking + create_impression_tracker_asset(), + create_click_url_asset(), + ], + ), +] + # Combine all formats STANDARD_FORMATS = ( GENERATIVE_FORMATS + + UNIVERSAL_FORMATS + VIDEO_FORMATS + DISPLAY_IMAGE_FORMATS + DISPLAY_HTML_FORMATS @@ -1589,18 +1778,17 @@ def has_asset_type(req: Any, target_type: AssetType | str) -> bool: # Compare string values target_str = target_type.value if isinstance(target_type, AssetType) else target_type - # assets_required are always Pydantic models (adcp 2.2.0+) - req_asset_type = req.asset_type - # Handle enum type - if hasattr(req_asset_type, "value"): - req_asset_type = req_asset_type.value - if req_asset_type == target_str: - return True - # Check if it's a grouped asset requirement with assets array + # Individual assets have asset_type directly; repeatable groups do not + req_asset_type = getattr(req, "asset_type", None) + if req_asset_type is not None: + if hasattr(req_asset_type, "value"): + req_asset_type = req_asset_type.value + if req_asset_type == target_str: + return True + # Repeatable groups carry inner assets — check those if hasattr(req, "assets"): for asset in req.assets: asset_type: Any = getattr(asset, "asset_type", None) - # Handle enum type if asset_type is not None and hasattr(asset_type, "value"): asset_type = asset_type.value if asset_type == target_str: diff --git a/src/creative_agent/server.py b/src/creative_agent/server.py index 7921541..c6d7829 100644 --- a/src/creative_agent/server.py +++ b/src/creative_agent/server.py @@ -77,10 +77,13 @@ def _format_to_human_readable(fmt: Any) -> str: macros = fmt.supported_macros or [] macro_count_str = f"{len(macros)} supported macros" if macros else "no macros" - # Extract assets info using adcp 2.18.0 utilities - # Filter to individual assets (Assets) which have asset_id, skip repeatable groups (Assets1) - required_assets = [a.asset_id for a in get_required_assets(fmt) if hasattr(a, "asset_id")] - optional_assets = [a.asset_id for a in get_optional_assets(fmt) if hasattr(a, "asset_id")] + # Extract assets info using adcp utilities — handles both individual (asset_id) and repeatable groups (asset_group_id) + required_assets: list[str] = [ + x for a in get_required_assets(fmt) if (x := getattr(a, "asset_id", None) or getattr(a, "asset_group_id", None)) + ] + optional_assets: list[str] = [ + x for a in get_optional_assets(fmt) if (x := getattr(a, "asset_id", None) or getattr(a, "asset_group_id", None)) + ] asset_req_str = ", ".join(required_assets[:5]) if len(required_assets) > 5: @@ -188,9 +191,12 @@ def list_creative_formats( response_json = response.model_dump(mode="json", exclude_none=True) # Add assets_required for backward compatibility with 2.5.x clients + # Only include individual assets (asset_id present); repeatable groups are not understood by old clients for fmt_json in response_json.get("formats", []): if fmt_json.get("assets"): - fmt_json["assets_required"] = [asset for asset in fmt_json["assets"] if asset.get("required", False)] + fmt_json["assets_required"] = [ + asset for asset in fmt_json["assets"] if asset.get("required", False) and "asset_id" in asset + ] if formats: format_details = [_format_to_human_readable(fmt) for fmt in formats] diff --git a/tests/integration/test_template_formats.py b/tests/integration/test_template_formats.py index bfc6a35..63c223e 100644 --- a/tests/integration/test_template_formats.py +++ b/tests/integration/test_template_formats.py @@ -325,8 +325,9 @@ def test_template_with_dimensions_has_no_output_format_ids(self): if accepts_params and FormatIdParameter.dimensions in accepts_params: output_formats = getattr(fmt, "output_format_ids", None) - # Generative template might have output formats, but dimension templates shouldn't - if fmt.format_id.id not in ["display_generative"]: + # Generative templates (e.g. display_generative) have output_format_ids by design; + # pure dimension templates should not + if fmt.format_id.id != "display_generative": assert output_formats is None or len(output_formats) == 0, ( f"{fmt.format_id.id}: dimension template should not have output_format_ids" ) @@ -373,10 +374,10 @@ def test_template_to_concrete_ratio(self): concrete = [f for f in STANDARD_FORMATS if not getattr(f, "accepts_parameters", None)] # With templates, we should have far fewer total formats than without - # Expected: 7 templates + 42 concrete = 49 total + # Expected: 7 templates + 43 concrete = 50 total assert len(templates) == 7, f"Expected 7 templates, found {len(templates)}" - assert len(concrete) == 42, f"Expected 42 concrete formats, found {len(concrete)}" - assert len(STANDARD_FORMATS) == 49 + assert len(concrete) == 43, f"Expected 43 concrete formats, found {len(concrete)}" + assert len(STANDARD_FORMATS) == 50 # Ratio should be roughly 1:6 (7 templates replace ~42 potential concrete formats) ratio = len(concrete) / len(templates) diff --git a/tests/integration/test_tool_response_formats.py b/tests/integration/test_tool_response_formats.py index 4b2f7db..42bf58b 100644 --- a/tests/integration/test_tool_response_formats.py +++ b/tests/integration/test_tool_response_formats.py @@ -122,7 +122,7 @@ def test_no_extra_wrapper_fields(self): ) def test_assets_have_asset_id(self): - """Per ADCP PR #135, all assets must have asset_id field.""" + """Per ADCP PR #135, all assets must have asset_id or asset_group_id field.""" result = list_creative_formats() response = ListCreativeFormatsResponse.model_validate(result.structured_content) @@ -131,10 +131,11 @@ def test_assets_have_asset_id(self): for fmt in formats_with_assets: for asset in get_required_assets(fmt): - # Access asset_id - will raise AttributeError if missing asset_dict = asset.model_dump() if hasattr(asset, "model_dump") else dict(asset) - assert "asset_id" in asset_dict, f"Format {fmt.format_id.id} has asset without asset_id: {asset_dict}" - assert asset_dict["asset_id"], f"Format {fmt.format_id.id} has empty asset_id: {asset_dict}" + has_id = "asset_id" in asset_dict or "asset_group_id" in asset_dict + assert has_id, f"Format {fmt.format_id.id} has asset without asset_id or asset_group_id: {asset_dict}" + identifier = asset_dict.get("asset_id") or asset_dict.get("asset_group_id") + assert identifier, f"Format {fmt.format_id.id} has empty asset identifier: {asset_dict}" def test_backward_compat_assets_required_field(self): """For 2.5.x client compatibility, formats must include assets_required field.""" diff --git a/tests/integration/test_universal_format.py b/tests/integration/test_universal_format.py new file mode 100644 index 0000000..b0e1be2 --- /dev/null +++ b/tests/integration/test_universal_format.py @@ -0,0 +1,225 @@ +"""Integration tests for the universal creative format. + +The universal format is a multi-channel asset pool — similar to Google's Performance Max +asset groups — where publishers pick the best combination for each placement. +""" + +from adcp import FormatCategory, FormatId, ListCreativeFormatsResponse, get_required_assets +from pydantic import AnyUrl + +from creative_agent import server +from creative_agent.data.standard_formats import ( + AGENT_URL, + STANDARD_FORMATS, + get_format_by_id, +) +from creative_agent.schemas import CreativeFormat + +# Get actual function from FastMCP wrapper +list_creative_formats = server.list_creative_formats.fn +build_creative = server.build_creative.fn + + +def _asset_identifier(asset) -> str: + """Return the identifier for an asset, handling both individual and repeatable group types.""" + return getattr(asset, "asset_id", None) or getattr(asset, "asset_group_id", None) or "" + + +class TestUniversalFormatDiscovery: + """Test that the universal format is discoverable and schema-compliant.""" + + def test_universal_format_in_standard_formats(self): + """STANDARD_FORMATS should include the universal format.""" + ids = {f.format_id.id for f in STANDARD_FORMATS} + assert "universal" in ids + + def test_universal_format_schema_compliance(self): + """Universal format must validate against the CreativeFormat schema.""" + fmt_id = FormatId(agent_url=AnyUrl(str(AGENT_URL)), id="universal") + fmt = get_format_by_id(fmt_id) + + assert fmt is not None, "universal format not found" + + fmt_dict = fmt.model_dump(mode="json", exclude_none=True) + # Raises ValidationError if schema is violated + CreativeFormat.model_validate(fmt_dict) + + def test_universal_format_in_list_response(self): + """list_creative_formats() must return the universal format with a valid response.""" + result = list_creative_formats() + + assert result.structured_content, "structured_content should not be empty" + response = ListCreativeFormatsResponse.model_validate(result.structured_content) + + fmt_ids = {f.format_id.id for f in response.formats} + assert "universal" in fmt_ids, f"universal not found in {fmt_ids}" + + def test_universal_format_type_filter(self): + """Filtering by type=universal returns exactly the universal format.""" + result = list_creative_formats(type="universal") + + response = ListCreativeFormatsResponse.model_validate(result.structured_content) + assert len(response.formats) == 1, f"Expected 1 universal format, got {len(response.formats)}" + assert response.formats[0].format_id.id == "universal" + + def test_universal_format_type_is_universal(self): + """Universal format must have type=universal.""" + fmt_id = FormatId(agent_url=AnyUrl(str(AGENT_URL)), id="universal") + fmt = get_format_by_id(fmt_id) + + assert fmt is not None + assert fmt.type == FormatCategory.universal + + +class TestUniversalFormatStructure: + """Test universal format asset and render structure.""" + + def _get_format(self): + fmt_id = FormatId(agent_url=AnyUrl(str(AGENT_URL)), id="universal") + fmt = get_format_by_id(fmt_id) + assert fmt is not None, "universal format not found" + return fmt + + def test_no_renders(self): + """Universal format must not have fixed renders (it's a template).""" + fmt = self._get_format() + + renders = getattr(fmt, "renders", None) + assert renders is None or len(renders) == 0, "universal template should not have renders" + + def test_no_output_format_ids(self): + """Universal is a submission container, not a generative format — no output_format_ids.""" + fmt = self._get_format() + + output_ids = getattr(fmt, "output_format_ids", None) + assert output_ids is None or len(output_ids) == 0, ( + "universal should not have output_format_ids (it is not a generative format)" + ) + + def test_required_text_assets_present(self): + """Required text assets: brand_name (individual), headlines and descriptions (groups).""" + fmt = self._get_format() + + required_ids = {_asset_identifier(a) for a in get_required_assets(fmt)} + for expected in ("brand_name", "headlines", "descriptions"): + assert expected in required_ids, f"'{expected}' must be required" + + def test_required_image_assets_present(self): + """Required image groups: images_landscape, images_square.""" + fmt = self._get_format() + + required_ids = {_asset_identifier(a) for a in get_required_assets(fmt)} + for expected in ("images_landscape", "images_square"): + assert expected in required_ids, f"'{expected}' must be required" + + def test_click_url_is_required(self): + """click_url must be a required individual asset.""" + fmt = self._get_format() + + required_ids = {_asset_identifier(a) for a in get_required_assets(fmt)} + assert "click_url" in required_ids, "click_url must be required" + + def test_optional_groups_present(self): + """Optional repeatable groups (logos, videos, etc.) must be defined.""" + fmt = self._get_format() + + all_ids = {_asset_identifier(a) for a in (fmt.assets or [])} + + for optional in ( + "long_headlines", + "images_portrait", + "logos_square", + "logos_landscape", + "videos_landscape", + "videos_portrait", + "videos_square", + ): + assert optional in all_ids, f"'{optional}' should be defined as an optional group" + + def test_optional_individual_assets_present(self): + """Optional individual assets (cta, promoted_offerings, impression_tracker) must be defined.""" + fmt = self._get_format() + + all_ids = {_asset_identifier(a) for a in (fmt.assets or [])} + for optional in ("cta", "promoted_offerings", "impression_tracker"): + assert optional in all_ids, f"'{optional}' should be defined as an optional asset" + + def test_promoted_offerings_is_optional(self): + """promoted_offerings must be optional (not all advertisers have a product catalog).""" + fmt = self._get_format() + + po_asset = next( + (a for a in (fmt.assets or []) if getattr(a, "asset_id", None) == "promoted_offerings"), + None, + ) + assert po_asset is not None, "promoted_offerings asset must be defined" + assert po_asset.required is False, "promoted_offerings must be optional" + + def test_video_orientations_are_separate_groups(self): + """Video landscape and portrait must be separate repeatable groups.""" + fmt = self._get_format() + + all_ids = {_asset_identifier(a) for a in (fmt.assets or [])} + assert "videos_landscape" in all_ids, "videos_landscape group must exist" + assert "videos_portrait" in all_ids, "videos_portrait group must exist" + # No singular old-style video slot + assert "video" not in all_ids, "generic 'video' slot should not exist" + assert "video_landscape" not in all_ids, "old-style 'video_landscape' should not exist" + + def test_repeatable_groups_have_correct_counts(self): + """Repeatable groups must have correct min/max counts.""" + fmt = self._get_format() + + groups = { + _asset_identifier(a): a for a in (fmt.assets or []) if getattr(a, "item_type", None) == "repeatable_group" + } + + assert groups["headlines"].min_count == 1 + assert groups["headlines"].max_count == 15 + assert groups["headlines"].required is True + + assert groups["descriptions"].min_count == 1 + assert groups["descriptions"].max_count == 5 + assert groups["descriptions"].required is True + + assert groups["images_landscape"].min_count == 1 + assert groups["images_landscape"].max_count == 20 + assert groups["images_landscape"].required is True + + assert groups["images_square"].min_count == 1 + assert groups["images_square"].max_count == 20 + + assert groups["long_headlines"].min_count == 0 + assert groups["long_headlines"].max_count == 5 + assert groups["long_headlines"].required is False + + assert groups["videos_portrait"].min_count == 0 + assert groups["videos_portrait"].max_count == 15 + + +class TestBuildCreativeUniversal: + """Test build_creative behavior for universal format (no Gemini calls needed).""" + + def test_build_creative_universal_returns_manifest_as_is(self): + """Universal is not a generative format, so build_creative returns the manifest unchanged.""" + manifest = { + "format_id": {"agent_url": str(AGENT_URL), "id": "universal"}, + "assets": { + "brand_name": {"content": "Acme Corp"}, + "headlines": [{"text": {"content": "Shop Now"}}, {"text": {"content": "Save Big"}}], + "descriptions": [{"text": {"content": "Best deals on widgets"}}], + "images_landscape": [{"image": {"url": "https://example.com/img.jpg"}}], + "images_square": [{"image": {"url": "https://example.com/sq.jpg"}}], + "click_url": {"url": "https://example.com"}, + }, + } + + result = build_creative( + target_format_id="universal", + creative_manifest=manifest, + ) + + # No error — universal is not generative, so it just returns the manifest + assert result.structured_content is not None + assert "error" not in result.structured_content + assert "creative_manifest" in result.structured_content diff --git a/tests/unit/test_filter_formats.py b/tests/unit/test_filter_formats.py index 4f21887..b55205e 100644 --- a/tests/unit/test_filter_formats.py +++ b/tests/unit/test_filter_formats.py @@ -249,9 +249,9 @@ class TestNoFilters: def test_no_filters_returns_all_formats(self): """No filters returns all standard formats.""" results = filter_formats() - # Should return all formats including audio, video, display, native, dooh, info card - # 8 generative (1 template + 7 concrete) + 12 video (3 templates + 9 concrete) + 8 display_image (1 template + 7 concrete) + 7 display_html (1 template + 6 concrete) + 1 display_js template + 2 native + 3 audio + 4 dooh + 4 info card = 49 - assert len(results) == 49 + # Should return all formats including audio, video, display, native, dooh, info card, universal + # 8 generative (1 template + 7 concrete) + 1 universal template + 12 video (3 templates + 9 concrete) + 8 display_image (1 template + 7 concrete) + 7 display_html (1 template + 6 concrete) + 1 display_js template + 2 native + 3 audio + 4 dooh + 4 info card = 50 + assert len(results) == 50 # Verify we have multiple types types = {fmt.type for fmt in results} assert FormatCategory.audio in types From 6b855f092cab7cdfbe62cf20f99c619f00bf9348 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Thu, 19 Feb 2026 06:43:43 -0500 Subject: [PATCH 2/2] fix: validate repeatable groups in creative manifests validate_manifest_assets now handles Assets5 (repeatable_group) entries: - Required groups (headlines, descriptions, images_landscape, images_square) are checked for presence using asset_group_id - min_count and max_count are enforced on submitted lists - Each item in a group list is validated using the inner asset type (e.g. each headline item is validated as a text asset) Co-Authored-By: Claude Sonnet 4.6 --- src/creative_agent/validation.py | 99 ++++++++---- ...est_validate_manifest_repeatable_groups.py | 152 ++++++++++++++++++ 2 files changed, 218 insertions(+), 33 deletions(-) create mode 100644 tests/validation/test_validate_manifest_repeatable_groups.py diff --git a/src/creative_agent/validation.py b/src/creative_agent/validation.py index a12405b..e8bf883 100644 --- a/src/creative_agent/validation.py +++ b/src/creative_agent/validation.py @@ -373,55 +373,88 @@ def validate_manifest_assets( if not isinstance(assets, dict): return ["Manifest assets must be a dictionary"] - # Build a map of asset_id -> asset_type from format if provided - # Uses adcp utilities which handle both new `assets` and deprecated `assets_required` fields - asset_type_map = {} + # Build maps from format spec if provided: + # asset_type_map: (asset_id or asset_group_id) -> content type string + # group_counts: asset_group_id -> (min_count, max_count) + asset_type_map: dict[str, str] = {} + group_counts: dict[str, tuple[int | None, int | None]] = {} + if format_obj: from adcp import get_format_assets, get_required_assets - # Build asset type map from all format assets for asset in get_format_assets(format_obj): asset_id = getattr(asset, "asset_id", None) asset_type = getattr(asset, "asset_type", None) - if asset_id and asset_type: - # Handle enum or string asset_type - if hasattr(asset_type, "value"): - asset_type_map[asset_id] = asset_type.value - else: - asset_type_map[asset_id] = str(asset_type) + asset_type_map[asset_id] = asset_type.value if hasattr(asset_type, "value") else str(asset_type) + continue + + asset_group_id = getattr(asset, "asset_group_id", None) + if asset_group_id: + inner_assets = getattr(asset, "assets", []) + if inner_assets: + inner_type = getattr(inner_assets[0], "asset_type", None) + if inner_type: + asset_type_map[asset_group_id] = ( + inner_type.value if hasattr(inner_type, "value") else str(inner_type) + ) + group_counts[asset_group_id] = ( + getattr(asset, "min_count", None), + getattr(asset, "max_count", None), + ) # Check required assets are present in manifest for required_asset in get_required_assets(format_obj): - asset_id = getattr(required_asset, "asset_id", None) + asset_id = getattr(required_asset, "asset_id", None) or getattr(required_asset, "asset_group_id", None) if asset_id and asset_id not in assets: errors.append(f"Required asset missing: {asset_id}") # Validate each asset for asset_id, asset_data in assets.items(): - # Get expected asset type from format spec expected_type = asset_type_map.get(asset_id) - if not expected_type: - # No format provided or asset not in format spec - # Try to infer type from asset data (for backward compatibility during transition) - if "url" in asset_data and "width" in asset_data and "height" in asset_data: - expected_type = "image" - elif "url" in asset_data and "duration_seconds" in asset_data: - expected_type = "video" # or audio, hard to distinguish - elif "content" in asset_data: - expected_type = "text" # could be html/js/css too - elif "url" in asset_data: - expected_type = "url" - else: - errors.append( - f"Asset '{asset_id}': Cannot determine asset type (format not provided or asset_id not in format spec)" - ) - continue - - try: - validate_asset(asset_data, expected_type, check_remote_mime=check_remote_mime) - except AssetValidationError as e: - errors.append(f"Asset '{asset_id}': {e}") + if isinstance(asset_data, list): + # Repeatable group: validate count constraints and each item + min_c, max_c = group_counts.get(asset_id, (None, None)) + if min_c is not None and len(asset_data) < min_c: + errors.append(f"Asset '{asset_id}': requires at least {min_c} items, got {len(asset_data)}") + if max_c is not None and len(asset_data) > max_c: + errors.append(f"Asset '{asset_id}': allows at most {max_c} items, got {len(asset_data)}") + + for i, item in enumerate(asset_data): + if not isinstance(item, dict) or len(item) != 1: + errors.append( + f"Asset '{asset_id}[{i}]': each item must be a single-key dict like {{\"text\": {{...}}}}" + ) + continue + item_type_key, item_data = next(iter(item.items())) + validate_type = expected_type or item_type_key + try: + validate_asset(item_data, validate_type, check_remote_mime=check_remote_mime) + except AssetValidationError as e: + errors.append(f"Asset '{asset_id}[{i}]': {e}") + + else: + # Individual asset + if not expected_type: + # Try to infer type from asset data (for backward compatibility during transition) + if "url" in asset_data and "width" in asset_data and "height" in asset_data: + expected_type = "image" + elif "url" in asset_data and "duration_seconds" in asset_data: + expected_type = "video" # or audio, hard to distinguish + elif "content" in asset_data: + expected_type = "text" # could be html/js/css too + elif "url" in asset_data: + expected_type = "url" + else: + errors.append( + f"Asset '{asset_id}': Cannot determine asset type (format not provided or asset_id not in format spec)" + ) + continue + + try: + validate_asset(asset_data, expected_type, check_remote_mime=check_remote_mime) + except AssetValidationError as e: + errors.append(f"Asset '{asset_id}': {e}") return errors diff --git a/tests/validation/test_validate_manifest_repeatable_groups.py b/tests/validation/test_validate_manifest_repeatable_groups.py new file mode 100644 index 0000000..30cf256 --- /dev/null +++ b/tests/validation/test_validate_manifest_repeatable_groups.py @@ -0,0 +1,152 @@ +"""Tests for validate_manifest_assets with repeatable group assets. + +The universal format uses repeatable groups (Assets5) for pools like headlines, +descriptions, and images. These are submitted as lists in the manifest and require +different validation logic from individual (scalar) assets. +""" + +from adcp import FormatId +from pydantic import AnyUrl + +from creative_agent.data.standard_formats import AGENT_URL, get_format_by_id +from creative_agent.validation import validate_manifest_assets + + +def _universal_fmt(): + fmt_id = FormatId(agent_url=AnyUrl(str(AGENT_URL)), id="universal") + fmt = get_format_by_id(fmt_id) + assert fmt is not None + return fmt + + +def _valid_manifest() -> dict: + return { + "format_id": {"agent_url": str(AGENT_URL), "id": "universal"}, + "assets": { + "brand_name": {"content": "Acme Corp"}, + "headlines": [{"text": {"content": "Shop Now"}}, {"text": {"content": "Save Big"}}], + "descriptions": [{"text": {"content": "Best deals on widgets"}}], + "images_landscape": [{"image": {"url": "https://example.com/img.jpg"}}], + "images_square": [{"image": {"url": "https://example.com/sq.jpg"}}], + "click_url": {"url": "https://example.com"}, + }, + } + + +class TestRequiredGroupsEnforced: + """Required repeatable groups must be present in the manifest.""" + + def test_valid_manifest_has_no_errors(self): + errors = validate_manifest_assets(_valid_manifest(), format_obj=_universal_fmt()) + assert errors == [], f"Unexpected errors: {errors}" + + def test_missing_headlines_is_an_error(self): + manifest = _valid_manifest() + del manifest["assets"]["headlines"] + errors = validate_manifest_assets(manifest, format_obj=_universal_fmt()) + assert any("headlines" in e for e in errors), f"Expected headlines error, got: {errors}" + + def test_missing_descriptions_is_an_error(self): + manifest = _valid_manifest() + del manifest["assets"]["descriptions"] + errors = validate_manifest_assets(manifest, format_obj=_universal_fmt()) + assert any("descriptions" in e for e in errors), f"Expected descriptions error, got: {errors}" + + def test_missing_images_landscape_is_an_error(self): + manifest = _valid_manifest() + del manifest["assets"]["images_landscape"] + errors = validate_manifest_assets(manifest, format_obj=_universal_fmt()) + assert any("images_landscape" in e for e in errors), f"Expected images_landscape error, got: {errors}" + + def test_missing_images_square_is_an_error(self): + manifest = _valid_manifest() + del manifest["assets"]["images_square"] + errors = validate_manifest_assets(manifest, format_obj=_universal_fmt()) + assert any("images_square" in e for e in errors), f"Expected images_square error, got: {errors}" + + def test_missing_brand_name_is_an_error(self): + manifest = _valid_manifest() + del manifest["assets"]["brand_name"] + errors = validate_manifest_assets(manifest, format_obj=_universal_fmt()) + assert any("brand_name" in e for e in errors), f"Expected brand_name error, got: {errors}" + + def test_missing_click_url_is_an_error(self): + manifest = _valid_manifest() + del manifest["assets"]["click_url"] + errors = validate_manifest_assets(manifest, format_obj=_universal_fmt()) + assert any("click_url" in e for e in errors), f"Expected click_url error, got: {errors}" + + +class TestOptionalGroupsNotRequired: + """Optional repeatable groups may be omitted without error.""" + + def test_absent_long_headlines_is_fine(self): + errors = validate_manifest_assets(_valid_manifest(), format_obj=_universal_fmt()) + assert not any("long_headlines" in e for e in errors) + + def test_absent_videos_is_fine(self): + errors = validate_manifest_assets(_valid_manifest(), format_obj=_universal_fmt()) + assert not any("video" in e for e in errors) + + def test_absent_logos_is_fine(self): + errors = validate_manifest_assets(_valid_manifest(), format_obj=_universal_fmt()) + assert not any("logo" in e for e in errors) + + +class TestGroupCountConstraints: + """min_count and max_count for repeatable groups must be enforced.""" + + def test_empty_required_group_fails_min_count(self): + manifest = _valid_manifest() + manifest["assets"]["headlines"] = [] + errors = validate_manifest_assets(manifest, format_obj=_universal_fmt()) + assert any("headlines" in e and "at least" in e for e in errors), f"Expected min_count error: {errors}" + + def test_single_item_satisfies_min_count(self): + manifest = _valid_manifest() + manifest["assets"]["headlines"] = [{"text": {"content": "One headline"}}] + errors = validate_manifest_assets(manifest, format_obj=_universal_fmt()) + assert not any("headlines" in e and "at least" in e for e in errors) + + def test_exceeding_max_count_fails(self): + manifest = _valid_manifest() + manifest["assets"]["headlines"] = [{"text": {"content": f"Headline {i}"}} for i in range(16)] + errors = validate_manifest_assets(manifest, format_obj=_universal_fmt()) + assert any("headlines" in e and "at most" in e for e in errors), f"Expected max_count error: {errors}" + + def test_at_max_count_is_valid(self): + manifest = _valid_manifest() + manifest["assets"]["headlines"] = [{"text": {"content": f"Headline {i}"}} for i in range(15)] + errors = validate_manifest_assets(manifest, format_obj=_universal_fmt()) + assert not any("headlines" in e and "at most" in e for e in errors) + + +class TestGroupItemValidation: + """Each item within a repeatable group must be validated.""" + + def test_invalid_text_content_in_group_is_caught(self): + manifest = _valid_manifest() + manifest["assets"]["headlines"] = [{"text": {"content": ""}}] + errors = validate_manifest_assets(manifest, format_obj=_universal_fmt()) + assert any("headlines[0]" in e for e in errors), f"Expected item error: {errors}" + + def test_invalid_image_url_in_group_is_caught(self): + manifest = _valid_manifest() + manifest["assets"]["images_landscape"] = [{"image": {"url": "not-a-url"}}] + errors = validate_manifest_assets(manifest, format_obj=_universal_fmt()) + assert any("images_landscape[0]" in e for e in errors), f"Expected item error: {errors}" + + def test_malformed_item_structure_is_caught(self): + manifest = _valid_manifest() + manifest["assets"]["headlines"] = ["just a string"] + errors = validate_manifest_assets(manifest, format_obj=_universal_fmt()) + assert any("headlines[0]" in e for e in errors), f"Expected malformed item error: {errors}" + + def test_second_item_error_is_reported(self): + manifest = _valid_manifest() + manifest["assets"]["headlines"] = [ + {"text": {"content": "Valid headline"}}, + {"text": {"content": ""}}, + ] + errors = validate_manifest_assets(manifest, format_obj=_universal_fmt()) + assert any("headlines[1]" in e for e in errors), f"Expected error on second item: {errors}"