maskanx_cm_backend/tests/test_campaign_crm_sync.py

# -*- coding: utf-8 -*-
"""Pushing leads and campaign figures into Maskan CRM.

The properties worth protecting: every lead in the window is re-sent so a
half-failed run heals itself, a lead without Meta's id is never sent (it
would duplicate on every run), campaign figures come from stored insights
rather than Meta, and one campaign failing does not stop the rest.
"""
from datetime import datetime, timedelta, timezone

import pytest

from adclaw.campaigns.crm_sync import (
    DEFAULT_INTERVAL_SECONDS,
    crm_sync_interval_seconds,
    lead_window_days,
    push_campaign_leads,
    push_campaign_metrics,
    sync_to_crm,
)
from adclaw.campaigns.models import CampaignSpec


class FakeCRM:
    def __init__(self):
        self.leads: list[tuple[dict, str]] = []
        self.campaigns: list[dict] = []
        self.fail_on_lead: str | None = None

    def create_lead(self, payload, idempotency_key):
        if payload.get("external_id") == self.fail_on_lead:
            raise RuntimeError("CRM rejected the lead")
        self.leads.append((payload, idempotency_key))
        return {"lead_id": "crm_1", "created": True}

    def push_campaign(self, payload):
        self.campaigns.append(payload)
        return {"campaign_id": "crm_camp_1", "created": True}


class FakeMeta:
    def __init__(self, leads=None):
        self.leads = leads if leads is not None else []
        self.calls: list[dict] = []

    async def get_leads(self, object_id, *, since=None, limit=200):
        self.calls.append({"object_id": object_id, "since": since})
        return self.leads


class FakeInsightsRepo:
    """Returns totals the way the real repository does: already normalised.

    The real `totals_between` sums in SQL over stored rows, whose metrics
    were normalised on the way in. Returning raw Meta-shaped values here
    would let a double-conversion bug pass unnoticed.
    """

    def __init__(self, totals=None):
        self.totals = totals if totals is not None else {}

    async def totals_between(self, since, until, campaign_id=None):
        return dict(self.totals)


class FakeRepo:
    def __init__(self, campaigns=()):
        self.items = list(campaigns)
        self.events: list[dict] = []

    async def list_campaigns(self, company_id=None, status=None):
        return list(self.items)

    async def update_campaign_with_event(
        self, spec, event_type, actor=None, reason=None, payload=None,
    ):
        self.events.append({"event_type": event_type, "payload": payload})
        return spec


def _campaign(**overrides) -> CampaignSpec:
    data = {
        "id": "camp_1",
        "name": "Q3 lead gen",
        "status": "live",
        "objective": "OUTCOME_LEADS",
        "ad_account_id": "act_1",
        "meta_campaign_id": "meta_camp_1",
        "budget": {"daily_budget": 50000, "currency": "INR"},
        "channels": ["facebook"],
    }
    data.update(overrides)
    return CampaignSpec(**data)


def _meta_lead(lead_id="lead_1", **overrides):
    lead = {
        "id": lead_id,
        "campaign_id": "meta_camp_1",
        "adset_id": "meta_set_1",
        "ad_id": "meta_ad_1",
        "field_data": [
            {"name": "full_name", "values": ["Asha Menon"]},
            {"name": "email", "values": ["asha@example.com"]},
        ],
    }
    lead.update(overrides)
    return lead


# --- configuration ---


def test_the_interval_defaults_to_five_minutes(monkeypatch):
    monkeypatch.delenv("MASKANX_CRM_SYNC_SECONDS", raising=False)

    assert crm_sync_interval_seconds() == DEFAULT_INTERVAL_SECONDS


def test_a_nonsense_interval_falls_back_to_the_default(monkeypatch):
    monkeypatch.setenv("MASKANX_CRM_SYNC_SECONDS", "often")

    assert crm_sync_interval_seconds() == DEFAULT_INTERVAL_SECONDS


def test_the_lead_window_is_never_shorter_than_a_day(monkeypatch):
    monkeypatch.setenv("MASKANX_CRM_LEAD_WINDOW_DAYS", "0")

    assert lead_window_days() == 1


