diff --git a/briefs/example_linkedin_brief.md b/briefs/example_linkedin_brief.md index 06bb5b0..569e9a2 100644 --- a/briefs/example_linkedin_brief.md +++ b/briefs/example_linkedin_brief.md @@ -11,11 +11,11 @@ Generate qualified inbound leads for the Lattice Cloud free-tier signup at `lattice.example/cloud`. Hand-off to SDR queue. ## KPIs -- **Primary:** Cost per Marketing Qualified Lead (MQL) ≤ $120 +- **Primary:** Cost per Marketing Qualified Lead (MQL) ≤ €110 - **Secondary:** CTR ≥ 0.6% on Sponsored Content ## Budget -$40,000 USD total, across the flight. +€36,000 EUR total, across the flight. ## Flight 2026-07-06 → 2026-08-31 (≈8 weeks). @@ -57,6 +57,10 @@ Three creatives in market at launch: LinkedIn. Mirror flights to Google Search are scoped separately. ## Notes +To advertise a post you published by hand on the Company Page, add a line like +`Existing post: urn:li:share:7340000000000000000` under that creative — the agent +will reference it directly instead of creating a new Direct Sponsored Content post. + Hold the benchmark-report creative until week 3 so the engineering-leader story sets context first. Keep all initial drafts in **DRAFT** status on LinkedIn — nothing gets activated in Campaign Manager without sign-off from the demand-gen diff --git a/docs/linkedin-integration.md b/docs/linkedin-integration.md index 22c9435..0fa8b8c 100644 --- a/docs/linkedin-integration.md +++ b/docs/linkedin-integration.md @@ -43,7 +43,7 @@ Optional: | Variable | Default | Meaning | |---|---|---| -| `LINKEDIN_API_VERSION` | `202405` | Versioned API header (`LinkedIn-Version`). | +| `LINKEDIN_API_VERSION` | `202605` | Versioned API header (`LinkedIn-Version`). | | `LINKEDIN_ALLOWED_AD_ACCOUNTS` | unset | Comma-separated allowlist of ad account ids. The configured account must be in the list, or `YIELDAGENT_ALLOW_LIVE=1` must be set. | | `YIELDAGENT_ALLOW_LIVE` | unset | Set to `1` to bypass the allowlist. Disables the only safety net LinkedIn offers through this integration — use sparingly. | diff --git a/src/yieldagent/agents/campaign_setup/prompts.py b/src/yieldagent/agents/campaign_setup/prompts.py index 46de853..42b34c3 100644 --- a/src/yieldagent/agents/campaign_setup/prompts.py +++ b/src/yieldagent/agents/campaign_setup/prompts.py @@ -12,6 +12,8 @@ empty rather than inventing values. Convert currency symbols to ISO codes (e.g. "$" -> "USD"). Convert dates to ISO 8601. Use lowercase 'meta' for Facebook/Instagram, 'google' for Google Ads, 'tiktok' for TikTok. +If a creative cites the URN of an already-published post (e.g. 'urn:li:share:123' or +'urn:li:ugcPost:123'), copy it verbatim into that creative's existing_post_urn field. """ PLAN_CAMPAIGN_SYSTEM = """\ @@ -28,4 +30,5 @@ - Carry the Brief's audience through to the LineItem's Targeting unchanged unless the Brief specifies sub-audience splits. - Use the Brief's notes section to inform naming (e.g. "Midnight Brew Launch — June"). +- Preserve each creative's existing_post_urn unchanged when the Brief sets it. """ diff --git a/src/yieldagent/agents/linkedin_setup/cli.py b/src/yieldagent/agents/linkedin_setup/cli.py index 40b76e1..4bc8cc7 100644 --- a/src/yieldagent/agents/linkedin_setup/cli.py +++ b/src/yieldagent/agents/linkedin_setup/cli.py @@ -22,6 +22,7 @@ from langgraph.types import Command from yieldagent.agents.campaign_setup.nodes import DEFAULT_MODEL +from yieldagent.env import load_dotenv from .graph import build_graph @@ -119,6 +120,7 @@ async def _run(brief_path: Path, *, auto_approve: bool, dry_run: bool, model_nam def main() -> int: + load_dotenv() parser = argparse.ArgumentParser(prog="yieldagent-linkedin-setup") parser.add_argument("brief", type=Path, help="Path to a markdown campaign brief") parser.add_argument( diff --git a/src/yieldagent/domain/brief.py b/src/yieldagent/domain/brief.py index 084d773..35760a1 100644 --- a/src/yieldagent/domain/brief.py +++ b/src/yieldagent/domain/brief.py @@ -92,6 +92,14 @@ class CreativeAsset(BaseModel): video_url: str | None = None call_to_action: str | None = None landing_url: str | None = None + existing_post_urn: str | None = Field( + default=None, + description=( + "URN of an already-published LinkedIn post/share (e.g. 'urn:li:share:123'). " + "When set, the ad references this post directly instead of creating a new " + "Direct Sponsored Content post — use it to advertise content published by hand." + ), + ) class Brief(BaseModel): diff --git a/src/yieldagent/integrations/linkedin/client.py b/src/yieldagent/integrations/linkedin/client.py index bd1b1d1..d5d546d 100644 --- a/src/yieldagent/integrations/linkedin/client.py +++ b/src/yieldagent/integrations/linkedin/client.py @@ -20,6 +20,9 @@ _BASE_URL = "https://api.linkedin.com/rest" _FORBIDDEN_STATUSES = {"ACTIVE", "COMPLETED"} +# LinkedIn's politicalIntent is a String enum, not a boolean. Sending a bool +# fails with "enum type is not backed by a String". +_POLITICAL_INTENT_VALUES = {"POLITICAL", "NOT_POLITICAL", "NOT_DECLARED"} class LinkedInError(RuntimeError): @@ -37,7 +40,7 @@ def __init__(self, config: LinkedInConfig, http: httpx.AsyncClient | None = None self._http = http or httpx.AsyncClient(timeout=30.0) self._owns_http = http is None - async def __aenter__(self) -> "LinkedInClient": + async def __aenter__(self) -> LinkedInClient: return self async def __aexit__(self, *_exc: object) -> None: @@ -147,7 +150,11 @@ async def create_campaign_group( payload["totalBudget"] = total_budget if run_schedule is not None: payload["runSchedule"] = run_schedule - return await self._request("POST", "/adCampaignGroups", json=payload) + return await self._request( + "POST", + f"/adAccounts/{self.config.ad_account_id}/adCampaignGroups", + json=payload, + ) async def create_campaign( self, @@ -164,9 +171,16 @@ async def create_campaign( unit_cost: dict[str, str] | None = None, cost_type: str = "CPC", status: str | None = None, + offsite_delivery_enabled: bool = False, + political_intent: str = "NOT_POLITICAL", ) -> dict[str, Any]: if (daily_budget is None) == (total_budget is None): raise ValueError("Provide exactly one of daily_budget or total_budget") + if political_intent not in _POLITICAL_INTENT_VALUES: + raise ValueError( + f"political_intent must be one of {sorted(_POLITICAL_INTENT_VALUES)}, " + f"got {political_intent!r}" + ) payload: dict[str, Any] = { "account": self.config.account_urn, "campaignGroup": campaign_group_urn, @@ -178,6 +192,10 @@ async def create_campaign( "runSchedule": run_schedule, "targetingCriteria": targeting_criteria, "locale": locale, + # Both fields became required in current API versions. Safe defaults: + # LinkedIn-only delivery (no Audience Network) and non-political. + "offsiteDeliveryEnabled": offsite_delivery_enabled, + "politicalIntent": political_intent, } if daily_budget is not None: payload["dailyBudget"] = daily_budget @@ -185,19 +203,80 @@ async def create_campaign( payload["totalBudget"] = total_budget if unit_cost is not None: payload["unitCost"] = unit_cost - return await self._request("POST", "/adCampaigns", json=payload) + return await self._request( + "POST", + f"/adAccounts/{self.config.ad_account_id}/adCampaigns", + json=payload, + ) + + async def create_post( + self, + *, + author_urn: str, + commentary: str, + article: dict[str, Any] | None = None, + dsc_ad_account_urn: str | None = None, + feed_distribution: str = "NONE", + ) -> dict[str, Any]: + """Create a Post via the (non-account-scoped) Posts API. + + A LinkedIn Creative cannot carry inline copy — it must reference a real + Post (share / ugcPost). For ads we create a *dark post* (Direct Sponsored + Content): authored by the advertiser org, `feedDistribution=NONE` so it + never shows on the page's organic feed, and an `adContext` tying it to the + sponsored account. Returns the new post URN under `id` (from `x-restli-id`). + + `feedDistribution=NONE` makes LinkedIn treat the post as DSC, which *requires* + `adContext.dscAdAccount`. `dscAdType` must NOT be sent — it is read-only and + a 422 ("ReadOnly field present in a create request") results otherwise. + """ + payload: dict[str, Any] = { + "author": author_urn, + "commentary": commentary, + "visibility": "PUBLIC", + "distribution": { + "feedDistribution": feed_distribution, + "targetEntities": [], + "thirdPartyDistributionChannels": [], + }, + "lifecycleState": "PUBLISHED", + "isReshareDisabledByAuthor": False, + } + if article is not None: + payload["content"] = {"article": article} + if dsc_ad_account_urn is not None: + payload["adContext"] = {"dscAdAccount": dsc_ad_account_urn} + return await self._request("POST", "/posts", json=payload) async def create_creative( self, *, campaign_urn: str, content: dict[str, Any], - status: str | None = None, + intended_status: str | None = None, ) -> dict[str, Any]: + # The Creatives API rejects `account` (read-only) and uses `intendedStatus` + # (an enum) rather than `status`. `content` must reference a Post URN, e.g. + # {"reference": "urn:li:share:..."}. payload: dict[str, Any] = { - "account": self.config.account_urn, "campaign": campaign_urn, "content": content, - "status": self._check_status(status), + "intendedStatus": self._check_status(intended_status), } - return await self._request("POST", "/creatives", json=payload) + return await self._request( + "POST", + f"/adAccounts/{self.config.ad_account_id}/creatives", + json=payload, + ) + + async def list_campaigns(self) -> dict[str, Any]: + """List campaigns under the configured ad account. + + Read-only. Useful for smoke tests and for agents that need to know what + already exists before planning a new campaign. + """ + return await self._request( + "GET", + f"/adAccounts/{self.config.ad_account_id}/adCampaigns", + params={"q": "search"}, + ) diff --git a/src/yieldagent/integrations/linkedin/config.py b/src/yieldagent/integrations/linkedin/config.py index 58c9a10..e383821 100644 --- a/src/yieldagent/integrations/linkedin/config.py +++ b/src/yieldagent/integrations/linkedin/config.py @@ -13,9 +13,13 @@ class LinkedInConfig: api_version: str allow_live: bool allowed_accounts: frozenset[str] = field(default_factory=frozenset) + # Organization (Company Page) that authors the Direct Sponsored Content posts + # backing each creative. Optional: if unset, the server resolves it from the + # ad account's `reference` field at publish time. + organization_urn: str | None = None @classmethod - def from_env(cls) -> "LinkedInConfig": + def from_env(cls) -> LinkedInConfig: token = os.environ.get("LINKEDIN_ACCESS_TOKEN") account = os.environ.get("LINKEDIN_AD_ACCOUNT_ID") if not token: @@ -31,12 +35,19 @@ def from_env(cls) -> "LinkedInConfig": ) raw_allow = os.environ.get("LINKEDIN_ALLOWED_AD_ACCOUNTS", "") allowed = frozenset(a.strip() for a in raw_allow.split(",") if a.strip()) + # Accept a bare org id or a full URN. + org = os.environ.get("LINKEDIN_ORGANIZATION_URN") or os.environ.get( + "LINKEDIN_ORGANIZATION_ID" + ) + if org and not org.startswith("urn:li:organization:"): + org = f"urn:li:organization:{org.strip()}" return cls( access_token=token, ad_account_id=account, - api_version=os.environ.get("LINKEDIN_API_VERSION", "202405"), + api_version=os.environ.get("LINKEDIN_API_VERSION", "202605"), allow_live=os.environ.get("YIELDAGENT_ALLOW_LIVE") == "1", allowed_accounts=allowed, + organization_urn=org, ) @property diff --git a/src/yieldagent/integrations/linkedin/mapping.py b/src/yieldagent/integrations/linkedin/mapping.py index 5038089..55ab825 100644 --- a/src/yieldagent/integrations/linkedin/mapping.py +++ b/src/yieldagent/integrations/linkedin/mapping.py @@ -73,6 +73,19 @@ def flight_to_run_schedule(flight: Flight) -> dict[str, int]: return {"start": int(start.timestamp() * 1000), "end": int(end.timestamp() * 1000)} +def campaign_run_schedule(flights: list[Flight]) -> dict[str, int]: + """Build a Campaign Group runSchedule that spans all its child Campaigns. + + LinkedIn now requires `runSchedule` on `POST /adAccounts/{id}/adCampaignGroups`. + The group must cover the earliest start and latest end across its line items. + """ + if not flights: + raise ValueError("Cannot compute runSchedule from empty list of flights") + earliest = min(f.start_date for f in flights) + latest = max(f.end_date for f in flights) + return flight_to_run_schedule(Flight(start_date=earliest, end_date=latest)) + + def audience_to_targeting(audience: Audience) -> dict[str, Any]: """Build a LinkedIn `targetingCriteria` payload. @@ -153,12 +166,16 @@ def line_item_payload( } -def creative_content(creative: CreativeAsset) -> dict[str, Any]: - """Build a minimal Sponsored Content `content` block. +def post_article_content(creative: CreativeAsset) -> dict[str, Any]: + """Build the `article` block for a Posts API dark post. + + A Creative can't hold inline copy — it references a Post. We model each ad as + an article post pointing at the landing URL. The Posts API does *not* scrape + the URL, so we set title/description explicitly. - Image/video uploads are out of scope for the first slice — the agent will - pass `image_url`/`video_url` references; production use will need to upload - these assets via the `/assets` endpoint and reference the returned URNs. + Thumbnail is intentionally omitted: it must be an `urn:li:image:{id}` from the + Images API, not a plain URL. Uploading creative imagery is a follow-up; until + then posts render with LinkedIn's default link preview. """ article: dict[str, Any] = { "source": creative.landing_url or "https://example.com", @@ -167,17 +184,17 @@ def creative_content(creative: CreativeAsset) -> dict[str, Any]: article["title"] = creative.headline if creative.description: article["description"] = creative.description - if creative.image_url: - article["thumbnail"] = creative.image_url - - content: dict[str, Any] = {"article": article} - if creative.primary_text: - content["commentary"] = creative.primary_text - if creative.call_to_action: - content["callToAction"] = { - "label": creative.call_to_action.upper().replace(" ", "_") - } - return content + return article + + +def post_commentary(creative: CreativeAsset) -> str: + """The text shown above the post. Falls back through primary_text → headline → name.""" + return creative.primary_text or creative.headline or creative.name + + +def creative_content_reference(post_urn: str) -> dict[str, Any]: + """Wrap a Post URN as a Creative `content` reference.""" + return {"reference": post_urn} __all__ = [ @@ -186,9 +203,12 @@ def creative_content(creative: CreativeAsset) -> dict[str, Any]: "OBJECTIVE_TO_LINKEDIN", "audience_to_targeting", "campaign_objective", - "creative_content", + "campaign_run_schedule", + "creative_content_reference", "flight_to_run_schedule", "line_item_locale", "line_item_payload", "money_to_linkedin_amount", + "post_article_content", + "post_commentary", ] diff --git a/src/yieldagent/integrations/linkedin/server.py b/src/yieldagent/integrations/linkedin/server.py index 8f1e91d..2cc5a4e 100644 --- a/src/yieldagent/integrations/linkedin/server.py +++ b/src/yieldagent/integrations/linkedin/server.py @@ -3,7 +3,7 @@ Run with: `python -m yieldagent.integrations.linkedin.server` Required env: LINKEDIN_ACCESS_TOKEN, LINKEDIN_AD_ACCOUNT_ID -Optional env: LINKEDIN_API_VERSION (default 202405), +Optional env: LINKEDIN_API_VERSION (default 202605), LINKEDIN_ALLOWED_AD_ACCOUNTS (comma-separated allowlist), YIELDAGENT_ALLOW_LIVE (set to 1 to bypass the allowlist) """ @@ -16,6 +16,7 @@ from mcp.server.fastmcp import FastMCP from yieldagent.domain import Campaign +from yieldagent.env import load_dotenv from .client import LinkedInClient from .config import LinkedInConfig @@ -23,10 +24,13 @@ DEFAULT_CAMPAIGN_TYPE, audience_to_targeting, campaign_objective, - creative_content, + campaign_run_schedule, + creative_content_reference, flight_to_run_schedule, line_item_locale, money_to_linkedin_amount, + post_article_content, + post_commentary, ) mcp = FastMCP("yieldagent-linkedin") @@ -136,6 +140,8 @@ async def publish_draft_campaign(campaign: dict[str, Any]) -> dict[str, Any]: group = await client.create_campaign_group( name=parsed.name, total_budget=money_to_linkedin_amount(group_amount, group_currency), + # LinkedIn now requires runSchedule on the group; span all line items. + run_schedule=campaign_run_schedule([li.flight for li in parsed.line_items]), ) group_urn = f"urn:li:sponsoredCampaignGroup:{group['id']}" result["campaign_id"] = group["id"] @@ -143,6 +149,23 @@ async def publish_draft_campaign(campaign: dict[str, Any]) -> dict[str, Any]: objective_type = campaign_objective(parsed) + # Creatives reference a real Post. Ads carrying `existing_post_urn` reuse a + # hand-published post; the rest mint a new Direct Sponsored Content post, + # which must be authored by a Company Page. Only resolve/require the org URN + # when at least one ad needs a fresh post. + org_urn = config.organization_urn + if any(not ad.creative.existing_post_urn for ad in parsed.ads): + if org_urn is None: + account = await client.get_ad_account() + org_urn = account.get("reference") + if not org_urn or not str(org_urn).startswith("urn:li:organization:"): + raise ValueError( + "No organization (Company Page) is associated with this ad account, " + "so Direct Sponsored Content posts cannot be authored for creatives. " + "Set LINKEDIN_ORGANIZATION_URN, or use an ad account linked to a page, " + "or set `existing_post_urn` on every ad to reuse hand-published posts." + ) + line_item_urns: dict[str, str] = {} unresolved_by_li: dict[str, dict[str, Any]] = {} for li in parsed.line_items: @@ -172,12 +195,28 @@ async def publish_draft_campaign(campaign: dict[str, Any]) -> dict[str, Any]: raise ValueError( f"Ad {ad.name!r} references unknown line_item_name {ad.line_item_name!r}" ) + # Either reference a hand-published post, or mint a dark post (DSC). + if ad.creative.existing_post_urn: + post_urn = ad.creative.existing_post_urn + else: + post = await client.create_post( + author_urn=org_urn, + commentary=post_commentary(ad.creative), + article=post_article_content(ad.creative), + dsc_ad_account_urn=config.account_urn, + ) + post_urn = post.get("id") created = await client.create_creative( campaign_urn=campaign_urn, - content=creative_content(ad.creative), + content=creative_content_reference(post_urn), ) result["ads"].append( - {"name": ad.name, "id": created.get("id"), "campaign_urn": campaign_urn} + { + "name": ad.name, + "id": created.get("id"), + "campaign_urn": campaign_urn, + "post_urn": post_urn, + } ) if unresolved_by_li: @@ -192,6 +231,7 @@ async def publish_draft_campaign(campaign: dict[str, Any]) -> dict[str, Any]: def main() -> None: + load_dotenv() asyncio.run(mcp.run_stdio_async()) diff --git a/tests/integrations/__init__.py b/tests/integrations/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/integrations/test_linkedin_client.py b/tests/integrations/test_linkedin_client.py new file mode 100644 index 0000000..6b2c225 --- /dev/null +++ b/tests/integrations/test_linkedin_client.py @@ -0,0 +1,251 @@ +"""Tests for the LinkedIn Marketing API client. + +These pin the endpoint paths the client uses for writes. LinkedIn deprecated +the global write endpoints (`/adCampaignGroups`, `/adCampaigns`, `/creatives`) +in favour of account-scoped ones (`/adAccounts/{id}/...`). The old endpoints +return 400 with an explicit migration message — see +`docs/claude_docs/linkedin_campaign_manager_api_debug_prompt.md`. + +The tests intercept HTTP at the transport layer (no network), assert the +exact path each method posts to, and confirm the payload shape stays intact. +""" + +from __future__ import annotations + +import httpx +import pytest + +from yieldagent.integrations.linkedin.client import LinkedInClient, LinkedInError +from yieldagent.integrations.linkedin.config import LinkedInConfig + +_AD_ACCOUNT_ID = "537690018" + + +def _make_config() -> LinkedInConfig: + return LinkedInConfig( + access_token="test-token", + ad_account_id=_AD_ACCOUNT_ID, + api_version="202605", + allow_live=False, + allowed_accounts=frozenset({_AD_ACCOUNT_ID}), + ) + + +class _Recorder: + """Collects every outbound httpx request so tests can assert on the URL.""" + + def __init__(self) -> None: + self.requests: list[httpx.Request] = [] + + def handler(self, request: httpx.Request) -> httpx.Response: + self.requests.append(request) + # LinkedIn returns 201 with the new id in this header for write endpoints. + return httpx.Response( + status_code=201, + headers={"x-restli-id": "urn:li:sponsoredCampaignGroup:123"}, + ) + + +@pytest.fixture +def recorded_client(): + recorder = _Recorder() + transport = httpx.MockTransport(recorder.handler) + http = httpx.AsyncClient(transport=transport, timeout=5.0) + client = LinkedInClient(_make_config(), http=http) + yield client, recorder + + +async def test_create_campaign_group_uses_account_scoped_path(recorded_client) -> None: + client, recorder = recorded_client + await client.create_campaign_group( + name="smoke", + total_budget={"amount": "10", "currencyCode": "EUR"}, + ) + assert len(recorder.requests) == 1 + request = recorder.requests[0] + assert request.method == "POST" + assert request.url.path == f"/rest/adAccounts/{_AD_ACCOUNT_ID}/adCampaignGroups" + + +async def test_create_campaign_uses_account_scoped_path(recorded_client) -> None: + client, recorder = recorded_client + await client.create_campaign( + campaign_group_urn="urn:li:sponsoredCampaignGroup:1", + name="smoke", + objective_type="WEBSITE_VISITS", + campaign_type="SPONSORED_UPDATES", + total_budget={"amount": "10", "currencyCode": "EUR"}, + run_schedule={"start": 1, "end": 2}, + targeting_criteria={"include": {"and": []}}, + locale={"country": "US", "language": "en"}, + ) + assert len(recorder.requests) == 1 + request = recorder.requests[0] + assert request.method == "POST" + assert request.url.path == f"/rest/adAccounts/{_AD_ACCOUNT_ID}/adCampaigns" + + +async def test_create_creative_uses_account_scoped_path(recorded_client) -> None: + client, recorder = recorded_client + await client.create_creative( + campaign_urn="urn:li:sponsoredCampaign:1", + content={"reference": "urn:li:share:1"}, + ) + assert len(recorder.requests) == 1 + request = recorder.requests[0] + assert request.method == "POST" + assert request.url.path == f"/rest/adAccounts/{_AD_ACCOUNT_ID}/creatives" + + +async def test_create_creative_omits_account_and_uses_intended_status(recorded_client) -> None: + """The Creatives API rejects a `account` field (read-only) and requires + `intendedStatus` (not `status`). See the 422 errors: + "/account :: ReadOnly field present", "/status :: unrecognized field", + "/intendedStatus :: field is required". + """ + import json + + client, recorder = recorded_client + await client.create_creative( + campaign_urn="urn:li:sponsoredCampaign:1", + content={"reference": "urn:li:share:1"}, + ) + body = json.loads(recorder.requests[0].read()) + assert "account" not in body + assert "status" not in body + assert body["intendedStatus"] == "DRAFT" + assert body["campaign"] == "urn:li:sponsoredCampaign:1" + assert body["content"] == {"reference": "urn:li:share:1"} + + +async def test_create_creative_refuses_active_intended_status(recorded_client) -> None: + client, _ = recorded_client + with pytest.raises(LinkedInError): + await client.create_creative( + campaign_urn="urn:li:sponsoredCampaign:1", + content={"reference": "urn:li:share:1"}, + intended_status="ACTIVE", + ) + + +async def test_create_post_uses_posts_endpoint_as_dark_post(recorded_client) -> None: + """Creatives must reference a real Post. We create it as a dark (DSC) post: + org author, feedDistribution NONE, lifecycleState PUBLISHED, and an adContext + tying it to the sponsored account. + """ + import json + + client, recorder = recorded_client + await client.create_post( + author_urn="urn:li:organization:80050982", + commentary="How Northwind migrated to Lattice Cloud.", + article={"source": "https://lattice.example/cloud", "title": "We replaced our warehouse"}, + dsc_ad_account_urn=f"urn:li:sponsoredAccount:{_AD_ACCOUNT_ID}", + ) + request = recorder.requests[0] + assert request.method == "POST" + assert request.url.path == "/rest/posts" + body = json.loads(request.read()) + assert body["author"] == "urn:li:organization:80050982" + assert body["commentary"].startswith("How Northwind") + assert body["visibility"] == "PUBLIC" + assert body["lifecycleState"] == "PUBLISHED" + assert body["distribution"]["feedDistribution"] == "NONE" + assert body["content"]["article"]["source"] == "https://lattice.example/cloud" + assert body["adContext"]["dscAdAccount"] == f"urn:li:sponsoredAccount:{_AD_ACCOUNT_ID}" + # dscAdType is read-only — sending it on create returns 422 + # "ReadOnly field present in a create request". + assert "dscAdType" not in body["adContext"] + + +async def test_get_ad_account_path_unchanged(recorded_client) -> None: + """Reads were already account-scoped — make sure we don't accidentally regress.""" + client, recorder = recorded_client + await client.get_ad_account() + assert recorder.requests[0].url.path == f"/rest/adAccounts/{_AD_ACCOUNT_ID}" + + +async def test_list_campaigns_uses_account_scoped_path_with_search(recorded_client) -> None: + """New helper for the read path used in smoke testing.""" + client, recorder = recorded_client + await client.list_campaigns() + request = recorder.requests[0] + assert request.method == "GET" + assert request.url.path == f"/rest/adAccounts/{_AD_ACCOUNT_ID}/adCampaigns" + assert request.url.params.get("q") == "search" + + +async def test_create_campaign_includes_offsite_and_political_defaults(recorded_client) -> None: + """LinkedIn now requires offsiteDeliveryEnabled + politicalIntent on Campaign create. + + Defaults must be safe: LinkedIn-only delivery (no Audience Network) and + non-political. Callers can override via kwargs. + """ + client, recorder = recorded_client + await client.create_campaign( + campaign_group_urn="urn:li:sponsoredCampaignGroup:1", + name="smoke", + objective_type="WEBSITE_VISITS", + campaign_type="SPONSORED_UPDATES", + total_budget={"amount": "10", "currencyCode": "EUR"}, + run_schedule={"start": 1, "end": 2}, + targeting_criteria={"include": {"and": []}}, + locale={"country": "US", "language": "en"}, + ) + payload = recorder.requests[0].read() + import json + body = json.loads(payload) + assert body["offsiteDeliveryEnabled"] is False + # politicalIntent is a STRING enum (POLITICAL | NOT_POLITICAL | NOT_DECLARED), + # NOT a boolean. LinkedIn rejects a bool with + # "enum type is not backed by a String". + assert body["politicalIntent"] == "NOT_POLITICAL" + + +async def test_create_campaign_offsite_and_political_overrideable(recorded_client) -> None: + client, recorder = recorded_client + await client.create_campaign( + campaign_group_urn="urn:li:sponsoredCampaignGroup:1", + name="smoke", + objective_type="WEBSITE_VISITS", + campaign_type="SPONSORED_UPDATES", + total_budget={"amount": "10", "currencyCode": "EUR"}, + run_schedule={"start": 1, "end": 2}, + targeting_criteria={"include": {"and": []}}, + locale={"country": "US", "language": "en"}, + offsite_delivery_enabled=True, + political_intent="POLITICAL", + ) + payload = recorder.requests[0].read() + import json + body = json.loads(payload) + assert body["offsiteDeliveryEnabled"] is True + assert body["politicalIntent"] == "POLITICAL" + + +async def test_create_campaign_rejects_invalid_political_intent(recorded_client) -> None: + """Guard against passing a value LinkedIn's enum doesn't accept.""" + client, _ = recorded_client + with pytest.raises(ValueError, match="political_intent"): + await client.create_campaign( + campaign_group_urn="urn:li:sponsoredCampaignGroup:1", + name="smoke", + objective_type="WEBSITE_VISITS", + campaign_type="SPONSORED_UPDATES", + total_budget={"amount": "10", "currencyCode": "EUR"}, + run_schedule={"start": 1, "end": 2}, + targeting_criteria={"include": {"and": []}}, + locale={"country": "US", "language": "en"}, + political_intent="MAYBE", + ) + + +async def test_required_headers_present(recorded_client) -> None: + """All writes must include the LinkedIn-Version + restli protocol headers.""" + client, recorder = recorded_client + await client.create_campaign_group(name="smoke") + request = recorder.requests[0] + # Headers are case-insensitive in httpx. + assert request.headers["authorization"] == "Bearer test-token" + assert request.headers["linkedin-version"] == "202605" + assert request.headers["x-restli-protocol-version"] == "2.0.0" diff --git a/tests/integrations/test_linkedin_mapping.py b/tests/integrations/test_linkedin_mapping.py new file mode 100644 index 0000000..b5dfaa7 --- /dev/null +++ b/tests/integrations/test_linkedin_mapping.py @@ -0,0 +1,76 @@ +"""Tests for the LinkedIn payload mapper.""" + +from __future__ import annotations + +from datetime import date + +from yieldagent.domain import CreativeAsset, Flight +from yieldagent.integrations.linkedin.mapping import ( + campaign_run_schedule, + creative_content_reference, + flight_to_run_schedule, + post_article_content, +) + + +def test_flight_to_run_schedule_emits_epoch_millis() -> None: + flight = Flight(start_date=date(2026, 7, 1), end_date=date(2026, 7, 31)) + out = flight_to_run_schedule(flight) + # Values are epoch milliseconds; start should be 13-digit, end strictly later. + assert out["start"] > 1_000_000_000_000 + assert out["end"] > out["start"] + # 31-day flight: end - start is roughly 31 days in ms, within 1 day tolerance. + span_days = (out["end"] - out["start"]) / 1000 / 86400 + assert 30 < span_days < 32 + + +def test_campaign_run_schedule_spans_earliest_start_to_latest_end() -> None: + """LinkedIn Campaign Group's runSchedule must cover all child Campaigns.""" + flights = [ + Flight(start_date=date(2026, 7, 6), end_date=date(2026, 8, 31)), + Flight(start_date=date(2026, 6, 15), end_date=date(2026, 7, 15)), + Flight(start_date=date(2026, 8, 1), end_date=date(2026, 9, 30)), + ] + out = campaign_run_schedule(flights) + # earliest start = 2026-06-15 + earliest = flight_to_run_schedule( + Flight(start_date=date(2026, 6, 15), end_date=date(2026, 6, 15)) + ) + assert out["start"] == earliest["start"] + # latest end = 2026-09-30 + latest = flight_to_run_schedule( + Flight(start_date=date(2026, 9, 30), end_date=date(2026, 9, 30)) + ) + assert out["end"] == latest["end"] + + +def test_campaign_run_schedule_single_flight_is_passthrough() -> None: + flight = Flight(start_date=date(2026, 7, 1), end_date=date(2026, 7, 31)) + out = campaign_run_schedule([flight]) + assert out == flight_to_run_schedule(flight) + + +def test_post_article_content_maps_landing_headline_description() -> None: + creative = CreativeAsset( + name="Engineering-leader story", + headline="We replaced our warehouse in 30 days.", + primary_text="How Northwind migrated to Lattice Cloud.", + description="A migration story.", + landing_url="https://lattice.example/cloud", + ) + article = post_article_content(creative) + assert article["source"] == "https://lattice.example/cloud" + assert article["title"] == "We replaced our warehouse in 30 days." + assert article["description"] == "A migration story." + # image_url is a plain URL, not an urn:li:image — thumbnail needs the Images + # API, so it must NOT be set here. + assert "thumbnail" not in article + + +def test_post_article_content_defaults_source_when_no_landing_url() -> None: + article = post_article_content(CreativeAsset(name="x")) + assert article["source"].startswith("http") + + +def test_creative_content_reference_wraps_post_urn() -> None: + assert creative_content_reference("urn:li:share:123") == {"reference": "urn:li:share:123"} diff --git a/tests/integrations/test_linkedin_publish.py b/tests/integrations/test_linkedin_publish.py new file mode 100644 index 0000000..ab133a4 --- /dev/null +++ b/tests/integrations/test_linkedin_publish.py @@ -0,0 +1,141 @@ +"""Tests for the publish_draft_campaign orchestration. + +The focus here is the creative-backing-post branch: an ad may reference a +hand-published post via `existing_post_urn` (no posting permission needed), or +let the server mint a Direct Sponsored Content "dark" post. These pin which +path each ad takes without standing up a real MCP server or hitting LinkedIn. +""" + +from __future__ import annotations + +from datetime import date + +import pytest + +from yieldagent.domain import ( + Audience, + Campaign, + CreativeAsset, + Flight, + LineItem, + Money, + Objective, + Targeting, +) +from yieldagent.domain.campaign import Ad +from yieldagent.integrations.linkedin import server as srv +from yieldagent.integrations.linkedin.config import LinkedInConfig + +_AD_ACCOUNT_ID = "537690018" +_ORG_URN = "urn:li:organization:80050982" + + +def _config() -> LinkedInConfig: + return LinkedInConfig( + access_token="t", + ad_account_id=_AD_ACCOUNT_ID, + api_version="202605", + allow_live=True, + organization_urn=_ORG_URN, + ) + + +class _FakeClient: + """Records create_post / create_creative calls; returns plausible ids.""" + + def __init__(self, config: LinkedInConfig): + self.config = config + self.create_post_calls: list[dict] = [] + self.create_creative_calls: list[dict] = [] + + async def __aenter__(self) -> _FakeClient: + return self + + async def __aexit__(self, *_exc: object) -> None: + pass + + def assert_account_allowed(self) -> None: + pass + + async def get_ad_account(self) -> dict: + return {"reference": _ORG_URN} + + async def create_campaign_group(self, **_kw) -> dict: + return {"id": "111"} + + async def create_campaign(self, **_kw) -> dict: + return {"id": "222"} + + async def create_post(self, **kw) -> dict: + self.create_post_calls.append(kw) + return {"id": "urn:li:share:MINTED"} + + async def create_creative(self, **kw) -> dict: + self.create_creative_calls.append(kw) + return {"id": "333"} + + +@pytest.fixture +def patched(monkeypatch): + captured: dict[str, _FakeClient] = {} + + def _factory(config: LinkedInConfig) -> _FakeClient: + client = _FakeClient(config) + captured["client"] = client + return client + + monkeypatch.setattr(srv, "LinkedInClient", _factory) + monkeypatch.setattr(srv.LinkedInConfig, "from_env", classmethod(lambda cls: _config())) + return captured + + +def _campaign(*creatives: CreativeAsset) -> dict: + line_item = LineItem( + name="LI-1", + budget=Money(amount=100, currency="EUR"), + flight=Flight(start_date=date(2026, 7, 1), end_date=date(2026, 7, 31)), + targeting=Targeting(audience=Audience(description="engineers", geos=["US"])), + ) + ads = [ + Ad(name=c.name, line_item_name="LI-1", creative=c) for c in creatives + ] + return Campaign( + name="C", + objective=Objective.traffic, + line_items=[line_item], + ads=ads, + ).model_dump(mode="json") + + +async def test_existing_post_urn_skips_create_post(patched) -> None: + campaign = _campaign( + CreativeAsset(name="reuse", existing_post_urn="urn:li:share:HANDMADE") + ) + result = await srv.publish_draft_campaign(campaign) + client = patched["client"] + assert client.create_post_calls == [] + assert len(client.create_creative_calls) == 1 + assert client.create_creative_calls[0]["content"] == {"reference": "urn:li:share:HANDMADE"} + assert result["ads"][0]["post_urn"] == "urn:li:share:HANDMADE" + + +async def test_missing_post_urn_mints_dark_post(patched) -> None: + campaign = _campaign( + CreativeAsset(name="fresh", landing_url="https://example.com") + ) + await srv.publish_draft_campaign(campaign) + client = patched["client"] + assert len(client.create_post_calls) == 1 + assert client.create_creative_calls[0]["content"] == {"reference": "urn:li:share:MINTED"} + + +async def test_mixed_ads_only_mint_for_missing_urn(patched) -> None: + campaign = _campaign( + CreativeAsset(name="reuse", existing_post_urn="urn:li:share:HANDMADE"), + CreativeAsset(name="fresh", landing_url="https://example.com"), + ) + await srv.publish_draft_campaign(campaign) + client = patched["client"] + assert len(client.create_post_calls) == 1 + refs = {c["content"]["reference"] for c in client.create_creative_calls} + assert refs == {"urn:li:share:HANDMADE", "urn:li:share:MINTED"}