From 0cacf62e254cfa3e7df14a0783b8a43f03b259fa Mon Sep 17 00:00:00 2001 From: Ken Lewerentz Date: Fri, 3 Apr 2026 12:15:42 +0700 Subject: [PATCH 1/3] perf: increase job concurrency, add channel routing and identity keys Increase parallel-safe channel limits from 1 to 4 (cycle, eligibility_manager, program_manager) now that INSERT ON CONFLICT makes these operations safe for concurrent execution. Add two serial channels: - entitlement_approval (limit=1): fund balance tracking must be serial - statistics_refresh (limit=1): avoid concurrent refresh storms Route entitlement approval/validation jobs to entitlement_approval channel. Route all completion handlers (mark_*_as_done) to statistics_refresh channel. Add identity_key to all async dispatch methods to prevent duplicate job submission when users double-click action buttons. --- spp_programs/data/queue_data.xml | 14 +++++++-- .../models/managers/cycle_manager_base.py | 29 +++++++++++++++---- .../models/managers/eligibility_manager.py | 9 +++--- .../managers/entitlement_manager_base.py | 16 ++++++++-- .../managers/entitlement_manager_cash.py | 6 +++- .../managers/entitlement_manager_inkind.py | 10 +++++-- .../models/managers/program_manager.py | 9 +++--- 7 files changed, 71 insertions(+), 22 deletions(-) diff --git a/spp_programs/data/queue_data.xml b/spp_programs/data/queue_data.xml index 9dd6f2df..afc44466 100644 --- a/spp_programs/data/queue_data.xml +++ b/spp_programs/data/queue_data.xml @@ -1,16 +1,26 @@ cycle - 1 + 4 0 eligibility_manager - 1 + 4 0 program_manager + 4 + 0 + + + entitlement_approval + 1 + 0 + + + statistics_refresh 1 0 diff --git a/spp_programs/models/managers/cycle_manager_base.py b/spp_programs/models/managers/cycle_manager_base.py index c3914b96..f9fb56e8 100644 --- a/spp_programs/models/managers/cycle_manager_base.py +++ b/spp_programs/models/managers/cycle_manager_base.py @@ -527,9 +527,14 @@ def _check_eligibility_async(self, cycle, beneficiaries_count): jobs = [] for min_id, max_id in id_ranges: - jobs.append(self.delayable(channel="cycle")._check_eligibility(cycle, min_id=min_id, max_id=max_id)) + jobs.append( + self.delayable( + channel="cycle", + identity_key=f"check_elig_{cycle.id}_{min_id}", + )._check_eligibility(cycle, min_id=min_id, max_id=max_id) + ) main_job = group(*jobs) - main_job.on_done(self.delayable(channel="cycle").mark_check_eligibility_as_done(cycle)) + main_job.on_done(self.delayable(channel="statistics_refresh").mark_check_eligibility_as_done(cycle)) main_job.delay() def _check_eligibility( @@ -607,10 +612,17 @@ def _prepare_entitlements_async(self, cycle, beneficiaries_count): jobs = [] for min_id, max_id in id_ranges: - jobs.append(self.delayable(channel="cycle")._prepare_entitlements(cycle, min_id=min_id, max_id=max_id)) + jobs.append( + self.delayable( + channel="cycle", + identity_key=f"prepare_ent_{cycle.id}_{min_id}", + )._prepare_entitlements(cycle, min_id=min_id, max_id=max_id) + ) main_job = group(*jobs) main_job.on_done( - self.delayable(channel="cycle").mark_prepare_entitlement_as_done(cycle, _("Entitlement Ready.")) + self.delayable(channel="statistics_refresh").mark_prepare_entitlement_as_done( + cycle, _("Entitlement Ready.") + ) ) main_job.delay() @@ -844,7 +856,10 @@ def _add_beneficiaries_async(self, cycle, beneficiaries, state): jobs = [] for i in range(0, beneficiaries_count, self.MAX_ROW_JOB_QUEUE): jobs.append( - self.delayable(channel="cycle")._add_beneficiaries( + self.delayable( + channel="cycle", + identity_key=f"add_benef_{cycle.id}_{i}", + )._add_beneficiaries( cycle, beneficiaries[i : i + self.MAX_ROW_JOB_QUEUE], state, @@ -852,7 +867,9 @@ def _add_beneficiaries_async(self, cycle, beneficiaries, state): ) main_job = group(*jobs) - main_job.on_done(self.delayable(channel="cycle").mark_import_as_done(cycle, _("Beneficiary import finished."))) + main_job.on_done( + self.delayable(channel="statistics_refresh").mark_import_as_done(cycle, _("Beneficiary import finished.")) + ) main_job.delay() def _add_beneficiaries(self, cycle, beneficiaries, state="draft", do_count=False): diff --git a/spp_programs/models/managers/eligibility_manager.py b/spp_programs/models/managers/eligibility_manager.py index 54dde250..c93a65e8 100644 --- a/spp_programs/models/managers/eligibility_manager.py +++ b/spp_programs/models/managers/eligibility_manager.py @@ -155,12 +155,13 @@ def _import_registrants_async(self, new_beneficiaries, state="draft"): jobs = [] for i in range(0, len(new_beneficiaries), 10000): jobs.append( - self.delayable(channel="eligibility_manager")._import_registrants( - new_beneficiaries[i : i + 10000], state - ) + self.delayable( + channel="eligibility_manager", + identity_key=f"import_reg_{program.id}_{i}", + )._import_registrants(new_beneficiaries[i : i + 10000], state) ) main_job = group(*jobs) - main_job.on_done(self.delayable(channel="eligibility_manager").mark_import_as_done()) + main_job.on_done(self.delayable(channel="statistics_refresh").mark_import_as_done()) main_job.delay() def mark_import_as_done(self): diff --git a/spp_programs/models/managers/entitlement_manager_base.py b/spp_programs/models/managers/entitlement_manager_base.py index 8a8d6483..5ddb61b6 100644 --- a/spp_programs/models/managers/entitlement_manager_base.py +++ b/spp_programs/models/managers/entitlement_manager_base.py @@ -89,7 +89,9 @@ def _set_pending_validation_entitlements_async(self, cycle, entitlements): jobs = [] for i in range(0, entitlements_count, self.MAX_ROW_JOB_QUEUE): jobs.append( - self.delayable()._set_pending_validation_entitlements(entitlements[i : i + self.MAX_ROW_JOB_QUEUE]) + self.delayable(channel="entitlement_approval")._set_pending_validation_entitlements( + entitlements[i : i + self.MAX_ROW_JOB_QUEUE] + ) ) main_job = group(*jobs) main_job.on_done(self.delayable().mark_job_as_done(cycle, _("Entitlements Set to Pending Validation."))) @@ -137,7 +139,11 @@ def _validate_entitlements_async(self, cycle, entitlements, entitlements_count): jobs = [] for i in range(0, entitlements_count, self.MAX_ROW_JOB_QUEUE): - jobs.append(self.delayable()._validate_entitlements(entitlements[i : i + self.MAX_ROW_JOB_QUEUE])) + jobs.append( + self.delayable(channel="entitlement_approval")._validate_entitlements( + entitlements[i : i + self.MAX_ROW_JOB_QUEUE] + ) + ) main_job = group(*jobs) main_job.on_done(self.delayable().mark_job_as_done(cycle, _("Entitlements Validated and Approved."))) main_job.delay() @@ -197,7 +203,11 @@ def _cancel_entitlements_async(self, cycle, entitlements, entitlements_count): jobs = [] for i in range(0, entitlements_count, self.MAX_ROW_JOB_QUEUE): - jobs.append(self.delayable()._cancel_entitlements(entitlements[i : i + self.MAX_ROW_JOB_QUEUE])) + jobs.append( + self.delayable(channel="entitlement_approval")._cancel_entitlements( + entitlements[i : i + self.MAX_ROW_JOB_QUEUE] + ) + ) main_job = group(*jobs) main_job.on_done(self.delayable().mark_job_as_done(cycle, _("Entitlements Cancelled."))) main_job.delay() diff --git a/spp_programs/models/managers/entitlement_manager_cash.py b/spp_programs/models/managers/entitlement_manager_cash.py index c17c7ec4..7449263d 100644 --- a/spp_programs/models/managers/entitlement_manager_cash.py +++ b/spp_programs/models/managers/entitlement_manager_cash.py @@ -319,7 +319,11 @@ def _validate_entitlements_async(self, cycle, entitlements, entitlements_count): jobs = [] for i in range(0, entitlements_count, self.MAX_ROW_JOB_QUEUE): # Needs to override - jobs.append(self.delayable()._validate_entitlements(cycle, entitlements[i : i + self.MAX_ROW_JOB_QUEUE])) + jobs.append( + self.delayable(channel="entitlement_approval")._validate_entitlements( + cycle, entitlements[i : i + self.MAX_ROW_JOB_QUEUE] + ) + ) main_job = group(*jobs) main_job.on_done(self.delayable().mark_job_as_done(cycle, _("Entitlements Validated and Approved."))) main_job.delay() diff --git a/spp_programs/models/managers/entitlement_manager_inkind.py b/spp_programs/models/managers/entitlement_manager_inkind.py index cdb09077..f3630ee2 100644 --- a/spp_programs/models/managers/entitlement_manager_inkind.py +++ b/spp_programs/models/managers/entitlement_manager_inkind.py @@ -216,7 +216,9 @@ def _set_pending_validation_entitlements_async(self, cycle, entitlements_count): jobs = [] for i in range(0, entitlements_count, self.MAX_ROW_JOB_QUEUE): jobs.append( - self.delayable()._set_pending_validation_entitlements(cycle, offset=i, limit=self.MAX_ROW_JOB_QUEUE) + self.delayable(channel="entitlement_approval")._set_pending_validation_entitlements( + cycle, offset=i, limit=self.MAX_ROW_JOB_QUEUE + ) ) main_job = group(*jobs) main_job.on_done(self.delayable().mark_job_as_done(cycle, _("Entitlements Set to Pending Validation."))) @@ -315,7 +317,11 @@ def _validate_entitlements_async(self, cycle, entitlements_count): jobs = [] for i in range(0, entitlements_count, self.MAX_ROW_JOB_QUEUE): - jobs.append(self.delayable()._validate_entitlements(cycle, offset=i, limit=self.MAX_ROW_JOB_QUEUE)) + jobs.append( + self.delayable(channel="entitlement_approval")._validate_entitlements( + cycle, offset=i, limit=self.MAX_ROW_JOB_QUEUE + ) + ) main_job = group(*jobs) main_job.on_done(self.delayable().mark_job_as_done(cycle, _("Entitlements Validated and Approved."))) main_job.delay() diff --git a/spp_programs/models/managers/program_manager.py b/spp_programs/models/managers/program_manager.py index e309e360..7a799664 100644 --- a/spp_programs/models/managers/program_manager.py +++ b/spp_programs/models/managers/program_manager.py @@ -199,12 +199,13 @@ def _enroll_eligible_registrants_async(self, states, members_count): jobs = [] for min_id, max_id in id_ranges: jobs.append( - self.delayable(channel="program_manager")._enroll_eligible_registrants( - states, min_id=min_id, max_id=max_id - ) + self.delayable( + channel="program_manager", + identity_key=f"enroll_eligible_{program.id}_{min_id}", + )._enroll_eligible_registrants(states, min_id=min_id, max_id=max_id) ) main_job = group(*jobs) - main_job.on_done(self.delayable(channel="program_manager").mark_enroll_eligible_as_done()) + main_job.on_done(self.delayable(channel="statistics_refresh").mark_enroll_eligible_as_done()) main_job.delay() def _enroll_eligible_registrants(self, states, offset=0, limit=None, min_id=None, max_id=None, do_count=False): From c0af41cf8c9a78274950ed5f48c4b1ce21dda8d8 Mon Sep 17 00:00:00 2001 From: Edwin Gonzales Date: Fri, 17 Apr 2026 11:23:54 +0800 Subject: [PATCH 2/3] docs: bump version to 19.0.2.0.10, add changelog for concurrency tuning --- spp_programs/README.rst | 12 +++++++++ spp_programs/__manifest__.py | 2 +- spp_programs/readme/HISTORY.md | 7 +++++ spp_programs/static/description/index.html | 31 +++++++++++++++------- 4 files changed, 42 insertions(+), 10 deletions(-) diff --git a/spp_programs/README.rst b/spp_programs/README.rst index fc223252..269543fb 100644 --- a/spp_programs/README.rst +++ b/spp_programs/README.rst @@ -254,6 +254,18 @@ Dependencies Changelog ========= +19.0.2.0.10 +~~~~~~~~~~~ + +- Increase parallel-safe channel limits (cycle, eligibility_manager, + program_manager) from 1 to 4 +- Add serial ``entitlement_approval`` channel (limit=1) for fund balance + safety +- Add serial ``statistics_refresh`` channel (limit=1) to prevent + concurrent refresh storms +- Add ``identity_key`` to async job dispatchers to prevent duplicate + submission on double-click + 19.0.2.0.9 ~~~~~~~~~~ diff --git a/spp_programs/__manifest__.py b/spp_programs/__manifest__.py index b32ff972..0f61ce1c 100644 --- a/spp_programs/__manifest__.py +++ b/spp_programs/__manifest__.py @@ -4,7 +4,7 @@ "name": "OpenSPP Programs", "summary": "Manage programs, cycles, beneficiary enrollment, entitlements (cash and in-kind), payments, and fund tracking for social protection.", "category": "OpenSPP/Core", - "version": "19.0.2.0.9", + "version": "19.0.2.0.10", "sequence": 1, "author": "OpenSPP.org", "website": "https://github.com/OpenSPP/OpenSPP2", diff --git a/spp_programs/readme/HISTORY.md b/spp_programs/readme/HISTORY.md index bd3fbc33..2ca0bedb 100644 --- a/spp_programs/readme/HISTORY.md +++ b/spp_programs/readme/HISTORY.md @@ -1,3 +1,10 @@ +### 19.0.2.0.10 + +- Increase parallel-safe channel limits (cycle, eligibility_manager, program_manager) from 1 to 4 +- Add serial `entitlement_approval` channel (limit=1) for fund balance safety +- Add serial `statistics_refresh` channel (limit=1) to prevent concurrent refresh storms +- Add `identity_key` to async job dispatchers to prevent duplicate submission on double-click + ### 19.0.2.0.9 - Add context flags (`skip_registrant_statistics`, `skip_program_statistics`) to suppress expensive computed field recomputation during bulk operations diff --git a/spp_programs/static/description/index.html b/spp_programs/static/description/index.html index a9f08e62..fc5367a3 100644 --- a/spp_programs/static/description/index.html +++ b/spp_programs/static/description/index.html @@ -658,6 +658,19 @@

Changelog

+

19.0.2.0.10

+
    +
  • Increase parallel-safe channel limits (cycle, eligibility_manager, +program_manager) from 1 to 4
    • +
  • Add serial entitlement_approval channel (limit=1) for fund balance +safety
    • +
  • Add serial statistics_refresh channel (limit=1) to prevent +concurrent refresh storms
    • +
  • Add identity_key to async job dispatchers to prevent duplicate +submission on double-click
    • +
+
+

19.0.2.0.9

  • Add context flags (skip_registrant_statistics, @@ -670,7 +683,7 @@

    19.0.2.0.9

    _compute_has_members
-
+

19.0.2.0.8

  • Replace OFFSET pagination with NTILE-based ID-range batching in all @@ -681,7 +694,7 @@

    19.0.2.0.8

    program and cycle
-
+

19.0.2.0.7

  • Bulk membership creation using raw SQL INSERT ON CONFLICT DO NOTHING @@ -690,7 +703,7 @@

    19.0.2.0.7

    _add_beneficiaries with bulk SQL path
-
+

19.0.2.0.6

  • Remove unused entitlement_base_model.py (dead code, never imported)
    • @@ -699,34 +712,34 @@

    19.0.2.0.6

    payment, and fund tests (172 → 492 tests)
-
+

19.0.2.0.5

  • Batch create entitlements and payments instead of one-by-one ORM creates
-
+

19.0.2.0.4

  • Fetch fund balance once per approval batch instead of per entitlement
-
+

19.0.2.0.3

  • Replace cycle computed fields (total_amount, entitlements_count, approval flags) with SQL aggregation queries
-
+

19.0.2.0.2

  • Add composite indexes for frequent query patterns on entitlements and program memberships
-
+

19.0.2.0.1

  • Replace Python-level uniqueness checks with SQL UNIQUE constraints for @@ -735,7 +748,7 @@

    19.0.2.0.1

    constraint creation
-
+

19.0.2.0.0

  • Initial migration to OpenSPP2
    • From 832c4979c3d861ba9a23a44525358de8bb37e8a9 Mon Sep 17 00:00:00 2001 From: Edwin Gonzales Date: Fri, 17 Apr 2026 11:37:14 +0800 Subject: [PATCH 3/3] test: add coverage for channel routing and identity keys in async dispatchers --- spp_programs/tests/__init__.py | 1 + spp_programs/tests/test_concurrency.py | 297 +++++++++++++++++++++++++ 2 files changed, 298 insertions(+) create mode 100644 spp_programs/tests/test_concurrency.py diff --git a/spp_programs/tests/__init__.py b/spp_programs/tests/__init__.py index aef67f51..873da4ce 100644 --- a/spp_programs/tests/__init__.py +++ b/spp_programs/tests/__init__.py @@ -35,3 +35,4 @@ from . import test_bulk_membership from . import test_keyset_pagination from . import test_canary_patterns +from . import test_concurrency diff --git a/spp_programs/tests/test_concurrency.py b/spp_programs/tests/test_concurrency.py new file mode 100644 index 00000000..c35a224a --- /dev/null +++ b/spp_programs/tests/test_concurrency.py @@ -0,0 +1,297 @@ +# Part of OpenSPP. See LICENSE file for full copyright and licensing details. +"""Tests for Phase 9: Job concurrency, channel routing, and identity keys. + +Verify that async dispatchers pass correct channel and identity_key to +delayable(), and that completion handlers route to statistics_refresh. +""" + +import uuid +from unittest.mock import MagicMock, patch + +from odoo import fields +from odoo.tests import TransactionCase + + +class TestCycleManagerChannelRouting(TransactionCase): + """Test channel routing and identity_key in cycle manager async methods.""" + + def setUp(self): + super().setUp() + self.program = self.env["spp.program"].create({"name": f"Test Program {uuid.uuid4().hex[:8]}"}) + self.cycle = self.env["spp.cycle"].create( + { + "name": "Test Cycle", + "program_id": self.program.id, + "start_date": fields.Date.today(), + "end_date": fields.Date.today(), + } + ) + self.cycle_manager = self.env["spp.cycle.manager.default"].create( + { + "name": "Test Cycle Manager", + "program_id": self.program.id, + } + ) + + def test_check_eligibility_async_uses_identity_key(self): + """_check_eligibility_async must pass identity_key to delayable.""" + partners = self.env["res.partner"].create( + [{"name": f"Registrant {i}", "is_registrant": True} for i in range(3)] + ) + self.env["spp.cycle.membership"].create( + [{"partner_id": p.id, "cycle_id": self.cycle.id, "state": "draft"} for p in partners] + ) + + delayable_calls = [] + original_delayable = type(self.cycle_manager).delayable + + def mock_delayable(self_inner, **kwargs): + delayable_calls.append(kwargs) + return original_delayable(self_inner, **kwargs) + + with patch.object(type(self.cycle_manager), "delayable", mock_delayable): + try: + self.cycle_manager._check_eligibility_async(self.cycle, 3) + except Exception: + pass + + # Should have at least one call with identity_key containing "check_elig_" + identity_keys = [c.get("identity_key", "") for c in delayable_calls] + has_check_elig_key = any("check_elig_" in k for k in identity_keys) + self.assertTrue(has_check_elig_key, f"Expected identity_key with 'check_elig_', got: {identity_keys}") + + # Completion handler should route to statistics_refresh + channels = [c.get("channel", "") for c in delayable_calls] + self.assertIn("statistics_refresh", channels) + + def test_prepare_entitlements_async_uses_identity_key(self): + """_prepare_entitlements_async must pass identity_key to delayable.""" + partners = self.env["res.partner"].create( + [{"name": f"Registrant {i}", "is_registrant": True} for i in range(3)] + ) + self.env["spp.cycle.membership"].create( + [{"partner_id": p.id, "cycle_id": self.cycle.id, "state": "enrolled"} for p in partners] + ) + + delayable_calls = [] + original_delayable = type(self.cycle_manager).delayable + + def mock_delayable(self_inner, **kwargs): + delayable_calls.append(kwargs) + return original_delayable(self_inner, **kwargs) + + with patch.object(type(self.cycle_manager), "delayable", mock_delayable): + try: + self.cycle_manager._prepare_entitlements_async(self.cycle, 3) + except Exception: + pass + + identity_keys = [c.get("identity_key", "") for c in delayable_calls] + has_prepare_key = any("prepare_ent_" in k for k in identity_keys) + self.assertTrue(has_prepare_key, f"Expected identity_key with 'prepare_ent_', got: {identity_keys}") + + channels = [c.get("channel", "") for c in delayable_calls] + self.assertIn("statistics_refresh", channels) + + def test_add_beneficiaries_async_uses_identity_key(self): + """_add_beneficiaries_async must pass identity_key to delayable.""" + partners = self.env["res.partner"].create( + [{"name": f"Registrant {i}", "is_registrant": True} for i in range(3)] + ) + + delayable_calls = [] + original_delayable = type(self.cycle_manager).delayable + + def mock_delayable(self_inner, **kwargs): + delayable_calls.append(kwargs) + return original_delayable(self_inner, **kwargs) + + with patch.object(type(self.cycle_manager), "delayable", mock_delayable): + try: + self.cycle_manager._add_beneficiaries_async(self.cycle, partners.ids, "draft") + except Exception: + pass + + identity_keys = [c.get("identity_key", "") for c in delayable_calls] + has_add_key = any("add_benef_" in k for k in identity_keys) + self.assertTrue(has_add_key, f"Expected identity_key with 'add_benef_', got: {identity_keys}") + + channels = [c.get("channel", "") for c in delayable_calls] + self.assertIn("statistics_refresh", channels) + + +class TestProgramManagerChannelRouting(TransactionCase): + """Test channel routing and identity_key in program manager async methods.""" + + def setUp(self): + super().setUp() + self.program = self.env["spp.program"].create({"name": f"Test Program {uuid.uuid4().hex[:8]}"}) + self.manager = self.env["spp.program.manager.default"].create( + { + "name": "Test Manager", + "program_id": self.program.id, + } + ) + + def test_enroll_eligible_async_uses_identity_key(self): + """_enroll_eligible_registrants_async must pass identity_key to delayable.""" + partners = self.env["res.partner"].create( + [{"name": f"Registrant {i}", "is_registrant": True} for i in range(3)] + ) + self.env["spp.program.membership"].create( + [{"partner_id": p.id, "program_id": self.program.id, "state": "draft"} for p in partners] + ) + + delayable_calls = [] + original_delayable = type(self.manager).delayable + + def mock_delayable(self_inner, **kwargs): + delayable_calls.append(kwargs) + return original_delayable(self_inner, **kwargs) + + with patch.object(type(self.manager), "delayable", mock_delayable): + try: + self.manager._enroll_eligible_registrants_async(["draft"], 3) + except Exception: + pass + + identity_keys = [c.get("identity_key", "") for c in delayable_calls] + has_enroll_key = any("enroll_eligible_" in k for k in identity_keys) + self.assertTrue(has_enroll_key, f"Expected identity_key with 'enroll_eligible_', got: {identity_keys}") + + channels = [c.get("channel", "") for c in delayable_calls] + self.assertIn("statistics_refresh", channels) + + +class TestEligibilityManagerChannelRouting(TransactionCase): + """Test channel routing and identity_key in eligibility manager async methods.""" + + def setUp(self): + super().setUp() + self.program = self.env["spp.program"].create({"name": f"Test Program {uuid.uuid4().hex[:8]}"}) + self.elig_manager = self.env["spp.program.membership.manager.default"].create( + { + "name": "Test Elig Manager", + "program_id": self.program.id, + } + ) + + def test_import_registrants_async_uses_identity_key(self): + """_import_registrants_async must pass identity_key to delayable.""" + partners = self.env["res.partner"].create( + [{"name": f"Registrant {i}", "is_registrant": True} for i in range(3)] + ) + + delayable_calls = [] + original_delayable = type(self.elig_manager).delayable + + def mock_delayable(self_inner, **kwargs): + delayable_calls.append(kwargs) + return original_delayable(self_inner, **kwargs) + + with patch.object(type(self.elig_manager), "delayable", mock_delayable): + try: + self.elig_manager._import_registrants_async(partners, "draft") + except Exception: + pass + + identity_keys = [c.get("identity_key", "") for c in delayable_calls] + has_import_key = any("import_reg_" in k for k in identity_keys) + self.assertTrue(has_import_key, f"Expected identity_key with 'import_reg_', got: {identity_keys}") + + channels = [c.get("channel", "") for c in delayable_calls] + self.assertIn("statistics_refresh", channels) + + +class TestEntitlementManagerChannelRouting(TransactionCase): + """Test that entitlement async methods route to entitlement_approval channel.""" + + def setUp(self): + super().setUp() + self.program = self.env["spp.program"].create({"name": f"Test Program {uuid.uuid4().hex[:8]}"}) + self.cycle = self.env["spp.cycle"].create( + { + "name": "Test Cycle", + "program_id": self.program.id, + "start_date": fields.Date.today(), + "end_date": fields.Date.today(), + } + ) + + def _get_entitlement_manager(self): + """Get a cash entitlement manager for testing.""" + return self.env["spp.program.entitlement.manager.cash"].create( + { + "name": "Test Entitlement Manager", + "program_id": self.program.id, + } + ) + + def test_set_pending_validation_async_routes_to_entitlement_approval(self): + """_set_pending_validation_entitlements_async must use entitlement_approval channel.""" + ent_manager = self._get_entitlement_manager() + mock_entitlements = MagicMock() + mock_entitlements.__len__ = MagicMock(return_value=5) + mock_entitlements.__getitem__ = MagicMock(return_value=mock_entitlements) + + delayable_calls = [] + original_delayable = type(ent_manager).delayable + + def mock_delayable(self_inner, **kwargs): + delayable_calls.append(kwargs) + return original_delayable(self_inner, **kwargs) + + with patch.object(type(ent_manager), "delayable", mock_delayable): + try: + ent_manager._set_pending_validation_entitlements_async(self.cycle, mock_entitlements) + except Exception: + pass + + channels = [c.get("channel", "") for c in delayable_calls] + self.assertIn("entitlement_approval", channels) + + def test_validate_entitlements_async_routes_to_entitlement_approval(self): + """_validate_entitlements_async must use entitlement_approval channel.""" + ent_manager = self._get_entitlement_manager() + mock_entitlements = MagicMock() + mock_entitlements.__len__ = MagicMock(return_value=5) + mock_entitlements.__getitem__ = MagicMock(return_value=mock_entitlements) + + delayable_calls = [] + original_delayable = type(ent_manager).delayable + + def mock_delayable(self_inner, **kwargs): + delayable_calls.append(kwargs) + return original_delayable(self_inner, **kwargs) + + with patch.object(type(ent_manager), "delayable", mock_delayable): + try: + ent_manager._validate_entitlements_async(self.cycle, mock_entitlements, 5) + except Exception: + pass + + channels = [c.get("channel", "") for c in delayable_calls] + self.assertIn("entitlement_approval", channels) + + def test_cancel_entitlements_async_routes_to_entitlement_approval(self): + """_cancel_entitlements_async must use entitlement_approval channel.""" + ent_manager = self._get_entitlement_manager() + mock_entitlements = MagicMock() + mock_entitlements.__len__ = MagicMock(return_value=5) + mock_entitlements.__getitem__ = MagicMock(return_value=mock_entitlements) + + delayable_calls = [] + original_delayable = type(ent_manager).delayable + + def mock_delayable(self_inner, **kwargs): + delayable_calls.append(kwargs) + return original_delayable(self_inner, **kwargs) + + with patch.object(type(ent_manager), "delayable", mock_delayable): + try: + ent_manager._cancel_entitlements_async(self.cycle, mock_entitlements, 5) + except Exception: + pass + + channels = [c.get("channel", "") for c in delayable_calls] + self.assertIn("entitlement_approval", channels)