# --- leads ---


async def test_each_lead_is_sent_keyed_on_metas_lead_id():
    """That key is what lets the CRM ignore a lead it already has."""
    repo = FakeRepo()
    crm, meta = FakeCRM(), FakeMeta([_meta_lead("lead_1"), _meta_lead("lead_2")])

    sent = await push_campaign_leads(repo, crm, meta, _campaign())

    assert sent == 2
    assert [key for _, key in crm.leads] == ["meta-lead-lead_1", "meta-lead-lead_2"]


async def test_the_lead_carries_its_campaign_attribution():
    crm, meta = FakeCRM(), FakeMeta([_meta_lead()])

    await push_campaign_leads(FakeRepo(), crm, meta, _campaign())

    payload, _ = crm.leads[0]
    assert payload["campaign"]["ad_id"] == "meta_ad_1"
    assert payload["campaign"]["campaign_name"] == "Q3 lead gen"
    assert payload["first_name"] == "Asha"


async def test_a_lead_without_an_id_is_never_sent():
    """With no stable external id it would duplicate on every run."""
    crm, meta = FakeCRM(), FakeMeta([_meta_lead(lead_id="")])

    assert await push_campaign_leads(FakeRepo(), crm, meta, _campaign()) == 0
    assert crm.leads == []


async def test_an_unsynced_campaign_asks_meta_for_nothing():
    crm, meta = FakeCRM(), FakeMeta()

    sent = await push_campaign_leads(
        FakeRepo(), crm, meta, _campaign(meta_campaign_id=None),
    )

    assert sent == 0
    assert meta.calls == []


# --- the cursor ---


async def test_a_campaign_never_synced_reads_the_whole_window():
    """A campaign adopted today should still bring its recent leads across."""
    crm, meta = FakeCRM(), FakeMeta()

    await push_campaign_leads(FakeRepo(), crm, meta, _campaign())

    since = datetime.fromtimestamp(int(meta.calls[0]["since"]), tz=timezone.utc)
    age = datetime.now(timezone.utc) - since
    assert timedelta(days=6) < age < timedelta(days=9)


async def test_the_next_run_reads_from_where_the_last_one_finished():
    """Re-sending the whole window every five minutes is thousands of
    requests a day for leads the CRM already has."""
    repo = FakeRepo()
    campaign = _campaign()
    crm, meta = FakeCRM(), FakeMeta([_meta_lead()])

    await push_campaign_leads(repo, crm, meta, campaign)
    await push_campaign_leads(repo, crm, meta, campaign)

    second_since = datetime.fromtimestamp(
        int(meta.calls[1]["since"]), tz=timezone.utc,
    )
    # Roughly an hour back, the deliberate overlap — not seven days.
    age = datetime.now(timezone.utc) - second_since
    assert timedelta(minutes=50) < age < timedelta(hours=2)


async def test_the_cursor_only_advances_after_every_lead_is_accepted():
    """A run that fails half-way must be retried in full by the next one."""
    repo = FakeRepo()
    campaign = _campaign()
    crm = FakeCRM()
    crm.fail_on_lead = "lead_2"
    meta = FakeMeta([_meta_lead("lead_1"), _meta_lead("lead_2")])

    with pytest.raises(RuntimeError):
        await push_campaign_leads(repo, crm, meta, campaign)

    assert "crm_sync" not in campaign.advanced


async def test_nothing_is_written_when_there_were_no_leads():
    """Otherwise every campaign gets an event row every five minutes."""
    repo = FakeRepo()

    await push_campaign_leads(repo, FakeCRM(), FakeMeta([]), _campaign())

    assert repo.events == []


async def test_a_successful_send_is_recorded():
    repo = FakeRepo()

    await push_campaign_leads(repo, FakeCRM(), FakeMeta([_meta_lead()]), _campaign())

    assert repo.events[-1]["event_type"] == "campaign.crm_leads_synced"
    assert repo.events[-1]["payload"]["leads_sent"] == 1


