diff --git a/src/subscriptions/__init__.py b/src/subscriptions/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/subscriptions/client.py b/src/subscriptions/client.py new file mode 100644 index 0000000..4ce2ad2 --- /dev/null +++ b/src/subscriptions/client.py @@ -0,0 +1,80 @@ +import os + +import httpx + +# NB : base URL et schémas d'endpoints à confirmer dans la doc Frisbii lors de la +# première intégration en environnement de test. Frisbii Billing/Pay s'appuie sur +# l'API Reepay (api.reepay.com) ; ajuster FRISBII_API_BASE_URL si besoin. +_DEFAULT_BASE_URL = "https://api.reepay.com" +_TIMEOUT = 15.0 + + +class FrisbiiError(Exception): + def __init__(self, status_code: int, body): + super().__init__(f"Frisbii API error {status_code}: {body}") + self.status_code = status_code + self.body = body + + +def _api_key() -> str: + return os.getenv("FRISBII_API_KEY", "") + + +def _base_url() -> str: + return os.getenv("FRISBII_API_BASE_URL") or _DEFAULT_BASE_URL + + +def _call(method: str, path: str, json: dict | None = None) -> dict: + resp = httpx.request( + method, + f"{_base_url()}{path}", + auth=(_api_key(), ""), + json=json, + timeout=_TIMEOUT, + ) + if resp.status_code >= 400: + raise FrisbiiError(resp.status_code, resp.text) + return resp.json() + + +def get_or_create_customer(handle: str, email: str) -> dict: + try: + return _call("GET", f"/v1/customer/{handle}") + except FrisbiiError as exc: + if exc.status_code != 404: + raise + return _call("POST", "/v1/customer", json={"handle": handle, "email": email}) + + +def create_subscription_session( + plan_handle: str, + customer_handle: str, + accept_url: str, + cancel_url: str, + no_trial: bool = False, +) -> str: + prepare = {"plan": plan_handle, "customer": customer_handle} + if no_trial: + prepare["no_trial"] = True + data = _call( + "POST", + "/v1/session/subscription", + json={ + "accept_url": accept_url, + "cancel_url": cancel_url, + "prepare_subscription": prepare, + }, + ) + return data["url"] + + +def cancel_subscription(subscription_handle: str) -> dict: + return _call("POST", f"/v1/subscription/{subscription_handle}/cancel") + + +def get_subscription(subscription_handle: str) -> dict: + return _call("GET", f"/v1/subscription/{subscription_handle}") + + +def get_plan(plan_handle: str) -> dict: + return _call("GET", f"/v1/plan/{plan_handle}") diff --git a/tests/subscriptions/__init__.py b/tests/subscriptions/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/subscriptions/conftest.py b/tests/subscriptions/conftest.py new file mode 100644 index 0000000..a3d4ad1 --- /dev/null +++ b/tests/subscriptions/conftest.py @@ -0,0 +1,36 @@ +import json as _json + +import pytest + + +class FakeResponse: + def __init__(self, status_code: int, payload=None): + self.status_code = status_code + self._payload = payload if payload is not None else {} + + def json(self): + return self._payload + + @property + def text(self): + return _json.dumps(self._payload) + + +@pytest.fixture +def fake_httpx(monkeypatch): + """Capture les appels httpx.request du client Frisbii et renvoie des réponses programmées.""" + from src.subscriptions import client + + calls = [] + queue = [] + + def _fake_request(method, url, **kwargs): + calls.append({"method": method, "url": url, **kwargs}) + if queue: + return queue.pop(0) + return FakeResponse(200, {}) + + monkeypatch.setenv("FRISBII_API_KEY", "priv_test") + monkeypatch.setenv("FRISBII_API_BASE_URL", "https://api.test") + monkeypatch.setattr(client.httpx, "request", _fake_request) + return {"calls": calls, "queue": queue, "Response": FakeResponse} diff --git a/tests/subscriptions/test_client.py b/tests/subscriptions/test_client.py new file mode 100644 index 0000000..18e52c2 --- /dev/null +++ b/tests/subscriptions/test_client.py @@ -0,0 +1,61 @@ +import pytest + +from src.subscriptions import client + + +def test_auth_and_base_url_used(fake_httpx): + fake_httpx["queue"].append(fake_httpx["Response"](200, {"handle": "decpinfo-1"})) + client.get_or_create_customer("decpinfo-1", "a@b.fr") + call = fake_httpx["calls"][0] + assert call["url"] == "https://api.test/v1/customer/decpinfo-1" + assert call["auth"] == ("priv_test", "") + + +def test_get_or_create_customer_existing(fake_httpx): + fake_httpx["queue"].append(fake_httpx["Response"](200, {"handle": "decpinfo-1"})) + result = client.get_or_create_customer("decpinfo-1", "a@b.fr") + assert result == {"handle": "decpinfo-1"} + assert len(fake_httpx["calls"]) == 1 # pas de POST de création + + +def test_get_or_create_customer_creates_on_404(fake_httpx): + fake_httpx["queue"].append(fake_httpx["Response"](404, {"error": "not found"})) + fake_httpx["queue"].append(fake_httpx["Response"](200, {"handle": "decpinfo-1"})) + result = client.get_or_create_customer("decpinfo-1", "a@b.fr") + assert result == {"handle": "decpinfo-1"} + assert fake_httpx["calls"][1]["method"] == "POST" + assert fake_httpx["calls"][1]["json"] == {"handle": "decpinfo-1", "email": "a@b.fr"} + + +def test_create_subscription_session_returns_url(fake_httpx): + fake_httpx["queue"].append( + fake_httpx["Response"](200, {"id": "cs_1", "url": "https://pay.test/cs_1"}) + ) + url = client.create_subscription_session( + "plan_simple", "decpinfo-1", "https://app/ok", "https://app/ko" + ) + assert url == "https://pay.test/cs_1" + body = fake_httpx["calls"][0]["json"] + assert body["prepare_subscription"] == { + "plan": "plan_simple", + "customer": "decpinfo-1", + } + assert body["accept_url"] == "https://app/ok" + assert body["cancel_url"] == "https://app/ko" + + +def test_create_subscription_session_no_trial(fake_httpx): + fake_httpx["queue"].append( + fake_httpx["Response"](200, {"url": "https://pay.test/cs_2"}) + ) + client.create_subscription_session( + "plan_simple", "decpinfo-1", "https://app/ok", "https://app/ko", no_trial=True + ) + assert fake_httpx["calls"][0]["json"]["prepare_subscription"]["no_trial"] is True + + +def test_http_error_raises_frisbii_error(fake_httpx): + fake_httpx["queue"].append(fake_httpx["Response"](500, {"error": "boom"})) + with pytest.raises(client.FrisbiiError) as exc: + client.get_plan("plan_simple") + assert exc.value.status_code == 500