From 4b1d8bf5c387ba5fe6d219cc847a632c62d17520 Mon Sep 17 00:00:00 2001
From: Colin Maudry <colin@maudry.com>
Date: Thu, 25 Jun 2026 18:56:26 +0200
Subject: [PATCH] feat(abonnement): signature + mapping des webhooks Frisbii
 (#90)

---
 src/subscriptions/webhooks.py        | 48 ++++++++++++++++++++
 tests/subscriptions/test_webhooks.py | 67 ++++++++++++++++++++++++++++
 2 files changed, 115 insertions(+)
 create mode 100644 src/subscriptions/webhooks.py
 create mode 100644 tests/subscriptions/test_webhooks.py

diff --git a/src/subscriptions/webhooks.py b/src/subscriptions/webhooks.py
new file mode 100644
index 0000000..2a5c1ff
--- /dev/null
+++ b/src/subscriptions/webhooks.py
@@ -0,0 +1,48 @@
+import hashlib
+import hmac
+from datetime import datetime, timezone
+
+
+def verify_signature(payload: dict, secret: str) -> bool:
+    # Schéma Reepay/Frisbii : HMAC-SHA256(secret, timestamp + id), hex.
+    # À confirmer dans la doc webhooks Frisbii lors de l'intégration.
+    if not secret:
+        return False
+    timestamp = payload.get("timestamp", "")
+    event_id = payload.get("id", "")
+    received = payload.get("signature", "")
+    expected = hmac.new(
+        secret.encode(), (timestamp + event_id).encode(), hashlib.sha256
+    ).hexdigest()
+    return hmac.compare_digest(expected, received)
+
+
+def _in_future(value) -> bool:
+    if not value:
+        return False
+    try:
+        return datetime.fromisoformat(value.replace("Z", "+00:00")) > datetime.now(
+            timezone.utc
+        )
+    except ValueError:
+        return False
+
+
+def map_subscription(sub: dict) -> tuple[str, str | None]:
+    """Mappe un objet subscription Frisbii vers (status, current_period_end).
+
+    Noms de champs Reepay (à confirmer) : state, trial_end, expires,
+    is_cancelled, next_period_start.
+    """
+    state = sub.get("state")
+    if state == "expired":
+        return "expired", sub.get("expires")
+    if sub.get("is_cancelled") or state == "cancelled":
+        return "cancelled", sub.get("expires")
+    if _in_future(sub.get("trial_end")):
+        return "trial", sub.get("trial_end")
+    if state == "active":
+        return "active", sub.get("next_period_start")
+    if state == "pending":
+        return "pending", None
+    return (state or "pending"), sub.get("next_period_start")
diff --git a/tests/subscriptions/test_webhooks.py b/tests/subscriptions/test_webhooks.py
new file mode 100644
index 0000000..de0e6bb
--- /dev/null
+++ b/tests/subscriptions/test_webhooks.py
@@ -0,0 +1,67 @@
+import hashlib
+import hmac
+from datetime import datetime, timedelta, timezone
+
+from src.subscriptions import webhooks
+
+
+def _sign(secret, timestamp, event_id):
+    return hmac.new(
+        secret.encode(), (timestamp + event_id).encode(), hashlib.sha256
+    ).hexdigest()
+
+
+def test_verify_signature_accepts_valid():
+    payload = {"id": "evt_1", "timestamp": "2026-06-25T10:00:00Z"}
+    payload["signature"] = _sign("s3cr3t", payload["timestamp"], payload["id"])
+    assert webhooks.verify_signature(payload, "s3cr3t") is True
+
+
+def test_verify_signature_rejects_tampered():
+    payload = {
+        "id": "evt_1",
+        "timestamp": "2026-06-25T10:00:00Z",
+        "signature": "deadbeef",
+    }
+    assert webhooks.verify_signature(payload, "s3cr3t") is False
+
+
+def test_verify_signature_rejects_without_secret():
+    payload = {"id": "evt_1", "timestamp": "t", "signature": "x"}
+    assert webhooks.verify_signature(payload, "") is False
+
+
+def _future():
+    return (datetime.now(timezone.utc) + timedelta(days=2)).isoformat()
+
+
+def test_map_trial():
+    status, end = webhooks.map_subscription(
+        {"state": "active", "trial_end": _future(), "next_period_start": _future()}
+    )
+    assert status == "trial"
+
+
+def test_map_active():
+    nxt = _future()
+    status, end = webhooks.map_subscription(
+        {"state": "active", "next_period_start": nxt}
+    )
+    assert status == "active"
+    assert end == nxt
+
+
+def test_map_cancelled():
+    exp = _future()
+    status, end = webhooks.map_subscription(
+        {"state": "active", "is_cancelled": True, "expires": exp}
+    )
+    assert status == "cancelled"
+    assert end == exp
+
+
+def test_map_expired():
+    status, end = webhooks.map_subscription(
+        {"state": "expired", "expires": "2020-01-01T00:00:00Z"}
+    )
+    assert status == "expired"