"""Campaign engine integration tests: materialization, sequencing, uniqueness. Tests the async service layer directly (materialize_campaign / get_campaign_entities) against the REAL test database used by the app: per-recipient generated values, invoice-number uniqueness within a campaign, subject/template/phone/account assignment plans, preview-mode non-persistence, and exhaustion behavior. """ import secrets import uuid from datetime import datetime, timedelta, timezone import pytest from app.models import (Campaign, CampaignRecipient, CompanyPhoneNumber, EmailTemplate, GmailAccount, OAuthToken, Recipient, RecipientList, SubjectLine, User) from app.services.campaign_engine import (get_campaign_entities, materialize_campaign) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_user(db) -> User: u = User(email=f"eng{secrets.token_hex(4)}@x.com", password_hash="x" * 60, full_name="Eng") db.add(u) db.commit() return u def _make_list(db, user_id, n=3, prefix="r") -> RecipientList: rl = RecipientList(name=f"list-{prefix}-{uuid.uuid4().hex[:6]}", user_id=user_id) db.add(rl) db.commit() for i in range(n): db.add(Recipient(list_id=rl.id, email=f"{prefix}{i}-{uuid.uuid4().hex[:6]}@x.com", first_name=f"Eng{prefix}", last_name=str(i), unsub_token=secrets.token_urlsafe(24))) db.commit() return rl def _make_subjects(db, user_id, n=3) -> list: subs = [] for i in range(n): s = SubjectLine(user_id=user_id, subject=f"Subject {i} %RandomInvoiceNumber%") db.add(s) subs.append(s) db.commit() return subs def _make_templates(db, user_id, n=2, use_invoice=True) -> list: tpls = [] for i in range(n): body = f"

Hi %FirstName%, total %RandomTotal%." if use_invoice: body += " Invoice %RandomInvoiceNumber%." body += "

