diff --git a/backend/app/commands/sync_subscriptions.py b/backend/app/commands/sync_subscriptions.py index 6fd7aff..e87b7ba 100644 --- a/backend/app/commands/sync_subscriptions.py +++ b/backend/app/commands/sync_subscriptions.py @@ -19,7 +19,7 @@ EXPIRED_SUBSCRIPTION_STATUS, TERMINATED_SUBSCRIPTION_STATUS, ) -from app.events.subscriptions.utils import get_datasource_id +from app.events.subscriptions.utils import get_datasource_id, get_live_subscription from app.exceptions import InvalidDataSourceIdError from app.notifications import ColumnHeader, NotificationDetails, send_error, send_info from app.telemetry import capture_telemetry_cli_command @@ -101,6 +101,7 @@ def build_query(products: list[str], status: list[str], since: datetime, until: async def sync_subscription( + client: MPTClient, entitlement_handler: EntitlementHandler, account: Account, subscription: dict[str, Any], @@ -139,12 +140,29 @@ async def sync_subscription( message=f"Created new entitlement {created.id}.", ) else: - if entitlements: + bound = [ + entitlement + for entitlement in entitlements + if entitlement.affiliate_external_id == sub_id + ] + if bound: actions = [] - for entitlement in entitlements: - if entitlement.affiliate_external_id != sub_id: - continue - if entitlement.status == EntitlementStatus.NEW: + # Another live subscription for the same datasource takes the entitlement over. + replacement = await get_live_subscription(client, account, bound[0].datasource_id) + for entitlement in bound: + if replacement is not None: + await entitlement_handler.update( + entitlement, data={"affiliate_external_id": replacement["id"]} + ) + actions.append( + f"The entitlement {entitlement.id} was moved to the subscription " + f"{replacement['id']}." + ) + logger.info( + f"Moved entitlement {entitlement.id} from subscription {sub_id} " + f"to the subscription {replacement['id']}." + ) + elif entitlement.status == EntitlementStatus.NEW: await entitlement_handler.delete(entitlement) actions.append(f"The entitlement {entitlement.id} was deleted.") logger.info(f"Deleted entitlement {entitlement.id} for subscription {sub_id}.") @@ -154,12 +172,11 @@ async def sync_subscription( logger.info( f"Terminated entitlement {entitlement.id} for subscription {sub_id}." ) - if actions: - return SyncResult( - subscription_id=sub_id, - account_id=account.id, - message="\n".join(actions), - ) + return SyncResult( + subscription_id=sub_id, + account_id=account.id, + message="\n".join(actions), + ) return None @@ -240,6 +257,7 @@ async def sync_page( ], ) result = await sync_subscription( + client, entitlement_handler, account, subscription, diff --git a/backend/app/events/subscriptions/processing.py b/backend/app/events/subscriptions/processing.py index bd32360..0c91cd2 100644 --- a/backend/app/events/subscriptions/processing.py +++ b/backend/app/events/subscriptions/processing.py @@ -17,7 +17,11 @@ SubscriptionNotFoundError, SubscriptionNotValidError, ) -from app.events.subscriptions.utils import get_datasource_id, is_product_supported +from app.events.subscriptions.utils import ( + get_datasource_id, + get_live_subscription, + is_product_supported, +) from app.exceptions import InvalidDataSourceIdError from app.utils import validate_datasource_id_format @@ -31,10 +35,12 @@ class SubscriptionProcessor(EventProcessor): def __init__( self, + installation_client: MPTClient, entitlement_repo: EntitlementRepository, account: Account, subscription: dict[str, Any], ): + self.installation_client = installation_client self.entitlement_repo = entitlement_repo self.account = account self.subscription = subscription @@ -105,10 +111,27 @@ async def handle_active(self, entitlement: Entitlement | None) -> ProcessingResu return self.completed(f"The entitlement {entitlement.id} was created.") async def handle_inactive(self, entitlement: Entitlement | None) -> ProcessingResult: - """Drop an entitlement that was never redeemed, terminate the one in use.""" + """ + Drop an entitlement that was never redeemed, terminate the one in use. + + When another live subscription covers the same datasource, the entitlement is kept + and moved to it instead. + """ if entitlement is None: return self.completed(f"No entitlement is bound to the subscription {self.object_id}.") + replacement = await get_live_subscription( + self.installation_client, self.account, entitlement.datasource_id + ) + if replacement is not None: + await self.entitlement_repo.update( + entitlement, data={"affiliate_external_id": replacement["id"]} + ) + return self.completed( + f"The entitlement {entitlement.id} was moved to the subscription " + f"{replacement['id']}." + ) + if entitlement.status == EntitlementStatus.NEW: await self.entitlement_repo.delete(entitlement) return self.completed(f"The entitlement {entitlement.id} was deleted.") @@ -150,6 +173,7 @@ async def _get_processor(self, object_id: str) -> SubscriptionProcessor: if not subscription: raise SubscriptionNotFoundError(f"No subscription found for {object_id}.") return SubscriptionProcessor( + installation_client=self.installation_client, entitlement_repo=self.entitlement_repo, account=self.account, subscription=subscription, diff --git a/backend/app/events/subscriptions/utils.py b/backend/app/events/subscriptions/utils.py index 2c595d7..39357d3 100644 --- a/backend/app/events/subscriptions/utils.py +++ b/backend/app/events/subscriptions/utils.py @@ -1,7 +1,12 @@ import logging from typing import Any +from app.api_clients.mpt import MPTClient from app.db.models import Account +from app.events.subscriptions.constants import ( + EXPIRED_SUBSCRIPTION_STATUS, + TERMINATED_SUBSCRIPTION_STATUS, +) logger = logging.getLogger(__name__) @@ -19,3 +24,29 @@ def is_product_supported(subscription: dict[str, Any], account: Account) -> bool if product.strip() } return subscription["product"]["id"].lower() in products + + +async def get_live_subscription( + client: MPTClient, + account: Account, + datasource_id: str, +) -> dict[str, Any] | None: + """ + Return a marketplace subscription for the datasource that has not ended, if any. + + A subscription can be replaced by a new one for the same datasource (e.g. split billing or + a change of client, buyer or licensee): the entitlement must follow the live one instead + of being terminated with the old one. + """ + query = ( + "and(" + f"eq(externalIds.vendor,{datasource_id})," + f"out(status,({TERMINATED_SUBSCRIPTION_STATUS},{EXPIRED_SUBSCRIPTION_STATUS}))," + f"in(product.id,({account.products}))" + ")" + ) + return await client.get_first( + "commerce/subscriptions", + query=query, + select=["id", "name", "status", "externalIds"], + ) diff --git a/backend/tests/commands/test_sync_subscriptions.py b/backend/tests/commands/test_sync_subscriptions.py index 644cae9..920a9dc 100644 --- a/backend/tests/commands/test_sync_subscriptions.py +++ b/backend/tests/commands/test_sync_subscriptions.py @@ -35,10 +35,16 @@ TERMINATED_SUBSCRIPTION_STATUS, ) from app.notifications import ColumnHeader -from tests.types import ModelFactory, MPTSubscriptionFactory, SubscriptionsPageMocker +from tests.types import ( + ModelFactory, + MPTSubscriptionFactory, + SubscriptionMocker, + SubscriptionsPageMocker, +) SUBSCRIPTION_ID = "SUB-1234-5678" DATASOURCE_ID = "ds0001234" +REPLACEMENT_ID = "SUB-8765-4321" SINCE = datetime(2026, 9, 12, 9, 0, 0, tzinfo=UTC) UNTIL = datetime(2026, 9, 13, 9, 0, 0, tzinfo=UTC) WINDOW = "changed between 2026-09-12 09:00:00 UTC and 2026-09-13 09:00:00 UTC" @@ -112,6 +118,7 @@ def test_build_query_filters_by_product_status_and_window( async def test_sync_subscription_creates_missing_entitlement( + mpt_client: MPTClient, entitlement_handler: EntitlementHandler, subscription_account: Account, mpt_subscription_factory: MPTSubscriptionFactory, @@ -119,7 +126,9 @@ async def test_sync_subscription_creates_missing_entitlement( ) -> None: subscription = mpt_subscription_factory() - result = await sync_subscription(entitlement_handler, subscription_account, subscription, []) + result = await sync_subscription( + mpt_client, entitlement_handler, subscription_account, subscription, [] + ) entitlement = (await get_entitlements(db_session, subscription_account))[0] assert entitlement.status == EntitlementStatus.NEW @@ -135,6 +144,7 @@ async def test_sync_subscription_creates_missing_entitlement( async def test_sync_subscription_skips_an_active_subscription_with_active_entitlement( + mpt_client: MPTClient, entitlement_handler: EntitlementHandler, subscription_account: Account, mpt_subscription_factory: MPTSubscriptionFactory, @@ -148,13 +158,14 @@ async def test_sync_subscription_skips_an_active_subscription_with_active_entitl subscription = mpt_subscription_factory() result = await sync_subscription( - entitlement_handler, subscription_account, subscription, [entitlement] + mpt_client, entitlement_handler, subscription_account, subscription, [entitlement] ) assert result is None async def test_sync_subscription_reports_more_than_one_live_entitlement( + mpt_client: MPTClient, entitlement_handler: EntitlementHandler, subscription_account: Account, mpt_subscription_factory: MPTSubscriptionFactory, @@ -170,7 +181,7 @@ async def test_sync_subscription_reports_more_than_one_live_entitlement( subscription = mpt_subscription_factory() result = await sync_subscription( - entitlement_handler, subscription_account, subscription, [first, second] + mpt_client, entitlement_handler, subscription_account, subscription, [first, second] ) assert result is not None @@ -187,6 +198,7 @@ async def test_sync_subscription_reports_more_than_one_live_entitlement( [TERMINATED_SUBSCRIPTION_STATUS, EXPIRED_SUBSCRIPTION_STATUS], ) async def test_sync_subscription_ignore_non_active_subscription_without_entitlements( + mpt_client: MPTClient, entitlement_handler: EntitlementHandler, subscription_account: Account, mpt_subscription_factory: MPTSubscriptionFactory, @@ -194,7 +206,9 @@ async def test_sync_subscription_ignore_non_active_subscription_without_entitlem ) -> None: subscription = mpt_subscription_factory(status=subscription_status) - result = await sync_subscription(entitlement_handler, subscription_account, subscription, []) + result = await sync_subscription( + mpt_client, entitlement_handler, subscription_account, subscription, [] + ) assert result is None @@ -204,9 +218,11 @@ async def test_sync_subscription_ignore_non_active_subscription_without_entitlem [TERMINATED_SUBSCRIPTION_STATUS, EXPIRED_SUBSCRIPTION_STATUS], ) async def test_sync_subscription_actions_for_non_active_subscription( + mpt_client: MPTClient, entitlement_handler: EntitlementHandler, subscription_account: Account, mpt_subscription_factory: MPTSubscriptionFactory, + mock_live_subscription_lookup: SubscriptionMocker, entitlement_factory: ModelFactory[Entitlement], db_session: AsyncSession, subscription_status: str, @@ -225,8 +241,10 @@ async def test_sync_subscription_actions_for_non_active_subscription( ) subscription = mpt_subscription_factory(status=subscription_status) + mock_live_subscription_lookup([]) + result = await sync_subscription( - entitlement_handler, subscription_account, subscription, [new_ent, active_ent] + mpt_client, entitlement_handler, subscription_account, subscription, [new_ent, active_ent] ) await db_session.refresh(new_ent) @@ -247,6 +265,7 @@ async def test_sync_subscription_actions_for_non_active_subscription( [TERMINATED_SUBSCRIPTION_STATUS, EXPIRED_SUBSCRIPTION_STATUS], ) async def test_sync_subscription_ignores_entitlements_of_other_subscriptions( + mpt_client: MPTClient, entitlement_handler: EntitlementHandler, subscription_account: Account, mpt_subscription_factory: MPTSubscriptionFactory, @@ -270,7 +289,7 @@ async def test_sync_subscription_ignores_entitlements_of_other_subscriptions( subscription = mpt_subscription_factory(status=subscription_status) result = await sync_subscription( - entitlement_handler, subscription_account, subscription, [new_ent, active_ent] + mpt_client, entitlement_handler, subscription_account, subscription, [new_ent, active_ent] ) assert result is None @@ -282,9 +301,11 @@ async def test_sync_subscription_ignores_entitlements_of_other_subscriptions( async def test_sync_subscription_only_acts_on_entitlements_of_the_subscription( + mpt_client: MPTClient, entitlement_handler: EntitlementHandler, subscription_account: Account, mpt_subscription_factory: MPTSubscriptionFactory, + mock_live_subscription_lookup: SubscriptionMocker, entitlement_factory: ModelFactory[Entitlement], db_session: AsyncSession, ) -> None: @@ -303,8 +324,10 @@ async def test_sync_subscription_only_acts_on_entitlements_of_the_subscription( ) subscription = mpt_subscription_factory(status=TERMINATED_SUBSCRIPTION_STATUS) + mock_live_subscription_lookup([]) + result = await sync_subscription( - entitlement_handler, subscription_account, subscription, [own_ent, other_ent] + mpt_client, entitlement_handler, subscription_account, subscription, [own_ent, other_ent] ) await db_session.refresh(own_ent) @@ -315,6 +338,56 @@ async def test_sync_subscription_only_acts_on_entitlements_of_the_subscription( assert result.message == f"The entitlement {own_ent.id} was terminated." +@pytest.mark.parametrize( + "subscription_status", + [TERMINATED_SUBSCRIPTION_STATUS, EXPIRED_SUBSCRIPTION_STATUS], +) +async def test_sync_subscription_moves_entitlements_to_the_live_replacement( + mpt_client: MPTClient, + entitlement_handler: EntitlementHandler, + subscription_account: Account, + mpt_subscription_factory: MPTSubscriptionFactory, + mock_live_subscription_lookup: SubscriptionMocker, + entitlement_factory: ModelFactory[Entitlement], + db_session: AsyncSession, + subscription_status: str, +) -> None: + """A live subscription for the same datasource keeps the entitlements and takes them over.""" + new_ent = await entitlement_factory( + owner=subscription_account, + datasource_id=DATASOURCE_ID, + affiliate_external_id=SUBSCRIPTION_ID, + status=EntitlementStatus.NEW, + ) + active_ent = await entitlement_factory( + owner=subscription_account, + datasource_id=DATASOURCE_ID, + affiliate_external_id=SUBSCRIPTION_ID, + status=EntitlementStatus.ACTIVE, + ) + subscription = mpt_subscription_factory(status=subscription_status) + mock_live_subscription_lookup([mpt_subscription_factory(subscription_id=REPLACEMENT_ID)]) + + result = await sync_subscription( + mpt_client, entitlement_handler, subscription_account, subscription, [new_ent, active_ent] + ) + + await db_session.refresh(new_ent) + assert new_ent.status == EntitlementStatus.NEW + assert new_ent.affiliate_external_id == REPLACEMENT_ID + await db_session.refresh(active_ent) + assert active_ent.status == EntitlementStatus.ACTIVE + assert active_ent.affiliate_external_id == REPLACEMENT_ID + assert active_ent.terminated_at is None + + assert result is not None + assert result.succeeded is True + assert result.message == ( + f"The entitlement {new_ent.id} was moved to the subscription {REPLACEMENT_ID}.\n" + f"The entitlement {active_ent.id} was moved to the subscription {REPLACEMENT_ID}." + ) + + async def test_sync_page( mpt_client: MPTClient, mock_subscriptions_api: SubscriptionsPageMocker, diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index a1aac9a..cbcaab8 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -91,6 +91,7 @@ MPTAccountMocker, MPTSubscriptionFactory, OrderFactory, + SubscriptionMocker, ) @@ -2773,6 +2774,36 @@ def _subscription( return _subscription +@pytest.fixture +def mock_live_subscription_lookup( + httpx_mock: HTTPXMock, test_settings: Settings +) -> SubscriptionMocker: + """Make the marketplace answer the live-subscription lookup for a datasource.""" + + def _mock( + subscriptions: list[dict[str, Any]], + datasource_id: str = "ds0001234", + products: str = "PRD-1111-1111,PRD-2222-2222", + ) -> None: + query = ( + f"and(eq(externalIds.vendor,{datasource_id}),out(status,(Terminated,Expired))," + f"in(product.id,({products})))" + ) + httpx_mock.add_response( + method="GET", + url=( + f"{test_settings.mpt_api_base_url}/commerce/subscriptions?{query}" + "&select=id,name,status,externalIds&limit=1&offset=0" + ), + json={ + "data": subscriptions, + "$meta": {"pagination": {"total": len(subscriptions), "limit": 1, "offset": 0}}, + }, + ) + + return _mock + + @pytest.fixture async def subscription_account(account_factory: ModelFactory[Account]) -> Account: """An affiliate account entitled to the product the subscription fixtures use.""" diff --git a/backend/tests/events/test_subscription.py b/backend/tests/events/test_subscription.py index 7ca2fe5..1d8ca57 100644 --- a/backend/tests/events/test_subscription.py +++ b/backend/tests/events/test_subscription.py @@ -9,6 +9,7 @@ from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession +from app.api_clients.mpt import MPTClient from app.conf import Settings from app.db.handlers import EntitlementHandler from app.db.models import Account, CloudProvider, Entitlement @@ -28,6 +29,7 @@ SUBSCRIPTION_ID = "SUB-1234-5678" DATASOURCE_ID = "ds0001234" TASK_ID = "TSK-0014-5578-6577-4688" +REPLACEMENT_ID = "SUB-8765-4321" @pytest.fixture @@ -141,6 +143,7 @@ async def test_claim_task_starts_the_task_without_an_ownership_check( async def test_process_completes_for_a_non_affiliate_account( + mpt_client: MPTClient, mpt_subscription_factory: MPTSubscriptionFactory, account_factory: ModelFactory[Account], db_session: AsyncSession, @@ -152,6 +155,7 @@ async def test_process_completes_for_a_non_affiliate_account( products="PRD-1111-1111", ) processor = SubscriptionProcessor( + installation_client=mpt_client, entitlement_repo=EntitlementHandler(db_session), account=account, subscription=mpt_subscription_factory(), @@ -318,6 +322,7 @@ async def test_inactive_subscription_deletes_not_redeemed_entitlement( subscription_event_handler: SubscriptionEventHandler, mpt_subscription_factory: MPTSubscriptionFactory, mock_get_subscription: SubscriptionMocker, + mock_live_subscription_lookup: SubscriptionMocker, entitlement_factory: ModelFactory[Entitlement], subscription_account: Account, db_session: AsyncSession, @@ -328,6 +333,7 @@ async def test_inactive_subscription_deletes_not_redeemed_entitlement( owner=subscription_account, datasource_id=DATASOURCE_ID, status=EntitlementStatus.NEW ) mock_get_subscription(mpt_subscription_factory(status=subscription_status)) + mock_live_subscription_lookup([]) processor = await subscription_event_handler.get_processor(object_id=SUBSCRIPTION_ID) result = await processor.process() @@ -348,6 +354,7 @@ async def test_inactive_subscription_terminates_an_active_entitlement( subscription_event_handler: SubscriptionEventHandler, mpt_subscription_factory: MPTSubscriptionFactory, mock_get_subscription: SubscriptionMocker, + mock_live_subscription_lookup: SubscriptionMocker, entitlement_factory: ModelFactory[Entitlement], subscription_account: Account, db_session: AsyncSession, @@ -358,6 +365,7 @@ async def test_inactive_subscription_terminates_an_active_entitlement( owner=subscription_account, datasource_id=DATASOURCE_ID, status=EntitlementStatus.ACTIVE ) mock_get_subscription(mpt_subscription_factory(status=subscription_status)) + mock_live_subscription_lookup([]) processor = await subscription_event_handler.get_processor(object_id=SUBSCRIPTION_ID) result = await processor.process() @@ -369,6 +377,47 @@ async def test_inactive_subscription_terminates_an_active_entitlement( assert result.message == f"The entitlement {redeemed.id} was terminated." +@pytest.mark.parametrize( + "subscription_status", + [TERMINATED_SUBSCRIPTION_STATUS, EXPIRED_SUBSCRIPTION_STATUS], +) +@pytest.mark.parametrize("entitlement_status", [EntitlementStatus.NEW, EntitlementStatus.ACTIVE]) +async def test_inactive_subscription_moves_the_entitlement_to_the_live_replacement( + subscription_event_handler: SubscriptionEventHandler, + mpt_subscription_factory: MPTSubscriptionFactory, + mock_get_subscription: SubscriptionMocker, + mock_live_subscription_lookup: SubscriptionMocker, + entitlement_factory: ModelFactory[Entitlement], + subscription_account: Account, + db_session: AsyncSession, + subscription_status: str, + entitlement_status: EntitlementStatus, +) -> None: + """A live subscription for the same datasource keeps the entitlement and takes it over.""" + entitlement = await entitlement_factory( + owner=subscription_account, + datasource_id=DATASOURCE_ID, + affiliate_external_id=SUBSCRIPTION_ID, + status=entitlement_status, + ) + mock_get_subscription(mpt_subscription_factory(status=subscription_status)) + mock_live_subscription_lookup([mpt_subscription_factory(subscription_id=REPLACEMENT_ID)]) + processor = await subscription_event_handler.get_processor(object_id=SUBSCRIPTION_ID) + + result = await processor.process() + + await db_session.refresh(entitlement) + assert entitlement.status is entitlement_status + assert entitlement.affiliate_external_id == REPLACEMENT_ID + assert entitlement.terminated_at is None + assert entitlement.deleted_at is None + assert result.status is ProcessingStatus.COMPLETE + assert result.severity == "Info" + assert result.message == ( + f"The entitlement {entitlement.id} was moved to the subscription {REPLACEMENT_ID}." + ) + + @pytest.mark.parametrize("status", [EntitlementStatus.TERMINATED, EntitlementStatus.DELETED]) async def test_terminated_subscription_does_nothing_when_with_stale_entitlement( subscription_event_handler: SubscriptionEventHandler,