acc = GmailAccount(user_id=user_id, email=email or f"g{uuid.uuid4().hex[:8]}@gmail.com", display_name="Test Account", status=status, sending_method=method, is_active=(status != "revoked")) db.add(acc) db.flush() if with_token: db.add(OAuthToken(account_id=acc.id, access_token="fake-access-token", refresh_token=refresh, expires_at=datetime.now(timezone.utc) + timedelta(hours=1))) db.commit() return acc.id def _seed_campaign_fixture(db, user_id, n_recipients=3): """RecipientList + recipients + subject + template + authorized account. Returns ids dict for building CampaignIn payloads.""" from app.models import (EmailTemplate, GmailAccount, OAuthToken, Recipient, RecipientList, SubjectLine) rl = RecipientList(user_id=user_id, name=f"camp-list-{uuid.uuid4().hex[:6]}") db.add(rl) db.flush() for i in range(1, n_recipients + 1): db.add(Recipient(list_id=rl.id, email=f"camp{i}@x.com", first_name=f"Camp{i}", last_name="User", unsub_token=secrets.token_urlsafe(24), is_valid=True)) subj = SubjectLine(user_id=user_id, subject="Your invoice %RandomInvoiceNumber% is ready") db.add(subj) tpl = EmailTemplate(user_id=user_id, name=f"tpl-{uuid.uuid4().hex[:6]}", subject_hint="Invoice", body_html="
Hi %FirstName%, pay %RandomTotal% now.
") db.add(tpl) acc = GmailAccount(user_id=user_id, email=f"camp-acc-{uuid.uuid4().hex[:6]}@gmail.com", status="authorized", sending_method="api") db.add(acc) db.flush() db.add(OAuthToken(account_id=acc.id, access_token="fake-access", refresh_token="fake-refresh")) db.commit() return {"list_id": rl.id, "subject_id": subj.id, "template_id": tpl.id, "account_id": acc.id} def _cIn(fixture, **over): base = { "name": "Test Campaign", "recipient_list_id": fixture["list_id"], "subject_set_ids": [fixture["subject_id"]], "template_ids": [fixture["template_id"]], "message_type": "html", "attach_mode": "none", "account_ids": [fixture["account_id"]], "rotation_strategy": "sequential", "subject_rotation": "round_robin", "template_sequence_mode": "repeat", "phone_selection": "round_robin", } base.update(over)