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,
                                                  is_valid__eq=True).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
        invs1 = {p["invoice_number"] for p in res1["plan"]}
        invs2 = {p["campaign_id"] if "campaign_id" in p else p["invoice"] for p in res2["plan"]}
        assert len(invs1) == 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_persists_campaign_recipients(self, db_engine):  # noqa: F811
        pass

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