""" Tests verifying that the from_number pool isolates numbers per telephony_configuration_id within an organization. When an org has multiple telephony configurations (each with its own pool of caller IDs), a campaign pinned to config A must never be handed a from_number that belongs to config B. Otherwise the call is placed via provider A using a DID owned by config B and either fails or originates from the wrong number. These tests cover both: - The rate_limiter, which owns the Redis-backed pool. - The CampaignCallDispatcher, which must thread the campaign's telephony_configuration_id through acquire / release / mapping calls. """ import os import uuid from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch import pytest from api.services.call_concurrency import CallConcurrencySlot from api.services.call_concurrency.rate_limiter import RateLimiter from api.services.campaign.campaign_call_dispatcher import CampaignCallDispatcher def _unique_id() -> int: """A stable-but-unique positive int derived from a uuid for keying tests.""" return uuid.uuid4().int % 10_000_000 @pytest.fixture async def isolated_rate_limiter(): """A RateLimiter wired to the same Redis as production but using unique ids per test, with cleanup of any keys it touched.""" rl = RateLimiter() redis_client = await rl._get_redis() created_keys: list[str] = [] original_eval = redis_client.eval original_zadd = redis_client.zadd async def tracking_eval(script, numkeys, *args, **kwargs): if numkeys >= 1: created_keys.append(args[0]) return await original_eval(script, numkeys, *args, **kwargs) async def tracking_zadd(name, *args, **kwargs): created_keys.append(name) return await original_zadd(name, *args, **kwargs) redis_client.eval = tracking_eval # type: ignore[assignment] redis_client.zadd = tracking_zadd # type: ignore[assignment] yield rl redis_client.eval = original_eval # type: ignore[assignment] redis_client.zadd = original_zadd # type: ignore[assignment] if created_keys: await redis_client.delete(*set(created_keys)) await rl.close() # --------------------------------------------------------------------------- # Rate limiter pool isolation # --------------------------------------------------------------------------- class TestRateLimiterFromNumberPoolIsolation: """The rate_limiter pool keys must include telephony_configuration_id.""" @pytest.mark.skipif( "REDIS_URL" not in os.environ, reason="Requires Redis (set REDIS_URL via .env.test)", ) @pytest.mark.asyncio async def test_acquire_only_returns_numbers_for_requested_config( self, isolated_rate_limiter ): rl = isolated_rate_limiter org_id = _unique_id() config_a = _unique_id() config_b = _unique_id() numbers_a = [f"+1555111{i:04d}" for i in range(3)] numbers_b = [f"+1555222{i:04d}" for i in range(3)] await rl.initialize_from_number_pool( org_id, numbers_a, telephony_configuration_id=config_a ) await rl.initialize_from_number_pool( org_id, numbers_b, telephony_configuration_id=config_b ) # Drain and cycle config_a's pool many times; the acquire should never # hand out a config_b number. seen: set[str] = set() for _ in range(20): n = await rl.acquire_from_number( org_id, telephony_configuration_id=config_a ) if n is None: break seen.add(n) await rl.release_from_number(org_id, n, telephony_configuration_id=config_a) assert seen == set(numbers_a), ( f"Expected only config_a numbers, but acquire returned: {seen}. " f"Cross-config leak: {seen - set(numbers_a)}" ) @pytest.mark.skipif( "REDIS_URL" not in os.environ, reason="Requires Redis (set REDIS_URL via .env.test)", ) @pytest.mark.asyncio async def test_release_returns_number_to_owning_config_pool( self, isolated_rate_limiter ): rl = isolated_rate_limiter org_id = _unique_id() config_a = _unique_id() config_b = _unique_id() numbers_a = ["+15551110001", "+15551110002"] numbers_b = ["+15552220001", "+15552220002"] await rl.initialize_from_number_pool( org_id, numbers_a, telephony_configuration_id=config_a ) await rl.initialize_from_number_pool( org_id, numbers_b, telephony_configuration_id=config_b ) # Acquire all of config_a's numbers (none released). first = await rl.acquire_from_number( org_id, telephony_configuration_id=config_a ) second = await rl.acquire_from_number( org_id, telephony_configuration_id=config_a ) assert {first, second} == set(numbers_a) # config_a is now exhausted — config_b is fully untouched. # Acquiring for config_a must return None, NOT spill into config_b. none_for_a = await rl.acquire_from_number( org_id, telephony_configuration_id=config_a ) assert none_for_a is None, ( f"Pool for config_a is exhausted but acquire returned {none_for_a} — " "this indicates a cross-config leak." ) # config_b's pool is fully available. b_acquired = [] for _ in range(2): n = await rl.acquire_from_number( org_id, telephony_configuration_id=config_b ) assert n is not None b_acquired.append(n) assert set(b_acquired) == set(numbers_b) @pytest.mark.skipif( "REDIS_URL" not in os.environ, reason="Requires Redis (set REDIS_URL via .env.test)", ) @pytest.mark.asyncio async def test_workflow_from_number_mapping_round_trips_config( self, isolated_rate_limiter ): """The mapping stored at dispatch must include the config so cleanup can release back to the correct pool.""" rl = isolated_rate_limiter workflow_run_id = _unique_id() org_id = _unique_id() config_id = _unique_id() from_number = "+15553330001" await rl.store_workflow_from_number_mapping( workflow_run_id, org_id, from_number, telephony_configuration_id=config_id, ) mapping = await rl.get_workflow_from_number_mapping(workflow_run_id) assert mapping is not None # Tuple shape: (org_id, from_number, telephony_configuration_id) assert len(mapping) == 3, ( f"mapping must include telephony_configuration_id, got: {mapping}" ) assert mapping == (org_id, from_number, config_id) await rl.delete_workflow_from_number_mapping(workflow_run_id) # --------------------------------------------------------------------------- # Dispatcher: must thread telephony_configuration_id end-to-end # --------------------------------------------------------------------------- def _make_campaign( *, organization_id: int, telephony_configuration_id: int | None, workflow_id: int = 1, campaign_id: int = 99, ) -> SimpleNamespace: return SimpleNamespace( id=campaign_id, organization_id=organization_id, workflow_id=workflow_id, created_by=1, telephony_configuration_id=telephony_configuration_id, rate_limit_per_second=100, processed_rows=0, orchestrator_metadata={}, ) def _make_queued_run( *, queued_run_id: int = 1, phone_number: str = "+15559990001", ) -> SimpleNamespace: return SimpleNamespace( id=queued_run_id, source_uuid=f"src-{queued_run_id}", context_variables={"phone_number": phone_number}, ) class TestDispatcherThreadsTelephonyConfig: """The dispatcher must pass telephony_configuration_id when acquiring, storing the mapping, and releasing the from_number.""" @pytest.mark.asyncio async def test_legacy_campaign_pins_the_ready_resolved_config(self): campaign = _make_campaign( organization_id=7, telephony_configuration_id=None, ) provider = MagicMock() resolver = AsyncMock(return_value=4242) provider_factory = AsyncMock(return_value=provider) dispatcher = CampaignCallDispatcher() with ( patch( "api.services.telephony.outbound_readiness.resolve_outbound_configuration_id", resolver, ), patch( "api.services.telephony.factory.get_telephony_provider_by_id", provider_factory, ), ): result = await dispatcher.get_provider_for_campaign(campaign) assert result is provider assert campaign.telephony_configuration_id == 4242 assert resolver.await_args.args == (None, 7) provider_factory.assert_awaited_once_with(4242, 7) @pytest.mark.asyncio async def test_dispatch_call_acquires_from_number_for_campaign_config(self): org_id = 7 config_id = 4242 campaign = _make_campaign( organization_id=org_id, telephony_configuration_id=config_id ) queued_run = _make_queued_run() provider = MagicMock() provider.PROVIDER_NAME = "twilio" provider.WEBHOOK_ENDPOINT = "twilio/voice" provider.from_numbers = ["+15551110001"] provider.initiate_call = AsyncMock( return_value=SimpleNamespace(call_id="call-1", provider_metadata={}) ) workflow_run = SimpleNamespace(id=555, logs={}) dispatcher = CampaignCallDispatcher() with ( patch.object( dispatcher, "get_provider_for_campaign", AsyncMock(return_value=provider), ), patch( "api.services.campaign.campaign_call_dispatcher.db_client" ) as mock_db, patch( "api.services.campaign.campaign_call_dispatcher.rate_limiter" ) as mock_rl, patch( "api.services.campaign.campaign_call_dispatcher.call_concurrency" ) as mock_concurrency, patch( "api.services.campaign.campaign_call_dispatcher.get_backend_endpoints", AsyncMock(return_value=("https://example.com", None)), ), patch( "api.services.campaign.campaign_call_dispatcher.authorize_workflow_run_start", AsyncMock( return_value=SimpleNamespace(has_quota=True, error_message="") ), ), ): mock_db.get_workflow = AsyncMock(return_value=SimpleNamespace(id=1)) mock_db.get_telephony_configuration_for_org = AsyncMock( return_value=SimpleNamespace( id=config_id, name="Twilio campaign", provider="twilio", credentials={}, ) ) mock_db.list_phone_numbers_for_config = AsyncMock( return_value=[ SimpleNamespace( is_active=True, inbound_workflow_id=None, telephony_trunk_id=None, ) ] ) mock_db.create_workflow_run = AsyncMock(return_value=workflow_run) mock_db.update_workflow_run = AsyncMock() mock_concurrency.bind_workflow_run = AsyncMock() mock_concurrency.release_slot = AsyncMock() mock_concurrency.release_workflow_run_slot = AsyncMock() mock_rl.acquire_from_number = AsyncMock(return_value="+15551110001") mock_rl.release_from_number = AsyncMock() mock_rl.store_workflow_from_number_mapping = AsyncMock() slot = CallConcurrencySlot( organization_id=org_id, slot_id="slot-1", max_concurrent=1, source="test", ) await dispatcher.dispatch_call(queued_run, campaign, slot) mock_db.get_workflow.assert_awaited_once_with( campaign.workflow_id, organization_id=org_id, ) # acquire_from_number on rate_limiter must be called with the # campaign's telephony_configuration_id. assert mock_rl.acquire_from_number.await_count == 1 call = mock_rl.acquire_from_number.await_args kwargs = call.kwargs args = call.args received_config = kwargs.get("telephony_configuration_id") or ( args[1] if len(args) > 1 else None ) assert received_config == config_id, ( "dispatch_call must pass campaign.telephony_configuration_id " f"({config_id}) to rate_limiter.acquire_from_number, got " f"args={args}, kwargs={kwargs}" ) # The workflow→from_number mapping must also remember the config so # the cleanup path can release back to the right pool. assert mock_rl.store_workflow_from_number_mapping.await_count == 1 store_call = mock_rl.store_workflow_from_number_mapping.await_args store_kwargs = store_call.kwargs store_args = store_call.args stored_config = store_kwargs.get("telephony_configuration_id") or ( store_args[3] if len(store_args) > 3 else None ) assert stored_config == config_id, ( "store_workflow_from_number_mapping must persist the " f"telephony_configuration_id ({config_id}); got args={store_args}, " f"kwargs={store_kwargs}" ) assert provider.initiate_call.await_count == 1 webhook_url = provider.initiate_call.await_args.kwargs["webhook_url"] assert "campaign_id=" not in webhook_url, ( "campaign outbound answer_url should not include campaign_id; " f"got {webhook_url}" ) @pytest.mark.asyncio async def test_dispatch_rejects_incomplete_setup_before_creating_run(self): from api.services.telephony.outbound_readiness import ( OutboundSetupIncompleteError, ) org_id = 7 config_id = 4242 campaign = _make_campaign( organization_id=org_id, telephony_configuration_id=config_id ) queued_run = _make_queued_run() provider = MagicMock(PROVIDER_NAME="twilio") dispatcher = CampaignCallDispatcher() with ( patch.object( dispatcher, "get_provider_for_campaign", AsyncMock( side_effect=OutboundSetupIncompleteError( config_id, "Twilio campaign", "Add a caller ID before placing calls.", ) ), ), patch( "api.services.campaign.campaign_call_dispatcher.db_client" ) as mock_db, patch( "api.services.campaign.campaign_call_dispatcher.call_concurrency" ) as mock_concurrency, ): mock_db.get_workflow = AsyncMock(return_value=SimpleNamespace(id=1)) mock_db.create_workflow_run = AsyncMock() mock_concurrency.release_slot = AsyncMock() slot = CallConcurrencySlot( organization_id=org_id, slot_id="slot-1", max_concurrent=1, source="test", ) with pytest.raises(OutboundSetupIncompleteError): await dispatcher.dispatch_call(queued_run, campaign, slot) mock_db.create_workflow_run.assert_not_awaited() provider.initiate_call.assert_not_called() mock_concurrency.release_slot.assert_awaited_once_with(slot) @pytest.mark.asyncio async def test_release_call_slot_uses_stored_telephony_config(self): """When a call completes, release_call_slot must release the from_number to the same telephony config it was acquired from.""" org_id = 7 config_id = 4242 from_number = "+15551110001" workflow_run_id = 555 dispatcher = CampaignCallDispatcher() with ( patch( "api.services.campaign.campaign_call_dispatcher.rate_limiter" ) as mock_rl, patch( "api.services.campaign.campaign_call_dispatcher.call_concurrency" ) as mock_concurrency, ): mock_concurrency.release_workflow_run_slot = AsyncMock(return_value=False) mock_rl.get_workflow_from_number_mapping = AsyncMock( return_value=(org_id, from_number, config_id) ) mock_rl.release_from_number = AsyncMock(return_value=True) mock_rl.delete_workflow_from_number_mapping = AsyncMock(return_value=True) await dispatcher.release_call_slot(workflow_run_id) assert mock_rl.release_from_number.await_count == 1 call = mock_rl.release_from_number.await_args args = call.args kwargs = call.kwargs released_config = kwargs.get("telephony_configuration_id") or ( args[2] if len(args) > 2 else None ) assert released_config == config_id, ( "release_call_slot must pass telephony_configuration_id " f"({config_id}) so the number is returned to its pool; got " f"args={args}, kwargs={kwargs}" )