File size: 5,954 Bytes
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 | 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):
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()
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_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 -> they already bought
return [{"created_at": datetime.now(UTC).isoformat()}]
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
|