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()
            res1 = _async_materialize(db_engine, f["campaign"], recs, preview=True)
            res2 = _async_materialize(db_engine, f["campaign"], recs, preview=True)
        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"])