diff --git a/src/adloop/ads/write.py b/src/adloop/ads/write.py index ed25b96..54d0770 100644 --- a/src/adloop/ads/write.py +++ b/src/adloop/ads/write.py @@ -509,6 +509,175 @@ def draft_responsive_search_ad( return preview +# Fires when an in-place RSA update replaces headlines or descriptions. This is +# NOT a free edit: swapping the creative text resets the ad's optimization even +# though the ad keeps its ID. This is the single most important thing to relay +# to the user before applying such a change. +_RSA_TEXT_UPDATE_WARNING = ( + "Replacing headlines/descriptions is NOT a free in-place edit. Even though " + "the ad keeps its ID, Google treats the creative as new: this resets the " + "ad's asset-combination learning and wipes its performance history, and the " + "ad is sent back through policy review before it can serve again. Only " + "replace headlines/descriptions when the copy genuinely needs to change — " + "for URL or display-path tweaks these effects do not apply." +) + + +def update_responsive_search_ad( + config: AdLoopConfig, + *, + customer_id: str = "", + ad_id: str = "", + headlines: list[str | dict] | None = None, + descriptions: list[str | dict] | None = None, + final_url: str = "", + path1: str = "", + path2: str = "", + clear_path1: bool = False, + clear_path2: bool = False, +) -> dict: + """Draft an in-place update on an existing RSA — returns a PREVIEW. + + Mutates fields on an existing Responsive Search Ad in place via the + Google Ads API v23 ``AdService.MutateAds``. The ad keeps its ID. The + following fields are mutable in place: ``final_urls``, + ``responsive_search_ad.path1``, ``responsive_search_ad.path2``, + ``responsive_search_ad.headlines``, ``responsive_search_ad.descriptions``. + + IMPORTANT — replacing headlines or descriptions is NOT a cost-free edit. + Even though the ad ID is preserved, swapping the creative text RESETS the + ad's asset-combination learning and performance reporting and sends the ad + BACK THROUGH policy review — Google treats the creative as new for + optimization. URL-only and display-path-only edits do not incur this reset. + The returned preview carries a prominent warning whenever headlines or + descriptions are being replaced; relay it to the user before applying. + + Headlines/descriptions are list-replace: when provided, the entire list + swaps in. Google's RSA constraints still apply — 3-15 headlines, + 2-4 descriptions, 30/90 char caps, pin-slot rules — and are validated + here before the plan is stored. Each entry may be a plain string + (unpinned) or ``{"text": "...", "pinned_field": "HEADLINE_1"}``. + + Argument semantics: + - ``headlines`` / ``descriptions`` None or [] -> no change + - ``final_url`` empty -> no change; non-empty -> replaces final_urls + - ``path1`` / ``path2`` empty -> no change; non-empty -> sets value + - ``clear_path1`` / ``clear_path2`` True -> set the path to empty + (overrides the corresponding path string argument) + + At least one mutation must be requested. Call ``confirm_and_apply`` with + the returned plan_id to execute. + """ + from adloop.safety.guards import SafetyViolation, check_blocked_operation + from adloop.safety.preview import ChangePlan, store_plan + + try: + check_blocked_operation("update_responsive_search_ad", config.safety) + except SafetyViolation as e: + return {"error": str(e)} + + errors: list[str] = [] + + if not ad_id: + errors.append("ad_id is required") + elif not str(ad_id).isdigit(): + errors.append("ad_id must be a numeric ID") + + final_url = (final_url or "").strip() + path1 = (path1 or "").strip() + path2 = (path2 or "").strip() + + if path1 and len(path1) > 15: + errors.append(f"path1 must be 15 chars or fewer (got {len(path1)})") + if path2 and len(path2) > 15: + errors.append(f"path2 must be 15 chars or fewer (got {len(path2)})") + + norm_headlines: list[dict] = [] + norm_descriptions: list[dict] = [] + if headlines: + try: + norm_headlines = _normalize_rsa_assets(headlines) + except ValueError as e: + errors.append(str(e)) + if descriptions: + try: + norm_descriptions = _normalize_rsa_assets(descriptions) + except ValueError as e: + errors.append(str(e)) + + if norm_headlines or norm_descriptions: + # Only enforce count on the list being replaced. The other list is + # untouched on the live ad, so its size on the wire is whatever the + # ad already has — not our problem to validate. + errors.extend( + _validate_rsa_assets( + norm_headlines, + norm_descriptions, + enforce_headline_count=bool(norm_headlines), + enforce_description_count=bool(norm_descriptions), + ) + ) + + has_url_change = bool(final_url) + has_path1_change = bool(path1) or clear_path1 + has_path2_change = bool(path2) or clear_path2 + has_headlines_change = bool(norm_headlines) + has_descriptions_change = bool(norm_descriptions) + + if not ( + has_url_change + or has_path1_change + or has_path2_change + or has_headlines_change + or has_descriptions_change + ): + errors.append( + "No changes specified — provide final_url, path1, path2, " + "clear_path1, clear_path2, headlines, or descriptions" + ) + + if errors: + return {"error": "Validation failed", "details": errors} + + if has_url_change: + url_check = _validate_urls([final_url]) + if url_check.get(final_url): + return { + "error": "URL validation failed", + "details": [ + f"final_url '{final_url}' is not reachable: " + f"{url_check[final_url]}. Ads MUST point to working URLs." + ], + } + + changes: dict = {"ad_id": str(ad_id)} + if has_url_change: + changes["final_url"] = final_url + if has_path1_change: + changes["path1"] = "" if clear_path1 else path1 + if has_path2_change: + changes["path2"] = "" if clear_path2 else path2 + if has_headlines_change: + changes["headlines"] = norm_headlines + if has_descriptions_change: + changes["descriptions"] = norm_descriptions + + plan = ChangePlan( + operation="update_responsive_search_ad", + entity_type="ad", + entity_id=str(ad_id), + customer_id=customer_id, + changes=changes, + ) + store_plan(plan) + preview = plan.to_preview() + # Only warn when the creative text is actually being replaced — URL/path + # edits do not trigger the learning reset or policy re-review. + if has_headlines_change or has_descriptions_change: + preview["warnings"] = [_RSA_TEXT_UPDATE_WARNING] + return preview + + def draft_keywords( config: AdLoopConfig, *, @@ -2125,25 +2294,32 @@ def _check_broad_match_safety( return [] -def _validate_rsa( - ad_group_id: str, +def _validate_rsa_assets( headlines: list[dict], descriptions: list[dict], - final_url: str, + enforce_headline_count: bool = True, + enforce_description_count: bool = True, ) -> list[str]: - errors = [] - if not ad_group_id: - errors.append("ad_group_id is required") - if not final_url: - errors.append("final_url is required") - if len(headlines) < 3: - errors.append(f"Need at least 3 headlines, got {len(headlines)}") - if len(headlines) > 15: - errors.append(f"Maximum 15 headlines, got {len(headlines)}") - if len(descriptions) < 2: - errors.append(f"Need at least 2 descriptions, got {len(descriptions)}") - if len(descriptions) > 4: - errors.append(f"Maximum 4 descriptions, got {len(descriptions)}") + """Validate RSA headline/description content: count, char limits, pin slots. + + Shared by ``_validate_rsa`` (full RSA create) and + ``update_responsive_search_ad`` (in-place replace of headlines/descriptions). + Google enforces 3-15 headlines / 2-4 descriptions whenever a list is sent. + The update path passes ``enforce_*_count=False`` for a list it is NOT + replacing, since an omitted list stays untouched on the live ad — only the + supplied list is gated on count. + """ + errors: list[str] = [] + if enforce_headline_count: + if len(headlines) < 3: + errors.append(f"Need at least 3 headlines, got {len(headlines)}") + if len(headlines) > 15: + errors.append(f"Maximum 15 headlines, got {len(headlines)}") + if enforce_description_count: + if len(descriptions) < 2: + errors.append(f"Need at least 2 descriptions, got {len(descriptions)}") + if len(descriptions) > 4: + errors.append(f"Maximum 4 descriptions, got {len(descriptions)}") headline_pin_counts: dict[str, int] = {} for i, h in enumerate(headlines): @@ -2188,6 +2364,22 @@ def _validate_rsa( return errors +def _validate_rsa( + ad_group_id: str, + headlines: list[dict], + descriptions: list[dict], + final_url: str, +) -> list[str]: + errors = [] + if not ad_group_id: + errors.append("ad_group_id is required") + if not final_url: + errors.append("final_url is required") + errors.extend(_validate_rsa_assets(headlines, descriptions)) + + return errors + + _VALID_BIDDING_STRATEGIES = { "MAXIMIZE_CONVERSIONS", "MAXIMIZE_CONVERSION_VALUE", @@ -2539,6 +2731,7 @@ def _execute_plan(config: AdLoopConfig, plan: object) -> dict: "update_campaign": _apply_update_campaign, "update_ad_group": _apply_update_ad_group, "create_responsive_search_ad": _apply_create_rsa, + "update_responsive_search_ad": _apply_update_rsa, "add_keywords": _apply_add_keywords, "add_negative_keywords": _apply_add_negative_keywords, "add_negative_locations": _apply_add_negative_locations, @@ -2999,6 +3192,68 @@ def _apply_create_rsa(client: object, cid: str, changes: dict) -> dict: return {"resource_name": response.results[0].resource_name} +def _apply_update_rsa(client: object, cid: str, changes: dict) -> dict: + """Update mutable fields on an existing RSA in place. + + Builds a sparse ``AdOperation.update`` with only the fields the caller + asked to change, attached to a FieldMask so Google Ads ignores everything + else. Mutable in place on RSAs via ``AdService.MutateAds`` (API v23): + ``final_urls``, ``responsive_search_ad.path1``, + ``responsive_search_ad.path2``, ``responsive_search_ad.headlines``, + ``responsive_search_ad.descriptions``. Headlines/descriptions are + list-replace — the supplied list fully replaces the existing one. Note + that replacing the creative text resets the ad's learning and re-triggers + policy review even though the ad keeps its ID (see + ``update_responsive_search_ad``). + """ + from google.protobuf import field_mask_pb2 + + service = client.get_service("AdService") + operation = client.get_type("AdOperation") + ad = operation.update + ad.resource_name = service.ad_path(cid, changes["ad_id"]) + + field_paths: list[str] = [] + + if "final_url" in changes: + ad.final_urls.append(changes["final_url"]) + field_paths.append("final_urls") + + if "path1" in changes: + ad.responsive_search_ad.path1 = changes["path1"] + field_paths.append("responsive_search_ad.path1") + + if "path2" in changes: + ad.responsive_search_ad.path2 = changes["path2"] + field_paths.append("responsive_search_ad.path2") + + if "headlines" in changes: + for entry in changes["headlines"]: + asset = client.get_type("AdTextAsset") + asset.text = entry["text"] + if entry.get("pinned_field"): + asset.pinned_field = client.enums.ServedAssetFieldTypeEnum[ + entry["pinned_field"] + ] + ad.responsive_search_ad.headlines.append(asset) + field_paths.append("responsive_search_ad.headlines") + + if "descriptions" in changes: + for entry in changes["descriptions"]: + asset = client.get_type("AdTextAsset") + asset.text = entry["text"] + if entry.get("pinned_field"): + asset.pinned_field = client.enums.ServedAssetFieldTypeEnum[ + entry["pinned_field"] + ] + ad.responsive_search_ad.descriptions.append(asset) + field_paths.append("responsive_search_ad.descriptions") + + operation.update_mask = field_mask_pb2.FieldMask(paths=field_paths) + response = service.mutate_ads(customer_id=cid, operations=[operation]) + return {"resource_name": response.results[0].resource_name} + + def _apply_add_keywords(client: object, cid: str, changes: dict) -> dict: service = client.get_service("AdGroupCriterionService") ad_group_path = client.get_service("AdGroupService").ad_group_path( diff --git a/src/adloop/server.py b/src/adloop/server.py index 5ebe0ff..06ba6f0 100644 --- a/src/adloop/server.py +++ b/src/adloop/server.py @@ -57,6 +57,9 @@ def _coerce_json_string_to_list(value): _StrOrDictList = Annotated[ list[str | dict], BeforeValidator(_coerce_json_string_to_list) ] +_StrOrDictListOpt = Annotated[ + list[str | dict] | None, BeforeValidator(_coerce_json_string_to_list) +] def _build_orchestration_instructions() -> str: """Compact orchestration hint sent via MCP ``InitializeResult.instructions``. @@ -1568,6 +1571,65 @@ def draft_responsive_search_ad( ) +@mcp.tool(annotations=_WRITE) +@_safe +def update_responsive_search_ad( + ad_id: str, + customer_id: str = "", + headlines: _StrOrDictListOpt = None, + descriptions: _StrOrDictListOpt = None, + final_url: str = "", + path1: str = "", + path2: str = "", + clear_path1: bool = False, + clear_path2: bool = False, +) -> dict: + """Update mutable fields on an existing RSA in place — returns a PREVIEW. + + Edits an existing RSA without creating a new ad; the ad keeps its ID. + Google Ads API v23 (``AdService.MutateAds``) permits in-place mutation of + ``final_urls``, ``path1``, ``path2``, ``headlines``, and ``descriptions``. + + IMPORTANT: replacing headlines or descriptions is NOT a free in-place + edit. Even though the ad ID is preserved, swapping the creative text + RESETS the ad's asset-combination learning and performance history and + sends the ad BACK THROUGH Google policy review — Google treats the + creative as new for optimization. URL-only and path-only edits do not + incur this. When headlines/descriptions change, the returned preview + includes a ``warnings`` entry — surface it to the user before applying. + + Headlines/descriptions are LIST-REPLACE — when provided, the supplied + list fully swaps in for the existing one, and Google's RSA constraints + apply (3-15 headlines, 2-4 descriptions, 30/90 char limits, pin-slot + rules). Each entry may be a plain string (unpinned) or + ``{"text": "...", "pinned_field": "HEADLINE_1"}``. + + Argument semantics: + - ``headlines`` / ``descriptions``: None or [] -> no change; + non-empty list -> replaces the existing list in full + - ``final_url``: empty -> no change; non-empty -> replaces final URL + - ``path1`` / ``path2``: empty -> no change; non-empty -> sets value + - ``clear_path1`` / ``clear_path2``: True -> set to empty string + + At least one mutation must be requested. Call confirm_and_apply with the + returned plan_id to execute. + """ + from adloop.ads.write import update_responsive_search_ad as _impl + + return _impl( + current_config(), + customer_id=customer_id or current_config().ads.customer_id, + ad_id=ad_id, + headlines=headlines, + descriptions=descriptions, + final_url=final_url, + path1=path1, + path2=path2, + clear_path1=clear_path1, + clear_path2=clear_path2, + ) + + @mcp.tool(annotations=_WRITE) @_safe def draft_keywords( diff --git a/tests/test_server.py b/tests/test_server.py index ea02a97..d6e7927 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -312,3 +312,42 @@ async def test_add_negative_locations_tool_resolves_runtime_config(self): finally: runtime.set_default_config(None) preview_store.set_plan_store(InMemoryPlanStore()) + + @pytest.mark.asyncio + async def test_update_responsive_search_ad_tool_resolves_runtime_config(self): + from adloop import runtime + from adloop.config import AdLoopConfig, AdsConfig, SafetyConfig + from adloop.safety import preview as preview_store + from adloop.safety.preview import InMemoryPlanStore + from adloop.server import mcp + + preview_store.set_plan_store(InMemoryPlanStore()) + runtime.set_default_config( + AdLoopConfig( + ads=AdsConfig(customer_id="123-456-7890"), + safety=SafetyConfig(require_dry_run=True), + ) + ) + try: + tool = await mcp.get_tool("update_responsive_search_ad") + # A path-only edit avoids URL reachability checks and the + # learning-reset warning — keeps the call-through hermetic. + result = tool.fn(ad_id="999", path1="Pricing") + assert "error" not in result, result.get("error") + assert result["operation"] == "update_responsive_search_ad" + assert result["plan_id"] + # headlines/descriptions accept both plain strings and pinned dicts + # via the _StrOrDictListOpt coercion alias. + result2 = tool.fn( + ad_id="1000", + headlines=[ + "Fast Free Shipping", + {"text": "Shop the Sale", "pinned_field": "HEADLINE_1"}, + "Save 20% Now", + ], + ) + assert "error" not in result2, result2.get("error") + assert result2.get("warnings"), "text replace must warn" + finally: + runtime.set_default_config(None) + preview_store.set_plan_store(InMemoryPlanStore()) diff --git a/tests/test_update_rsa.py b/tests/test_update_rsa.py new file mode 100644 index 0000000..fde6837 --- /dev/null +++ b/tests/test_update_rsa.py @@ -0,0 +1,1151 @@ +"""Tests for ``update_responsive_search_ad`` — both the draft (preview) layer +and the ``_apply_update_rsa`` mutation layer. + +The Google Ads client is faked: we never hit the network, but we do verify +that the AdOperation we build carries the correct resource_name, the correct +update_mask paths, and the right field values. URL reachability is also +faked so the tests don't depend on the public internet. +""" + +from __future__ import annotations + +from types import SimpleNamespace + +import pytest +from google.ads.googleads.client import GoogleAdsClient + +from adloop.ads import write +from adloop.ads.client import GOOGLE_ADS_API_VERSION +from adloop.config import AdLoopConfig, AdsConfig, SafetyConfig +from adloop.safety import preview as preview_store + + +# --------------------------------------------------------------------------- +# Fake Google Ads client + AdService +# --------------------------------------------------------------------------- + + +class _FakeAdService: + """Captures the operations passed to ``mutate_ads`` for assertion. + + Returns a fake response shaped like the real one so callers that read + ``response.results[0].resource_name`` continue to work. + """ + + def __init__(self) -> None: + self.captured_operations: list[object] | None = None + self.captured_customer_id: str | None = None + + def ad_path(self, customer_id: str, ad_id: str) -> str: + return f"customers/{customer_id}/ads/{ad_id}" + + def mutate_ads( + self, + customer_id: str, + operations: list[object], + ) -> object: + self.captured_operations = operations + self.captured_customer_id = customer_id + first_op = operations[0] + return SimpleNamespace( + results=[SimpleNamespace(resource_name=first_op.update.resource_name)] + ) + + +class _FakeClient: + """Shim around the real client to swap in our fake AdService. + + Reuses the real client's ``enums`` and ``get_type`` for proto wiring; + only ``get_service`` is intercepted. + """ + + def __init__(self, ad_service: _FakeAdService): + self._base = GoogleAdsClient( + credentials=None, + developer_token="test-token", + use_proto_plus=True, + version=GOOGLE_ADS_API_VERSION, + ) + self.enums = self._base.enums + self.get_type = self._base.get_type + self._services = {"AdService": ad_service} + + def get_service(self, name: str) -> object: + return self._services[name] + + +# --------------------------------------------------------------------------- +# Shared fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture(autouse=True) +def clear_pending_plans(): + preview_store.set_plan_store(preview_store.InMemoryPlanStore()) + yield + preview_store.set_plan_store(preview_store.InMemoryPlanStore()) + + +@pytest.fixture(autouse=True) +def stub_url_validation(monkeypatch): + """Default: every URL passes. Tests can override to inject failures.""" + monkeypatch.setattr( + write, + "_validate_urls", + lambda urls, timeout=10: {u: None for u in urls}, + ) + + +@pytest.fixture +def config(tmp_path) -> AdLoopConfig: + return AdLoopConfig( + ads=AdsConfig(customer_id="123-456-7890"), + safety=SafetyConfig( + require_dry_run=False, + log_file=str(tmp_path / "audit.log"), + ), + ) + + +@pytest.fixture +def dry_run_config(tmp_path) -> AdLoopConfig: + return AdLoopConfig( + ads=AdsConfig(customer_id="123-456-7890"), + safety=SafetyConfig( + require_dry_run=True, + log_file=str(tmp_path / "audit.log"), + ), + ) + + +# --------------------------------------------------------------------------- +# Validation +# --------------------------------------------------------------------------- + + +class TestValidation: + def test_rejects_missing_ad_id(self, config): + result = write.update_responsive_search_ad( + config, customer_id="1234567890", final_url="https://example.com" + ) + assert result["error"] == "Validation failed" + assert any("ad_id is required" in d for d in result["details"]) + + def test_rejects_non_numeric_ad_id(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="abc123", + path1="Pricing", + ) + assert result["error"] == "Validation failed" + assert any("numeric" in d for d in result["details"]) + + def test_rejects_when_no_change_provided(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + ) + assert result["error"] == "Validation failed" + assert any("No changes specified" in d for d in result["details"]) + + def test_no_change_error_mentions_headlines_and_descriptions(self, config): + # The error message must hint at the headlines/descriptions options too — + # otherwise users won't know they exist. + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + ) + joined = " ".join(result["details"]) + assert "headlines" in joined + assert "descriptions" in joined + + def test_rejects_path1_too_long(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + path1="this-is-way-too-long-for-a-path", + ) + assert result["error"] == "Validation failed" + assert any("path1 must be 15 chars" in d for d in result["details"]) + + def test_rejects_path2_too_long(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + path2="X" * 16, + ) + assert result["error"] == "Validation failed" + assert any("path2 must be 15 chars" in d for d in result["details"]) + + def test_accepts_path_at_max_length_15(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + path1="X" * 15, + ) + assert result.get("error") is None + assert result["operation"] == "update_responsive_search_ad" + + def test_rejects_unreachable_url(self, config, monkeypatch): + monkeypatch.setattr( + write, + "_validate_urls", + lambda urls, timeout=10: {u: "HTTP 404" for u in urls}, + ) + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + final_url="https://example.com/missing", + ) + assert result["error"] == "URL validation failed" + assert any("not reachable" in d for d in result["details"]) + + def test_blocked_operation_rejected_before_validation( + self, config, monkeypatch + ): + config.safety.blocked_operations = ["update_responsive_search_ad"] + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + path1="OK", + ) + assert "blocked" in result["error"] + + def test_url_validation_skipped_when_only_paths_changed( + self, config, monkeypatch + ): + called = {"count": 0} + + def spy_validate(urls, timeout=10): + called["count"] += 1 + return {u: None for u in urls} + + monkeypatch.setattr(write, "_validate_urls", spy_validate) + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + path1="Pricing", + ) + assert result.get("error") is None + # No URL was supplied — we shouldn't have hit the validator. + assert called["count"] == 0 + + def test_multiple_validation_errors_returned_together(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="abc", # non-numeric + path1="X" * 16, # too long + ) + assert result["error"] == "Validation failed" + assert len(result["details"]) >= 2 + + +# --------------------------------------------------------------------------- +# Plan construction +# --------------------------------------------------------------------------- + + +class TestPlanConstruction: + def test_plan_metadata(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + path1="Sale", + ) + assert result["operation"] == "update_responsive_search_ad" + assert result["entity_type"] == "ad" + assert result["entity_id"] == "999" + assert result["customer_id"] == "1234567890" + assert result["status"] == "PENDING_CONFIRMATION" + assert "plan_id" in result + + def test_plan_stored_for_retrieval(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + path1="Sale", + ) + plan = preview_store.get_plan(result["plan_id"]) + assert plan is not None + assert plan.operation == "update_responsive_search_ad" + + def test_url_only_change_does_not_include_paths(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + final_url="https://example.com/x", + ) + changes = result["changes"] + assert changes["final_url"] == "https://example.com/x" + assert "path1" not in changes + assert "path2" not in changes + + def test_path1_only_change_does_not_include_url_or_path2(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + path1="Pricing", + ) + changes = result["changes"] + assert changes["path1"] == "Pricing" + assert "final_url" not in changes + assert "path2" not in changes + + def test_path2_only_change_does_not_include_url_or_path1(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + path2="Downtown", + ) + changes = result["changes"] + assert changes["path2"] == "Downtown" + assert "final_url" not in changes + assert "path1" not in changes + + def test_all_three_fields_set(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + final_url="https://example.com/x", + path1="A", + path2="B", + ) + changes = result["changes"] + assert changes["final_url"] == "https://example.com/x" + assert changes["path1"] == "A" + assert changes["path2"] == "B" + + def test_clear_path1_writes_empty_string_into_changes(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + clear_path1=True, + ) + changes = result["changes"] + # ``"path1" in changes`` is critical — apply uses presence to decide + # whether to mutate. Empty string is the *value*, not "no change". + assert "path1" in changes + assert changes["path1"] == "" + assert "path2" not in changes + + def test_clear_path2_writes_empty_string_into_changes(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + clear_path2=True, + ) + changes = result["changes"] + assert "path2" in changes + assert changes["path2"] == "" + assert "path1" not in changes + + def test_clear_path1_overrides_path1_argument(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + path1="Pricing", + clear_path1=True, + ) + # When clear_path1 is True we ignore the path1 string and clear it. + assert result["changes"]["path1"] == "" + + def test_paths_are_stripped_of_whitespace(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + path1=" Sale ", + path2=" \t Pricing\n", + ) + assert result["changes"]["path1"] == "Sale" + assert result["changes"]["path2"] == "Pricing" + + def test_ad_id_coerced_to_string(self, config): + # FastMCP types ad_id as str, but defensive coercion is cheap. + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + path1="X", + ) + assert isinstance(result["changes"]["ad_id"], str) + assert result["changes"]["ad_id"] == "999" + + +# --------------------------------------------------------------------------- +# Learning-reset warning — the single most important thing to relay before +# applying an in-place headline/description replacement. +# --------------------------------------------------------------------------- + + +class TestLearningResetWarning: + def test_warning_present_when_headlines_replaced(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + headlines=_valid_headlines(), + ) + assert result.get("error") is None + warnings = result.get("warnings", []) + assert warnings, "expected a warning when headlines are replaced" + joined = " ".join(warnings).lower() + assert "learning" in joined + assert "policy review" in joined + + def test_warning_present_when_descriptions_replaced(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + descriptions=_valid_descriptions(), + ) + assert result.get("error") is None + warnings = result.get("warnings", []) + assert warnings, "expected a warning when descriptions are replaced" + assert "learning" in " ".join(warnings).lower() + + def test_warning_absent_for_url_only_update(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + final_url="https://example.com/x", + ) + assert result.get("error") is None + # A URL-only edit is a true in-place change — no learning reset, so + # no warning must be attached. + assert "warnings" not in result + + def test_warning_absent_for_path_only_update(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + path1="Pricing", + path2="Sale", + ) + assert result.get("error") is None + assert "warnings" not in result + + def test_warning_absent_for_clear_path_only_update(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + clear_path1=True, + ) + assert result.get("error") is None + assert "warnings" not in result + + def test_warning_present_when_headlines_change_alongside_url(self, config): + # Mixing a headline replacement with a URL edit still resets learning, + # so the warning must fire. + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + headlines=_valid_headlines(), + final_url="https://example.com/x", + ) + assert result.get("error") is None + assert result.get("warnings") + + +# --------------------------------------------------------------------------- +# Apply / mutate +# --------------------------------------------------------------------------- + + +class TestApply: + def test_calls_mutate_ads_with_correct_resource_name(self, config): + ad_service = _FakeAdService() + client = _FakeClient(ad_service) + + write._apply_update_rsa( + client, + "1234567890", + {"ad_id": "999", "final_url": "https://example.com"}, + ) + + op = ad_service.captured_operations[0] + assert op.update.resource_name == "customers/1234567890/ads/999" + + def test_url_only_field_mask_has_only_final_urls(self, config): + ad_service = _FakeAdService() + client = _FakeClient(ad_service) + + write._apply_update_rsa( + client, + "1234567890", + {"ad_id": "999", "final_url": "https://example.com"}, + ) + + op = ad_service.captured_operations[0] + assert list(op.update_mask.paths) == ["final_urls"] + assert list(op.update.final_urls) == ["https://example.com"] + + def test_path1_only_field_mask(self, config): + ad_service = _FakeAdService() + client = _FakeClient(ad_service) + + write._apply_update_rsa( + client, + "1234567890", + {"ad_id": "999", "path1": "Sale"}, + ) + + op = ad_service.captured_operations[0] + assert list(op.update_mask.paths) == ["responsive_search_ad.path1"] + assert op.update.responsive_search_ad.path1 == "Sale" + + def test_path2_only_field_mask(self, config): + ad_service = _FakeAdService() + client = _FakeClient(ad_service) + + write._apply_update_rsa( + client, + "1234567890", + {"ad_id": "999", "path2": "Downtown"}, + ) + + op = ad_service.captured_operations[0] + assert list(op.update_mask.paths) == ["responsive_search_ad.path2"] + assert op.update.responsive_search_ad.path2 == "Downtown" + + def test_all_three_fields_in_mask(self, config): + ad_service = _FakeAdService() + client = _FakeClient(ad_service) + + write._apply_update_rsa( + client, + "1234567890", + { + "ad_id": "999", + "final_url": "https://example.com", + "path1": "Sale", + "path2": "New", + }, + ) + + op = ad_service.captured_operations[0] + assert set(op.update_mask.paths) == { + "final_urls", + "responsive_search_ad.path1", + "responsive_search_ad.path2", + } + + def test_clear_path_writes_empty_string_to_proto(self, config): + ad_service = _FakeAdService() + client = _FakeClient(ad_service) + + write._apply_update_rsa( + client, + "1234567890", + {"ad_id": "999", "path1": ""}, # the "clear" semantic + ) + + op = ad_service.captured_operations[0] + assert "responsive_search_ad.path1" in list(op.update_mask.paths) + assert op.update.responsive_search_ad.path1 == "" + + def test_returns_resource_name(self, config): + ad_service = _FakeAdService() + client = _FakeClient(ad_service) + + result = write._apply_update_rsa( + client, + "1234567890", + {"ad_id": "999", "path1": "X"}, + ) + + assert result == {"resource_name": "customers/1234567890/ads/999"} + + def test_passes_customer_id_to_mutate_call(self, config): + ad_service = _FakeAdService() + client = _FakeClient(ad_service) + + write._apply_update_rsa( + client, + "1234567890", + {"ad_id": "999", "path1": "X"}, + ) + + assert ad_service.captured_customer_id == "1234567890" + + +# --------------------------------------------------------------------------- +# Confirm-and-apply integration +# --------------------------------------------------------------------------- + + +class TestConfirmAndApplyIntegration: + def test_dry_run_returns_dry_run_success(self, config): + draft = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + path1="Pricing", + ) + result = write.confirm_and_apply( + config, plan_id=draft["plan_id"], dry_run=True + ) + assert result["status"] == "DRY_RUN_SUCCESS" + assert result["operation"] == "update_responsive_search_ad" + + def test_require_dry_run_overrides_dry_run_false(self, dry_run_config): + draft = write.update_responsive_search_ad( + dry_run_config, + customer_id="1234567890", + ad_id="999", + path1="Pricing", + ) + result = write.confirm_and_apply( + dry_run_config, plan_id=draft["plan_id"], dry_run=False + ) + assert result["status"] == "DRY_RUN_SUCCESS" + assert result.get("dry_run_forced_by") == "config.safety.require_dry_run" + + def test_unknown_plan_id_returns_error(self, config): + result = write.confirm_and_apply( + config, plan_id="does-not-exist", dry_run=True + ) + assert "error" in result + + def test_apply_routes_to_update_rsa(self, config, monkeypatch): + ad_service = _FakeAdService() + client = _FakeClient(ad_service) + + monkeypatch.setattr( + "adloop.ads.client.get_ads_client", lambda _cfg: client + ) + + draft = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + final_url="https://example.com", + path1="Sale", + ) + result = write.confirm_and_apply( + config, plan_id=draft["plan_id"], dry_run=False + ) + + assert result["status"] == "APPLIED" + assert result["operation"] == "update_responsive_search_ad" + assert ad_service.captured_operations is not None + assert len(ad_service.captured_operations) == 1 + + def test_apply_writes_audit_log(self, config, monkeypatch, tmp_path): + ad_service = _FakeAdService() + client = _FakeClient(ad_service) + monkeypatch.setattr( + "adloop.ads.client.get_ads_client", lambda _cfg: client + ) + + draft = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + path1="Sale", + ) + write.confirm_and_apply( + config, plan_id=draft["plan_id"], dry_run=False + ) + + log_path = config.safety.log_file + from pathlib import Path + contents = Path(log_path).read_text() + assert "update_responsive_search_ad" in contents + assert "success" in contents + + def test_apply_writes_dry_run_audit_log(self, config): + draft = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + path1="Sale", + ) + write.confirm_and_apply( + config, plan_id=draft["plan_id"], dry_run=True + ) + + from pathlib import Path + contents = Path(config.safety.log_file).read_text() + assert "dry_run_success" in contents + assert '"dry_run": true' in contents + + def test_plan_removed_after_successful_apply(self, config, monkeypatch): + ad_service = _FakeAdService() + client = _FakeClient(ad_service) + monkeypatch.setattr( + "adloop.ads.client.get_ads_client", lambda _cfg: client + ) + + draft = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + path1="Sale", + ) + plan_id = draft["plan_id"] + assert preview_store.get_plan(plan_id) is not None + + write.confirm_and_apply(config, plan_id=plan_id, dry_run=False) + assert preview_store.get_plan(plan_id) is None + + +# --------------------------------------------------------------------------- +# Headline / description mutation (in-place text replacement via AdService) +# --------------------------------------------------------------------------- + + +def _valid_headlines() -> list[str]: + """Minimum-viable headline set: 3 entries, each under 30 chars.""" + return ["Fast Free Shipping", "Shop the Sale Today", "Save 20% Now"] + + +def _valid_descriptions() -> list[str]: + """Minimum-viable description set: 2 entries, each under 90 chars.""" + return [ + "Free shipping on every order. Shop the sale and save today.", + "Thousands of five-star reviews. Money-back guarantee on all items.", + ] + + +class TestHeadlineDescriptionValidation: + """Covers the count / char-limit / pin-slot validation paths added when + update_responsive_search_ad gained headlines & descriptions support. + """ + + def test_rejects_too_few_headlines(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + headlines=["Only one headline"], + descriptions=_valid_descriptions(), + ) + assert result["error"] == "Validation failed" + assert any("at least 3 headlines" in d for d in result["details"]) + + def test_rejects_too_many_headlines(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + headlines=[f"H{i}" for i in range(16)], + descriptions=_valid_descriptions(), + ) + assert result["error"] == "Validation failed" + assert any("Maximum 15 headlines" in d for d in result["details"]) + + def test_rejects_headline_over_30_chars(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + headlines=[ + "OK headline one", + "OK headline two", + "X" * 31, # over the cap by 1 + ], + descriptions=_valid_descriptions(), + ) + assert result["error"] == "Validation failed" + assert any("exceeds 30 chars" in d for d in result["details"]) + + def test_accepts_headline_at_max_30_chars(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + headlines=["X" * 30, "OK", "Also OK"], + descriptions=_valid_descriptions(), + ) + assert result.get("error") is None + + def test_rejects_invalid_headline_pin(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + headlines=[ + {"text": "Brand", "pinned_field": "HEADLINE_99"}, + "OK two", + "OK three", + ], + descriptions=_valid_descriptions(), + ) + assert result["error"] == "Validation failed" + assert any( + "HEADLINE_99" in d and "invalid" in d for d in result["details"] + ) + + def test_rejects_too_many_headlines_per_pin_slot(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + headlines=[ + {"text": "Brand A", "pinned_field": "HEADLINE_1"}, + {"text": "Brand B", "pinned_field": "HEADLINE_1"}, + {"text": "Brand C", "pinned_field": "HEADLINE_1"}, # 3rd in slot + ], + descriptions=_valid_descriptions(), + ) + assert result["error"] == "Validation failed" + assert any( + "At most 2 headlines may pin to HEADLINE_1" in d + for d in result["details"] + ) + + def test_rejects_too_few_descriptions(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + headlines=_valid_headlines(), + descriptions=["Only one description here."], + ) + assert result["error"] == "Validation failed" + assert any("at least 2 descriptions" in d for d in result["details"]) + + def test_rejects_too_many_descriptions(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + headlines=_valid_headlines(), + descriptions=[f"Description {i} for the ad." for i in range(5)], + ) + assert result["error"] == "Validation failed" + assert any("Maximum 4 descriptions" in d for d in result["details"]) + + def test_rejects_description_over_90_chars(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + headlines=_valid_headlines(), + descriptions=[ + "Short OK description.", + "Y" * 91, # over the cap by 1 + ], + ) + assert result["error"] == "Validation failed" + assert any("exceeds 90 chars" in d for d in result["details"]) + + def test_rejects_invalid_description_pin(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + headlines=_valid_headlines(), + descriptions=[ + {"text": "Legal disclaimer.", "pinned_field": "DESCRIPTION_9"}, + "Second description.", + ], + ) + assert result["error"] == "Validation failed" + assert any( + "DESCRIPTION_9" in d and "invalid" in d for d in result["details"] + ) + + def test_rejects_too_many_descriptions_per_pin_slot(self, config): + # Descriptions: cap is 1 per slot (vs 2 for headlines) + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + headlines=_valid_headlines(), + descriptions=[ + {"text": "First desc.", "pinned_field": "DESCRIPTION_1"}, + {"text": "Second desc.", "pinned_field": "DESCRIPTION_1"}, + ], + ) + assert result["error"] == "Validation failed" + assert any( + "At most 1 description may pin to DESCRIPTION_1" in d + for d in result["details"] + ) + + def test_headlines_only_does_not_require_descriptions(self, config): + # Common use case: rewriting just headlines without touching descriptions. + # List-replace semantics mean if you provide only headlines, Google + # still enforces the 3-15 cap on the new headlines but doesn't touch + # descriptions at all. So validation should pass here. + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + headlines=_valid_headlines(), + ) + assert result.get("error") is None + assert "headlines" in result["changes"] + assert "descriptions" not in result["changes"] + + def test_descriptions_only_does_not_require_headlines(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + descriptions=_valid_descriptions(), + ) + assert result.get("error") is None + assert "descriptions" in result["changes"] + assert "headlines" not in result["changes"] + + +class TestHeadlineDescriptionPlanConstruction: + def test_normalizes_string_headlines_to_dicts(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + headlines=_valid_headlines(), + ) + for h in result["changes"]["headlines"]: + assert isinstance(h, dict) + assert h["text"] # non-empty + assert h["pinned_field"] is None # unpinned + + def test_preserves_pinned_field_dict_input(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + headlines=[ + {"text": "Fast Free Shipping", "pinned_field": "HEADLINE_1"}, + "Shop the Sale Today", + "Save 20% Now", + ], + ) + pinned = result["changes"]["headlines"][0] + assert pinned == { + "text": "Fast Free Shipping", + "pinned_field": "HEADLINE_1", + } + + def test_mixed_string_and_dict_entries_normalize(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + headlines=[ + {"text": "Pinned brand", "pinned_field": "HEADLINE_1"}, + "Unpinned plain", + "Third headline", + ], + ) + hs = result["changes"]["headlines"] + assert hs[0]["pinned_field"] == "HEADLINE_1" + assert hs[1]["pinned_field"] is None + assert hs[2]["pinned_field"] is None + + def test_combined_headlines_paths_url(self, config): + result = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + headlines=_valid_headlines(), + descriptions=_valid_descriptions(), + final_url="https://example.com/x", + path1="Pricing", + path2="Sale", + ) + changes = result["changes"] + assert changes["final_url"] == "https://example.com/x" + assert changes["path1"] == "Pricing" + assert changes["path2"] == "Sale" + assert len(changes["headlines"]) == 3 + assert len(changes["descriptions"]) == 2 + + +class TestHeadlineDescriptionApply: + def test_headlines_field_mask_and_assets_populated(self, config): + ad_service = _FakeAdService() + client = _FakeClient(ad_service) + + write._apply_update_rsa( + client, + "1234567890", + { + "ad_id": "999", + "headlines": [ + {"text": "Fast Free Shipping", "pinned_field": None}, + {"text": "Shop the Sale Today", "pinned_field": None}, + {"text": "Save 20% Now", "pinned_field": None}, + ], + }, + ) + + op = ad_service.captured_operations[0] + assert "responsive_search_ad.headlines" in list(op.update_mask.paths) + # Each AdTextAsset should carry the text through to the proto + headlines = list(op.update.responsive_search_ad.headlines) + assert len(headlines) == 3 + assert headlines[0].text == "Fast Free Shipping" + assert headlines[1].text == "Shop the Sale Today" + assert headlines[2].text == "Save 20% Now" + + def test_descriptions_field_mask_and_assets_populated(self, config): + ad_service = _FakeAdService() + client = _FakeClient(ad_service) + + write._apply_update_rsa( + client, + "1234567890", + { + "ad_id": "999", + "descriptions": [ + {"text": "Desc one.", "pinned_field": None}, + {"text": "Desc two.", "pinned_field": None}, + ], + }, + ) + + op = ad_service.captured_operations[0] + assert "responsive_search_ad.descriptions" in list(op.update_mask.paths) + descs = list(op.update.responsive_search_ad.descriptions) + assert len(descs) == 2 + assert descs[0].text == "Desc one." + + def test_pinned_field_propagates_to_proto_enum(self, config): + ad_service = _FakeAdService() + client = _FakeClient(ad_service) + + write._apply_update_rsa( + client, + "1234567890", + { + "ad_id": "999", + "headlines": [ + {"text": "Brand", "pinned_field": "HEADLINE_1"}, + {"text": "Plain A", "pinned_field": None}, + {"text": "Plain B", "pinned_field": None}, + ], + }, + ) + + op = ad_service.captured_operations[0] + headlines = list(op.update.responsive_search_ad.headlines) + # The pinned headline carries the HEADLINE_1 enum value; unpinned + # carries UNSPECIFIED (the proto default). + expected_pin = client.enums.ServedAssetFieldTypeEnum.HEADLINE_1 + assert headlines[0].pinned_field == expected_pin + # Unpinned entries should not set pinned_field (proto default). + unspecified = client.enums.ServedAssetFieldTypeEnum.UNSPECIFIED + assert headlines[1].pinned_field == unspecified + assert headlines[2].pinned_field == unspecified + + def test_combined_paths_headlines_descriptions_in_mask(self, config): + ad_service = _FakeAdService() + client = _FakeClient(ad_service) + + write._apply_update_rsa( + client, + "1234567890", + { + "ad_id": "999", + "path1": "Sale", + "headlines": [ + {"text": "H1", "pinned_field": None}, + {"text": "H2", "pinned_field": None}, + {"text": "H3", "pinned_field": None}, + ], + "descriptions": [ + {"text": "D1.", "pinned_field": None}, + {"text": "D2.", "pinned_field": None}, + ], + }, + ) + + op = ad_service.captured_operations[0] + paths = set(op.update_mask.paths) + assert paths == { + "responsive_search_ad.path1", + "responsive_search_ad.headlines", + "responsive_search_ad.descriptions", + } + + def test_headlines_only_does_not_include_url_or_path_in_mask(self, config): + ad_service = _FakeAdService() + client = _FakeClient(ad_service) + + write._apply_update_rsa( + client, + "1234567890", + { + "ad_id": "999", + "headlines": [ + {"text": "H1", "pinned_field": None}, + {"text": "H2", "pinned_field": None}, + {"text": "H3", "pinned_field": None}, + ], + }, + ) + + op = ad_service.captured_operations[0] + paths = list(op.update_mask.paths) + assert paths == ["responsive_search_ad.headlines"] + # final_urls must remain empty when no URL change is requested + assert list(op.update.final_urls) == [] + + +class TestHeadlineDescriptionIntegration: + """End-to-end via confirm_and_apply — the same path the MCP tool takes.""" + + def test_headline_rewrite_routes_through_dispatch_to_apply( + self, config, monkeypatch + ): + ad_service = _FakeAdService() + client = _FakeClient(ad_service) + monkeypatch.setattr( + "adloop.ads.client.get_ads_client", lambda _cfg: client + ) + + draft = write.update_responsive_search_ad( + config, + customer_id="1234567890", + ad_id="999", + headlines=[ + "Replacement Headline One", + "Replacement Headline Two", + "Replacement Headline Three", + ], + ) + assert draft.get("error") is None + + result = write.confirm_and_apply( + config, plan_id=draft["plan_id"], dry_run=False + ) + assert result["status"] == "APPLIED" + assert ad_service.captured_operations is not None + op = ad_service.captured_operations[0] + # Verify the round trip carried headlines all the way to the proto + texts = [h.text for h in op.update.responsive_search_ad.headlines] + assert texts == [ + "Replacement Headline One", + "Replacement Headline Two", + "Replacement Headline Three", + ]