"""Unit tests for Contract Mechanical Deduction endpoints.""" from datetime import datetime from decimal import Decimal from typing import Any from unittest.mock import MagicMock, patch import pytest from abacus_common_logic.utils.authorization import permissions_authorize_many_accounts from flask.testing import FlaskClient from owsresponse import response from abacus_contract.constants.constants import ( MECHANICAL_DEDUCTION_ADMIN_TYPES, MECHANICAL_DEDUCTION_TERRITORIES, MECHANICAL_DEDUCTION_TYPES, ) from abacus_contract.schemas.contract_mechanical_deduction import ( ContractMechanicalDeductionDetailSchema, ) from abacus_contract.tests.utils.factories import ( AccountContractFactory, ContractFactory, ContractMechanicalDeductionFactory, ) @patch('abacus_contract.blueprints.contract_mechanical_deduction.logic') def test_create_contract_mechanical_deduction(mock_logic, fixture_client): """Test creating contract mechanical deduction.""" mock_contract = ContractFactory.create() contract_id = mock_contract.contract_id mock_logic.create_contract_mechanical_deduction.return_value = response.Response( message='OK', status=201 ) mock_payload_request = { 'admin_fee': 1.09, 'admin_type': MECHANICAL_DEDUCTION_ADMIN_TYPES.BOTH, 'territory': MECHANICAL_DEDUCTION_TERRITORIES.ROW, 'mechanical_type': [MECHANICAL_DEDUCTION_TYPES.PHYSICAL], } res = fixture_client.post( f'/contract/{contract_id}/contract-mechanical-deduction/', json=mock_payload_request, ) assert res.status_code == 201 mock_logic.create_contract_mechanical_deduction.assert_called_once_with( contract_id=contract_id, admin_fee=Decimal('1.09'), admin_type=MECHANICAL_DEDUCTION_ADMIN_TYPES.BOTH, territory=MECHANICAL_DEDUCTION_TERRITORIES.ROW, mechanical_type=[MECHANICAL_DEDUCTION_TYPES.PHYSICAL], ) @patch('abacus_contract.blueprints.contract_mechanical_deduction.logic') def test_update_contract_mechanical_deduction(mock_logic, fixture_client): """Test updating contract_mechanical_deduction.""" mock_contract_mechanical_deduction = ContractMechanicalDeductionFactory.create( admin_fee=2.0, admin_type=MECHANICAL_DEDUCTION_ADMIN_TYPES.BOTH, mechanical_type=[MECHANICAL_DEDUCTION_TYPES.DIGITAL], territory=MECHANICAL_DEDUCTION_TERRITORIES.USA, ) contract_mechanical_deduction_id = ( mock_contract_mechanical_deduction.contract_mechanical_deduction_id ) mock_put_request_body = { 'admin_fee': 1.78, 'admin_type': MECHANICAL_DEDUCTION_ADMIN_TYPES.BUSINESS, 'mechanical_type': [MECHANICAL_DEDUCTION_TYPES.PHYSICAL], } mock_logic.update_contract_mechanical_deduction.return_value = response.Response( message='ok', status=200 ) res = fixture_client.put( f'/contract-mechanical-deduction/{contract_mechanical_deduction_id}', json=mock_put_request_body, ) assert res.status_code == 200 mock_logic.update_contract_mechanical_deduction.assert_called_once_with( mock_contract_mechanical_deduction, admin_fee=Decimal('1.78'), admin_type=MECHANICAL_DEDUCTION_ADMIN_TYPES.BUSINESS, mechanical_type=[MECHANICAL_DEDUCTION_TYPES.PHYSICAL], ) @patch('abacus_contract.blueprints.contract_mechanical_deduction.logic') def test_delete_contract_mechanical_deduction_by_id(mock_logic, fixture_client): """Test deleting contract mechanical deduction by id.""" mock_contract_mechanical_deduction = ContractMechanicalDeductionFactory.create() contract_mechanical_deduction_id = ( mock_contract_mechanical_deduction.contract_mechanical_deduction_id ) mock_logic.soft_delete_contract_mechanical_deduction.return_value = ( response.Response(status=204) ) res = fixture_client.delete( f'/contract-mechanical-deduction/{contract_mechanical_deduction_id}' ) assert res.status_code == 204 mock_logic.soft_delete_contract_mechanical_deduction.assert_called_once_with( mock_contract_mechanical_deduction ) @patch('abacus_contract.blueprints.contract_mechanical_deduction.logic') def test_create_contract_mechanical_deductions_worldwide(mock_logic, fixture_client): """Test creating contract mechanical deductions worldwide.""" mock_contract = ContractFactory.create() contract_id = mock_contract.contract_id mock_logic.create_contract_mechanical_deductions_worldwide.return_value = ( response.Response(message='OK', status=201) ) mock_payload_request = { 'admin_fee': 1.09, 'admin_type': MECHANICAL_DEDUCTION_ADMIN_TYPES.BOTH, 'mechanical_type': [MECHANICAL_DEDUCTION_TYPES.PHYSICAL], } res = fixture_client.post( f'/contract/{contract_id}/contract-mechanical-deductions/worldwide/', json=mock_payload_request, ) assert res.status_code == 201 mock_logic.create_contract_mechanical_deductions_worldwide.assert_called_once_with( contract_id, admin_fee=Decimal('1.09'), admin_type=MECHANICAL_DEDUCTION_ADMIN_TYPES.BOTH, mechanical_type=[MECHANICAL_DEDUCTION_TYPES.PHYSICAL], ) @pytest.fixture def mock_contract_mechanical_deduction(fixture_client): """Fixture for creating a mock contract mechanical deduction.""" mock_contract = ContractFactory.build(contract_id=1) AccountContractFactory.build(contract=mock_contract, account_id=3) mock_contract_mechanical_deduction = ContractMechanicalDeductionFactory.build( contract=mock_contract ) return mock_contract_mechanical_deduction @pytest.mark.parametrize( ( 'profile_type', 'profile_role', 'pdp_authorize_return', 'permissions_authorize_return', 'expected_status', ), [ pytest.param( 'LabelProfile', 'catalog', None, True, 200, id='Standalone access, permissions check OK', ), pytest.param( 'LabelProfile', 'catalog', None, False, 403, id='Standalone access, permissions check not OK', ), pytest.param( 'Account360Profile', 'account360', True, True, 200, id='standalone access fail, pdp check OK', ), pytest.param( 'Account360Profile', 'account360', False, None, 403, id='standalone access fail, pdp check not OK', ), ], ) @patch( 'abacus_contract.blueprints.contract_mechanical_deduction.logic.get_contract_mechanical_deductions_by_contract_id' ) @patch( 'abacus_contract.blueprints.contract_mechanical_deduction.pdp_authorize_many_accounts' ) @patch( 'abacus_contract.blueprints.contract_mechanical_deduction.permissions_authorize_many_accounts' ) @patch('abacus_contract.blueprints.contract_mechanical_deduction.ows_client') def test_get_contract_mechanical_deductions_by_contract_id( mock_ows_client: MagicMock, mock_permission_authorize_many_accounts: MagicMock, mock_pdp_authorize_many_accounts: MagicMock, mock_get_contract_mechanical_deductions: MagicMock, fixture_client: FlaskClient, mock_contract_mechanical_deduction: Any, profile_type: str, profile_role: str, pdp_authorize_return: bool, permissions_authorize_return: bool, expected_status: int, ) -> None: """Test GET /contract//contract-mechanical-deductions.""" mock_pdp_authorize_many_accounts.return_value = pdp_authorize_return mock_permission_authorize_many_accounts.return_value = permissions_authorize_return mock_get_contract_mechanical_deductions.return_value = ( response.Response( message=ContractMechanicalDeductionDetailSchema(many=True).dump( [mock_contract_mechanical_deduction] ), status=200, ), mock_contract_mechanical_deduction.contract, ) contract_id = mock_contract_mechanical_deduction.contract.contract_id res = fixture_client.get( f'/contract/{contract_id}/contract-mechanical-deductions', headers={ 'Orchard-Requestor-Service': 'graphql-abacus', 'Orchard-Profile-Type': profile_type, 'Orchard-Profile-Id': '1234', 'Orchard-Roles': profile_role, 'Orchard-Identity-Id': '1234', }, ) assert res.status_code == expected_status # Check if authorize_many_accounts was called when standalone check failed if profile_type == 'Account360Profile': mock_pdp_authorize_many_accounts.assert_called_once_with([3]) else: mock_pdp_authorize_many_accounts.assert_not_called() if pdp_authorize_return is not False: mock_permission_authorize_many_accounts.assert_called_once_with( mock_ows_client, profile_type, '1234', [3] ) else: mock_permission_authorize_many_accounts.assert_not_called() @patch('abacus_contract.blueprints.contract_mechanical_deduction.logic') def test_get_active_contracts_with_mechanical_deductions( mock_logic, fixture_client: FlaskClient ): """Test GET /contracts/mechanical-deductions/active.""" mocked_response = [ { 'account_id': 12345, 'contract_id': 67890, 'term_type': 'product', 'attachments': ['1111111111111'], 'mechanical_type': ['digital', 'physical'], }, { 'account_id': 54321, 'contract_id': 9876, 'term_type': 'track', 'attachments': ['2222222222222'], 'mechanical_type': ['digital'], }, ] mock_logic.get_active_contracts_with_mechanical_deductions.return_value = ( response.Response(message=mocked_response, status=200) ) res = fixture_client.get('/contracts/mechanical-deductions/active?date=2025-05-21') assert res.status_code == 200 assert res.get_json() == mocked_response mock_logic.get_active_contracts_with_mechanical_deductions.assert_called_once_with( datetime(2025, 5, 21).date() ) @pytest.mark.parametrize( ( 'profile_type', 'profile_role', 'permissions_authorize_return', 'expected_status', ), [ pytest.param( 'LabelProfile', 'catalog', True, 200, id='Standalone access, permissions check OK', ), pytest.param( 'LabelProfile', 'catalog', False, 403, id='Standalone access, permissions check not OK', ), pytest.param( 'InsightsProfile', 'analytics', None, 401, id='Standalone access fail', ), ], ) @patch( 'abacus_contract.blueprints.contract_mechanical_deduction.logic.get_mechadmin_for_account' ) @patch( 'abacus_contract.blueprints.contract_mechanical_deduction.permissions_authorize_many_accounts' ) @patch('abacus_contract.blueprints.contract_mechanical_deduction.ows_client') def test_get_contract_mechanical_deductions_by_contract_id( mock_ows_client: MagicMock, mock_permission_authorize_many_accounts: MagicMock, mock_get_mechadmin_for_account: MagicMock, fixture_client: FlaskClient, profile_type: str, profile_role: str, permissions_authorize_return: bool, expected_status: int, ) -> None: """Test GET /contracts/mechanical-deductions/account//mechadmin.""" account_id = 1 mock_permission_authorize_many_accounts.return_value = permissions_authorize_return mock_get_mechadmin_for_account.return_value = response.Response( message={'mechadmin_physical': True, 'mechadmin_digital': True}, status=200, ) res = fixture_client.get( f'/contracts/mechanical-deductions/account/{account_id}/mechadmin', headers={ 'Orchard-Requestor-Service': 'graphql-abacus', 'Orchard-Profile-Type': profile_type, 'Orchard-Profile-Id': '1234', 'Orchard-Roles': profile_role, 'Orchard-Identity-Id': '1234', }, ) assert res.status_code == expected_status if permissions_authorize_return is not None: mock_permission_authorize_many_accounts.assert_called_once_with( mock_ows_client, profile_type, '1234', [account_id] ) else: mock_permission_authorize_many_accounts.assert_not_called()