File size: 10,808 Bytes
2994a94 2322bed 2994a94 2322bed 2994a94 2322bed 2994a94 2322bed 2994a94 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 | from __future__ import annotations
from datetime import UTC, datetime, timedelta
from sqlalchemy import select
from app import carts, mailer
from app.models import AbandonedCart, ChatSession, Tenant
async def _session(db, tenant_id, lang="es"):
s = ChatSession(tenant_id=tenant_id, lang=lang)
db.add(s)
await db.flush()
return s
async def test_capture_one_per_session_and_updates(db_session):
s = await _session(db_session, 1)
c1 = await carts.capture(
db_session, 1, session_id=s.id, checkout_url="https://x/cart/1:1", items_text="1 art."
)
# same session -> SAME row updated (no duplicate), latest link wins
c2 = await carts.capture(
db_session, 1, session_id=s.id, checkout_url="https://x/cart/2:1", items_text="2 art."
)
assert c2.id == c1.id
assert c2.checkout_url.endswith("2:1") and c2.items_text == "2 art."
assert c2.email is None # nothing captured yet
async def test_capture_autofills_known_email(db_session):
s = await _session(db_session, 1)
c = await carts.capture(
db_session, 1, session_id=s.id, checkout_url="https://x/cart/9:1",
email="Buyer@X.com", name="Leo",
)
assert c.email == "buyer@x.com" and c.name == "Leo" # normalized
async def test_attach_contact_requires_existing_cart(db_session):
s = await _session(db_session, 1)
# no cart built yet in this session -> nothing to attach
assert await carts.attach_contact(db_session, 1, session_id=s.id, email="a@b.com") is None
await carts.capture(db_session, 1, session_id=s.id, checkout_url="https://x/cart/3:1")
c = await carts.attach_contact(db_session, 1, session_id=s.id, email="a@b.com", name="Ann")
assert c is not None and c.email == "a@b.com" and c.name == "Ann"
async def test_capture_is_tenant_isolated(db_session):
sa = await _session(db_session, 1)
sb = await _session(db_session, 2)
a = await carts.capture(db_session, 1, session_id=sa.id, checkout_url="https://a/cart/1:1")
b = await carts.capture(db_session, 2, session_id=sb.id, checkout_url="https://b/cart/1:1")
assert a.id != b.id and a.tenant_id == 1 and b.tenant_id == 2
def test_recovery_email_language():
es_subject, es_body = carts._recovery_email(
AbandonedCart(name="María García", lang="es", checkout_url="https://x/cart/1:1"), "MiTienda"
)
assert "MiTienda" in es_subject and "María" in es_body and "https://x/cart/1:1" in es_body
en_subject, en_body = carts._recovery_email(
AbandonedCart(name="John", lang="en", checkout_url="https://x/cart/1:1"), "MyShop"
)
assert "saved" in en_subject.lower() and en_body.startswith("Hi John")
async def test_recover_due_carts_sends_and_skips(db_session, monkeypatch):
import app.shopify.orders as orders_mod
import app.tenancy as tenancy_mod
db_session.add(Tenant(slug="cr", name="CR", brand_name="CartShop",
allow_cart_recovery=True, support_email="shop@cr.com"))
db_session.add(Tenant(slug="off", name="OFF", allow_cart_recovery=False))
await db_session.flush()
cr = (await db_session.execute(
select(Tenant).where(Tenant.slug == "cr"))).scalar_one()
off = (await db_session.execute(
select(Tenant).where(Tenant.slug == "off"))).scalar_one()
old = datetime.now(UTC) - timedelta(hours=5) # past the 4h delay, within 48h
# due: opted-in tenant, has email, old enough
db_session.add(AbandonedCart(tenant_id=cr.id, email="a@b.com", checkout_url="https://x/cart/1:1",
created_at=old))
# not due: no email captured
db_session.add(AbandonedCart(tenant_id=cr.id, checkout_url="https://x/cart/2:1", created_at=old))
# not due: too recent
db_session.add(AbandonedCart(tenant_id=cr.id, email="c@d.com", checkout_url="https://x/cart/3:1",
created_at=datetime.now(UTC)))
# opted-out tenant must NEVER be emailed
db_session.add(AbandonedCart(tenant_id=off.id, email="e@f.com", checkout_url="https://x/cart/4:1",
created_at=old))
await db_session.flush()
# opted-in tenants have Shopify connected; no orders => nobody has bought yet
monkeypatch.setattr(tenancy_mod, "tenant_shopify_client", lambda tenant: object())
async def no_orders(client, email):
return []
monkeypatch.setattr(orders_mod, "find_orders_by_email", no_orders)
sent: list[tuple] = []
async def fake_send(settings, to, subject, body, **kw):
sent.append((to, subject, kw.get("reply_to")))
return True
monkeypatch.setattr(mailer, "send_email", fake_send)
n = await carts.recover_due_carts(db_session, object(), now=datetime.now(UTC))
assert n == 1
assert sent == [("a@b.com", sent[0][1], "shop@cr.com")] # only the one due cart, with reply-to
async def test_recover_fails_closed_without_shopify_client(db_session, monkeypatch):
"""Fail closed: a tenant whose Shopify client can't be resolved (no creds) must NOT
be emailed — we can't check whether the shopper already bought."""
import app.tenancy as tenancy_mod
db_session.add(Tenant(slug="noshop", name="NS", allow_cart_recovery=True))
await db_session.flush()
t = (await db_session.execute(select(Tenant).where(Tenant.slug == "noshop"))).scalar_one()
db_session.add(AbandonedCart(tenant_id=t.id, email="a@b.com", checkout_url="https://x/cart/1:1",
created_at=datetime.now(UTC) - timedelta(hours=5)))
await db_session.flush()
monkeypatch.setattr(tenancy_mod, "tenant_shopify_client", lambda tenant: None)
sent = []
monkeypatch.setattr(mailer, "send_email", lambda *a, **k: sent.append(a))
n = await carts.recover_due_carts(db_session, object(), now=datetime.now(UTC))
assert n == 0 and sent == []
async def test_recover_dedupes_per_recipient(db_session, monkeypatch):
"""Many carts (sessions) with the SAME email get at most ONE recovery email."""
import app.shopify.orders as orders_mod
import app.tenancy as tenancy_mod
db_session.add(Tenant(slug="dd", name="DD", brand_name="DDShop", allow_cart_recovery=True))
await db_session.flush()
t = (await db_session.execute(select(Tenant).where(Tenant.slug == "dd"))).scalar_one()
old = datetime.now(UTC) - timedelta(hours=5)
for i in range(3): # three abandoned carts, same victim address
db_session.add(AbandonedCart(tenant_id=t.id, email="victim@x.com",
checkout_url=f"https://x/cart/{i}:1", created_at=old))
await db_session.flush()
monkeypatch.setattr(tenancy_mod, "tenant_shopify_client", lambda tenant: object())
async def no_orders(client, email):
return []
monkeypatch.setattr(orders_mod, "find_orders_by_email", no_orders)
sent = []
async def fake_send(settings, to, *a, **k):
sent.append(to)
return True
monkeypatch.setattr(mailer, "send_email", fake_send)
n = await carts.recover_due_carts(db_session, object(), now=datetime.now(UTC))
assert n == 1 and sent == ["victim@x.com"] # ONE email despite 3 carts
async def test_recover_only_stamps_nudged_on_success(db_session, monkeypatch):
"""A transient send failure leaves nudged_at NULL so the cart is retried, not
silently dropped."""
import app.shopify.orders as orders_mod
import app.tenancy as tenancy_mod
db_session.add(Tenant(slug="tx", name="TX", allow_cart_recovery=True))
await db_session.flush()
t = (await db_session.execute(select(Tenant).where(Tenant.slug == "tx"))).scalar_one()
cart = AbandonedCart(tenant_id=t.id, email="a@b.com", checkout_url="https://x/cart/1:1",
created_at=datetime.now(UTC) - timedelta(hours=5))
db_session.add(cart)
await db_session.flush()
monkeypatch.setattr(tenancy_mod, "tenant_shopify_client", lambda tenant: object())
async def no_orders(client, email):
return []
monkeypatch.setattr(orders_mod, "find_orders_by_email", no_orders)
async def fail_send(*a, **k):
return False # transient Brevo failure (no exception)
monkeypatch.setattr(mailer, "send_email", fail_send)
n = await carts.recover_due_carts(db_session, object(), now=datetime.now(UTC))
assert n == 0 and cart.nudged_at is None # not consumed; will retry
async def test_purge_stale_carts(db_session):
db_session.add(Tenant(slug="pg", name="PG", allow_cart_recovery=True))
await db_session.flush()
t = (await db_session.execute(select(Tenant).where(Tenant.slug == "pg"))).scalar_one()
db_session.add(AbandonedCart(tenant_id=t.id, email="old@x.com", checkout_url="u",
created_at=datetime.now(UTC) - timedelta(hours=60))) # past 48h
db_session.add(AbandonedCart(tenant_id=t.id, email="fresh@x.com", checkout_url="u",
created_at=datetime.now(UTC) - timedelta(hours=2)))
await db_session.flush()
purged = await carts.purge_stale_carts(db_session, now=datetime.now(UTC))
assert purged == 1
rows = (await db_session.execute(
select(AbandonedCart).where(AbandonedCart.tenant_id == t.id))).scalars().all()
assert [r.email for r in rows] == ["fresh@x.com"]
async def test_recover_due_skips_already_bought(db_session, monkeypatch):
import app.shopify.orders as orders_mod
import app.tenancy as tenancy_mod
db_session.add(Tenant(slug="cr2", name="CR2", brand_name="Shop2", allow_cart_recovery=True))
await db_session.flush()
t = (await db_session.execute(
select(Tenant).where(Tenant.slug == "cr2"))).scalar_one()
old = datetime.now(UTC) - timedelta(hours=5)
cart = AbandonedCart(tenant_id=t.id, email="paid@x.com", checkout_url="https://x/cart/1:1",
created_at=old)
db_session.add(cart)
await db_session.flush()
monkeypatch.setattr(tenancy_mod, "tenant_shopify_client", lambda tenant: object())
async def fake_orders(client, email):
# an order placed AFTER the cart was built, by THIS shopper -> already bought
return [{"created_at": datetime.now(UTC).isoformat(), "email": "paid@x.com"}]
monkeypatch.setattr(orders_mod, "find_orders_by_email", fake_orders)
sent = []
async def fake_send(*a, **k):
sent.append(a)
return True
monkeypatch.setattr(mailer, "send_email", fake_send)
n = await carts.recover_due_carts(db_session, object(), now=datetime.now(UTC))
assert n == 0 and sent == [] # no email — sale already closed
# same identity-mapped row recover_due_carts mutated (caller commits; don't refresh
# here or the un-flushed change is discarded)
assert cart.recovered is True and cart.nudged_at is None
|