Files
maskanx_cm_backend/tests/test_meta_writes.py
T
AFFAANhandClaude Opus 5 f752c4fdf3 fix(meta): send the budget-sharing flag Graph requires on every campaign
The connection test finally said what was wrong: "Must specify True or
False in is_adset_budget_sharing_enabled ... if you are not using campaign
budget." Code 100, subcode 4834011 — the same failure that has been
blocking a live sync, previously reported only as "Invalid parameter".

MaskanX always puts the budget on the ad set, so the campaign never
carries one and Graph never treats this field as optional. It was never
sent at all, so no campaign could be created in this account.

Defaulting to false, not true. True lets ad sets lend each other up to 20%
of their budget, which means an ad set can outspend the daily budget we
set for it — and validate_guardrails treats that number as a ceiling.
Predictable spend beats Meta's optimisation. Overridable per call.

Sent as the lowercase literal "false": form-encoding a Python bool would
put "True" on the wire, which Graph rejects.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-07 13:29:16 +05:30

605 lines
20 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 (
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"]},
}
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"]}}
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"