from __future__ import annotations import hashlib import hmac from app.wa import ( Deduper, cards_to_text, parse_incoming, session_id_for, to_whatsapp_text, verify_signature, ) def _sig(secret: str, body: bytes) -> str: return "sha256=" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest() def test_verify_signature_roundtrip(): body = b'{"hola": 1}' assert verify_signature("s3cret", body, _sig("s3cret", body)) is True assert verify_signature("s3cret", body, _sig("wrong", body)) is False assert verify_signature("s3cret", body, None) is False assert verify_signature("s3cret", body, "md5=abc") is False def test_to_whatsapp_text_converts_markdown(): md = "**Manguera plana** de PVC\n\n- [Ver carrito](https://x.com/cart/1:1)\n## Detalles" out = to_whatsapp_text(md) assert "*Manguera plana*" in out assert "**" not in out assert "Ver carrito: https://x.com/cart/1:1" in out assert "##" not in out def test_cards_to_text_caps_and_links(): cards = [ {"title": f"Prod {i}", "price": "9,99 €", "add_url": f"https://s/cart/{i}:1"} for i in range(5) ] out = cards_to_text(cards) assert out.count("•") == 3 # capped assert "*Prod 0*" in out and "https://s/cart/0:1" in out def test_session_id_deterministic_and_fits_column(): a = session_id_for(1, "34600111222") assert a == session_id_for(1, "34600111222") # stable per phone assert a != session_id_for(2, "34600111222") # tenant-scoped assert a != session_id_for(1, "34600999888") # phone-scoped assert len(a) <= 40 and a.startswith("wa") def test_deduper_remembers_and_evicts(): d = Deduper(maxlen=2) assert d.seen("a") is False assert d.seen("a") is True d.seen("b"), d.seen("c") # evicts "a" assert d.seen("a") is False def test_parse_incoming_extracts_text_and_skips_statuses(): payload = { "entry": [ { "changes": [ { "value": { "metadata": {"phone_number_id": "111"}, "messages": [ {"id": "wamid.1", "from": "34600111222", "type": "text", "text": {"body": "hola"}}, {"id": "wamid.2", "from": "34600111222", "type": "image"}, ], } }, {"value": {"metadata": {"phone_number_id": "111"}, "statuses": [{"id": "wamid.1", "status": "delivered"}]}}, ] } ] } msgs = parse_incoming(payload) assert len(msgs) == 1 assert msgs[0].wamid == "wamid.1" assert msgs[0].phone_number_id == "111" assert msgs[0].from_phone == "34600111222" assert msgs[0].text == "hola"