AFFAANhandClaude Opus 5
ddd07d5d16
Fourth failure in the same chain, one object further each time: "Advantage audience flag required ... setting the advantage_audience flag to either 1 or 0 within the targeting_automation field" (code 100, subcode 1870227). We were never sending targeting_automation at all. Defaulting to 0 (off) for the same reason bid_strategy defaults to LOWEST_COST_WITHOUT_CAP: an ad set should reach the audience it was told to — the countries/age/gender actually set on the campaign — not whatever Meta's Advantage+ expansion additionally decides to include. advanced.advantage_audience (0 or 1) overrides it per campaign. Ordering note: targeting_automation is added unconditionally, so the existing "no audience" guard (raises when targeting is empty) had to move before it — otherwise targeting would never be empty and that guard would go silently dead. A test pins this: an empty CampaignSpec.targeting must still be refused. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
769 lines
26 KiB
Python
769 lines
26 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Meta write helpers. No live calls: the transport is injected.
|
|
|
|
Covers the money-safety properties Task 4 exists to enforce:
|
|
1. Every create_* helper hard-codes status="PAUSED" and raises ValueError
|
|
(with zero network calls) if asked for anything else.
|
|
2. update_object_status is the only method allowed to send ACTIVE.
|
|
3. Dict params Graph expects as JSON strings (targeting,
|
|
object_story_spec, special_ad_categories) are json.dumps-encoded.
|
|
4. Meta's code/error_subcode survive on MetaError.
|
|
5. No test makes a live network call or creates a real Meta object -
|
|
every client here is constructed with a FakeTransport.
|
|
"""
|
|
import base64
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from adclaw.meta.client import MetaClient, MetaError
|
|
from adclaw.meta.objects import (
|
|
ALLOWED_OPTIMIZATION_GOALS,
|
|
DEFAULT_OPTIMIZATION_GOAL,
|
|
DEFAULT_OPTIMIZATION_GOAL_BY_OBJECTIVE,
|
|
build_ad_set_payload,
|
|
build_campaign_payload,
|
|
build_creative_payload,
|
|
)
|
|
from adclaw.campaigns.models import CampaignSpec
|
|
|
|
|
|
class FakeTransport:
|
|
"""Records calls and returns queued responses. Never touches a network."""
|
|
|
|
def __init__(self, responses):
|
|
self.responses = list(responses)
|
|
self.calls = []
|
|
|
|
async def __call__(self, method, url, params):
|
|
self.calls.append({"method": method, "url": url, "params": params})
|
|
return self.responses.pop(0)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# create_campaign
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_campaign_is_always_paused():
|
|
transport = FakeTransport([{"id": "23851234567890"}])
|
|
client = MetaClient(access_token="tok", transport=transport)
|
|
|
|
campaign_id = await client.create_campaign(
|
|
"act_1", name="Q3", objective="OUTCOME_LEADS",
|
|
)
|
|
|
|
assert campaign_id == "23851234567890"
|
|
call = transport.calls[0]
|
|
assert call["method"] == "POST"
|
|
assert call["url"].endswith("/act_1/campaigns")
|
|
assert call["params"]["status"] == "PAUSED"
|
|
assert call["params"]["objective"] == "OUTCOME_LEADS"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_campaign_refuses_active_status():
|
|
transport = FakeTransport([{"id": "1"}])
|
|
client = MetaClient(access_token="tok", transport=transport)
|
|
|
|
with pytest.raises(ValueError):
|
|
await client.create_campaign(
|
|
"act_1", name="Q3", objective="OUTCOME_LEADS", status="ACTIVE",
|
|
)
|
|
assert transport.calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_campaign_serialises_special_ad_categories():
|
|
transport = FakeTransport([{"id": "c1"}])
|
|
client = MetaClient(access_token="tok", transport=transport)
|
|
|
|
await client.create_campaign(
|
|
"act_1",
|
|
name="Q3",
|
|
objective="OUTCOME_LEADS",
|
|
special_ad_categories=["HOUSING"],
|
|
)
|
|
|
|
sent = transport.calls[0]["params"]["special_ad_categories"]
|
|
assert isinstance(sent, str)
|
|
assert json.loads(sent) == ["HOUSING"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# create_ad_set
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_ad_set_sends_paused_and_serialises_targeting():
|
|
transport = FakeTransport([{"id": "adset_1"}])
|
|
client = MetaClient(access_token="tok", transport=transport)
|
|
|
|
await client.create_ad_set(
|
|
"act_1",
|
|
campaign_id="c1",
|
|
name="Ad set",
|
|
daily_budget=10000,
|
|
targeting={"geo_locations": {"countries": ["IN"]}},
|
|
optimization_goal="LEAD_GENERATION",
|
|
billing_event="IMPRESSIONS",
|
|
)
|
|
|
|
params = transport.calls[0]["params"]
|
|
assert params["status"] == "PAUSED"
|
|
assert params["daily_budget"] == 10000
|
|
assert isinstance(params["targeting"], str)
|
|
assert json.loads(params["targeting"]) == {
|
|
"geo_locations": {"countries": ["IN"]},
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_ad_set_refuses_active_status():
|
|
transport = FakeTransport([{"id": "adset_1"}])
|
|
client = MetaClient(access_token="tok", transport=transport)
|
|
|
|
with pytest.raises(ValueError):
|
|
await client.create_ad_set(
|
|
"act_1",
|
|
campaign_id="c1",
|
|
name="Ad set",
|
|
daily_budget=10000,
|
|
targeting={},
|
|
optimization_goal="LEAD_GENERATION",
|
|
billing_event="IMPRESSIONS",
|
|
status="ACTIVE",
|
|
)
|
|
assert transport.calls == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# create_ad
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_ad_is_always_paused():
|
|
transport = FakeTransport([{"id": "ad_1"}])
|
|
client = MetaClient(access_token="tok", transport=transport)
|
|
|
|
ad_id = await client.create_ad(
|
|
"act_1", name="Ad", adset_id="adset_1", creative_id="creative_1",
|
|
)
|
|
|
|
assert ad_id == "ad_1"
|
|
params = transport.calls[0]["params"]
|
|
assert params["status"] == "PAUSED"
|
|
assert params["adset_id"] == "adset_1"
|
|
assert json.loads(params["creative"]) == {"creative_id": "creative_1"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_ad_refuses_active_status():
|
|
transport = FakeTransport([{"id": "ad_1"}])
|
|
client = MetaClient(access_token="tok", transport=transport)
|
|
|
|
with pytest.raises(ValueError):
|
|
await client.create_ad(
|
|
"act_1",
|
|
name="Ad",
|
|
adset_id="adset_1",
|
|
creative_id="creative_1",
|
|
status="ACTIVE",
|
|
)
|
|
assert transport.calls == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# create_ad_creative
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_ad_creative_serialises_object_story_spec():
|
|
transport = FakeTransport([{"id": "creative_1"}])
|
|
client = MetaClient(access_token="tok", transport=transport)
|
|
|
|
creative_id = await client.create_ad_creative(
|
|
"act_1",
|
|
name="Creative",
|
|
page_id="page_1",
|
|
message="Hello",
|
|
headline="Headline",
|
|
description="Description",
|
|
link="https://example.com",
|
|
image_hash="hash123",
|
|
)
|
|
|
|
assert creative_id == "creative_1"
|
|
sent = transport.calls[0]["params"]["object_story_spec"]
|
|
assert isinstance(sent, str)
|
|
decoded = json.loads(sent)
|
|
assert decoded["page_id"] == "page_1"
|
|
assert decoded["link_data"]["image_hash"] == "hash123"
|
|
assert decoded["link_data"]["message"] == "Hello"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# upload_ad_image
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upload_ad_image_encodes_bytes_and_returns_hash(tmp_path):
|
|
image_path = tmp_path / "creative.png"
|
|
raw_bytes = b"\x89PNG\r\n\x1a\nfake-image-bytes"
|
|
image_path.write_bytes(raw_bytes)
|
|
|
|
transport = FakeTransport(
|
|
[{"images": {"creative.png": {"hash": "abc123", "url": "https://x"}}}],
|
|
)
|
|
client = MetaClient(access_token="tok", transport=transport)
|
|
|
|
image_hash = await client.upload_ad_image("act_1", str(image_path))
|
|
|
|
assert image_hash == "abc123"
|
|
call = transport.calls[0]
|
|
assert call["method"] == "POST"
|
|
assert call["url"].endswith("/act_1/adimages")
|
|
assert base64.b64decode(call["params"]["bytes"]) == raw_bytes
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upload_ad_image_raises_meta_error_when_no_hash(tmp_path):
|
|
image_path = tmp_path / "creative.png"
|
|
image_path.write_bytes(b"data")
|
|
|
|
transport = FakeTransport([{"images": {}}])
|
|
client = MetaClient(access_token="tok", transport=transport)
|
|
|
|
with pytest.raises(MetaError):
|
|
await client.upload_ad_image("act_1", str(image_path))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _post error handling
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_post_error_preserves_meta_code_and_subcode():
|
|
transport = FakeTransport([
|
|
{"error": {"message": "Invalid budget", "code": 100, "error_subcode": 1487079}},
|
|
])
|
|
client = MetaClient(access_token="tok", transport=transport)
|
|
|
|
with pytest.raises(MetaError) as excinfo:
|
|
await client.create_campaign("act_1", name="Q3", objective="OUTCOME_LEADS")
|
|
|
|
assert excinfo.value.code == 100
|
|
assert excinfo.value.subcode == 1487079
|
|
assert "Invalid budget" in str(excinfo.value)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# update_object_status - the only path allowed to send ACTIVE
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_object_status_accepts_active():
|
|
"""Launch is the one path allowed to send ACTIVE."""
|
|
transport = FakeTransport([{"success": True}])
|
|
client = MetaClient(access_token="tok", transport=transport)
|
|
|
|
await client.update_object_status("c1", "ACTIVE")
|
|
|
|
call = transport.calls[0]
|
|
assert call["method"] == "POST"
|
|
assert call["url"].endswith("/c1")
|
|
assert call["params"]["status"] == "ACTIVE"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_object_status_returns_none():
|
|
transport = FakeTransport([{"success": True}])
|
|
client = MetaClient(access_token="tok", transport=transport)
|
|
|
|
result = await client.update_object_status("c1", "PAUSED")
|
|
|
|
assert result is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _post defence in depth - reachable directly, still refuses non-PAUSED
|
|
# creates on campaign/adset/ad paths, before any transport call
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_post_refuses_non_paused_status_on_campaign_create_path():
|
|
transport = FakeTransport([{"id": "c1"}])
|
|
client = MetaClient(access_token="tok", transport=transport)
|
|
|
|
with pytest.raises(ValueError):
|
|
await client._post(
|
|
"/act_1/campaigns",
|
|
{"name": "Q3", "objective": "OUTCOME_LEADS", "status": "ACTIVE"},
|
|
)
|
|
assert transport.calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_post_still_allows_update_object_status_to_send_active():
|
|
"""update_object_status posts to /{object_id}, which never matches the
|
|
campaign/adset/ad creation suffixes, so the _post guard above must not
|
|
block Launch."""
|
|
transport = FakeTransport([{"success": True}])
|
|
client = MetaClient(access_token="tok", transport=transport)
|
|
|
|
await client.update_object_status("c1", "ACTIVE")
|
|
|
|
assert transport.calls[0]["params"]["status"] == "ACTIVE"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# list_campaigns / list_ad_sets / list_ads
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_campaigns_returns_data_array():
|
|
transport = FakeTransport([{"data": [{"id": "c1", "name": "X", "status": "PAUSED"}]}])
|
|
client = MetaClient(access_token="tok", transport=transport)
|
|
|
|
result = await client.list_campaigns("act_1")
|
|
|
|
assert result[0]["id"] == "c1"
|
|
assert transport.calls[0]["method"] == "GET"
|
|
assert transport.calls[0]["url"].endswith("/act_1/campaigns")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_campaigns_returns_empty_list_when_no_data():
|
|
transport = FakeTransport([{}])
|
|
client = MetaClient(access_token="tok", transport=transport)
|
|
|
|
assert await client.list_campaigns("act_1") == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_ad_sets_returns_data_array():
|
|
transport = FakeTransport([{"data": [{"id": "as1"}]}])
|
|
client = MetaClient(access_token="tok", transport=transport)
|
|
|
|
result = await client.list_ad_sets("act_1")
|
|
|
|
assert result == [{"id": "as1"}]
|
|
assert transport.calls[0]["method"] == "GET"
|
|
assert transport.calls[0]["url"].endswith("/act_1/adsets")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_ads_returns_data_array():
|
|
transport = FakeTransport([{"data": [{"id": "ad1"}]}])
|
|
client = MetaClient(access_token="tok", transport=transport)
|
|
|
|
result = await client.list_ads("act_1")
|
|
|
|
assert result == [{"id": "ad1"}]
|
|
assert transport.calls[0]["method"] == "GET"
|
|
assert transport.calls[0]["url"].endswith("/act_1/ads")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# objects.py - pure payload builders, no transport at all
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _spec(**overrides) -> CampaignSpec:
|
|
defaults = dict(
|
|
id="camp_1",
|
|
name="Test Campaign",
|
|
objective="OUTCOME_LEADS",
|
|
budget={"daily_budget": 5000},
|
|
targeting={"age_min": 25, "age_max": 45, "countries": ["US", "CA"]},
|
|
advanced={},
|
|
)
|
|
defaults.update(overrides)
|
|
return CampaignSpec(**defaults)
|
|
|
|
|
|
def test_build_campaign_payload_maps_fields():
|
|
spec = _spec(advanced={"special_ad_categories": ["HOUSING"]})
|
|
|
|
payload = build_campaign_payload(spec)
|
|
|
|
assert payload == {
|
|
"name": "Test Campaign",
|
|
"objective": "OUTCOME_LEADS",
|
|
"special_ad_categories": ["HOUSING"],
|
|
}
|
|
|
|
|
|
def test_build_campaign_payload_defaults_special_ad_categories_to_empty():
|
|
spec = _spec()
|
|
|
|
payload = build_campaign_payload(spec)
|
|
|
|
assert payload["special_ad_categories"] == []
|
|
|
|
|
|
def test_build_campaign_payload_requires_objective():
|
|
spec = _spec(objective=None)
|
|
|
|
with pytest.raises(ValueError):
|
|
build_campaign_payload(spec)
|
|
|
|
|
|
def test_build_campaign_payload_normalises_bare_string_special_ad_category():
|
|
# advanced is unvalidated free-form JSON, so a bare string is a
|
|
# plausible client input for this compliance field. It must become a
|
|
# single-element list, not be exploded into characters by list().
|
|
spec = _spec(advanced={"special_ad_categories": "HOUSING"})
|
|
|
|
payload = build_campaign_payload(spec)
|
|
|
|
assert payload["special_ad_categories"] == ["HOUSING"]
|
|
|
|
|
|
def test_build_campaign_payload_passes_through_a_list_unchanged():
|
|
spec = _spec(advanced={"special_ad_categories": ["HOUSING", "EMPLOYMENT"]})
|
|
|
|
payload = build_campaign_payload(spec)
|
|
|
|
assert payload["special_ad_categories"] == ["HOUSING", "EMPLOYMENT"]
|
|
|
|
|
|
def test_build_ad_set_payload_maps_targeting_and_budget():
|
|
spec = _spec()
|
|
|
|
payload = build_ad_set_payload(spec, campaign_id="c1")
|
|
|
|
assert payload["campaign_id"] == "c1"
|
|
assert payload["daily_budget"] == 5000
|
|
assert payload["targeting"] == {
|
|
"age_min": 25,
|
|
"age_max": 45,
|
|
"geo_locations": {"countries": ["US", "CA"]},
|
|
"targeting_automation": {"advantage_audience": 0},
|
|
}
|
|
assert payload["optimization_goal"] == "LEAD_GENERATION"
|
|
assert payload["billing_event"] == "IMPRESSIONS"
|
|
|
|
|
|
def test_build_ad_set_payload_honours_advanced_overrides():
|
|
spec = _spec(
|
|
advanced={
|
|
"optimization_goal": "OFFSITE_CONVERSIONS",
|
|
"billing_event": "LINK_CLICKS",
|
|
},
|
|
)
|
|
|
|
payload = build_ad_set_payload(spec, campaign_id="c1")
|
|
|
|
assert payload["optimization_goal"] == "OFFSITE_CONVERSIONS"
|
|
assert payload["billing_event"] == "LINK_CLICKS"
|
|
|
|
|
|
def test_build_ad_set_payload_requires_daily_budget():
|
|
spec = _spec(budget={})
|
|
|
|
with pytest.raises(ValueError):
|
|
build_ad_set_payload(spec, campaign_id="c1")
|
|
|
|
|
|
def test_build_ad_set_payload_requires_an_audience():
|
|
"""Graph rejects an ad set with no targeting; say so here instead."""
|
|
spec = _spec(targeting={})
|
|
|
|
with pytest.raises(ValueError, match="no audience"):
|
|
build_ad_set_payload(spec, campaign_id="c1")
|
|
|
|
|
|
def test_build_ad_set_payload_accepts_any_single_targeting_field():
|
|
payload = build_ad_set_payload(_spec(targeting={"countries": ["IN"]}), "c1")
|
|
|
|
assert payload["targeting"] == {
|
|
"geo_locations": {"countries": ["IN"]},
|
|
"targeting_automation": {"advantage_audience": 0},
|
|
}
|
|
|
|
|
|
def test_build_ad_set_payload_accepts_allowlisted_optimization_goal():
|
|
spec = _spec(advanced={"optimization_goal": "LINK_CLICKS"})
|
|
|
|
payload = build_ad_set_payload(spec, campaign_id="c1")
|
|
|
|
assert payload["optimization_goal"] == "LINK_CLICKS"
|
|
|
|
|
|
def test_build_ad_set_payload_rejects_unknown_optimization_goal():
|
|
spec = _spec(advanced={"optimization_goal": "SOMETHING_MADE_UP"})
|
|
|
|
with pytest.raises(ValueError):
|
|
build_ad_set_payload(spec, campaign_id="c1")
|
|
|
|
|
|
def test_build_ad_set_payload_accepts_allowlisted_billing_event():
|
|
spec = _spec(advanced={"billing_event": "LINK_CLICKS"})
|
|
|
|
payload = build_ad_set_payload(spec, campaign_id="c1")
|
|
|
|
assert payload["billing_event"] == "LINK_CLICKS"
|
|
|
|
|
|
def test_build_ad_set_payload_rejects_unknown_billing_event():
|
|
# billing_event determines how the ad account is charged, so an
|
|
# unrecognised value from client-controlled `advanced` must never reach
|
|
# Meta - a spending-relevant field with no allowlist is the finding
|
|
# this guards against.
|
|
spec = _spec(advanced={"billing_event": "SOMETHING_MADE_UP"})
|
|
|
|
with pytest.raises(ValueError):
|
|
build_ad_set_payload(spec, campaign_id="c1")
|
|
|
|
|
|
def test_build_ad_set_payload_targeting_can_produce_json_dumpable_dict():
|
|
# This is what create_ad_set will json.dumps() before sending - make
|
|
# sure the mapping never emits anything that would break that step.
|
|
spec = _spec()
|
|
|
|
payload = build_ad_set_payload(spec, campaign_id="c1")
|
|
|
|
encoded = json.dumps(payload["targeting"])
|
|
assert json.loads(encoded) == payload["targeting"]
|
|
|
|
|
|
def test_build_creative_payload_maps_fields():
|
|
spec = _spec(
|
|
advanced={
|
|
"message": "Come see our homes",
|
|
"headline": "New Listings",
|
|
"description": "Fresh inventory weekly",
|
|
"link": "https://example.com/listings",
|
|
},
|
|
)
|
|
|
|
payload = build_creative_payload(spec, page_id="page_1", image_hash="hash123")
|
|
|
|
assert payload == {
|
|
"name": "Test Campaign - Creative",
|
|
"page_id": "page_1",
|
|
"message": "Come see our homes",
|
|
"headline": "New Listings",
|
|
"description": "Fresh inventory weekly",
|
|
"link": "https://example.com/listings",
|
|
"image_hash": "hash123",
|
|
}
|
|
|
|
|
|
def test_build_creative_payload_falls_back_to_campaign_name_for_headline():
|
|
spec = _spec(advanced={})
|
|
|
|
payload = build_creative_payload(spec, page_id="page_1", image_hash="hash123")
|
|
|
|
assert payload["headline"] == "Test Campaign"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# create_campaign: the field Graph requires when the campaign has no budget
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_campaign_declares_adset_budget_sharing():
|
|
"""Graph rejects a budget-less campaign that omits this field.
|
|
|
|
MaskanX always puts the budget on the ad set, so the campaign never
|
|
carries one and the field is never optional. Omitting it is what
|
|
produced code 100 / subcode 4834011 against the live account, with the
|
|
useless message "Invalid parameter".
|
|
|
|
It must be the lowercase literal: form-encoding a Python bool sends
|
|
"True", which Graph does not accept.
|
|
"""
|
|
transport = FakeTransport([{"id": "c1"}])
|
|
client = MetaClient(access_token="tok", transport=transport)
|
|
|
|
await client.create_campaign("act_1", name="Q3", objective="OUTCOME_LEADS")
|
|
|
|
sent = transport.calls[0]["params"]["is_adset_budget_sharing_enabled"]
|
|
assert sent == "false"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_adset_budget_sharing_is_off_by_default_but_overridable():
|
|
"""True lets ad sets lend each other up to 20% of their budget, so an
|
|
ad set can outspend the number we set for it. The guardrails treat that
|
|
number as a ceiling, so the default has to be off."""
|
|
transport = FakeTransport([{"id": "c1"}, {"id": "c2"}])
|
|
client = MetaClient(access_token="tok", transport=transport)
|
|
|
|
await client.create_campaign("act_1", name="A", objective="OUTCOME_LEADS")
|
|
await client.create_campaign(
|
|
"act_1", name="B", objective="OUTCOME_LEADS", adset_budget_sharing=True,
|
|
)
|
|
|
|
assert transport.calls[0]["params"]["is_adset_budget_sharing_enabled"] == "false"
|
|
assert transport.calls[1]["params"]["is_adset_budget_sharing_enabled"] == "true"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# bid strategy: the field Graph requires on a budget-carrying ad set
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_build_ad_set_payload_defaults_to_automatic_bidding():
|
|
"""Graph rejects an ad set that carries its own budget and no strategy.
|
|
|
|
Code 100 / subcode 2490487 — the failure that stopped the live sync
|
|
once the campaign-level one was fixed. LOWEST_COST_WITHOUT_CAP is the
|
|
only strategy needing no bid amount from us, and it never exceeds the
|
|
daily budget, which is what the guardrails assume.
|
|
"""
|
|
payload = build_ad_set_payload(_spec(), campaign_id="c1")
|
|
|
|
assert payload["bid_strategy"] == "LOWEST_COST_WITHOUT_CAP"
|
|
|
|
|
|
def test_build_ad_set_payload_refuses_a_strategy_that_needs_a_bid_amount():
|
|
"""These would be accepted here and rejected by Meta *after* the
|
|
campaign object exists, leaving a half-built chain in the account.
|
|
Refusing before the first network call keeps it clean."""
|
|
for strategy in (
|
|
"COST_CAP", "LOWEST_COST_WITH_BID_CAP", "LOWEST_COST_WITH_MIN_ROAS",
|
|
):
|
|
spec = _spec(advanced={"bid_strategy": strategy})
|
|
with pytest.raises(ValueError, match="bid_strategy"):
|
|
build_ad_set_payload(spec, campaign_id="c1")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_ad_set_sends_the_bid_strategy():
|
|
transport = FakeTransport([{"id": "set_1"}])
|
|
client = MetaClient(access_token="tok", transport=transport)
|
|
|
|
await client.create_ad_set(
|
|
"act_1",
|
|
campaign_id="c1",
|
|
name="Set",
|
|
daily_budget=5000,
|
|
targeting={"geo_locations": {"countries": ["IN"]}},
|
|
optimization_goal="LEAD_GENERATION",
|
|
billing_event="IMPRESSIONS",
|
|
)
|
|
|
|
sent = transport.calls[0]["params"]["bid_strategy"]
|
|
assert sent == "LOWEST_COST_WITHOUT_CAP"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# optimisation goal: Meta pairs it with the campaign objective
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_the_default_optimisation_goal_follows_the_objective():
|
|
"""A traffic campaign cannot optimise for leads.
|
|
|
|
Meta rejects the pair with code 100 / subcode 2490408, at ad-set
|
|
creation — after the campaign object exists. The default used to be
|
|
LEAD_GENERATION for every objective, which is correct for exactly one
|
|
of them and is what the live connection test hit.
|
|
"""
|
|
assert (
|
|
build_ad_set_payload(_spec(objective="OUTCOME_TRAFFIC"), campaign_id="c1")[
|
|
"optimization_goal"
|
|
]
|
|
== "LINK_CLICKS"
|
|
)
|
|
assert (
|
|
build_ad_set_payload(_spec(objective="OUTCOME_LEADS"), campaign_id="c1")[
|
|
"optimization_goal"
|
|
]
|
|
== "LEAD_GENERATION"
|
|
)
|
|
assert (
|
|
build_ad_set_payload(_spec(objective="OUTCOME_AWARENESS"), campaign_id="c1")[
|
|
"optimization_goal"
|
|
]
|
|
== "REACH"
|
|
)
|
|
|
|
|
|
def test_every_mapped_default_is_one_we_allow():
|
|
"""The table and the allowlist have to agree, or a perfectly ordinary
|
|
objective raises ValueError on its own default."""
|
|
for objective, goal in DEFAULT_OPTIMIZATION_GOAL_BY_OBJECTIVE.items():
|
|
assert goal in ALLOWED_OPTIMIZATION_GOALS, objective
|
|
assert DEFAULT_OPTIMIZATION_GOAL in ALLOWED_OPTIMIZATION_GOALS
|
|
|
|
|
|
def test_an_unknown_objective_falls_back_to_the_widest_goal():
|
|
"""Better a goal most objectives accept than a lead goal only one does."""
|
|
payload = build_ad_set_payload(_spec(objective="OUTCOME_SOMETHING_NEW"),
|
|
campaign_id="c1")
|
|
|
|
assert payload["optimization_goal"] == "LINK_CLICKS"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Advantage+ Audience: the field Graph requires on every ad set's targeting
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_targeting_automation_defaults_to_advantage_audience_off():
|
|
"""Graph rejects an ad set whose targeting spec omits this — code 100 /
|
|
subcode 1870227, the failure that stopped the live connection test
|
|
once optimisation_goal was fixed.
|
|
|
|
Off by default for the same reason as bid_strategy: an ad set matches
|
|
the audience actually specified, not whatever Meta additionally
|
|
expands it to.
|
|
"""
|
|
payload = build_ad_set_payload(_spec(), campaign_id="c1")
|
|
|
|
assert payload["targeting"]["targeting_automation"] == {
|
|
"advantage_audience": 0,
|
|
}
|
|
|
|
|
|
def test_advantage_audience_can_be_turned_on():
|
|
spec = _spec(advanced={"advantage_audience": 1})
|
|
|
|
payload = build_ad_set_payload(spec, campaign_id="c1")
|
|
|
|
assert payload["targeting"]["targeting_automation"] == {
|
|
"advantage_audience": 1,
|
|
}
|
|
|
|
|
|
def test_advantage_audience_rejects_anything_but_0_or_1():
|
|
spec = _spec(advanced={"advantage_audience": 2})
|
|
|
|
with pytest.raises(ValueError, match="advantage_audience"):
|
|
build_ad_set_payload(spec, campaign_id="c1")
|
|
|
|
|
|
def test_an_empty_audience_is_still_refused_even_though_targeting_automation_is_always_set():
|
|
"""targeting_automation is added unconditionally, so `targeting` is
|
|
never an empty dict any more. The no-audience guard has to run before
|
|
that, or this refusal silently stops firing."""
|
|
spec = _spec(targeting={})
|
|
|
|
with pytest.raises(ValueError, match="no audience"):
|
|
build_ad_set_payload(spec, campaign_id="c1")
|
|
|
|
|
|
def test_an_explicit_optimisation_goal_still_wins():
|
|
spec = _spec(
|
|
objective="OUTCOME_TRAFFIC",
|
|
advanced={"optimization_goal": "LANDING_PAGE_VIEWS"},
|
|
)
|
|
|
|
payload = build_ad_set_payload(spec, campaign_id="c1")
|
|
|
|
assert payload["optimization_goal"] == "LANDING_PAGE_VIEWS"
|