"""Unit tests for contract_flowthrough handlers.""" from decimal import Decimal from unittest.mock import MagicMock, patch import pytest from flask import testing as flask_testing from owsresponse import response from abacus_contract.constants.constants import CONTRACT_FLOWTHROUGH_STATUSES from abacus_contract.schemas.contract_flowthrough import ContractFlowthroughDetailSchema from abacus_contract.tests.utils.factories import ( ContractFactory, ContractFlowthroughFactory, ) from core.config import Config STANDALONE_HEADERS = { 'Orchard-Requestor-Service': 'graphql-abacus', 'Orchard-Profile-Type': 'ContentProfile', 'Orchard-Profile-Id': '1234', 'Orchard-Roles': 'manage_nr_ownership', 'Orchard-Identity-Id': '1234', } @patch('abacus_contract.blueprints.contract_flowthrough.logic') def test_get_contract_flowthrough_by_contract_id(mock_logic, fixture_client): """Test GET contract_flowthrough by contract_id.""" mock_contract_flowthrough = ContractFlowthroughFactory.create() mock_contract_id = mock_contract_flowthrough.contract_id mock_logic.get_contract_flowthrough_by_contract_id.return_value = response.Response( message=ContractFlowthroughDetailSchema().dump(mock_contract_flowthrough), status=200, ) res = fixture_client.get(f'/contract/{mock_contract_id}/contract-flowthrough/') assert res.status_code == 200 assert res.json == { 'contract_flowthrough_id': mock_contract_flowthrough.contract_flowthrough_id, 'contract_id': mock_contract_flowthrough.contract_id, 'reference_flowthrough_calculation_id': mock_contract_flowthrough.reference_flowthrough_calculation_id, 'flowthrough_rate': str(mock_contract_flowthrough.flowthrough_rate), 'flowthrough_status': mock_contract_flowthrough.flowthrough_status, 'has_automatic_shutoff': mock_contract_flowthrough.has_automatic_shutoff, 'recoupment_cap': mock_contract_flowthrough.recoupment_cap, 'calculation_comment': mock_contract_flowthrough.calculation_comment, 'previous_flowthrough_status': mock_contract_flowthrough.previous_flowthrough_status, 'status_last_modified_by': mock_contract_flowthrough.status_last_modified_by, 'status_last_modified': str( mock_contract_flowthrough.status_last_modified.date() ), } mock_logic.get_contract_flowthrough_by_contract_id.assert_called_once_with( mock_contract_id ) @pytest.mark.parametrize( [ 'standalone_check_result', 'get_account_id_result', 'pdp_check_result', 'expected_status_code', ], [ pytest.param(True, None, None, 200, id='Standalone check passed'), pytest.param(False, None, None, 403, id='Account id not found'), pytest.param(False, 999, False, 403, id='PDP check failed'), pytest.param(False, 999, True, 200, id='PDP check passed'), ], ) @patch('abacus_contract.blueprints.contract_flowthrough.logic') @patch('abacus_contract.blueprints.contract_flowthrough.authorization') @patch('abacus_contract.blueprints.contract_flowthrough.contract_logic') @patch('abacus_contract.blueprints.contract_flowthrough.flask_request') def test_get_contract_flowthrough_by_id_authorization( mock_flask_request: MagicMock, mock_contract_logic: MagicMock, mock_authorization: MagicMock, mock_logic: MagicMock, standalone_check_result: bool, get_account_id_result: int | None, pdp_check_result: bool | None, expected_status_code: int, fixture_client: flask_testing.FlaskClient, ) -> None: """Test GET contract_flowthrough by id with authorization check.""" mock_flask_request.verify_rules_access_standalone.return_value = ( standalone_check_result ) mock_contract_logic.get_account_id_by_contract_id.return_value = ( get_account_id_result ) mock_authorization.pdp_authorize_many_accounts.return_value = pdp_check_result mock_logic.get_contract_flowthrough_by_contract_id.return_value = ( response.Response() ) contract_id = 777 res = fixture_client.get(f'/contract/{contract_id}/contract-flowthrough/') assert res.status_code == expected_status_code mock_flask_request.verify_rules_access_standalone.assert_called_once() if not standalone_check_result: mock_contract_logic.get_account_id_by_contract_id.assert_called_once_with( contract_id ) if get_account_id_result: mock_authorization.pdp_authorize_many_accounts.assert_called_once_with( [get_account_id_result] ) @patch('abacus_contract.blueprints.contract_flowthrough.logic') def test_soft_delete_contract_flowthrough(mock_logic, fixture_client): """Test soft deleting contract_flowthrough by id.""" mock_contract_flowthrough = ContractFlowthroughFactory.create() contract_flowthrough_id = mock_contract_flowthrough.contract_flowthrough_id mock_logic.soft_delete_contract_flowthrough.return_value = response.Response( status=204 ) res = fixture_client.delete(f'/contract-flowthrough/{contract_flowthrough_id}') assert res.status_code == 204 mock_logic.soft_delete_contract_flowthrough.assert_called_once_with( mock_contract_flowthrough ) @patch('abacus_contract.blueprints.contract_flowthrough.logic') def test_update_contract_flowthrough(mock_logic, fixture_client): """Test updating contract_flowthough.""" mock_contract_flowthrough = ContractFlowthroughFactory.create() contract_flowthrough_id = mock_contract_flowthrough.contract_flowthrough_id mock_put_request_body = { 'flowthrough_rate': 1.78, 'flowthrough_status': CONTRACT_FLOWTHROUGH_STATUSES.PAUSED, } mock_logic.update_contract_flowthrough.return_value = response.Response( message='ok', status=200 ) res = fixture_client.put( f'/contract-flowthrough/{contract_flowthrough_id}', json=mock_put_request_body ) assert res.status_code == 200 mock_logic.update_contract_flowthrough.assert_called_once_with( mock_contract_flowthrough, flowthrough_rate=Decimal('1.78'), flowthrough_status=CONTRACT_FLOWTHROUGH_STATUSES.PAUSED, ) @patch('abacus_contract.blueprints.contract_flowthrough.logic') def test_create_contract_flowthrough(mock_logic, fixture_client): """Test creating contract_flowthrough.""" mock_contract = ContractFactory.create() contract_id = mock_contract.contract_id mock_logic.create_contract_flowthrough.return_value = response.Response( message='OK', status=201 ) mock_post_request = { 'reference_flowthrough_calculation_id': 2, 'flowthrough_rate': '90.78', 'recoupment_cap': 908786, } res = fixture_client.post( f'/contract/{contract_id}/contract-flowthrough/', json=mock_post_request ) assert res.status_code == 201 mock_logic.create_contract_flowthrough.assert_called_once_with( contract_id=mock_contract.contract_id, reference_flowthrough_calculation_id=2, flowthrough_rate=Decimal('90.78'), recoupment_cap=908786, ) @patch('abacus_contract.blueprints.contract_flowthrough.contract_logic') def test_get_flowthroughs_by_contract_ids_dataloader_over_cap( mock_contract_logic: MagicMock, fixture_client: flask_testing.FlaskClient, ) -> None: """Over-cap batches are rejected with a 400 before any lookup happens.""" ids = list(range(Config.OWS_BATCH_LIMIT + 1)) res = fixture_client.post('/contract-flowthroughs/dataloader', json=ids) assert res.status_code == 400 mock_contract_logic.get_account_id_map_by_contract_ids.assert_not_called() @patch('abacus_contract.utils.dataloader.ows_client') @patch('abacus_contract.utils.dataloader.permissions_authorize_many_accounts') @patch('abacus_contract.blueprints.contract_flowthrough.logic') @patch('abacus_contract.utils.dataloader.authorization') @patch('abacus_contract.blueprints.contract_flowthrough.contract_logic') def test_get_flowthroughs_by_contract_ids_dataloader( mock_contract_logic: MagicMock, mock_authorization: MagicMock, mock_logic: MagicMock, mock_permissions_authorize_many_accounts: MagicMock, mock_ows_client: MagicMock, fixture_client: flask_testing.FlaskClient, ) -> None: """One-to-one happy path: one record per contract id in request order. Only authorized ids reach fetch_records; an id that does not resolve to an account surfaces as data: null. """ # Contract 3 does not resolve to an account and must be dropped from fetch. mock_contract_logic.get_account_id_map_by_contract_ids.return_value = { 1: 9, 2: 9, } mock_permissions_authorize_many_accounts.return_value = True mock_logic.get_flowthrough_records_by_contract_ids.return_value = [ {'contract_flowthrough_id': 100, 'contract_id': 1, 'flowthrough_rate': '0.51'}, {'contract_flowthrough_id': 101, 'contract_id': 2, 'flowthrough_rate': '0.75'}, ] res = fixture_client.post( '/contract-flowthroughs/dataloader', json=[1, 2, 3], headers=STANDALONE_HEADERS, ) assert res.status_code == 200 assert res.json == [ { 'data': { 'contract_flowthrough_id': 100, 'contract_id': 1, 'flowthrough_rate': '0.51', } }, { 'data': { 'contract_flowthrough_id': 101, 'contract_id': 2, 'flowthrough_rate': '0.75', } }, {'data': None}, ] # Only the authorized (account-resolved) ids reach fetch_records. mock_logic.get_flowthrough_records_by_contract_ids.assert_called_once_with([1, 2])