Colin Maudry
127 lines
4.3 KiB
Python
127 lines
4.3 KiB
Python
import hashlib
|
|
import hmac
|
|
|
|
from src.subscriptions import client as frisbii_client
|
|
from src.subscriptions import db
|
|
|
|
|
|
def _sign(secret, timestamp, event_id):
|
|
return hmac.new(
|
|
secret.encode(), (timestamp + event_id).encode(), hashlib.sha256
|
|
).hexdigest()
|
|
|
|
|
|
def test_subscribe_redirects_to_hosted_url(logged_in_client, monkeypatch):
|
|
client, uid = logged_in_client
|
|
monkeypatch.setattr(
|
|
frisbii_client, "get_or_create_customer", lambda h, e: {"handle": h}
|
|
)
|
|
monkeypatch.setattr(
|
|
frisbii_client,
|
|
"create_subscription_session",
|
|
lambda plan, cust, ok, ko, no_trial=False: "https://pay.test/cs_1",
|
|
)
|
|
resp = client.post("/subscriptions/subscribe", data={"plan": "simple"})
|
|
assert resp.status_code == 303
|
|
assert resp.headers["Location"] == "https://pay.test/cs_1"
|
|
assert db.get_by_user(uid)["status"] == "pending"
|
|
|
|
|
|
def test_subscribe_disables_trial_after_first_use(logged_in_client, monkeypatch):
|
|
client, uid = logged_in_client
|
|
# L'utilisateur a déjà consommé un essai par le passé.
|
|
db.create_pending(uid, "decpinfo-%d" % uid, "simple")
|
|
db.update_from_webhook(
|
|
"decpinfo-%d" % uid, "sub_42", "trial", "2099-01-01T00:00:00+00:00"
|
|
)
|
|
captured = {}
|
|
monkeypatch.setattr(
|
|
frisbii_client, "get_or_create_customer", lambda h, e: {"handle": h}
|
|
)
|
|
|
|
def fake_session(plan, cust, ok, ko, no_trial=False):
|
|
captured["no_trial"] = no_trial
|
|
return "https://pay.test/cs_9"
|
|
|
|
monkeypatch.setattr(frisbii_client, "create_subscription_session", fake_session)
|
|
resp = client.post("/subscriptions/subscribe", data={"plan": "soutien"})
|
|
assert resp.status_code == 303
|
|
assert captured["no_trial"] is True
|
|
|
|
|
|
def test_subscribe_unknown_plan(logged_in_client):
|
|
client, _ = logged_in_client
|
|
resp = client.post("/subscriptions/subscribe", data={"plan": "bidon"})
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_subscribe_api_error_redirects_with_error(logged_in_client, monkeypatch):
|
|
client, _ = logged_in_client
|
|
|
|
def boom(h, e):
|
|
raise frisbii_client.FrisbiiError(500, "boom")
|
|
|
|
monkeypatch.setattr(frisbii_client, "get_or_create_customer", boom)
|
|
resp = client.post("/subscriptions/subscribe", data={"plan": "simple"})
|
|
assert resp.status_code == 302
|
|
assert "error=frisbii" in resp.headers["Location"]
|
|
|
|
|
|
def test_subscribe_requires_login(sub_app):
|
|
resp = sub_app.test_client().post(
|
|
"/subscriptions/subscribe", data={"plan": "simple"}
|
|
)
|
|
assert resp.status_code in (302, 401)
|
|
|
|
|
|
def test_cancel_calls_api_and_marks_cancelled(logged_in_client, monkeypatch):
|
|
client, uid = logged_in_client
|
|
db.create_pending(uid, "decpinfo-%d" % uid, "simple")
|
|
db.update_from_webhook(
|
|
"decpinfo-%d" % uid, "sub_42", "active", "2099-01-01T00:00:00+00:00"
|
|
)
|
|
monkeypatch.setattr(
|
|
frisbii_client,
|
|
"cancel_subscription",
|
|
lambda handle: {"expires": "2099-02-01T00:00:00+00:00"},
|
|
)
|
|
resp = client.post("/subscriptions/cancel")
|
|
assert resp.status_code == 302
|
|
assert "resiliation=ok" in resp.headers["Location"]
|
|
row = db.get_by_user(uid)
|
|
assert row["status"] == "cancelled"
|
|
assert row["current_period_end"] == "2099-02-01T00:00:00+00:00"
|
|
|
|
|
|
def test_webhook_invalid_signature(sub_app):
|
|
resp = sub_app.test_client().post(
|
|
"/frisbii/webhook", json={"id": "e", "signature": "x"}
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_webhook_updates_subscription(sub_app, monkeypatch):
|
|
from src.auth import db as auth_db
|
|
|
|
auth_db.init_schema()
|
|
uid = auth_db.create_user("wh@ex.fr", "hash")
|
|
db.create_pending(uid, "decpinfo-%d" % uid, "simple")
|
|
monkeypatch.setattr(
|
|
frisbii_client,
|
|
"get_subscription",
|
|
lambda h: {"state": "active", "next_period_start": "2099-01-01T00:00:00+00:00"},
|
|
)
|
|
payload = {
|
|
"id": "evt_1",
|
|
"timestamp": "2026-06-25T10:00:00Z",
|
|
"event_type": "subscription_created",
|
|
"customer": "decpinfo-%d" % uid,
|
|
"subscription": "sub_42",
|
|
}
|
|
payload["signature"] = _sign("s3cr3t", payload["timestamp"], payload["id"])
|
|
resp = sub_app.test_client().post("/frisbii/webhook", json=payload)
|
|
assert resp.status_code == 200
|
|
row = db.get_by_user(uid)
|
|
assert row["status"] == "active"
|
|
assert row["frisbii_subscription_handle"] == "sub_42"
|