From 272cb9bcc98e605d4954b36b317d5a7c6804c7a9 Mon Sep 17 00:00:00 2001 From: Vlad Stan Date: Tue, 12 May 2026 11:11:35 +0300 Subject: [PATCH] fix: plan --- lnbits/fiat/square.py | 152 +++++++++++++++++++++++++----- tests/unit/test_fiat_providers.py | 103 ++++++++++++++++++++ 2 files changed, 230 insertions(+), 25 deletions(-) diff --git a/lnbits/fiat/square.py b/lnbits/fiat/square.py index 5b69d2743..a3684d628 100644 --- a/lnbits/fiat/square.py +++ b/lnbits/fiat/square.py @@ -52,6 +52,11 @@ class SquareCreateInvoiceOptions(BaseModel): subscription: SquareSubscriptionOptions | None = None +class SquareSubscriptionCheckoutInfo(BaseModel): + plan_variation_id: str + price_money: dict[str, Any] + + class SquareWallet(FiatProvider): """https://developer.squareup.com/reference/square""" @@ -199,23 +204,20 @@ class SquareWallet(FiatProvider): payment_options.extra["subscription_request_id"] = ( payment_options.subscription_request_id ) - print("### create_subscription", subscription_id, quantity, payment_options) try: - price_money = await self._get_subscription_price_money(subscription_id) - print("### price_money", price_money) + checkout_info = await self._get_subscription_checkout_info(subscription_id) metadata = self._serialize_metadata(payment_options) - print("### metadata", metadata) payload = { "idempotency_key": payment_options.subscription_request_id, "description": metadata, "quick_pay": { "name": (payment_options.memo or "LNbits Subscription")[:255], - "price_money": price_money, + "price_money": checkout_info.price_money, "location_id": self.location_id, }, "checkout_options": { "redirect_url": success_url, - "subscription_plan_id": subscription_id, + "subscription_plan_id": checkout_info.plan_variation_id, }, "payment_note": metadata, } @@ -224,7 +226,6 @@ class SquareWallet(FiatProvider): ) r.raise_for_status() data = r.json() - print("### response data", data) payment_link = data.get("payment_link") or {} url = payment_link.get("url") if not url: @@ -338,33 +339,134 @@ class SquareWallet(FiatProvider): r.raise_for_status() return r.json().get("payment") or {} - async def _get_subscription_price_money( - self, plan_variation_id: str - ) -> dict[str, Any]: - r = await self.client.get(f"/v2/catalog/object/{plan_variation_id}") - r.raise_for_status() - catalog_object = r.json().get("object") or {} - if catalog_object.get("type") != "SUBSCRIPTION_PLAN_VARIATION": - raise ValueError("Square subscription ID must be a plan variation ID.") + async def _get_subscription_checkout_info( + self, subscription_plan_id: str + ) -> SquareSubscriptionCheckoutInfo: + catalog_object = await self._get_catalog_object(subscription_plan_id) + if catalog_object.get("type") == "SUBSCRIPTION_PLAN": + return await self._get_plan_checkout_info(catalog_object) - variation_data = catalog_object.get("subscription_plan_variation_data") or {} + if catalog_object.get("type") == "SUBSCRIPTION_PLAN_VARIATION": + price_money = await self._get_subscription_price_money( + catalog_object, + ) + plan_variation_id = catalog_object.get("id") + if not plan_variation_id: + raise ValueError("Square subscription plan variation is missing an ID.") + return SquareSubscriptionCheckoutInfo( + plan_variation_id=plan_variation_id, + price_money=price_money, + ) + + raise ValueError( + "Square subscription ID must be a plan ID or plan variation ID." + ) + + async def _get_plan_checkout_info( + self, catalog_object: dict[str, Any] + ) -> SquareSubscriptionCheckoutInfo: + plan_data = catalog_object.get("subscription_plan_data") or {} + plan_variations = plan_data.get("subscription_plan_variations") or [] + plan_variation = next( + ( + variation + for variation in plan_variations + if not variation.get("is_deleted") + ), + None, + ) + if not plan_variation: + raise ValueError("Square subscription plan is missing a variation.") + + price_money = await self._get_subscription_price_money( + plan_variation, + eligible_item_ids=plan_data.get("eligible_item_ids") or [], + ) + plan_variation_id = plan_variation.get("id") + if not plan_variation_id: + raise ValueError("Square subscription plan variation is missing an ID.") + + return SquareSubscriptionCheckoutInfo( + plan_variation_id=plan_variation_id, + price_money=price_money, + ) + + async def _get_catalog_object(self, object_id: str) -> dict[str, Any]: + r = await self.client.get(f"/v2/catalog/object/{object_id}") + r.raise_for_status() + return r.json().get("object") or {} + + async def _get_subscription_price_money( + self, + plan_variation: dict[str, Any], + eligible_item_ids: list[str] | None = None, + ) -> dict[str, Any]: + variation_data = plan_variation.get("subscription_plan_variation_data") or {} phases = variation_data.get("phases") or [] for phase in phases: pricing = phase.get("pricing") or {} price_money = pricing.get("price_money") or phase.get( "recurring_price_money" ) - if ( - price_money - and price_money.get("amount") is not None - and price_money.get("currency") - ): - return { - "amount": int(price_money["amount"]), - "currency": price_money["currency"].upper(), - } + parsed_price_money = self._parse_price_money(price_money) + if parsed_price_money: + return parsed_price_money + + if pricing.get("type") == "RELATIVE": + return await self._get_relative_subscription_price_money( + eligible_item_ids or [] + ) + raise ValueError("Square subscription plan variation is missing price_money.") + async def _get_relative_subscription_price_money( + self, eligible_item_ids: list[str] + ) -> dict[str, Any]: + if len(eligible_item_ids) != 1: + raise ValueError( + "Square relative subscription plan must have exactly one item." + ) + + item = await self._get_catalog_object(eligible_item_ids[0]) + item_variations: list[dict[str, Any]] = [] + if item.get("type") == "ITEM": + item_variations = (item.get("item_data") or {}).get("variations") or [] + elif item.get("type") == "ITEM_VARIATION": + item_variations = [item] + + item_variation = next( + ( + variation + for variation in item_variations + if not variation.get("is_deleted") + ), + None, + ) + if not item_variation: + raise ValueError("Square subscription item is missing a variation.") + + price_money = self._parse_price_money( + (item_variation.get("item_variation_data") or {}).get("price_money") + ) + if price_money: + return price_money + + raise ValueError("Square subscription item variation is missing price_money.") + + def _parse_price_money( + self, price_money: dict[str, Any] | None + ) -> dict[str, Any] | None: + if ( + price_money + and price_money.get("amount") is not None + and price_money.get("currency") + ): + return { + "amount": int(price_money["amount"]), + "currency": price_money["currency"].upper(), + } + return None + def _status_from_payment(self, payment: dict[str, Any]) -> FiatPaymentStatus: status = (payment.get("status") or "").upper() if status == "COMPLETED": diff --git a/tests/unit/test_fiat_providers.py b/tests/unit/test_fiat_providers.py index 9652f4151..6553159da 100644 --- a/tests/unit/test_fiat_providers.py +++ b/tests/unit/test_fiat_providers.py @@ -422,6 +422,7 @@ async def test_square_wallet_create_subscription(settings: Settings): json_data={ "object": { "type": "SUBSCRIPTION_PLAN_VARIATION", + "id": "PLAN_VARIATION_123", "subscription_plan_variation_data": { "phases": [ { @@ -481,6 +482,108 @@ async def test_square_wallet_create_subscription(settings: Settings): assert metadata[3:] == ["link-1", "Monthly Gold"] +@pytest.mark.anyio +async def test_square_wallet_create_subscription_from_plan_id(settings: Settings): + settings.square_api_endpoint = "https://connect.squareupsandbox.com" + settings.square_access_token = "square-token" + settings.square_location_id = "LOC123" + settings.square_api_version = "2026-01-22" + settings.square_payment_success_url = "https://lnbits.example/success" + + wallet = SquareWallet() + client = MockHTTPClient( + [ + MockHTTPResponse( + json_data={ + "object": { + "type": "SUBSCRIPTION_PLAN", + "id": "PLAN123", + "subscription_plan_data": { + "name": "LNbits Test Weekly Personal Plan", + "subscription_plan_variations": [ + { + "type": "SUBSCRIPTION_PLAN_VARIATION", + "id": "PLAN_VARIATION_123", + "subscription_plan_variation_data": { + "name": "LNbits Test Weekly Personal Plan", + "phases": [ + { + "uid": "PHASE123", + "cadence": "WEEKLY", + "ordinal": 0, + "pricing": {"type": "RELATIVE"}, + } + ], + "subscription_plan_id": "PLAN123", + }, + } + ], + "eligible_item_ids": ["ITEM123"], + "all_items": False, + }, + } + } + ), + MockHTTPResponse( + json_data={ + "object": { + "type": "ITEM", + "id": "ITEM123", + "item_data": { + "name": "LNbits Test Weekly Personal Plan", + "variations": [ + { + "type": "ITEM_VARIATION", + "id": "ITEM_VARIATION_123", + "item_variation_data": { + "item_id": "ITEM123", + "name": "Regular", + "pricing_type": "FIXED_PRICING", + "price_money": { + "amount": 1500, + "currency": "USD", + }, + }, + } + ], + }, + } + } + ), + MockHTTPResponse( + json_data={ + "payment_link": { + "id": "plink_123", + "url": "https://square.link/u/sub_123", + } + } + ), + ] + ) + wallet.client = client # type: ignore[assignment] + + response = await wallet.create_subscription( + "PLAN123", + 1, + FiatSubscriptionPaymentOptions( + wallet_id="wallet_1", + memo="Weekly Plan", + success_url="https://lnbits.example/success", + ), + ) + + assert response.ok is True + assert client.calls[0][0] == "/v2/catalog/object/PLAN123" + assert client.calls[1][0] == "/v2/catalog/object/ITEM123" + assert client.calls[2][0] == "/v2/online-checkout/payment-links" + payload = client.calls[2][1]["json"] + assert payload["quick_pay"]["price_money"] == {"amount": 1500, "currency": "USD"} + assert payload["checkout_options"] == { + "redirect_url": "https://lnbits.example/success", + "subscription_plan_id": "PLAN_VARIATION_123", + } + + @pytest.mark.anyio async def test_square_wallet_create_subscription_invoice(settings: Settings): settings.square_api_endpoint = "https://connect.squareupsandbox.com"