assert "Invalid link" in r.text

    def test_unknown_token_invalid_link(self, api):
        r = api.get("/unsubscribe/{target}".format(target="Z" * 32))
        assert r.status_code == 200
        assert "Invalid link" in r.text

    testdata_unknown_json = None

    def test_confirm_adds_suppression_and_event(self, api, db):
        # with campaign context: suppression + EmailEvent recorded
        from app.models import Campaign, CampaignRecipient
        rec = self._seed(db, user_id_of(api))
===
        assert rows[1][1] == "camp1@x.com"
        assert rows[1][2] == "queued"

    @requires_broker
    def test_sse_stream_first_events(self, api, db):
        f = _seed_campaign_fixture(db, user_id_of(api), n_recipients=2)
        cid = api.post("/campaigns/create", headers=H(api),
                       json=_cIn(f, name="SSE")).json()["id"]
        api.post(f"/campaigns/{cid}/start", headers=H(api))
        # consume the first SSE frames without waiting for the stream end
        frames = []
        with api.stream("GET", f"/campaigns/{cid}/stream") as resp:
            assert resp.status_code == 200
            assert resp.headers["content-type"].startswith("text/event-stream")
            for line in resp.iter_lines():
                frames.append(line)
                if "event: progress" in line or "event: connected" in line:
                    pass
                if len(frames) > 12:
                    break
        assert any("event: connected" in f for f in frames)
        assert any("keepalive" in f for f in frames) or any("event: progress" in f for f in frames)

    def test_campaign_isolation(self, api, db, second_user):
        f = _seed_campaign_fixture(db, user_id_of(api))
        cid = api.post("/campaigns/create", headers=H(api),
                       json=_cIn(f, name="Mine")).json()["id"]
        assert second_user.get(f"/campaigns/{cid}").status_code == 404
        r = second_user.post(f"/campaigns/{cid}/start", headers=H(second_user))
        assert r.status_code == 404

    def test_requires_auth_and_csrf(self, api):
        from starlette.testclient import TestClient
        anon = TestClient(api.app)
        assert anon.get("/campaigns/list").status_code == 401
        assert anon.post("/campaigns/create", json={"name": "x"}).status_code == 401
        r = api.post("/campaigns/create", json={"name": "NoCsrf"})
        assert r.status_code == 403
        r2 = api.post("/campaigns/validate-step", json={"step": 1, "data": {}})
        assert r2.status_code == 403