feat(billing): _api_cancel_subscription service method + unit tests

Plan 01 (NexaCloud cutover) Task 1: cancel/close a subscription with the same
service-scoped authorization as _api_record_usage (resolve via
_fc_resolve_subscription; partner must be linked to this service). Idempotent
(no-op if already 6_churn). 5 unit tests, verified green on fcb_test
(fresh + l10n_ca). DELETE route + HttpCase follow in Task 2.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

fusion_centralize_billing/models/service.py

@@ -247,3 +247,24 @@ class FusionBillingService(models.Model):
         sub.action_confirm()
         return {'status': 'ok', 'subscription_id': sub.id,
                 'subscription_state': sub.subscription_state}
+
+    def _api_cancel_subscription(self, external_ref):
+        """Cancel (close) the subscription identified by ``external_ref``.
+
+        Authorization mirrors ``_api_record_usage``: the resolved sale.order must
+        exist, be a subscription, and belong to a customer THIS service is linked
+        to. Idempotent — closing an already-churned subscription returns ok.
+        Validation (C3): an empty ref returns a 4xx-shaped error, never raises.
+        """
+        self.ensure_one()
+        if external_ref in (None, ''):
+            return {'status': 'error', 'error': 'subscription id required'}
+        sub = self._fc_resolve_subscription(external_ref)
+        linked_partners = self.account_link_ids.mapped('partner_id')
+        if not sub.exists() or not sub.is_subscription \
+                or sub.partner_id not in linked_partners:
+            return {'status': 'error', 'error': 'unknown subscription'}
+        if sub.subscription_state != '6_churn':
+            sub.set_close()
+        return {'status': 'ok', 'subscription_id': sub.id,
+                'subscription_state': sub.subscription_state}

fusion_centralize_billing/tests/__init__.py

@@ -6,3 +6,4 @@ from . import test_webhook
 from . import test_importer
 from . import test_reconciliation
 from . import test_invoice_ledger
+from . import test_subscription_cancel

fusion_centralize_billing/tests/test_subscription_cancel.py

@@ -0,0 +1,54 @@
+# -*- coding: utf-8 -*-
+from odoo.tests.common import TransactionCase, tagged
+
+
+@tagged('post_install', '-at_install')
+class TestSubscriptionCancel(TransactionCase):
+
+    def _service(self, code, name):
+        Svc = self.env['fusion.billing.service'].sudo()
+        return Svc.search([('code', '=', code)], limit=1) or Svc.create(
+            {'name': name, 'code': code})
+
+    def setUp(self):
+        super().setUp()
+        self.plan = self.env['sale.subscription.plan'].sudo().create(
+            {'name': 'Monthly', 'billing_period_value': 1, 'billing_period_unit': 'month'})
+        self.product = self.env['product.product'].sudo().create(
+            {'name': 'NexaCloud Plan', 'type': 'service',
+             'recurring_invoice': True, 'list_price': 49.0})
+        self.svc_a = self._service('nexacloud', 'NexaCloud')
+        self.svc_b = self._service('other_app', 'Other App')
+        self.svc_a._api_upsert_customer({'external_id': 'user-1', 'name': 'Acme'})
+        res = self.svc_a._api_create_subscription({
+            'external_customer_id': 'user-1', 'plan_id': self.plan.id,
+            'lines': [{'product_id': self.product.id, 'quantity': 1}]})
+        self.sub = self.env['sale.order'].browse(res['subscription_id'])
+
+    def test_cancel_closes_subscription(self):
+        self.assertEqual(self.sub.subscription_state, '3_progress')
+        res = self.svc_a._api_cancel_subscription(str(self.sub.id))
+        self.assertEqual(res['status'], 'ok')
+        self.assertEqual(self.sub.subscription_state, '6_churn')
+
+    def test_cancel_is_idempotent(self):
+        self.svc_a._api_cancel_subscription(str(self.sub.id))
+        res = self.svc_a._api_cancel_subscription(str(self.sub.id))
+        self.assertEqual(res['status'], 'ok')
+        self.assertEqual(self.sub.subscription_state, '6_churn')
+
+    def test_cancel_unknown_subscription_rejected(self):
+        res = self.svc_a._api_cancel_subscription('999999999')
+        self.assertEqual(res['status'], 'error')
+        self.assertEqual(res['error'], 'unknown subscription')
+
+    def test_cancel_cross_service_rejected(self):
+        # svc_b is not linked to the customer that owns self.sub
+        res = self.svc_b._api_cancel_subscription(str(self.sub.id))
+        self.assertEqual(res['status'], 'error')
+        self.assertEqual(res['error'], 'unknown subscription')
+        self.assertEqual(self.sub.subscription_state, '3_progress')
+
+    def test_cancel_missing_id_rejected(self):
+        res = self.svc_a._api_cancel_subscription('')
+        self.assertEqual(res['status'], 'error')