import pytest from typing import Any from uuid import UUID from unittest.mock import AsyncMock from fastapi import Request, HTTPException from pytest_mock import MockerFixture from python_pdp_sdk import ResourceWithAttributes, UnauthenticatedException from product_staging.api import auth, datasources from product_staging.api.auth import TenantLookup from product_staging.api.schemas.tenant import Tenant from product_staging.constants.error import ( ERROR_MESSAGE_CANNOT_ACCESS_VENDOR, ERROR_MESSAGE_NO_VENDOR_ID, ) def make_request_with_scope(scope_dict): class DummyReceive: async def __call__(self): pass # Ensure 'type': 'http' is always present in the scope scope = {**{"type": "http"}, **scope_dict} return Request(scope=scope, receive=DummyReceive()) def test_profiles_from_scope_fetches_from_ows_users(mocker: MockerFixture) -> None: """Test profiles_from_scope fetches profiles from ows-users by identity.""" mocker.patch( "product_staging.api.auth.jwt.get_identity_uuid", return_value="10436b38-5e11-472d-b6a4-bf1ee2b1b438", ) mock_profiles = [("LabelProfile", 789), ("ContentProfile", 101)] mocker.patch( "product_staging.api.auth.ows_users.get_profiles_for_identity", return_value=mock_profiles, ) req = make_request_with_scope({"token": {}}) result = auth.profiles_from_scope(req) assert result == mock_profiles def test_profiles_from_scope_returns_empty_when_no_profiles(mocker): """When ows-users returns no profiles, return empty list.""" mocker.patch( "product_staging.api.auth.jwt.get_identity_uuid", return_value="10436b38-5e11-472d-b6a4-bf1ee2b1b438", ) mocker.patch( "product_staging.api.auth.ows_users.get_profiles_for_identity", return_value=[], ) req = make_request_with_scope({"token": {}}) result = auth.profiles_from_scope(req) assert result == [] def test_profiles_from_scope_returns_empty_when_no_identity_uuid(mocker): """When JWT has no identity UUID, return empty list.""" mocker.patch( "product_staging.api.auth.jwt.get_identity_uuid", return_value=None, ) req = make_request_with_scope({"token": {}}) result = auth.profiles_from_scope(req) assert result == [] def test_profiles_from_scope_without_token(): req = make_request_with_scope({}) with pytest.raises(HTTPException) as exc: auth.profiles_from_scope(req) assert exc.value.status_code == 401 assert "JWT not decoded by JWTAuthenticationMiddleware" in str(exc.value.detail) @pytest.mark.parametrize( "profile_type, profile_id, required, allowed_types, status", [ ("LabelProfile", 1, True, None, 200), # success, required (None, None, False, None, 200), # success, not required ("LabelProfile", None, False, None, 400), # incomplete headers (None, None, True, None, 403), # absent required headers ("LabelProfile", 1, False, ("ContentProfile",), 403), # type is not allowed ], ) def test_verify_profile_headers( profile_type: str | None, profile_id: int | None, required: bool, allowed_types: tuple[str, ...] | None, status: int, ) -> None: """Check 'verify_profile_headers' with various inputs.""" result = auth.verify_profile_headers( profile_type, profile_id, required, allowed_types ) assert result.status == status async def test_check_vendor_access_with_valid_profile( mocker: MockerFixture, identity_uuid: str ) -> None: """Test that check_vendor_access_for_profiles returns vendor_id when profile is valid.""" vendor_uuid = UUID("35289af1-d291-44a9-96c1-72fce8ee1c48") vendor_id = 42 profiles = [("LabelProfile", 123)] allowed_types = ("LabelProfile", "ContentProfile") # Mock Redis client mock_redis = mocker.MagicMock(spec=datasources.RedisConnector) mock_redis.get = AsyncMock(return_value=None) mock_redis.set = AsyncMock(return_value=None) mocker.patch( "product_staging.api.datasources.get_redis_client", return_value=mock_redis, ) # Mock verify_profile_headers to return success mock_response = mocker.Mock() mock_response.status = 200 mocker.patch( "product_staging.api.auth.verify_profile_headers", return_value=mock_response, ) # Mock get_vendor_id to return vendor_id mocker.patch( "product_staging.connectors.ows_account.get_vendor_id", return_value=vendor_id, ) # Mock check_profile_access to return success mock_access_response = mocker.Mock() mock_access_response.status = 200 mocker.patch( "product_staging.logic.profile.check_profile_access", return_value=mock_access_response, ) result = await auth.check_vendor_access_for_profiles( identity_id=UUID(identity_uuid), vendor_uuid=vendor_uuid, profiles=profiles, allowed_types=allowed_types, ) assert result == vendor_id async def test_check_vendor_access_with_invalid_headers( mocker: MockerFixture, identity_uuid: str ) -> None: """Test that check_vendor_access raises 403 when profile headers are invalid.""" vendor_uuid = UUID("35289af1-d291-44a9-96c1-72fce8ee1c48") profiles = [("InvalidProfile", 999)] allowed_types = ("LabelProfile", "ContentProfile") # Mock Redis client mock_redis = mocker.MagicMock(spec=datasources.RedisConnector) mock_redis.get = AsyncMock(return_value=None) mock_redis.set = AsyncMock(return_value=None) mocker.patch( "product_staging.api.datasources.get_redis_client", return_value=mock_redis, ) # Mock verify_profile_headers to return failure mock_response = mocker.Mock() mock_response.status = 403 mocker.patch( "product_staging.api.auth.verify_profile_headers", return_value=mock_response, ) mocker.patch( "product_staging.connectors.ows_account.get_vendor_id", return_value=42, ) with pytest.raises(HTTPException) as exc: await auth.check_vendor_access_for_profiles( identity_id=UUID(identity_uuid), vendor_uuid=vendor_uuid, profiles=profiles, allowed_types=allowed_types, ) assert exc.value.status_code == 403 assert ERROR_MESSAGE_CANNOT_ACCESS_VENDOR in exc.value.detail async def test_check_vendor_access_with_nonexistent_vendor( mocker: MockerFixture, identity_uuid: str ) -> None: """Test that check_vendor_access raises 404 when vendor doesn't exist.""" vendor_uuid = UUID("35289af1-d291-44a9-96c1-72fce8ee1c48") profiles = [("LabelProfile", 123)] allowed_types = ("LabelProfile", "ContentProfile") # Mock Redis client mock_redis = mocker.MagicMock(spec=datasources.RedisConnector) mock_redis.get = AsyncMock(return_value=None) mock_redis.set = AsyncMock(return_value=None) mocker.patch( "product_staging.api.datasources.get_redis_client", return_value=mock_redis, ) # Mock verify_profile_headers to return success mock_response = mocker.Mock() mock_response.status = 200 mocker.patch( "product_staging.api.auth.verify_profile_headers", return_value=mock_response, ) # Mock get_vendor_id to return None (vendor not found) mocker.patch( "product_staging.connectors.ows_account.get_vendor_id", return_value=None, ) with pytest.raises(HTTPException) as exc: await auth.check_vendor_access_for_profiles( identity_id=UUID(identity_uuid), vendor_uuid=vendor_uuid, profiles=profiles, allowed_types=allowed_types, ) assert exc.value.status_code == 404 assert ERROR_MESSAGE_NO_VENDOR_ID in exc.value.detail async def test_check_vendor_access_with_no_profile_access( mocker: MockerFixture, identity_uuid: str ) -> None: """Test that check_vendor_access raises 403 when profile has no access to vendor.""" vendor_uuid = UUID("35289af1-d291-44a9-96c1-72fce8ee1c48") vendor_id = 42 profiles = [("LabelProfile", 123)] allowed_types = ("LabelProfile", "ContentProfile") # Mock Redis client mock_redis = mocker.MagicMock(spec=datasources.RedisConnector) mock_redis.get = AsyncMock(return_value=None) mock_redis.set = AsyncMock(return_value=None) mocker.patch( "product_staging.api.datasources.get_redis_client", return_value=mock_redis, ) # Mock verify_profile_headers to return success mock_response = mocker.Mock() mock_response.status = 200 mocker.patch( "product_staging.api.auth.verify_profile_headers", return_value=mock_response, ) # Mock get_vendor_id to return vendor_id mocker.patch( "product_staging.connectors.ows_account.get_vendor_id", return_value=vendor_id, ) # Mock check_profile_access to return failure mock_access_response = mocker.Mock() mock_access_response.status = 403 mocker.patch( "product_staging.logic.profile.check_profile_access", return_value=mock_access_response, ) with pytest.raises(HTTPException) as exc: await auth.check_vendor_access_for_profiles( identity_id=UUID(identity_uuid), vendor_uuid=vendor_uuid, profiles=profiles, allowed_types=allowed_types, ) assert exc.value.status_code == 403 assert ERROR_MESSAGE_CANNOT_ACCESS_VENDOR in exc.value.detail async def test_check_vendor_access_with_multiple_profiles_first_valid( mocker: MockerFixture, identity_uuid: str ) -> None: """Test that check_vendor_access returns vendor_id when first profile is valid.""" vendor_uuid = UUID("35289af1-d291-44a9-96c1-72fce8ee1c48") vendor_id = 42 profiles = [("LabelProfile", 123), ("ContentProfile", 456)] allowed_types = ("LabelProfile", "ContentProfile") # Mock Redis client mock_redis = mocker.MagicMock(spec=datasources.RedisConnector) mock_redis.get = AsyncMock(return_value=None) mock_redis.set = AsyncMock(return_value=None) mocker.patch( "product_staging.api.datasources.get_redis_client", return_value=mock_redis, ) # Mock verify_profile_headers to return success mock_response = mocker.Mock() mock_response.status = 200 mocker.patch( "product_staging.api.auth.verify_profile_headers", return_value=mock_response, ) # Mock get_vendor_id to return vendor_id mock_get_vendor = mocker.patch( "product_staging.connectors.ows_account.get_vendor_id", return_value=vendor_id, ) # Mock check_profile_access to return success mock_access_response = mocker.Mock() mock_access_response.status = 200 mock_check_access = mocker.patch( "product_staging.logic.profile.check_profile_access", return_value=mock_access_response, ) result = await auth.check_vendor_access_for_profiles( identity_id=UUID(identity_uuid), vendor_uuid=vendor_uuid, profiles=profiles, allowed_types=allowed_types, ) assert result == vendor_id # Should only call once since first profile is valid assert mock_get_vendor.call_count == 1 assert mock_check_access.call_count == 1 async def test_check_vendor_access_with_multiple_profiles_second_valid( mocker: MockerFixture, identity_uuid: str ) -> None: """Test that check_vendor_access tries multiple profiles and succeeds on second.""" vendor_uuid = UUID("35289af1-d291-44a9-96c1-72fce8ee1c48") vendor_id = 42 profiles = [("LabelProfile", 123), ("ContentProfile", 456)] allowed_types = ("LabelProfile", "ContentProfile") # Mock Redis client mock_redis = mocker.MagicMock(spec=datasources.RedisConnector) mock_redis.get = AsyncMock(return_value=None) mock_redis.set = AsyncMock(return_value=None) mocker.patch( "product_staging.api.datasources.get_redis_client", return_value=mock_redis, ) # Mock verify_profile_headers to fail first, succeed second call_count = [0] def mock_verify(profile_type, profile_id, required, allowed_types): call_count[0] += 1 mock_response = mocker.Mock() if call_count[0] == 1: mock_response.status = 403 # First profile fails else: mock_response.status = 200 # Second profile succeeds return mock_response mocker.patch( "product_staging.api.auth.verify_profile_headers", side_effect=mock_verify, ) # Mock get_vendor_id to return vendor_id mock_get_vendor = mocker.patch( "product_staging.connectors.ows_account.get_vendor_id", return_value=vendor_id, ) # Mock check_profile_access to return success mock_access_response = mocker.Mock() mock_access_response.status = 200 mock_check_access = mocker.patch( "product_staging.logic.profile.check_profile_access", return_value=mock_access_response, ) result = await auth.check_vendor_access_for_profiles( identity_id=UUID(identity_uuid), vendor_uuid=vendor_uuid, profiles=profiles, allowed_types=allowed_types, ) assert result == vendor_id # Should call once for second profile (first profile failed header check) assert mock_get_vendor.call_count == 1 assert mock_check_access.call_count == 1 async def test_check_vendor_access_with_empty_profiles( mocker: MockerFixture, identity_uuid: str ) -> None: """Test that check_vendor_access raises 403 when profiles list is empty.""" vendor_uuid = UUID("35289af1-d291-44a9-96c1-72fce8ee1c48") allowed_types = ("LabelProfile", "ContentProfile") # Mock Redis client mock_redis = mocker.MagicMock(spec=datasources.RedisConnector) mock_redis.get = AsyncMock(return_value=None) mock_redis.set = AsyncMock(return_value=None) mocker.patch( "product_staging.api.datasources.get_redis_client", return_value=mock_redis, ) mocker.patch( "product_staging.connectors.ows_account.get_vendor_id", return_value=42, ) with pytest.raises(HTTPException) as exc: await auth.check_vendor_access_for_profiles( identity_id=UUID(identity_uuid), vendor_uuid=vendor_uuid, profiles=[], allowed_types=allowed_types, ) assert exc.value.status_code == 403 assert ERROR_MESSAGE_CANNOT_ACCESS_VENDOR in exc.value.detail async def test_check_vendor_access_calls_functions_with_correct_params( mocker: MockerFixture, identity_uuid: str ) -> None: """Test that check_vendor_access calls dependent functions with correct parameters.""" vendor_uuid = UUID("35289af1-d291-44a9-96c1-72fce8ee1c48") vendor_id = 42 profile_type = "LabelProfile" profile_id = 123 profiles = [(profile_type, profile_id)] allowed_types = ("LabelProfile", "ContentProfile") # Mock Redis client mock_redis = mocker.MagicMock(spec=datasources.RedisConnector) mock_redis.get = AsyncMock(return_value=None) mock_redis.set = AsyncMock(return_value=None) mocker.patch( "product_staging.api.datasources.get_redis_client", return_value=mock_redis, ) # Mock verify_profile_headers mock_response = mocker.Mock() mock_response.status = 200 mock_verify = mocker.patch( "product_staging.api.auth.verify_profile_headers", return_value=mock_response, ) # Mock get_vendor_id mock_get_vendor = mocker.patch( "product_staging.connectors.ows_account.get_vendor_id", return_value=vendor_id, ) # Mock check_profile_access mock_access_response = mocker.Mock() mock_access_response.status = 200 mock_check_access = mocker.patch( "product_staging.logic.profile.check_profile_access", return_value=mock_access_response, ) result = await auth.check_vendor_access_for_profiles( identity_id=UUID(identity_uuid), vendor_uuid=vendor_uuid, profiles=profiles, allowed_types=allowed_types, ) # Verify verify_profile_headers was called correctly mock_verify.assert_called_once_with( profile_type, profile_id, required=True, allowed_types=allowed_types, ) # Verify get_vendor_id was called correctly mock_get_vendor.assert_called_once_with(vendor_uuid) # Verify check_profile_access was called correctly mock_check_access.assert_called_once_with( identity_id=UUID(identity_uuid), profile_id=profile_id, profile_type=profile_type, vendor_id=vendor_id, subaccount_id=None, ) assert result == vendor_id async def test_check_vendor_access_with_subaccount( mocker: MockerFixture, identity_uuid: str ) -> None: """Test that check_vendor_access calls dependent functions with correct parameters when subaccount is passed.""" subaccount_id = 12345 profile_type = "LabelProfile" profile_id = 123 profiles = [(profile_type, profile_id)] allowed_types = ("LabelProfile", "ContentProfile") # Mock Redis client mock_redis = mocker.MagicMock(spec=datasources.RedisConnector) mock_redis.get = AsyncMock(return_value=None) mock_redis.set = AsyncMock(return_value=None) mocker.patch( "product_staging.api.datasources.get_redis_client", return_value=mock_redis, ) # Mock verify_profile_headers mock_response = mocker.Mock() mock_response.status = 200 mock_verify = mocker.patch( "product_staging.api.auth.verify_profile_headers", return_value=mock_response, ) # Mock check_profile_access mock_access_response = mocker.Mock() mock_access_response.status = 200 mock_check_access = mocker.patch( "product_staging.logic.profile.check_profile_access", return_value=mock_access_response, ) result = await auth.check_vendor_access_for_profiles( identity_id=UUID(identity_uuid), vendor_uuid=None, profiles=profiles, allowed_types=allowed_types, subaccount_id=subaccount_id, ) # Verify verify_profile_headers was called correctly mock_verify.assert_called_once_with( profile_type, profile_id, required=True, allowed_types=allowed_types, ) # Verify check_profile_access was called correctly mock_check_access.assert_called_once_with( identity_id=UUID(identity_uuid), profile_id=profile_id, profile_type=profile_type, vendor_id=None, subaccount_id=subaccount_id, ) assert result == subaccount_id async def test_check_vendor_access_uses_cache_on_hit( mocker: MockerFixture, identity_uuid: str ) -> None: """Test that check_vendor_access_for_profiles uses cached validation result when available.""" vendor_uuid = UUID("35289af1-d291-44a9-96c1-72fce8ee1c48") vendor_id = 42 profile_type = "LabelProfile" profile_id = 123 profiles = [(profile_type, profile_id)] allowed_types = ("LabelProfile", "ContentProfile") # Mock verify_profile_headers to return success mock_response = mocker.Mock() mock_response.status = 200 mocker.patch( "product_staging.api.auth.verify_profile_headers", return_value=mock_response, ) # Mock get_vendor_id to return vendor_id mocker.patch( "product_staging.connectors.ows_account.get_vendor_id", return_value=vendor_id, ) # Mock Redis client with cached successful validation mock_redis = mocker.MagicMock(spec=datasources.RedisConnector) mock_redis.get = AsyncMock(return_value={"status": 200}) mock_redis.set = AsyncMock(return_value=None) mocker.patch( "product_staging.api.datasources.get_redis_client", return_value=mock_redis, ) # Mock check_profile_access (should NOT be called due to cache hit) mock_check_access = mocker.patch( "product_staging.logic.profile.check_profile_access" ) result = await auth.check_vendor_access_for_profiles( identity_id=UUID(identity_uuid), vendor_uuid=vendor_uuid, profiles=profiles, allowed_types=allowed_types, ) assert result == vendor_id # Verify cache was checked expected_cache_key = ( f"/profile/access/{identity_uuid}/{profile_type}/{profile_id}/{vendor_id}" ) mock_redis.get.assert_called_once_with(expected_cache_key) # Verify check_profile_access was NOT called (cache hit) mock_check_access.assert_not_called() # Verify cache was NOT written (already existed) mock_redis.set.assert_not_called() async def test_check_vendor_access_caches_validation_result_on_miss( mocker: MockerFixture, identity_uuid: str ) -> None: """Test that check_vendor_access_for_profiles caches validation result on cache miss.""" vendor_uuid = UUID("35289af1-d291-44a9-96c1-72fce8ee1c48") vendor_id = 42 profile_type = "LabelProfile" profile_id = 123 profiles = [(profile_type, profile_id)] allowed_types = ("LabelProfile", "ContentProfile") # Mock verify_profile_headers to return success mock_response = mocker.Mock() mock_response.status = 200 mocker.patch( "product_staging.api.auth.verify_profile_headers", return_value=mock_response, ) # Mock get_vendor_id to return vendor_id mocker.patch( "product_staging.connectors.ows_account.get_vendor_id", return_value=vendor_id, ) # Mock Redis client with no cached data (cache miss) mock_redis = mocker.MagicMock(spec=datasources.RedisConnector) mock_redis.get = AsyncMock(return_value=None) mock_redis.set = AsyncMock(return_value=None) mocker.patch( "product_staging.api.datasources.get_redis_client", return_value=mock_redis, ) # Mock check_profile_access to return success mock_access_response = mocker.Mock() mock_access_response.status = 200 mock_check_access = mocker.patch( "product_staging.logic.profile.check_profile_access", return_value=mock_access_response, ) result = await auth.check_vendor_access_for_profiles( identity_id=UUID(identity_uuid), vendor_uuid=vendor_uuid, profiles=profiles, allowed_types=allowed_types, ) assert result == vendor_id # Verify cache was checked expected_cache_key = ( f"/profile/access/{identity_uuid}/{profile_type}/{profile_id}/{vendor_id}" ) mock_redis.get.assert_called_once_with(expected_cache_key) # Verify check_profile_access WAS called (cache miss) mock_check_access.assert_called_once_with( identity_id=UUID(identity_uuid), profile_id=profile_id, profile_type=profile_type, vendor_id=vendor_id, subaccount_id=None, ) # Verify validation result was cached mock_redis.set.assert_called_once_with(expected_cache_key, {"status": 200}) async def test_check_vendor_access_caches_failed_validation( mocker: MockerFixture, identity_uuid: str ) -> None: """Test that check_vendor_access_for_profiles caches failed validation results.""" vendor_uuid = UUID("35289af1-d291-44a9-96c1-72fce8ee1c48") vendor_id = 42 profile_type = "LabelProfile" profile_id = 123 profiles = [(profile_type, profile_id)] allowed_types = ("LabelProfile", "ContentProfile") # Mock verify_profile_headers to return success mock_response = mocker.Mock() mock_response.status = 200 mocker.patch( "product_staging.api.auth.verify_profile_headers", return_value=mock_response, ) # Mock get_vendor_id to return vendor_id mocker.patch( "product_staging.connectors.ows_account.get_vendor_id", return_value=vendor_id, ) # Mock Redis client with no cached data mock_redis = mocker.MagicMock(spec=datasources.RedisConnector) mock_redis.get = AsyncMock(return_value=None) mock_redis.set = AsyncMock(return_value=None) mocker.patch( "product_staging.api.datasources.get_redis_client", return_value=mock_redis, ) # Mock check_profile_access to return failure mock_access_response = mocker.Mock() mock_access_response.status = 403 mocker.patch( "product_staging.logic.profile.check_profile_access", return_value=mock_access_response, ) with pytest.raises(HTTPException) as exc: await auth.check_vendor_access_for_profiles( identity_id=UUID(identity_uuid), vendor_uuid=vendor_uuid, profiles=profiles, allowed_types=allowed_types, ) assert exc.value.status_code == 403 # Verify failed validation result was still cached expected_cache_key = ( f"/profile/access/{identity_uuid}/{profile_type}/{profile_id}/{vendor_id}" ) mock_redis.set.assert_called_once_with(expected_cache_key, {"status": 403}) @pytest.mark.parametrize( "vendor_uuid, subaccount_id, lookup_result, expected_tenant, expected_exception", [ pytest.param( None, None, None, None, None, id="returns none when no vendor or subaccount is provided", ), pytest.param( UUID("11111111-1111-4111-8111-111111111111"), None, None, { "tenant_type": "account", "tenant_uuid": "11111111-1111-4111-8111-111111111111", "tenant_attributes": {}, }, None, id="returns account tenant when only vendor is provided", ), pytest.param( None, 42, [ { "uuid": "22222222-2222-4222-8222-222222222222", "parent_company_uuid": "p-1", "company_brand_uuid": "b-1", "vendor_uuid": "v-1", } ], { "tenant_type": "subaccount", "tenant_uuid": "22222222-2222-4222-8222-222222222222", "tenant_attributes": { "tenant_hierarchy": ["p-1", "b-1", "v-1"], }, }, None, id="returns subaccount tenant when subaccount lookup succeeds", ), pytest.param( None, 42, [], None, None, id="returns none when subaccount lookup returns empty list", ), pytest.param( None, 42, [None], None, None, id="returns none when first subaccount result is none", ), pytest.param( None, 42, [{"parent_company_uuid": "p-1"}], None, None, id="returns none when first subaccount result has no uuid", ), pytest.param( UUID("11111111-1111-4111-8111-111111111111"), 42, [ { "uuid": "22222222-2222-4222-8222-222222222222", "parent_company_uuid": "p-1", "company_brand_uuid": "b-1", "vendor_uuid": "v-1", } ], { "tenant_type": "subaccount", "tenant_uuid": "22222222-2222-4222-8222-222222222222", "tenant_attributes": { "tenant_hierarchy": ["p-1", "b-1", "v-1"], }, }, None, id="prefers subaccount tenant when both are provided and subaccount exists", ), pytest.param( UUID("11111111-1111-4111-8111-111111111111"), 42, [], { "tenant_type": "account", "tenant_uuid": "11111111-1111-4111-8111-111111111111", "tenant_attributes": {}, }, None, id="falls back to account tenant when subaccount lookup is empty", ), pytest.param( UUID("11111111-1111-4111-8111-111111111111"), 42, [None], { "tenant_type": "account", "tenant_uuid": "11111111-1111-4111-8111-111111111111", "tenant_attributes": {}, }, None, id="falls back to account tenant when subaccount lookup's first element is None", ), pytest.param( UUID("11111111-1111-4111-8111-111111111111"), 42, [{"no": "uuid"}], { "tenant_type": "account", "tenant_uuid": "11111111-1111-4111-8111-111111111111", "tenant_attributes": {}, }, None, id="falls back to account tenant when subaccount lookup's first element has no uuid", ), pytest.param( UUID("11111111-1111-4111-8111-111111111111"), 0, None, { "tenant_type": "account", "tenant_uuid": "11111111-1111-4111-8111-111111111111", "tenant_attributes": {}, }, None, id="treats subaccount id zero as falsy and skips subaccount lookup", ), pytest.param( None, 42, [ { "uuid": "22222222-2222-4222-8222-222222222222", "parent_company_uuid": "p-1", } ], { "tenant_type": "subaccount", "tenant_uuid": "22222222-2222-4222-8222-222222222222", "tenant_attributes": {}, }, None, id="omits tenant hierarchy when any hierarchy fields are missing", ), pytest.param( None, 42, [ { "uuid": "22222222-2222-4222-8222-222222222222", } ], { "tenant_type": "subaccount", "tenant_uuid": "22222222-2222-4222-8222-222222222222", "tenant_attributes": {}, }, None, id="omits tenant hierarchy when all hierarchy fields are missing", ), ], ) async def test_get_tenant( mocker: MockerFixture, vendor_uuid: UUID | None, subaccount_id: int | None, lookup_result: list[dict[str, Any] | None] | None, expected_tenant: dict[str, Any] | None, expected_exception: type[Exception] | None, ) -> None: """Test get_tenant behavior across account/subaccount branches and edge cases.""" lookup_subaccounts_by_id = mocker.patch( "product_staging.api.auth.ows_account.lookup_subaccounts_by_id", new=AsyncMock(return_value=lookup_result), ) if expected_exception: with pytest.raises(expected_exception): await auth.get_tenant(vendor_uuid=vendor_uuid, subaccount_id=subaccount_id) else: tenant = await auth.get_tenant( vendor_uuid=vendor_uuid, subaccount_id=subaccount_id ) if expected_tenant is None: assert tenant is None else: assert tenant is not None assert tenant.model_dump(mode="json") == expected_tenant if subaccount_id: lookup_subaccounts_by_id.assert_awaited_once_with([subaccount_id]) else: lookup_subaccounts_by_id.assert_not_awaited() @pytest.mark.parametrize( "is_authorized_value", [ pytest.param(True, id="identity is authorized"), pytest.param(False, id="identity is not authorized"), ], ) def test_is_authorized_for_tenant_returns_authorization_backend_result( mocker: MockerFixture, is_authorized_value: bool, ) -> None: """Test is_authorized_for_tenant uses authorization_backend.""" tenant_uuid = UUID("35289af1-d291-44a9-96c1-72fce8ee1c48") authorization_backend = mocker.MagicMock() authorization_backend.is_authorized.return_value = is_authorized_value mocker.patch( "product_staging.api.auth.datasources.get_authorization_backend", return_value=authorization_backend, ) forward_kwargs_getter = mocker.MagicMock() mocker.patch( "product_staging.api.auth.ForwardKwargsGetter", return_value=forward_kwargs_getter, ) result = auth.is_authorized_for_tenant(tenant_uuid=tenant_uuid) assert result == is_authorized_value authorization_backend.is_authorized.assert_called_once_with( action="bulk_create", resource_id=0, resource_type="digital_audio", resource_getter=forward_kwargs_getter, tenant={ "tenant_type": "account", "tenant_uuid": str(tenant_uuid), }, ) def test_is_authorized_for_tenant_uses_non_default_params( mocker: MockerFixture, ) -> None: """Test is_authorized_for_tenant forwards non-default function params to authorization backend.""" tenant_uuid = UUID("35289af1-d291-44a9-96c1-72fce8ee1c48") authorization_backend = mocker.Mock() authorization_backend.is_authorized.return_value = True mocker.patch( "product_staging.api.auth.datasources.get_authorization_backend", return_value=authorization_backend, ) forward_kwargs_getter = mocker.Mock() mocker.patch( "product_staging.api.auth.ForwardKwargsGetter", return_value=forward_kwargs_getter, ) auth.is_authorized_for_tenant( tenant_uuid=tenant_uuid, tenant_type="subaccount", resource_type="dog", action="hug", tenant_attributes={ "tenant_hierarchy": [1, 2, 3], "fancy": True, "tenant_type": "not-subaccount-this-is-naughty-client-behavior", }, ) authorization_backend.is_authorized.assert_called_once_with( action="hug", resource_id=0, resource_type="dog", resource_getter=forward_kwargs_getter, tenant={ "tenant_uuid": str(tenant_uuid), "tenant_hierarchy": [1, 2, 3], "fancy": True, "tenant_type": "not-subaccount-this-is-naughty-client-behavior", }, ) def test_is_authorized_for_tenant_raises_401_for_unauthenticated( mocker: MockerFixture, ) -> None: """Test is_authorized_for_tenant handles authorization_backend Unauthenticated.""" tenant_uuid = UUID("35289af1-d291-44a9-96c1-72fce8ee1c48") authorization_backend = mocker.Mock() authorization_backend.is_authorized.side_effect = UnauthenticatedException mocker.patch( "product_staging.api.auth.datasources.get_authorization_backend", return_value=authorization_backend, ) forward_kwargs_getter = mocker.Mock() mocker.patch( "product_staging.api.auth.ForwardKwargsGetter", return_value=forward_kwargs_getter, ) with pytest.raises(HTTPException) as exc: auth.is_authorized_for_tenant(tenant_uuid=tenant_uuid) assert exc.value.status_code == 401 assert exc.value.detail == auth.error.ERROR_MESSAGE_NOT_AUTHENTICATED authorization_backend.is_authorized.assert_called_once_with( action="bulk_create", resource_id=0, resource_type="digital_audio", resource_getter=forward_kwargs_getter, tenant={ "tenant_type": "account", "tenant_uuid": str(tenant_uuid), }, ) def test_is_authorized_for_tenants_returns_authorization_backend_results( mocker: MockerFixture, ) -> None: """Test is_authorized_for_tenants uses authorization_backend.is_authorized_many.""" tenant_uuid_1 = UUID("35289af1-d291-44a9-96c1-72fce8ee1c48") tenant_uuid_2 = UUID("ec1fd7e2-9c95-4e09-a037-e924e0244283") bulk_session_id_1 = UUID("11111111-1111-1111-1111-111111111111") bulk_session_id_2 = UUID("22222222-2222-2222-2222-222222222222") tenants = [ Tenant(tenant_type="account", tenant_uuid=tenant_uuid_1), Tenant(tenant_type="account", tenant_uuid=tenant_uuid_2), ] bulk_session_ids = [bulk_session_id_1, bulk_session_id_2] authorization_backend = mocker.MagicMock() authorization_backend.is_authorized_many.return_value = [True, False] mocker.patch( "product_staging.api.auth.datasources.get_authorization_backend", return_value=authorization_backend, ) result = auth.is_authorized_for_tenants( tenants=tenants, bulk_session_ids=bulk_session_ids ) assert result == [True, False] authorization_backend.is_authorized_many.assert_called_once_with( action="bulk_create", resource_type="digital_audio", resources_with_attributes=[ ResourceWithAttributes( resource_id=f"bulk_session_for_{tenant_uuid_1}", attributes={ "bulk_session_id": str(bulk_session_id_1), "tenant": { "tenant_type": "account", "tenant_uuid": str(tenant_uuid_1), }, }, ), ResourceWithAttributes( resource_id=f"bulk_session_for_{tenant_uuid_2}", attributes={ "bulk_session_id": str(bulk_session_id_2), "tenant": { "tenant_type": "account", "tenant_uuid": str(tenant_uuid_2), }, }, ), ], ) def test_is_authorized_for_tenants_uses_non_default_params( mocker: MockerFixture, ) -> None: """Test is_authorized_for_tenants forwards non-default params and tenant_attributes to authorization backend. Reserved keys (tenant_type, tenant_uuid) on tenant_attributes must not override the values declared on the Tenant model. """ tenant_uuid = UUID("35289af1-d291-44a9-96c1-72fce8ee1c48") bulk_session_id = UUID("11111111-1111-1111-1111-111111111111") tenants = [ Tenant( tenant_type="subaccount", tenant_uuid=tenant_uuid, tenant_attributes={ "tenant_hierarchy": [1, 2, 3], "fancy": True, "tenant_type": "not-subaccount-this-is-naughty-client-behavior", "tenant_uuid": "00000000-0000-0000-0000-000000000000", }, ), ] authorization_backend = mocker.MagicMock() authorization_backend.is_authorized_many.return_value = [True] mocker.patch( "product_staging.api.auth.datasources.get_authorization_backend", return_value=authorization_backend, ) result = auth.is_authorized_for_tenants( tenants=tenants, bulk_session_ids=[bulk_session_id], resource_type="flower", action="sniff", ) assert result == [True] authorization_backend.is_authorized_many.assert_called_once_with( action="sniff", resource_type="flower", resources_with_attributes=[ ResourceWithAttributes( resource_id=f"bulk_session_for_{tenant_uuid}", attributes={ "bulk_session_id": str(bulk_session_id), "tenant": { "tenant_hierarchy": [1, 2, 3], "fancy": True, "tenant_type": "subaccount", "tenant_uuid": str(tenant_uuid), }, }, ), ], ) def test_is_authorized_for_tenants_with_empty_list( mocker: MockerFixture, ) -> None: """Test is_authorized_for_tenants passes through an empty tenant list.""" authorization_backend = mocker.MagicMock() authorization_backend.is_authorized_many.return_value = [] mocker.patch( "product_staging.api.auth.datasources.get_authorization_backend", return_value=authorization_backend, ) result = auth.is_authorized_for_tenants(tenants=[], bulk_session_ids=[]) assert result == [] authorization_backend.is_authorized_many.assert_called_once_with( action="bulk_create", resource_type="digital_audio", resources_with_attributes=[], ) def test_is_authorized_for_tenants_raises_401_for_unauthenticated( mocker: MockerFixture, ) -> None: """Test is_authorized_for_tenants handles authorization_backend Unauthenticated.""" tenant_uuid = UUID("35289af1-d291-44a9-96c1-72fce8ee1c48") bulk_session_id = UUID("11111111-1111-1111-1111-111111111111") tenants = [Tenant(tenant_type="account", tenant_uuid=tenant_uuid)] authorization_backend = mocker.MagicMock() authorization_backend.is_authorized_many.side_effect = UnauthenticatedException mocker.patch( "product_staging.api.auth.datasources.get_authorization_backend", return_value=authorization_backend, ) with pytest.raises(HTTPException) as exc: auth.is_authorized_for_tenants( tenants=tenants, bulk_session_ids=[bulk_session_id] ) assert exc.value.status_code == 401 assert exc.value.detail == auth.error.ERROR_MESSAGE_NOT_AUTHENTICATED authorization_backend.is_authorized_many.assert_called_once_with( action="bulk_create", resource_type="digital_audio", resources_with_attributes=[ ResourceWithAttributes( resource_id=f"bulk_session_for_{tenant_uuid}", attributes={ "bulk_session_id": str(bulk_session_id), "tenant": { "tenant_type": "account", "tenant_uuid": str(tenant_uuid), }, }, ), ], ) def test_lookup_subaccount_to_tenant_full_hierarchy() -> None: """Full hierarchy is forwarded in tenant_attributes.""" subaccount = { "uuid": "aaaaaaaa-0000-4000-8000-000000000001", "subaccount_id": 1, "parent_company_uuid": "bbbbbbbb-0000-0000-0000-000000000002", "company_brand_uuid": "cccccccc-0000-0000-0000-000000000003", "vendor_uuid": "dddddddd-0000-0000-0000-000000000004", } tenant = auth._lookup_subaccount_to_tenant(subaccount) assert tenant.tenant_uuid == UUID("aaaaaaaa-0000-4000-8000-000000000001") assert tenant.tenant_type == "subaccount" assert tenant.tenant_attributes == { "tenant_hierarchy": [ "bbbbbbbb-0000-0000-0000-000000000002", "cccccccc-0000-0000-0000-000000000003", "dddddddd-0000-0000-0000-000000000004", ] } def test_lookup_subaccount_to_tenant_partial_hierarchy() -> None: """Partial hierarchy (any None) results in empty tenant_attributes.""" subaccount = { "uuid": "aaaaaaaa-0000-4000-8000-000000000001", "subaccount_id": 1, "parent_company_uuid": None, "company_brand_uuid": "cccccccc-0000-0000-0000-000000000003", "vendor_uuid": "dddddddd-0000-0000-0000-000000000004", } tenant = auth._lookup_subaccount_to_tenant(subaccount) assert tenant.tenant_uuid == UUID("aaaaaaaa-0000-4000-8000-000000000001") assert tenant.tenant_type == "subaccount" assert tenant.tenant_attributes == {} @pytest.mark.parametrize( "requests, lookup_result, expected_tenants", [ pytest.param( [], None, [], id="empty input returns empty list without calling lookup", ), pytest.param( [ TenantLookup( vendor_uuid=UUID("11111111-1111-4111-8111-111111111111"), subaccount_id=None, ) ], None, [ { "tenant_type": "account", "tenant_uuid": "11111111-1111-4111-8111-111111111111", "tenant_attributes": {}, } ], id="account-only request resolved without subaccount lookup", ), pytest.param( [TenantLookup(vendor_uuid=None, subaccount_id=None)], None, [None], id="no vendor and no subaccount returns None", ), pytest.param( [TenantLookup(vendor_uuid=None, subaccount_id=42)], [ { "subaccount_id": 42, "uuid": "22222222-2222-4222-8222-222222222222", "parent_company_uuid": "p-1", "company_brand_uuid": "b-1", "vendor_uuid": "v-1", } ], [ { "tenant_type": "subaccount", "tenant_uuid": "22222222-2222-4222-8222-222222222222", "tenant_attributes": {"tenant_hierarchy": ["p-1", "b-1", "v-1"]}, } ], id="subaccount request resolved with full hierarchy", ), pytest.param( [TenantLookup(vendor_uuid=None, subaccount_id=42)], [ { "subaccount_id": 42, "uuid": "22222222-2222-4222-8222-222222222222", "parent_company_uuid": "p-1", } ], [ { "tenant_type": "subaccount", "tenant_uuid": "22222222-2222-4222-8222-222222222222", "tenant_attributes": {}, } ], id="omits tenant_hierarchy when any hierarchy field is missing", ), pytest.param( [ TenantLookup( vendor_uuid=UUID("11111111-1111-4111-8111-111111111111"), subaccount_id=42, ) ], [], [ { "tenant_type": "account", "tenant_uuid": "11111111-1111-4111-8111-111111111111", "tenant_attributes": {}, } ], id="falls back to account when subaccount lookup returns empty", ), pytest.param( [ TenantLookup( vendor_uuid=UUID("11111111-1111-4111-8111-111111111111"), subaccount_id=42, ) ], [None], [ { "tenant_type": "account", "tenant_uuid": "11111111-1111-4111-8111-111111111111", "tenant_attributes": {}, } ], id="falls back to account when subaccount lookup's first element is None", ), pytest.param( [ TenantLookup( vendor_uuid=UUID("11111111-1111-4111-8111-111111111111"), subaccount_id=42, ) ], [{"no": "uuid"}], [ { "tenant_type": "account", "tenant_uuid": "11111111-1111-4111-8111-111111111111", "tenant_attributes": {}, } ], id="falls back to account when subaccount lookup's first element has no uuid", ), pytest.param( [TenantLookup(vendor_uuid=None, subaccount_id=42)], [{"subaccount_id": 42}], [None], id="returns None when subaccount lookup result has no uuid", ), ], ) async def test_get_tenants_many( mocker: MockerFixture, requests: list[TenantLookup], lookup_result: list | None, expected_tenants: list, ) -> None: """Test get_tenants_many across account/subaccount branches and edge cases.""" mock_lookup = mocker.patch( "product_staging.api.auth.ows_account.lookup_subaccounts_by_id", new=AsyncMock(return_value=lookup_result or []), ) result = await auth.get_tenants_many(requests) assert len(result) == len(expected_tenants) for tenant, expected in zip(result, expected_tenants): if expected is None: assert tenant is None else: assert tenant is not None assert tenant.model_dump(mode="json") == expected has_subaccount = any(r.subaccount_id for r in requests) if has_subaccount: mock_lookup.assert_awaited_once() else: mock_lookup.assert_not_awaited() async def test_get_tenants_many_deduplicates_subaccount_ids( mocker: MockerFixture, ) -> None: """Test that two requests for the same subaccount_id produce only one lookup call.""" subaccount_data = { "subaccount_id": 42, "uuid": "22222222-2222-4222-8222-222222222222", "parent_company_uuid": "p-1", "company_brand_uuid": "b-1", "vendor_uuid": "v-1", } mock_lookup = mocker.patch( "product_staging.api.auth.ows_account.lookup_subaccounts_by_id", new=AsyncMock(return_value=[subaccount_data]), ) result = await auth.get_tenants_many( [ TenantLookup(vendor_uuid=None, subaccount_id=42), TenantLookup(vendor_uuid=None, subaccount_id=42), ] ) mock_lookup.assert_awaited_once_with([42]) assert len(result) == 2 assert all(t is not None and t.tenant_type == "subaccount" for t in result) async def test_get_tenants_many_single_lookup_for_mixed_batch( mocker: MockerFixture, ) -> None: """Test that a mixed batch of account and subaccount requests issues one lookup call.""" mock_lookup = mocker.patch( "product_staging.api.auth.ows_account.lookup_subaccounts_by_id", new=AsyncMock( return_value=[ { "subaccount_id": 1, "uuid": "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", "parent_company_uuid": "pppppppp-pppp-pppp-pppp-pppppppppppp", "company_brand_uuid": "cccccccc-cccc-cccc-cccc-cccccccccccc", "vendor_uuid": "vvvvvvvv-vvvv-vvvv-vvvv-vvvvvvvvvvvv", }, { "subaccount_id": 2, "uuid": "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb", "parent_company_uuid": None, "company_brand_uuid": None, "vendor_uuid": None, }, ] ), ) result = await auth.get_tenants_many( [ TenantLookup( vendor_uuid=UUID("11111111-1111-4111-8111-111111111111"), subaccount_id=None, ), TenantLookup(vendor_uuid=None, subaccount_id=1), TenantLookup( vendor_uuid=UUID("33333333-3333-4333-8333-333333333333"), subaccount_id=None, ), TenantLookup(vendor_uuid=None, subaccount_id=2), ] ) mock_lookup.assert_awaited_once_with([1, 2]) assert result == [ Tenant( tenant_uuid=UUID("11111111-1111-4111-8111-111111111111"), tenant_type="account", ), Tenant( tenant_uuid=UUID("aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"), tenant_type="subaccount", tenant_attributes={ "tenant_hierarchy": [ "pppppppp-pppp-pppp-pppp-pppppppppppp", "cccccccc-cccc-cccc-cccc-cccccccccccc", "vvvvvvvv-vvvv-vvvv-vvvv-vvvvvvvvvvvv", ] }, ), Tenant( tenant_uuid=UUID("33333333-3333-4333-8333-333333333333"), tenant_type="account", ), Tenant( tenant_uuid=UUID("bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb"), tenant_type="subaccount", ), ] def test_is_authorized_for_tenants_raises_500_on_length_mismatch( mocker: MockerFixture, caplog ) -> None: """tenants and bulk_session_ids must be the same length — caller bug should surface as 500.""" tenant_uuid = UUID("35289af1-d291-44a9-96c1-72fce8ee1c48") tenants = [Tenant(tenant_type="account", tenant_uuid=tenant_uuid)] bulk_session_ids = [ UUID("11111111-1111-4111-8111-111111111111"), UUID("22222222-2222-4222-8222-222222222222"), ] authorization_backend = mocker.MagicMock() mocker.patch( "product_staging.api.auth.datasources.get_authorization_backend", return_value=authorization_backend, ) with pytest.raises(HTTPException) as exc: auth.is_authorized_for_tenants( tenants=tenants, bulk_session_ids=bulk_session_ids ) assert exc.value.status_code == 500 assert exc.value.detail == "Internal server error" assert "tenants has length: 1 but bulk_session_ids has length: 2" in caplog.text authorization_backend.is_authorized_many.assert_not_called() @pytest.mark.parametrize( "is_authorized_value", [ pytest.param(True, id="identity is authorized"), pytest.param(False, id="identity is not authorized"), ], ) def test_is_authorized_for_tenant_id_returns_authorization_backend_result( mocker: MockerFixture, is_authorized_value: bool, ) -> None: """Test is_authorized_for_tenant_id uses authorization_backend.""" authorization_backend = mocker.MagicMock() authorization_backend.is_authorized.return_value = is_authorized_value mocker.patch( "product_staging.api.auth.datasources.get_authorization_backend", return_value=authorization_backend, ) forward_kwargs_getter = mocker.MagicMock() mocker.patch( "product_staging.api.auth.ForwardKwargsGetter", return_value=forward_kwargs_getter, ) result = auth.is_authorized_for_tenant_id(tenant_id=123) assert result == is_authorized_value authorization_backend.is_authorized.assert_called_once_with( action="bulk_create", resource_id=0, resource_type="digital_audio", resource_getter=forward_kwargs_getter, id_to_uuid_exchange_tenant={ "tenant_type": "account", "tenant_id": 123, }, ) def test_is_authorized_for_tenant_id_uses_non_default_params( mocker: MockerFixture, ) -> None: """Test is_authorized_for_tenant_id forwards non-default params to authorization backend.""" authorization_backend = mocker.Mock() authorization_backend.is_authorized.return_value = True mocker.patch( "product_staging.api.auth.datasources.get_authorization_backend", return_value=authorization_backend, ) forward_kwargs_getter = mocker.Mock() mocker.patch( "product_staging.api.auth.ForwardKwargsGetter", return_value=forward_kwargs_getter, ) auth.is_authorized_for_tenant_id( tenant_id=456, tenant_type="subaccount", resource_type="dog", action="hug", tenant_attributes={ "tenant_hierarchy": [1, 2, 3], "fancy": True, }, ) authorization_backend.is_authorized.assert_called_once_with( action="hug", resource_id=0, resource_type="dog", resource_getter=forward_kwargs_getter, id_to_uuid_exchange_tenant={ "tenant_type": "subaccount", "tenant_id": 456, "tenant_hierarchy": [1, 2, 3], "fancy": True, }, ) def test_is_authorized_for_tenant_id_raises_401_for_unauthenticated( mocker: MockerFixture, ) -> None: """Test is_authorized_for_tenant_id handles authorization_backend Unauthenticated.""" authorization_backend = mocker.Mock() authorization_backend.is_authorized.side_effect = UnauthenticatedException mocker.patch( "product_staging.api.auth.datasources.get_authorization_backend", return_value=authorization_backend, ) forward_kwargs_getter = mocker.Mock() mocker.patch( "product_staging.api.auth.ForwardKwargsGetter", return_value=forward_kwargs_getter, ) with pytest.raises(HTTPException) as exc: auth.is_authorized_for_tenant_id(tenant_id=789) assert exc.value.status_code == 401 assert exc.value.detail == auth.error.ERROR_MESSAGE_NOT_AUTHENTICATED authorization_backend.is_authorized.assert_called_once_with( action="bulk_create", resource_id=0, resource_type="digital_audio", resource_getter=forward_kwargs_getter, id_to_uuid_exchange_tenant={ "tenant_type": "account", "tenant_id": 789, }, ) def test_is_authorized_many_for_tenant_id_returns_authorization_backend_results( mocker: MockerFixture, ) -> None: """Test is_authorized_many_for_tenant_id uses authorization_backend.is_authorized_many.""" authorization_backend = mocker.MagicMock() authorization_backend.is_authorized_many.return_value = [True, False] mocker.patch( "product_staging.api.auth.datasources.get_authorization_backend", return_value=authorization_backend, ) result = auth.is_authorized_many_for_tenant_id( tenant_ids=[123, 456], tenant_type="account", ) assert result == [True, False] authorization_backend.is_authorized_many.assert_called_once_with( action="bulk_create", resource_type="digital_audio", resources_with_attributes=[ ResourceWithAttributes( resource_id="123", attributes={ "id_to_uuid_exchange_tenant": { "tenant_type": "account", "tenant_id": "123", } }, ), ResourceWithAttributes( resource_id="456", attributes={ "id_to_uuid_exchange_tenant": { "tenant_type": "account", "tenant_id": "456", } }, ), ], ) def test_is_authorized_many_for_tenant_id_uses_non_default_params( mocker: MockerFixture, ) -> None: """Test is_authorized_many_for_tenant_id forwards non-default params to authorization backend.""" authorization_backend = mocker.Mock() authorization_backend.is_authorized_many.return_value = [True] mocker.patch( "product_staging.api.auth.datasources.get_authorization_backend", return_value=authorization_backend, ) result = auth.is_authorized_many_for_tenant_id( tenant_ids=[789], tenant_type="subaccount", resource_type="dog", action="hug", ) assert result == [True] authorization_backend.is_authorized_many.assert_called_once_with( action="hug", resource_type="dog", resources_with_attributes=[ ResourceWithAttributes( resource_id="789", attributes={ "id_to_uuid_exchange_tenant": { "tenant_type": "subaccount", "tenant_id": "789", } }, ), ], ) def test_is_authorized_many_for_tenant_id_with_empty_list( mocker: MockerFixture, ) -> None: """Test is_authorized_many_for_tenant_id short-circuits without calling the backend for an empty list.""" authorization_backend = mocker.MagicMock() mocker.patch( "product_staging.api.auth.datasources.get_authorization_backend", return_value=authorization_backend, ) result = auth.is_authorized_many_for_tenant_id(tenant_ids=[], tenant_type="account") assert result == [] authorization_backend.is_authorized_many.assert_not_called() def test_is_authorized_many_for_tenant_id_raises_401_for_unauthenticated( mocker: MockerFixture, ) -> None: """Test is_authorized_many_for_tenant_id handles authorization_backend Unauthenticated.""" authorization_backend = mocker.MagicMock() authorization_backend.is_authorized_many.side_effect = UnauthenticatedException mocker.patch( "product_staging.api.auth.datasources.get_authorization_backend", return_value=authorization_backend, ) with pytest.raises(HTTPException) as exc: auth.is_authorized_many_for_tenant_id(tenant_ids=[123], tenant_type="account") assert exc.value.status_code == 401 assert exc.value.detail == auth.error.ERROR_MESSAGE_NOT_AUTHENTICATED