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 with the app's own async engine (pytest-asyncio
    free: a private event loop drives the coroutine)."""
    import asyncio
    from app.db.session import AsyncSessionLocal

    async def runner():
        async with AsyncSessionLocal() 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]
            from sqlalchemy import select
            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

    return asyncio.run(runner())


# ---------------------------------------------------------------------------
# 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"]}