"""Functional tests for statement periods.""" from unittest.mock import patch from royalties.constants import constants, error from royalties.tests.utils.factories import ( AccountingPeriodFactory, ExchangeRateFactory, StatementPeriodFactory, StatementPeriodPaymentEntityFactory, ) def test_get_recent_statement_periods(fixture_client, statement_period_fixtures): """Get list of recent statement periods.""" result = fixture_client.get('/statement-periods/recent') assert result.json == [ { 'statement_period_id': 11, 'statement_period_name': 'Month 10', 'statement_month': 10, 'statement_year': 2024, 'closed_date': None, 'statement_period_status': 'current', 'exchange_rates_delivered': False, }, { 'statement_period_id': 10, 'statement_period_name': 'Month 9', 'statement_month': 9, 'statement_year': 2020, 'closed_date': '2021-01-01', 'statement_period_status': 'closed', 'exchange_rates_delivered': False, }, { 'statement_period_id': 9, 'statement_period_name': 'Month 8', 'statement_month': 8, 'statement_year': 2020, 'closed_date': '2021-01-01', 'statement_period_status': 'closed', 'exchange_rates_delivered': False, }, { 'statement_period_id': 8, 'statement_period_name': 'Month 7', 'statement_month': 7, 'statement_year': 2020, 'closed_date': '2021-01-01', 'statement_period_status': 'closed', 'exchange_rates_delivered': False, }, { 'statement_period_id': 7, 'statement_period_name': 'Month 6', 'statement_month': 6, 'statement_year': 2020, 'closed_date': '2021-01-01', 'statement_period_status': 'closed', 'exchange_rates_delivered': False, }, { 'statement_period_id': 6, 'statement_period_name': 'Month 5', 'statement_month': 5, 'statement_year': 2020, 'closed_date': '2021-01-01', 'statement_period_status': 'closed', 'exchange_rates_delivered': False, }, { 'statement_period_id': 5, 'statement_period_name': 'Month 4', 'statement_month': 4, 'statement_year': 2020, 'closed_date': '2021-01-01', 'statement_period_status': 'closed', 'exchange_rates_delivered': False, }, ] @patch( 'royalties.models.statement_period.is_statement_period_query_optimization_ff_enabled' ) def test_get_recent_statement_periods_ff_enabled(mock_feature, fixture_client): """Get a list of recent periods when FF is enabled.""" mock_feature.return_value = True for ind in range(3): StatementPeriodFactory.create( statement_period_name=f'Month {ind}', statement_period_status='closed', statement_month=ind, statement_year=2020, closed_date='2021-01-01', closed_by='admin', ) statement_period = StatementPeriodFactory.create( statement_period_name='Month 4', statement_period_status='current', statement_month=10, statement_year=2024, closed_date=None, closed_by=None, ) for ind in range(5, 7): StatementPeriodFactory.create( statement_period_name=f'Month {ind}', statement_month=ind, statement_year=2024, statement_period_status='open', closed_date=None, closed_by=None, ) ExchangeRateFactory.create(statement_period=statement_period) result = fixture_client.get('/statement-periods/recent') assert result.json == [ { 'statement_period_id': 4, 'statement_period_name': 'Month 4', 'statement_period_status': 'current', 'statement_month': 10, 'statement_year': 2024, 'closed_date': None, 'exchange_rates_delivered': True, }, { 'statement_period_id': 3, 'statement_period_name': 'Month 2', 'statement_period_status': 'closed', 'statement_month': 2, 'statement_year': 2020, 'closed_date': '2021-01-01', 'exchange_rates_delivered': False, }, { 'statement_period_id': 2, 'statement_period_name': 'Month 1', 'statement_period_status': 'closed', 'statement_month': 1, 'statement_year': 2020, 'closed_date': '2021-01-01', 'exchange_rates_delivered': False, }, { 'statement_period_id': 1, 'statement_period_name': 'Month 0', 'statement_period_status': 'closed', 'statement_month': 0, 'statement_year': 2020, 'closed_date': '2021-01-01', 'exchange_rates_delivered': False, }, ] def test_get_upcoming_statement_periods(fixture_client, statement_period_fixtures): """Get list of upcoming statement periods.""" result = fixture_client.get('/statement-periods/upcoming') assert result.status_code == 200 assert len(result.json) == constants.OPEN_STATEMENT_PERIODS_COUNT status_open = constants.STATEMENT_PERIOD_STATUSES.OPEN assert all([item['statement_period_status'] == status_open for item in result.json]) def test_get_current_statement_periods(fixture_client, statement_period_fixtures): """Get current statement period.""" result = fixture_client.get('/statement-period/current') assert result.status_code == 200 status_current = constants.STATEMENT_PERIOD_STATUSES.CURRENT assert result.json['statement_period_status'] == status_current @patch('royalties.logic.statement_period.close_bulk_dp_collaborator_statements') def test_close_statement_period( mock_close_bulk, fixture_client, statement_period_fixtures ): """Update statement period status to 'close' and make next as 'current'.""" mock_close_bulk.return_value = {} recent_periods = fixture_client.get('/statement-periods/recent') current_period = recent_periods.json[0] statement_period_id = current_period['statement_period_id'] statement_period_name = current_period['statement_period_name'] result = fixture_client.put(f'/statement-period/{statement_period_id}/close') assert result.status_code == 201 mock_close_bulk.assert_called_once_with( statement_period_name, statement_period_id + 1 ) new_recent_periods = fixture_client.get('/statement-periods/recent') current_period = new_recent_periods.json[0] updated_period = new_recent_periods.json[1] status_current = constants.STATEMENT_PERIOD_STATUSES.CURRENT status_closed = constants.STATEMENT_PERIOD_STATUSES.CLOSED assert updated_period['statement_period_status'] == status_closed assert updated_period['statement_period_id'] == statement_period_id assert current_period['statement_period_status'] == status_current assert current_period['statement_period_id'] == statement_period_id + 1 @patch('royalties.logic.statement_period.close_bulk_dp_collaborator_statements') def test_close_statement_period_bulk_close_failed( mock_close_bulk, fixture_client, statement_period_fixtures ): """Update statement period status to 'close' and make next as 'current'.""" recent_periods = fixture_client.get('/statement-periods/recent') current_period = recent_periods.json[0] statement_period_id = current_period['statement_period_id'] mock_close_bulk.side_effect = Exception(error.ERROR_STATEMENT_PERIOD_COLLABORATORS) result = fixture_client.put(f'/statement-period/{statement_period_id}/close') assert result.status_code == 400 assert result.json['message'] == error.ERROR_STATEMENT_PERIOD_COLLABORATORS new_recent_periods = fixture_client.get('/statement-periods/recent') current_period_after_request = new_recent_periods.json[0] status_current = constants.STATEMENT_PERIOD_STATUSES.CURRENT assert current_period_after_request['statement_period_id'] == statement_period_id assert current_period_after_request['statement_period_status'] == status_current @patch('royalties.logic.statement_period.close_bulk_dp_collaborator_statements') def test_close_statement_period_not_current_error( mock_close_bulk, fixture_client, statement_period_fixtures ): """Show error if specified period_id is not in 'current' status.""" recent_periods = fixture_client.get('/statement-periods/recent') closed_period = recent_periods.json[1] statement_period_id = closed_period['statement_period_id'] result = fixture_client.put(f'/statement-period/{statement_period_id}/close') assert result.status_code == 400 assert result.json['message'] == error.ERROR_STATEMENT_PERIOD_CURRENT_STATUS mock_close_bulk.assert_not_called() @patch('royalties.logic.statement_period.close_bulk_dp_collaborator_statements') def test_close_statement_period_has_acc_period_error(mock_close_bulk, fixture_client): """Show error if there are active acc period within statement period.""" statement_period = StatementPeriodFactory.create( statement_period_status=constants.STATEMENT_PERIOD_STATUSES.CURRENT ) AccountingPeriodFactory.create(statement_period=statement_period) statement_period_id = statement_period.statement_period_id result = fixture_client.put(f'/statement-period/{statement_period_id}/close') assert result.status_code == 400 assert result.json['message'] == error.ERROR_STATEMENT_PERIOD_HAS_ACC_PERIODS mock_close_bulk.assert_not_called() @patch('royalties.logic.statement_period.close_bulk_dp_collaborator_statements') def test_close_statement_period_close_balances_error( mock_close_bulk, fixture_client, reference_payment_entity_fixtures, ): """Show error if payment entitity doesn't have is_visible_to_customer field set to true.""" statement_period = StatementPeriodFactory.create( statement_period_status=constants.STATEMENT_PERIOD_STATUSES.CURRENT ) StatementPeriodPaymentEntityFactory.create( reference_payment_entity_id=1, statement_period=statement_period, is_visible_to_customer=False, ) AccountingPeriodFactory.create( statement_period=statement_period, accounting_period_status=constants.ACCOUNTING_PERIOD_STATUSES.CLOSED, ) statement_period_id = statement_period.statement_period_id result = fixture_client.put(f'/statement-period/{statement_period_id}/close') assert result.status_code == 400 assert result.json['message'] == error.ERROR_BALANCES_ARE_NOT_CLOSED mock_close_bulk.assert_not_called() @patch('royalties.logic.statement_period.close_bulk_dp_collaborator_statements') def test_close_statement_period_payment_entity_error( mock_close_bulk, fixture_client, reference_payment_entity_fixtures, payment_entities_actions_fixtures, ): """Show error if payment entitity doesn't have is_visible_to_customer field set to true.""" statement_period = StatementPeriodFactory.create( statement_period_status=constants.STATEMENT_PERIOD_STATUSES.CURRENT ) StatementPeriodPaymentEntityFactory.create( statement_period_payment_entity_id=1, reference_payment_entity_id=1, statement_period=statement_period, is_visible_to_customer=False, ) AccountingPeriodFactory.create( statement_period=statement_period, accounting_period_status=constants.ACCOUNTING_PERIOD_STATUSES.CLOSED, ) statement_period_id = statement_period.statement_period_id result = fixture_client.put(f'/statement-period/{statement_period_id}/close') assert result.status_code == 400 assert result.json['message'] == error.ERROR_STATEMENT_PERIOD_PAYMENT_ENTITY mock_close_bulk.assert_not_called() def test_statement_period_by_id(fixture_client, statement_period_fixtures): """Get statement period by an id.""" result = fixture_client.get('/statement-period/11') assert result.status_code == 200 assert result.json == { 'statement_period_id': 11, 'statement_period_name': 'Month 10', 'statement_month': 10, 'statement_year': 2024, 'closed_date': None, 'statement_period_status': 'current', 'exchange_rates_delivered': False, } @patch( 'royalties.models.statement_period.is_statement_period_query_optimization_ff_enabled' ) def test_statement_period_by_id_ff_enabled(mock_feature, fixture_client): """Get statement period by an id when FF enabled.""" mock_feature.return_value = True statement_period = StatementPeriodFactory.create( statement_period_name='statement period for feb 2025', statement_month=2, statement_year=2025, statement_period_status='current', closed_date=None, closed_by=None, ) ExchangeRateFactory.create(statement_period=statement_period) statement_period_id = statement_period.statement_period_id result = fixture_client.get(f'/statement-period/{statement_period_id}') assert result.status_code == 200 assert result.json == { 'statement_period_id': statement_period_id, 'statement_period_name': statement_period.statement_period_name, 'statement_month': statement_period.statement_month, 'statement_year': statement_period.statement_year, 'closed_date': statement_period.closed_date, 'statement_period_status': statement_period.statement_period_status, 'exchange_rates_delivered': False, } @patch( 'royalties.models.statement_period.is_statement_period_query_optimization_ff_enabled' ) def test_statement_period_by_id_ff_disabled(mock_feature, fixture_client): """Get statement period by an id when FF disabled.""" mock_feature.return_value = False statement_period = StatementPeriodFactory.create( statement_period_name='statement period for feb 2025', statement_month=2, statement_year=2025, statement_period_status='current', closed_date=None, closed_by=None, ) ExchangeRateFactory.create(statement_period=statement_period) statement_period_id = statement_period.statement_period_id result = fixture_client.get(f'/statement-period/{statement_period_id}') assert result.status_code == 200 assert result.json == { 'statement_period_id': statement_period_id, 'statement_period_name': statement_period.statement_period_name, 'statement_month': statement_period.statement_month, 'statement_year': statement_period.statement_year, 'closed_date': statement_period.closed_date, 'statement_period_status': statement_period.statement_period_status, 'exchange_rates_delivered': True, } def test_statement_period_by_invalid_id(fixture_client): """Show error if statement_period_id is invalid.""" result = fixture_client.get('/statement-period/11190') assert result.status_code == 404 assert result.json['message'] == error.ERROR_ENTITY_DOES_NOT_EXIST.format( object_type='StatementPeriod', object_id=11190 ) def test_exchange_rate_bulk_endpoint_json(fixture_client): """Test exchange rate bulk endpoint json output.""" statement_period = StatementPeriodFactory(statement_period_status='current') ExchangeRateFactory.reset_sequence(0) ExchangeRateFactory.create_batch(2, statement_period=statement_period) statement_period_id = statement_period.statement_period_id res = fixture_client.get( f'/statement-period/{statement_period_id}/bulk-exchange-rates' ) assert res.status_code == 200 assert sorted(res.json, key=lambda x: x['exchange_rate_id']) == [ { 'to_currency_code': 'CC1', 'exchange_rate_id': 1, 'from_currency_code': 'CC0', 'rate': '0.0000000000000000000', }, { 'to_currency_code': 'CC2', 'exchange_rate_id': 2, 'from_currency_code': 'CC1', 'rate': '0.0000010000000000000', }, ] def test_exchange_rate_bulk_endpoint_wrong_format(fixture_client): """Test exchange rate bulk endpoint returns 400 for unknown output format.""" statement_period = StatementPeriodFactory(statement_period_status='current') statement_period_id = statement_period.statement_period_id res = fixture_client.get( f'/statement-period/{statement_period_id}/bulk-exchange-rates?format=excel' ) assert res.status_code == 400 def test_get_statement_periods(fixture_client): """Get all statement periods.""" for ind in [1, 2]: StatementPeriodFactory.create( statement_period_name=f'Month {ind}', statement_period_status='closed', statement_month=ind, statement_year=2024, closed_date='2021-01-01', closed_by='admin', ) for ind in [3, 4]: StatementPeriodFactory.create( statement_period_name=f'Month {ind}', statement_period_status='open', closed_date=None, statement_month=ind, statement_year=2024, ) result = fixture_client.get('/statement-periods/?limit=3&offset=0') assert result.status_code == 200 assert result.json['total_count'] == 4 assert result.json['items'] == [ { 'statement_period_id': 1, 'statement_period_name': 'Month 1', 'statement_month': 1, 'statement_year': 2024, 'closed_date': '2021-01-01', 'statement_period_status': 'closed', 'exchange_rates_delivered': False, }, { 'statement_period_id': 2, 'statement_period_name': 'Month 2', 'statement_month': 2, 'statement_year': 2024, 'closed_date': '2021-01-01', 'statement_period_status': 'closed', 'exchange_rates_delivered': False, }, { 'statement_period_id': 3, 'statement_period_name': 'Month 3', 'statement_month': 3, 'statement_year': 2024, 'closed_date': None, 'statement_period_status': 'open', 'exchange_rates_delivered': False, }, ] def test_get_statement_periods_by_years(fixture_client): """Get statement periods by a list of statement_years.""" StatementPeriodFactory.create( statement_period_name='Month 1', statement_period_status='closed', statement_month=1, statement_year=2023, closed_date='2023-01-01', closed_by='admin', ) StatementPeriodFactory.create( statement_period_name='Month 2', statement_period_status='closed', statement_month=2, statement_year=2023, closed_date='2023-02-01', closed_by='admin', ) StatementPeriodFactory.create( statement_period_name='Month 1', statement_period_status='open', statement_month=1, statement_year=2024, closed_date=None, closed_by=None, ) StatementPeriodFactory.create( statement_period_name='Month 2', statement_period_status='open', statement_month=2, statement_year=2024, closed_date=None, closed_by=None, ) result = fixture_client.post('/statement-periods/', json=[2022, 2023, 2024, 2026]) assert result.status_code == 200 assert result.json == [ { 'statement_period_id': 1, 'statement_period_name': 'Month 1', 'statement_month': 1, 'statement_year': 2023, 'closed_date': '2023-01-01', 'statement_period_status': 'closed', 'exchange_rates_delivered': False, }, { 'statement_period_id': 2, 'statement_period_name': 'Month 2', 'statement_month': 2, 'statement_year': 2023, 'closed_date': '2023-02-01', 'statement_period_status': 'closed', 'exchange_rates_delivered': False, }, { 'statement_period_id': 3, 'statement_period_name': 'Month 1', 'statement_month': 1, 'statement_year': 2024, 'closed_date': None, 'statement_period_status': 'open', 'exchange_rates_delivered': False, }, { 'statement_period_id': 4, 'statement_period_name': 'Month 2', 'statement_month': 2, 'statement_year': 2024, 'closed_date': None, 'statement_period_status': 'open', 'exchange_rates_delivered': False, }, ] def test_get_statement_periods_by_years_error(fixture_client): """An error is thrown if an invalid year is passed to the POST request payload.""" result = fixture_client.post('/statement-periods/', json=[2022, 2023, 2024, 'Test']) assert result.status_code == 400 assert result.json['message'] == error.ERROR_INVALID_STATEMENT_YEARS