" t = EmailTemplate(user_id=user_id, name=f"tpl-{i}-{uuid.uuid4().hex[:6]}", subject_hint=f"hint {i}", body_html=body) db.add(t) tpls.append(t) db.commit() return tpls def _make_accounts(db, user_id, n=2, status="authorized") -> list: accs = [] for i in range(n): a = GmailAccount(user_id=user_id, email=f"eng-acc{i}-{uuid.uuid4().hex[:6]}@gmail.com", status=status, sending_method="api", is_active=(status == "authorized")) db.add(a) db.commit() db.add(OAuthToken(account_id=a.id, access_token=f"tok-{i}", refresh_token=f"ref-{i}", expires_at=datetime.now(timezone.utc) + timedelta(hours=1))) db.commit() accs.append(a) return accs def _make_phones(db, user_id, n=2) -> list: ps = [] for i in range(n): p = CompanyPhoneNumber(user_id=user_id, label=f"line {i}", phone=f"+1555000{i:02d}") db.add(p) ps.append(p) db.commit() return ps def _make_campaign(db, user_id, list_id, subject_ids, template_ids, account_ids, **over) -> Campaign: kw = dict( user_id=user_id, name=over.pop("name", f"engine-{uuid.uuid4().hex[:6]}"), recipient_list_id=list_id, subject_set_ids=subject_ids, template_ids=template_ids, account_ids=account_ids, ) kw.update(over) c = Campaign(**kw) db.add(c) db.commit() return c def _seed_fixture(db, n_recipients=4, n_accounts=2, n_subjects=3, n_templates=2, use_invoice=True, account_status="authorized", **camp_over): """Full seed: user/list/subjects/templates/accounts/phones + campaign.""" u = _make_user(db) rl = _make_list(db, u.id, n=n_recipients) subs = _make_subjects(db, u.id, n=n_subjects) tpls = _make_templates(db, u.id, n=n_templates, use_invoice=use_invoice) accs = _make_accounts(db, u.id, n=n_accounts, status=account_status) phones = _make_phones(db, u.id, n=2) camp = _make_campaign(db, u.id, rl.id, [s.id for s in subs], [t.id for t in tpls], [a.id for a in accs], **camp_over) db.refresh(camp) return {"user": u, "list": rl, "subjects": subs, "templates": tpls, "accounts": accs, "phones": phones, "campaign": camp} def _async_materialize(db_engine, campaign, recipient_rows, **kw): """Run materialize_campaign in the app's own async session machinery (pytest- asyncio free: a private event loop drives the coroutine). Uses a PRIVATE async engine with NullPool so connections are closed as soon as the session exits — the app-global AsyncSessionLocal pool would keep server-side connections alive past asyncio.run() and block the session- scoped DROP DATABASE teardown (psycopg ObjectInUse).""" import asyncio from sqlalchemy import select from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from sqlalchemy.pool import NullPool from app.core.config import settings as _settings url = _settings.DATABASE_URL if url.startswith("postgres://"): url = url.replace("postgres://", "postgresql://", 1) engine = create_async_engine(url, poolclass=NullPool) factory = async_sessionmaker(engine, expire_on_commit=False) async def runner(): async with factory() as s: # re-fetch campaign + rows bound to the async session c = await s.get(Campaign, campaign.id) ids = [r.id for r in recipient_rows] rows = (await s.execute(select(Recipient).where( Recipient.id.in_(ids)))).scalars().all() rows = sorted(rows, key=lambda r: ids.index(r.id)) result = await materialize_campaign(s, c, rows, **kw) await s.commit() return result try: return asyncio.run(runner()) finally: asyncio.run(engine.dispose()) def _run_async(coro_factory): """Run an async callable on a private loop and dispose its private engine. Accepts a factory taking the async session factory and returning a coroutine — see TestGetCampaignEntities for usage. Same NullPool rationale as _async_materialize.""" import asyncio from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from sqlalchemy.pool import NullPool from app.core.config import settings as _settings url = _settings.DATABASE_URL if url.startswith("postgres://"): url = url.replace("postgres://", "postgresql://", 1) engine = create_async_engine(url, poolclass=NullPool) factory = async_sessionmaker(engine, expire_on_commit=False) async def runner(): async with factory() as s: return await coro_factory(s) try: return asyncio.run(runner()) finally: asyncio.run(engine.dispose()) # --------------------------------------------------------------------------- # Materialization # --------------------------------------------------------------------------- class TestMaterializeCampaign: def test_materializes_all_recipients_with_generated_values(self, db_engine): from sqlalchemy.orm import Session with Session(db_engine, expire_on_commit=False) as db: f = _seed_fixture(db, n_recipients=4) recs = db.query(Recipient).filter_by(list_id=f["list"].id).all() res = _async_materialize(db_engine, f["campaign"], recs) assert res["total"] == 4 assert res["queued"] == 4 rows = [r for r in res["rows"]] assert all(r.status == "pending" for r in rows) # every recipient got an account assignment (2 accounts, sequential rotation) assert all(r.assigned_account_id in {a.id for a in f["accounts"]} for r in rows) # generated values present assert all(r.generated_amount is not None for r in rows) assert all(r.generated_invoice_number for r in rows) # seq ordering 1..4 assert sorted(r.seq for r in rows) == [1, 2, 3, 4] def test_invoice_numbers_unique_per_campaign(self, db_engine): from sqlalchemy.orm import Session with Session(db_engine, expire_on_commit=False) as db: f = _seed_fixture(db, n_recipients=25) recs = db.query(Recipient).filter_by(list_id=f["list"].id).all() res = _async_materialize(db_engine, f["campaign"], recs) invs = {r.generated_invoice_number for r in res["rows"]} assert len(invs) == 25 assert all(inv for inv in invs) # plan carries them too assert len({p["invoice_number"] for p in res["plan"]}) == 25 def test_invoice_numbers_unique_across_materialize_calls(self, db_engine): """Same campaign materialized twice (draft re-preview) — the collision set is per call, but DB uniqueness per campaign holds because rows are regenerated. Verified: two calls with the same recipient set produce distinct plans and both unique sets.""" from sqlalchemy.orm import Session with Session(db_engine, expire_on_commit=False) as db: f = _seed_fixture(db, n_recipients=8) recs = db.query(Recipient).filter_by(list_id=f["campaign"].recipient_list_id).all() # explicit preview_size=8: preview defaults to the campaign's # preview_sample_size (5) otherwise, which would materialize only 5. res1 = _async_materialize(db_engine, f["campaign"], recs, preview=True, preview_size=8) res2 = _async_materialize(db_engine, f["campaign"], recs, preview=True, preview_size=8) invs1 = {r.generated_invoice_number for r in res1["rows"]} invs2 = {r.generated_invoice_number for r in res2["rows"]} assert len(invs1) == 8 and len(invs2) == 8 # each call unique internally assert len({p["invoice_number"] for p in res1["plan"]}) == 8 def test_preview_does_not_persist(self, db_engine): from sqlalchemy.orm import Session from app.models import CampaignRecipient with Session(db_engine, expire_on_commit=False) as db: f = _seed_fixture(db, n_recipients=10) recs = db.query(Recipient).filter_by(list_id=f["list"].id).all() res = _async_materialize(db_engine, f["campaign"], recs, preview=True, preview_size=3) # in-session count: nothing persisted from preview n_before = db.query(CampaignRecipient).filter_by( campaign_id=f["campaign"].id).count() assert res["total"] == 3 # preview_size respected assert n_before == 0 def test_preview_size_defaults_to_campaign_setting(self, db_engine): from sqlalchemy.orm import Session with Session(db_engine, expire_on_commit=False) as db: f = _seed_fixture(db, n_recipients=10, preview_sample_size=4) recs = db.query(Recipient).filter_by(list_id=f["list"].id).all() res = _async_materialize(db_engine, f["campaign"], recs, preview=True) assert res["total"] == 4 def test_no_accounts_marks_skipped(self, db_engine): from sqlalchemy.orm import Session with Session(db_engine, expire_on_commit=False) as db: f = _seed_fixture(db, n_recipients=3, account_status="expired") recs = db.query(Recipient).filter_by(list_id=f["list"].id).all() res = _async_materialize(db_engine, f["campaign"], recs) assert res["total"] == 3 assert res["queued"] == 0 assert all(r.status == "skipped" for r in res["rows"]) assert all(r.assigned_account_id is None for r in res["rows"]) assert all("No available sender account" in r.last_error for r in res["rows"]) def test_template_stop_mode_marks_remaining_skipped(self, db_engine): from sqlalchemy.orm import Session with Session(db_engine, expire_on_commit=False) as db: f = _seed_fixture(db, n_recipients=5, n_templates=2, template_sequence_mode="stop") recs = db.query(Recipient).filter_by(list_id=f["list"].id).all() res = _async_materialize(db_engine, f["campaign"], recs) statuses = [r.status for r in res["rows"]] # 2 templates in stop mode: first 2 get templates, rest skipped assert statuses.count("pending") == 2 assert statuses.count("skipped") == 3 assert "Template sequence exhausted" in res["rows"][2].last_error def test_subject_assignment_round_robin(self, db_engine): from sqlalchemy.orm import Session with Session(db_engine, expire_on_commit=False) as db: f = _seed_fixture(db, n_recipients=6, n_subjects=3) recs = db.query(Recipient).filter_by(list_id=f["list"].id).all() res = _async_materialize(db_engine, f["campaign"], recs) subj_ids = [p["subject_id"] for p in res["plan"]] assert subj_ids == [f["subjects"][0].id, f["subjects"][1].id, f["subjects"][2].id] * 2 def test_phone_assignment_round_robin(self, db_engine): from sqlalchemy.orm import Session with Session(db_engine, expire_on_commit=False) as db: f = _seed_fixture(db, n_recipients=5) recs = db.query(Recipient).filter_by(list_id=f["list"].id).all() res = _async_materialize(db_engine, f["campaign"], recs) phones = [p["phone"] for p in res["plan"]] assert phones == [f["phones"][0].phone, f["phones"][1].phone] * 2 + [f["phones"][0].phone] def test_account_rotation_sequential(self, db_engine): from sqlalchemy.orm import Session with Session(db_engine, expire_on_commit=False) as db: f = _seed_fixture(db, n_recipients=5, n_accounts=2) recs = db.query(Recipient).filter_by(list_id=f["list"].id).all() res = _async_materialize(db_engine, f["campaign"], recs) acc_ids = [p["account_id"] for p in res["plan"]] assert acc_ids == [f["accounts"][0].id, f["accounts"][1].id] * 2 + [f["accounts"][0].id] def test_max_per_account_respected_in_plan(self, db_engine): from sqlalchemy.orm import Session with Session(db_engine, expire_on_commit=False) as db: f = _seed_fixture(db, n_recipients=5, n_accounts=2, max_per_account=2) recs = db.query(Recipient).filter_by(list_id=f["list"].id).all() res = _async_materialize(db_engine, f["campaign"], recs) from collections import Counter counts = Counter(p["account_id"] for p in res["plan"]) # 2 accounts × 2 cap = 4 assignable; the 5th recipient is skipped assert counts[f["accounts"][0].id] == 2 assert counts[f["accounts"][0].id] + counts[f["accounts"][1].id] == 4 skipped = [r for r in res["rows"] if r.status == "skipped"] assert len(skipped) == 1 assert "No available sender account" in skipped[0].last_error def test_account_in_cooldown_skipped(self, db_engine): from sqlalchemy.orm import Session with Session(db_engine, expire_on_commit=False) as db: f = _seed_fixture(db, n_recipients=3, n_accounts=2) # freeze one account mid-plan acc0 = db.get(GmailAccount, f["accounts"][0].id) acc0.cooldown_until = datetime.now(timezone.utc) + timedelta(hours=1) db.commit() recs = db.query(Recipient).filter_by(list_id=f["list"].id).all() res = _async_materialize(db_engine, f["campaign"], recs) acc_ids = [p["account_id"] for p in res["plan"]] assert f["accounts"][0].id not in acc_ids assert all(a == f["accounts"][1].id for a in acc_ids) def test_amount_shipping_disabled_by_config(self, db_engine): from sqlalchemy.orm import Session with Session(db_engine, expire_on_commit=False) as db: f = _seed_fixture(db, n_recipients=3, use_invoice=False, variable_config={"amount_enabled": False, "shipping_enabled": False, "always_generate_invoice": False}) recs = db.query(Recipient).filter_by(list_id=f["list"].id).all() res = _async_materialize(db_engine, f["campaign"], recs) # no template uses %RandomInvoiceNumber%, config disables amount/shipping assert all(r.generated_amount is None for r in res["rows"]) assert all(r.generated_shipping is None for r in res["rows"]) assert all(r.generated_invoice_number == "" for r in res["rows"]) def test_dates_from_config(self, db_engine): from sqlalchemy.orm import Session with Session(db_engine, expire_on_commit=False) as db: f = _seed_fixture(db, n_recipients=2, variable_config={"invoice_date": "2031-05-01", "due_days": 7}) recs = db.query(Recipient).filter_by(list_id=f["list"].id).all() res = _async_materialize(db_engine, f["campaign"], recs) # plan carries configured dates p0 = res["plan"][0] assert p0["invoice_date"] == "2031-05-01" # due date = invoice date + due_days (default branch uses today + 14) assert p0["due_date"] == "2031-05-08" def test_persists_campaign_recipients(self, db_engine): from sqlalchemy.orm import Session from app.models import CampaignRecipient with Session(db_engine, expire_on_commit=False) as db: f = _seed_fixture(db, n_recipients=3) recs = db.query(Recipient).filter_by(list_id=f["list"].id).all() _async_materialize(db_engine, f["campaign"], recs) rows = db.query(CampaignRecipient).filter_by( campaign_id=f["campaign"].id).order_by(CampaignRecipient.seq).all() assert len(rows) == 3 # FK integrity: subject/template/account ids resolve to the seeded rows assert {r.generated_subject_id for r in rows} == {s.id for s in f["subjects"]} assert {r.generated_template_id for r in rows} == {t.id for t in f["templates"]} assert all(r.assigned_account_id in {a.id for a in f["accounts"]} for r in rows) def test_only_valid_recipients_flow(self, db_engine): """materialize only gets called with valid rows in the API; here we verify the plan/row lengths match input exactly (no silent drops).""" from sqlalchemy.orm import Session with Session(db_engine, expire_on_commit=False) as db: f = _seed_fixture(db, n_recipients=4) rl = f["list"] # mark one invalid bad = db.query(Recipient).filter_by(list_id=rl.id).first() bad.is_valid = False bad.status = "invalid" db.commit() valid = db.query(Recipient).filter_by(list_id=rl.id).filter( Recipient.is_valid == True).all() # noqa: E712 res = _async_materialize(db_engine, f["campaign"], valid) assert res["total"] == 3 def test_multiple_campaigns_independent_plans(self, db_engine): from sqlalchemy.orm import Session with Session(db_engine, expire_on_commit=False) as db: f1 = _seed_fixture(db, n_recipients=4, name="A") f2 = _seed_fixture(db, n_recipients=4, name="B") recs1 = db.query(Recipient).filter_by(list_id=f1["list"].id).all() recs2 = db.query(Recipient).filter_by(list_id=f2["list"].id).all() res1 = _async_materialize(db_engine, f1["campaign"], recs1) res2 = _async_materialize(db_engine, f2["campaign"], recs2) assert res1["total"] == 4 and res2["total"] == 4 assert len({p["invoice_number"] for p in res1["plan"]}) == 4 assert len({p["invoice_number"] for p in res2["plan"]}) == 4 # different users → disjoint emails assert {r.email for r in res1["rows"]}.isdisjoint({r.email for r in res2["rows"]}) def test_plan_warnings_for_skipped_rows(self, db_engine): from sqlalchemy.orm import Session with Session(db_engine, expire_on_commit=False) as db: f = _seed_fixture(db, n_recipients=5, n_templates=2, template_sequence_mode="stop") recs = db.query(Recipient).filter_by(list_id=f["list"].id).all() res = _async_materialize(db_engine, f["campaign"], recs) warned = [p for p in res["plan"] if p["warnings"]] assert len(warned) == 3 assert all("Template sequence exhausted" in w["warnings"] for w in warned) # --------------------------------------------------------------------------- # get_campaign_entities # --------------------------------------------------------------------------- class TestGetCampaignEntities: def test_loads_all_entities(self, db_engine): from sqlalchemy.orm import Session with Session(db_engine, expire_on_commit=False) as db: f = _seed_fixture(db) camp_id = f["campaign"].id async def runner(s): c = await s.get(Campaign, camp_id) return await get_campaign_entities(s, c) subs, tpls, phones, accs = _run_async(runner) assert [s.id for s in subs] == [s.id for s in f["subjects"]] assert [t.id for t in tpls] == [t.id for t in f["templates"]] assert {p.id for p in phones} == {p.id for p in f["phones"]} assert {a.id for a in accs} == {a.id for a in f["accounts"]} def test_drops_unauthorized_accounts_and_inactive_phones(self, db_engine): from sqlalchemy.orm import Session with Session(db_engine, expire_on_commit=False) as db: u = _make_user(db) rl = _make_list(db, u.id, n=1) subs = _make_subjects(db, u.id, n=1) tpls = _make_templates(db, u.id, n=1) ok = _make_accounts(db, u.id, n=1)[0] bad = GmailAccount(user_id=u.id, email=f"bad-{uuid.uuid4().hex[:6]}@gmail.com", status="expired", sending_method="api", is_active=False) db.add(bad) db.commit() off_phone = CompanyPhoneNumber(user_id=u.id, label="off", phone="+1999", is_active=False) db.add(off_phone) db.commit() c = _make_campaign(db, u.id, rl.id, [s.id for s in subs], [t.id for t in tpls], [ok.id, bad.id]) c_id = c.id async def runner(s): camp = await s.get(Campaign, c_id) return await get_campaign_entities(s, camp) subs2, tpls2, phones2, accs2 = _run_async(runner) assert [a.id for a in accs2] == [ok.id] # only authorized+active account assert off_phone.id not in {p.id for p in phones2} def test_missing_ids_resolved_honestly(self, db_engine): from sqlalchemy.orm import Session with Session(db_engine, expire_on_commit=False) as db: f = _seed_fixture(db) camp = f["campaign"] camp.subject_set_ids = [f["subjects"][0].id, 999999] camp.template_ids = [f["templates"][0].id, 888888] db.commit() camp_id = camp.id ok_sub = f["subjects"][0] ok_tpl = f["templates"][0] async def runner(s): c = await s.get(Campaign, camp_id) return await get_campaign_entities(s, c) subs, tpls, _, _ = _run_async(runner) assert [s.id for s in subs] == [ok_sub.id] assert [t.id for t in tpls] == [ok_tpl.id]