"""Unit tests for auth dependency helpers.""" import uuid from unittest.mock import MagicMock import pytest from fastapi import HTTPException from pytest_mock import MockerFixture from python_pdp_sdk.backends.authorization_backend import AuthorizationBackend from starlette.requests import Request from contributor.api.auth import ( PROFILES_CLAIM, USER_METADATA_CLAIM, identity_uuid_from_scope, is_authorized, is_oa_user, label_profile_ids_from_profiles, profiles_from_scope, ) _IDENTITY_UUID = str(uuid.uuid4()) _LABEL_PROFILE = { "profile_type": "LabelProfile", "profile_id": 1, "roles": ["administrator"], } def _make_request(scope_extras: dict) -> Request: scope = {"type": "http", "method": "GET", "path": "/", **scope_extras} return Request(scope) class TestIdentityUuidFromScope: def test_returns_uuid_when_present(self) -> None: request = _make_request( {"token": {USER_METADATA_CLAIM: {"orchardIdentityId": _IDENTITY_UUID}}} ) result = identity_uuid_from_scope(request) assert result == uuid.UUID(_IDENTITY_UUID) def test_raises_401_when_no_token(self) -> None: request = _make_request({}) with pytest.raises(HTTPException) as exc_info: identity_uuid_from_scope(request) assert exc_info.value.status_code == 401 def test_raises_401_when_claim_missing(self) -> None: request = _make_request({"token": {}}) with pytest.raises(HTTPException) as exc_info: identity_uuid_from_scope(request) assert exc_info.value.status_code == 401 def test_raises_401_when_uuid_malformed(self) -> None: request = _make_request( {"token": {USER_METADATA_CLAIM: {"orchardIdentityId": "not-a-uuid"}}} ) with pytest.raises(HTTPException) as exc_info: identity_uuid_from_scope(request) assert exc_info.value.status_code == 401 class TestProfilesFromScope: def test_returns_profiles_when_present(self) -> None: request = _make_request({"token": {PROFILES_CLAIM: [_LABEL_PROFILE]}}) result = profiles_from_scope(request) assert result == [_LABEL_PROFILE] def test_returns_empty_list_when_claim_absent(self) -> None: request = _make_request({"token": {}}) result = profiles_from_scope(request) assert result == [] def test_raises_401_when_no_token(self) -> None: request = _make_request({}) with pytest.raises(HTTPException) as exc_info: profiles_from_scope(request) assert exc_info.value.status_code == 401 class TestLabelProfileIdsFromProfiles: def test_returns_ids_for_label_profiles(self) -> None: profiles = [ {"profile_type": "LabelProfile", "profile_id": 1, "roles": []}, {"profile_type": "LabelProfile", "profile_id": 2, "roles": []}, ] assert label_profile_ids_from_profiles(profiles) == [1, 2] def test_excludes_non_label_profiles(self) -> None: profiles = [ {"profile_type": "OtherProfile", "profile_id": 99}, {"profile_type": "LabelProfile", "profile_id": 7}, ] assert label_profile_ids_from_profiles(profiles) == [7] def test_returns_empty_for_no_profiles(self) -> None: assert label_profile_ids_from_profiles([]) == [] def test_excludes_profiles_missing_profile_id(self) -> None: profiles = [{"profile_type": "LabelProfile"}] assert label_profile_ids_from_profiles(profiles) == [] def test_returns_single_id(self) -> None: profiles = [{"profile_type": "LabelProfile", "profile_id": 7123, "roles": []}] assert label_profile_ids_from_profiles(profiles) == [7123] class TestIsAuthorized: _tenant_uuid = uuid.uuid4() def _patch_backend( self, mocker: MockerFixture, mock_authorization_backend: AuthorizationBackend, ) -> MagicMock: backend = MagicMock(wraps=mock_authorization_backend) mocker.patch( "contributor.api.auth.datasources.get_authorization_backend", return_value=backend, ) return backend def test_returns_true_when_authorized( self, mocker: MockerFixture, mock_authorization_backend: AuthorizationBackend, ) -> None: backend = self._patch_backend(mocker, mock_authorization_backend) backend.is_authorized.return_value = True assert is_authorized("read", "contributor", self._tenant_uuid) is True backend.is_authorized.assert_called_once_with( action="read", resource_id=0, resource_type="contributor", resource_getter=mocker.ANY, tenant={"tenant_type": "account", "tenant_uuid": str(self._tenant_uuid)}, ) def test_returns_true_with_explicit_resource_id( self, mocker: MockerFixture, mock_authorization_backend: AuthorizationBackend, ) -> None: backend = self._patch_backend(mocker, mock_authorization_backend) backend.is_authorized.return_value = True assert ( is_authorized("read", "contributor", self._tenant_uuid, resource_id=42) is True ) backend.is_authorized.assert_called_once_with( action="read", resource_id=42, resource_type="contributor", resource_getter=mocker.ANY, tenant={"tenant_type": "account", "tenant_uuid": str(self._tenant_uuid)}, ) def test_returns_false_when_not_authorized( self, mocker: MockerFixture, mock_authorization_backend: AuthorizationBackend, ) -> None: backend = self._patch_backend(mocker, mock_authorization_backend) backend.is_authorized.return_value = False assert is_authorized("write", "contributor", self._tenant_uuid) is False def test_raises_401_on_unauthenticated( self, mocker: MockerFixture, mock_authorization_backend: AuthorizationBackend, ) -> None: from python_pdp_sdk import UnauthenticatedException backend = self._patch_backend(mocker, mock_authorization_backend) backend.is_authorized.side_effect = UnauthenticatedException() with pytest.raises(HTTPException) as exc_info: is_authorized("read", "contributor", self._tenant_uuid) assert exc_info.value.status_code == 401 def test_raises_403_on_unauthorized( self, mocker: MockerFixture, mock_authorization_backend: AuthorizationBackend, ) -> None: from python_pdp_sdk import UnauthorizedException backend = self._patch_backend(mocker, mock_authorization_backend) backend.is_authorized.side_effect = UnauthorizedException() with pytest.raises(HTTPException) as exc_info: is_authorized("read", "contributor", self._tenant_uuid) assert exc_info.value.status_code == 403 class TestIsOaUser: def test_returns_true_when_header_matches_oa_with_digits(self) -> None: scope = { "type": "http", "method": "GET", "path": "/", "headers": [(b"orchard-user-id", b"oa:12345")], } request = Request(scope) assert is_oa_user(request) is True def test_returns_false_when_header_does_not_start_with_oa(self) -> None: scope = { "type": "http", "method": "GET", "path": "/", "headers": [(b"orchard-user-id", b"internal:some-user")], } request = Request(scope) assert is_oa_user(request) is False def test_returns_false_when_header_is_absent(self) -> None: scope = { "type": "http", "method": "GET", "path": "/", "headers": [], } request = Request(scope) assert is_oa_user(request) is False def test_returns_false_when_header_has_non_numeric_suffix(self) -> None: scope = { "type": "http", "method": "GET", "path": "/", "headers": [(b"orchard-user-id", b"oa:some-user")], } request = Request(scope) assert is_oa_user(request) is False def test_returns_false_when_header_value_is_just_oa_prefix(self) -> None: scope = { "type": "http", "method": "GET", "path": "/", "headers": [(b"orchard-user-id", b"oa:")], } request = Request(scope) assert is_oa_user(request) is False def test_returns_false_when_header_value_is_empty(self) -> None: scope = { "type": "http", "method": "GET", "path": "/", "headers": [(b"orchard-user-id", b"")], } request = Request(scope) assert is_oa_user(request) is False