# --- campaign figures ---


async def test_campaign_figures_come_from_stored_insights_not_meta():
    """This runs every few minutes; re-reading Meta would burn the quota."""
    crm = FakeCRM()
    insights = FakeInsightsRepo(
        {"spend": 46000, "clicks": 30, "leads": 4, "impressions": 1800},
    )

    await push_campaign_metrics(crm, insights, _campaign())

    payload = crm.campaigns[0]
    assert payload["spend"] == 46000
    assert payload["leads"] == 4
    assert payload["cost_per_lead"] == 11500


async def test_stored_spend_is_sent_as_is_not_converted_again():
    """Stored metrics are already in minor units.

    Passing them through the raw-row normaliser would multiply spend by a
    hundred, and the CRM would report a campaign that cost 460 rupees as
    having cost 46,000.
    """
    crm = FakeCRM()

    await push_campaign_metrics(
        crm, FakeInsightsRepo({"spend": 46000, "leads": 4}), _campaign(),
    )

    assert crm.campaigns[0]["spend"] == 46000


async def test_the_campaign_is_identified_by_its_maskanx_id():
    """The CRM upserts on it, so it has to be stable across pushes."""
    crm = FakeCRM()

    await push_campaign_metrics(crm, FakeInsightsRepo(), _campaign())

    assert crm.campaigns[0]["external_id"] == "camp_1"
    assert crm.campaigns[0]["provider"] == "maskanx"


async def test_a_campaign_with_no_insights_still_reports_zero():
    """Absent numbers and zero numbers should look the same in the CRM."""
    crm = FakeCRM()

    await push_campaign_metrics(crm, FakeInsightsRepo({}), _campaign())

    assert crm.campaigns[0]["spend"] == 0
    assert crm.campaigns[0]["cost_per_lead"] is None


async def test_the_budget_and_currency_are_carried_through():
    crm = FakeCRM()

    await push_campaign_metrics(crm, FakeInsightsRepo(), _campaign())

    assert crm.campaigns[0]["daily_budget"] == 50000
    assert crm.campaigns[0]["currency"] == "INR"


# --- the sweep ---


async def test_the_sweep_sends_both_figures_and_leads():
    repo = FakeRepo([_campaign()])
    crm, meta = FakeCRM(), FakeMeta([_meta_lead()])

    result = await sync_to_crm(repo, FakeInsightsRepo(), meta, crm)

    assert result.campaigns_sent == 1
    assert result.leads_sent == 1


async def test_campaigns_that_never_reached_meta_are_skipped():
    repo = FakeRepo([_campaign(meta_campaign_id=None)])

    result = await sync_to_crm(repo, FakeInsightsRepo(), FakeMeta(), FakeCRM())

    assert result.campaigns_sent == 0


async def test_one_campaign_failing_does_not_stop_the_rest():
    repo = FakeRepo([
        _campaign(id="c1", meta_campaign_id="meta_1"),
        _campaign(id="c2", meta_campaign_id="meta_2"),
    ])
    crm = FakeCRM()
    crm.fail_on_lead = "lead_1"

    class PerCampaignMeta(FakeMeta):
        async def get_leads(self, object_id, *, since=None, limit=200):
            # Only the first campaign has the lead the CRM will reject.
            return [_meta_lead("lead_1")] if object_id == "meta_1" else []

    result = await sync_to_crm(repo, FakeInsightsRepo(), PerCampaignMeta(), crm)

    assert result.failed == ["c1"]
    assert result.campaigns_sent == 2


@pytest.mark.parametrize("status", ["live", "paused", "stopped", "synced"])
async def test_paused_and_stopped_campaigns_are_still_reported(status):
    """They spent money; the CRM should keep showing what it bought."""
    repo = FakeRepo([_campaign(status=status)])
    crm = FakeCRM()

    result = await sync_to_crm(repo, FakeInsightsRepo(), FakeMeta(), crm)

    assert result.campaigns_sent == 1
    assert crm.campaigns[0]["status"] == status