"""Unit tests for transfer-job handlers + the PDP authorization decorator. These tests stub the PDP authorization backend, the project_transfer logic module, and project ownership; they exercise the routes through the Flask test client to confirm: * the decorator wires grass / JWT / job-fetch / grass-access / PDP correctly, * resource_id is forwarded to PDP when provided and omitted (i.e. 0) when not, * the execute_batch endpoint returns 501 (not 202) after the PDP gate passes, * the PDP gate is exercised regardless of which "role" the caller has (the test fakes role outcomes via the is_authorized boolean). """ from unittest.mock import MagicMock import pytest from project_manager import api from project_manager import handlers # noqa: F401 - registers routes from project_manager.logic import project_manager as pm_logic from project_manager.logic import project_transfer from project_manager.util import authorization as pdp_authorization _TEST_IDENTITY = '10436b38-5e11-472d-b6a4-bf1ee2b1b438' _PROJECT_DATA = { 'vendor_id': 7123, 'subaccount_id': 0, 'artist_id': 555, } _JOB = { 'job_id': 42, 'project_id': 10, 'originating_vendor_id': 7123, 'originating_subaccount_id': None, 'destination_vendor_id': 999, 'destination_subaccount_id': None, 'status': 'QUEUED', } @pytest.fixture def client(): """Flask test client.""" return api.app.test_client() @pytest.fixture(autouse=True) def mock_request_context(mocker): """Stub the JWT identity check so all tests receive a known identity. The transfer-job handlers reject requests without a valid JWT identity. These tests cover PDP-level authorization, not JWT parsing, so we patch _jwt_identity_or_error directly rather than faking the full request context (which would also affect g.resources setup used by the logger). """ mocker.patch.object( pdp_authorization, '_jwt_identity_or_error', return_value=(_TEST_IDENTITY, None), ) return _TEST_IDENTITY @pytest.fixture def transfer_headers(): """Common request headers for transfer endpoint tests.""" return { 'Correlation-Id': 'corr-001', 'Content-Type': 'application/json', } @pytest.fixture def mock_authorize(mocker): """Patch `is_authorized` on the imported authorization_backend. Returns a MagicMock that the test can configure (e.g. return_value True/False or `side_effect=[True, False]`). """ mock = MagicMock(return_value=True) mocker.patch.object( pdp_authorization.authorization_backend, 'is_authorized', mock, ) return mock @pytest.fixture def mock_fetch_job(mocker): """Stub fetch_job_for_auth to return the canonical test job dict.""" return mocker.patch.object( project_transfer, 'fetch_job_for_auth', return_value=_JOB) # ---- view (transfer_viewer) --------------------------------------------- def test_transfer_viewer_can_list_transfer_jobs( client, transfer_headers, mock_authorize, mocker): """transfer_viewer (PDP allow) -> GET /transfer/jobs -> 200.""" mocker.patch.object( project_transfer, 'list_transfer_jobs', return_value=_oto_ok({'items': [], 'total_count': 0})) mock_authorize.return_value = True res = client.get('/transfer/jobs', headers=transfer_headers) assert res.status_code == 200 assert mock_authorize.called # action=view passed through; resource_id=0 because non-job-scoped call = mock_authorize.call_args assert call.kwargs['action'] == 'view' assert call.kwargs['resource_type'] == 'project_transfer' assert call.kwargs['resource_id'] == 0 def test_transfer_viewer_cannot_create_transfer_job( client, transfer_headers, mock_authorize, mocker): """transfer_viewer (PDP deny on create) -> POST /transfer/job -> 403.""" mocker.patch.object(pm_logic, 'get_project_by_id', return_value=_PROJECT_DATA) mock_authorize.return_value = False res = client.post( '/transfer/job', headers=transfer_headers, json={'project_id': 10, 'destination_vendor_id': 999}) assert res.status_code == 403 # ---- create (transfer_creator) ------------------------------------------ def test_transfer_creator_can_create_transfer_job( client, transfer_headers, mock_authorize, mocker): """transfer_creator (PDP allow) -> POST /transfer/job -> 201.""" mocker.patch.object(pm_logic, 'get_project_by_id', return_value=_PROJECT_DATA) mocker.patch.object( project_transfer, 'create_transfer_job', return_value=_oto_ok({'job_id': 42}, status=201)) mock_authorize.return_value = True res = client.post( '/transfer/job', headers=transfer_headers, json={'project_id': 10, 'destination_vendor_id': 999}) assert res.status_code == 201 # Tenant is derived from the project's vendor_id (looked up by project_id). call = mock_authorize.call_args assert call.kwargs['action'] == 'create' assert call.kwargs['tenant'] == {'tenant_type': 'account'} assert call.kwargs['id_to_uuid_exchange_tenant'] == { 'tenant_type': 'account', 'tenant_id': 7123} def test_transfer_creator_can_delete_transfer_job( client, transfer_headers, mock_authorize, mock_fetch_job, mocker): """transfer_creator (PDP allow) -> DELETE /transfer/job/ -> 200.""" mocker.patch.object( project_transfer, 'delete_transfer_job', return_value=_oto_ok({'job_id': 42, 'status': 'DELETED'})) mock_authorize.return_value = True res = client.delete('/transfer/job/42', headers=transfer_headers) assert res.status_code == 200 # action=create, resource_id=42 (job-scoped), tenant from job owner. call = mock_authorize.call_args assert call.kwargs['action'] == 'create' assert call.kwargs['resource_id'] == 42 assert call.kwargs['tenant'] == {'tenant_type': 'account'} assert call.kwargs['id_to_uuid_exchange_tenant'] == { 'tenant_type': 'account', 'tenant_id': 7123} # ---- execute_batch (transfer_operator) ---------------------------------- def test_transfer_operator_execute_batch_returns_501( client, transfer_headers, mock_authorize): """transfer_operator (PDP allow) -> POST /transfer/batch/execute -> 501. The PDP gate passes (any-tenant transfer_operator derived role) but the SFN trigger is intentionally not implemented yet, so the endpoint returns 501 Not Implemented rather than 202. """ mock_authorize.return_value = True res = client.post('/transfer/batch/execute', headers=transfer_headers) assert res.status_code == 501 body = res.get_json() assert body['code'] == 'not_implemented' # Confirm PDP was called WITHOUT tenant attributes (any-tenant role). call = mock_authorize.call_args assert call.kwargs['action'] == 'execute_batch' assert call.kwargs['resource_id'] == 0 assert 'tenant' not in call.kwargs assert 'id_to_uuid_exchange_tenant' not in call.kwargs def test_non_operator_execute_batch_denied( client, transfer_headers, mock_authorize): """Non-operator (PDP deny) -> POST /transfer/batch/execute -> 403.""" mock_authorize.return_value = False res = client.post('/transfer/batch/execute', headers=transfer_headers) assert res.status_code == 403 # ---- attachments (transfer_operator) ------------------------------------ def test_transfer_operator_can_get_attachments( client, transfer_headers, mock_authorize, mocker): """transfer_operator (PDP allow) -> GET /transfer/job//attachments -> 200.""" mocker.patch.object( project_transfer, 'get_transfer_job_attachments', return_value=_oto_ok({'upcs': ['123456789012'], 'isrcs': ['USRC12345678']})) mock_authorize.return_value = True res = client.get('/transfer/job/42/attachments', headers=transfer_headers) assert res.status_code == 200 body = res.get_json() assert body['upcs'] == ['123456789012'] assert body['isrcs'] == ['USRC12345678'] call = mock_authorize.call_args assert call.kwargs['action'] == 'execute_batch' assert 'tenant' not in call.kwargs assert 'id_to_uuid_exchange_tenant' not in call.kwargs def test_non_operator_cannot_get_attachments( client, transfer_headers, mock_authorize): """Non-operator (PDP deny) -> GET /transfer/job//attachments -> 403.""" mock_authorize.return_value = False res = client.get('/transfer/job/42/attachments', headers=transfer_headers) assert res.status_code == 403 # ---- helpers ------------------------------------------------------------ def _oto_ok(message, status=200): """Build an oto.response.Response that flaskify will turn into 2xx.""" from oto import response return response.Response(message=message, status=status)