"""Tests for Vendor Contracts.""" from unittest.mock import MagicMock import pytest from owsrequest import request, test_utils from contracts import response from contracts.constants import query, service_name from contracts.logic import vendor_contracts from contracts.models import ( booked_vendor_contract, ows_account, ows_product, product_split, product_territory_split, track_split, vendor_contract, ) @pytest.fixture def fixture_active_vendor_contract(): """Fixture for active vendor contract.""" return response.Response({'vendor_id': 1, 'vendor_contract_id': 1}) @pytest.fixture def fixture_booked_contracts(): """Fixture for vendor booked contracts.""" booked_contracts = [ booked_vendor_contract.BookedVendorContract( snapshot_id=2, contract_id=106, period_id=100, vendor_id=100, payment_interval='quarter', currency_id=1, reserve_rate=0.1, number_of_months_before_payout=2, number_of_installments=3, contract_terms='Some Text for Test Contract Term', ), booked_vendor_contract.BookedVendorContract( snapshot_id=1, contract_id=105, period_id=99, vendor_id=100, payment_interval='month', currency_id=1, reserve_rate=0.6, number_of_months_before_payout=6, number_of_installments=6, contract_terms='Some Text for Test Contract Term', ), ] return response.Response(booked_contracts) def test_get_booked_contracts(monkeypatch, fixture_booked_contracts): """Test get_booked_contracts with contracts.""" vendor_id = 100 expected_items_count = 2 expected_items = [ { 'id': 2, 'contract_id': 106, 'period_id': 100, 'vendor_id': vendor_id, 'payment_interval': 'quarter', 'currency_id': 1, 'reserve_rate': 0.1, 'number_of_months_before_payout': 2, 'number_of_installments': 3, 'contract_terms': 'Some Text for Test Contract Term', }, { 'id': 1, 'contract_id': 105, 'period_id': 99, 'vendor_id': vendor_id, 'payment_interval': 'month', 'currency_id': 1, 'reserve_rate': 0.6, 'number_of_months_before_payout': 6, 'number_of_installments': 6, 'contract_terms': 'Some Text for Test Contract Term', }, ] expected_pagination = { 'type': 'standard', 'offset': query.PAGE_OFFSET_DEFAULT, 'limit': query.PAGE_LIMIT_DEFAULT, 'total_records': expected_items_count, 'sort_order': query.ASC, } monkeypatch.setattr( booked_vendor_contract, 'get_booked_contracts', MagicMock(return_value=fixture_booked_contracts), ) monkeypatch.setattr( booked_vendor_contract, 'get_booked_contracts_count', MagicMock(return_value=response.Response(expected_items_count)), ) result = vendor_contracts.get_booked_contracts( vendor_id, query.ASC, query.PAGE_LIMIT_DEFAULT, query.PAGE_OFFSET_DEFAULT ) assert booked_vendor_contract.get_booked_contracts.called assert booked_vendor_contract.get_booked_contracts_count.called assert result.status == 200 assert result.message['items'] == expected_items assert result.message['pagination'] == expected_pagination def test_get_booked_contracts_empty(monkeypatch): """Test get_booked_contracts without contracts.""" vendor_id = 100 expected_items_count = 0 expected_pagination = { 'type': 'standard', 'offset': query.PAGE_OFFSET_DEFAULT, 'limit': query.PAGE_LIMIT_DEFAULT, 'total_records': expected_items_count, 'sort_order': query.ASC, } monkeypatch.setattr( booked_vendor_contract, 'get_booked_contracts_count', MagicMock(return_value=response.Response(expected_items_count)), ) result = vendor_contracts.get_booked_contracts( vendor_id, query.ASC, query.PAGE_LIMIT_DEFAULT, query.PAGE_OFFSET_DEFAULT ) assert booked_vendor_contract.get_booked_contracts_count.called assert result.status == 200 assert result.message['items'] == [] assert result.message['pagination'] == expected_pagination def test_get_active_vendor_contract(monkeypatch, fixture_active_vendor_contract): """Test getting active vendor contract.""" monkeypatch.setattr( vendor_contract, 'get_active_contract', MagicMock(return_value=fixture_active_vendor_contract), ) result = vendor_contracts.get_active_vendor_contract(1) assert result.status == 200 def test_get_active_vendor_contract_missing(monkeypatch, create_not_found_response): """Test getting missing active vendor contract.""" monkeypatch.setattr( vendor_contract, 'get_active_contract', MagicMock(return_value=create_not_found_response), ) result = vendor_contracts.get_active_vendor_contract(1) assert result.status == 404 def test_get_vendor_contract(monkeypatch, fixture_vendor_contract_response): """Test getting vendor contract.""" monkeypatch.setattr( vendor_contract, 'get_contract', MagicMock(return_value=fixture_vendor_contract_response), ) result = vendor_contracts.get_vendor_contract(1) assert result.status == 200 def test_get_vendor_contract_missing(monkeypatch, create_not_found_response): """Test getting missing vendor contract.""" monkeypatch.setattr( vendor_contract, 'get_contract', MagicMock(return_value=create_not_found_response), ) result = vendor_contracts.get_vendor_contract(1) assert result.status == 404 def test_get_contract_service_type(monkeypatch, fixture_contract_service_type_response): """Test getting vendor contract service type.""" monkeypatch.setattr( vendor_contract, 'get_contract_service_type', MagicMock(return_value=fixture_contract_service_type_response), ) result = vendor_contracts.get_contract_service_type(1) assert result.status == 200 def test_get_product_split(monkeypatch, client, fixture_product_split_response): """Test getting product split.""" expected_paginated_response = response.Response( message={ 'items': [ {'upc': 192562596593, 'product_split': 0.65}, {'upc': 192562881774, 'product_split': 0.65}, ], 'pagination': { 'type': 'standard', 'offset': 0, 'limit': 50, 'total_records': 11, }, } ) page_offset = 1 page_limit = 2 monkeypatch.setattr( product_split, 'get_product_split', MagicMock(return_value=expected_paginated_response), ) result = vendor_contracts.get_product_split(3513135135, page_offset, page_limit) assert result.status == 200 def test_set_product_split_pass( monkeypatch, client, fixture_product_split_post_request_parameters ): """Test setting product split success.""" call_specs = [ { 'service': service_name.OWS_PRODUCT, 'path': '/vendor/25257/upcs', 'status': 200, } ] monkeypatch.setattr(request, 'post', test_utils.mock_ows_requests(call_specs)) request_params = [fixture_product_split_post_request_parameters] monkeypatch.setattr( ows_product, 'validate_upc', MagicMock(return_value=response.create_status_ok_response()), ) monkeypatch.setattr( product_split, 'set_product_split', MagicMock(return_value=response.create_status_ok_response()), ) result = vendor_contracts.set_product_split(request_params, '25257', 'vendor') assert result.status == 200 def test_set_product_split_for_invalid_upc( monkeypatch, client, fixture_product_split_post_request_parameters ): """Test setting product split for invalid upc.""" call_specs = [ { 'service': service_name.OWS_PRODUCT, 'path': '/vendor/25257/upcs', 'status': 200, } ] monkeypatch.setattr(request, 'post', test_utils.mock_ows_requests(call_specs)) request_params = [fixture_product_split_post_request_parameters] monkeypatch.setattr( ows_product, 'validate_upc', MagicMock(return_value=response.create_not_found_response()), ) result = vendor_contracts.set_product_split(request_params, '25257', 'vendor') assert result.status == 404 def test_set_product_split_for_upc_not_found( monkeypatch, client, fixture_product_split_post_request_parameters ): """Test setting product split for upc that doesn't belong to vendor.""" call_specs = [ { 'service': service_name.OWS_PRODUCT, 'path': '/vendor/25257/upcs', 'status': 404, } ] monkeypatch.setattr(request, 'post', test_utils.mock_ows_requests(call_specs)) request_params = [fixture_product_split_post_request_parameters] result = vendor_contracts.set_product_split(request_params, '25257', 'vendor') assert result.status == 404 def test_update_product_split_pass( monkeypatch, client, fixture_product_split_put_request_parameters ): """Test updating product split success.""" call_specs = [ { 'service': service_name.OWS_PRODUCT, 'path': '/vendor/25257/upcs', 'status': 200, } ] monkeypatch.setattr(request, 'post', test_utils.mock_ows_requests(call_specs)) request_params = [fixture_product_split_put_request_parameters] monkeypatch.setattr( product_split, 'update_product_split', MagicMock(return_value=response.create_status_ok_response()), ) result = vendor_contracts.update_product_split(request_params, '25257', 'vendor') assert result.status == 200 def test_update_product_split_for_invalid_upc( monkeypatch, client, fixture_product_split_put_request_parameters ): """Test updating product split for invalid upc.""" call_specs = [ { 'service': service_name.OWS_PRODUCT, 'path': '/vendor/25257/upcs', 'status': 200, } ] monkeypatch.setattr(request, 'post', test_utils.mock_ows_requests(call_specs)) request_params = [fixture_product_split_put_request_parameters] monkeypatch.setattr( product_split, 'update_product_split', MagicMock(return_value=response.create_not_found_response()), ) result = vendor_contracts.update_product_split(request_params, '25257', 'vendor') assert result.status == 404 def test_update_product_split_for_upc_not_found( monkeypatch, client, fixture_product_split_put_request_parameters ): """Test updating product split for upc that doesn't belong to vendor.""" call_specs = [ { 'service': service_name.OWS_PRODUCT, 'path': '/vendor/25257/upcs', 'status': 404, } ] monkeypatch.setattr(request, 'post', test_utils.mock_ows_requests(call_specs)) request_params = [fixture_product_split_put_request_parameters] result = vendor_contracts.update_product_split(request_params, '25257', 'vendor') assert result.status == 404 def test_check_product_ownership(monkeypatch): """Test response of product ownership function.""" call_specs = [ { 'service': service_name.OWS_PRODUCT, 'path': '/vendor/25257/upcs', 'status': 200, } ] monkeypatch.setattr(request, 'post', test_utils.mock_ows_requests(call_specs)) json_data = {'upcs': [123456, 5678908]} result = vendor_contracts.check_product_ownership(json_data, '25257', 'vendor') assert result.status_code == 200 def test_get_track_split(monkeypatch, fixture_track_split_response): """Test getting track split.""" expected_paginated_response = response.Response( message={ 'items': [ { 'isrc': 'ES20I1310348', 'track_split': 0.65, 'track_unique_id': 24442690, }, { 'isrc': 'QM4TW1664428', 'track_split': 0.75, 'track_unique_id': 24442689, }, ], 'pagination': { 'type': 'standard', 'offset': 1, 'limit': 2, 'total_records': 11, }, } ) page_offset = 1 page_limit = 2 monkeypatch.setattr( track_split, 'get_track_split', MagicMock(return_value=expected_paginated_response), ) result = vendor_contracts.get_track_split(123, page_offset, page_limit) assert result.status == 200 def test_delete_track_split(monkeypatch): """Test delete track split.""" monkeypatch.setattr( track_split, 'delete_track_split', MagicMock(return_value=response.create_status_ok_response()), ) vendor_id = 123 data = {'isrc': 'ES20I1310348', 'oa_user_id': 562, 'vend_contract_id': 2356} result = vendor_contracts.delete_track_split(vendor_id, data) assert result.status == 200 def test_set_track_split(monkeypatch, fixture_track_split_post_request_parameters): """Test set track split.""" monkeypatch.setattr( track_split, 'set_track_split', MagicMock(return_value=response.create_status_ok_response()), ) vendor_id = 123 request_params = fixture_track_split_post_request_parameters result = vendor_contracts.set_track_split(vendor_id, request_params) assert result.status == 200 def test_update_track_split(monkeypatch, fixture_track_split_put_request_parameters): """Test update track split.""" monkeypatch.setattr( track_split, 'update_track_split', MagicMock(return_value=response.create_status_ok_response()), ) vendor_id = 123 request_params = fixture_track_split_put_request_parameters result = vendor_contracts.update_track_split(vendor_id, request_params) assert result.status == 200 def test_get_product_territoty_split(monkeypatch): """Test getting track split.""" expected_paginated_response = response.Response( message={ 'items': [ { 'upc': 191018337490, 'vendor_id': 25824, 'product_territory_split': 0.56, 'country_details': { 'id': 2, 'name': 'Canada', 'country_code': 'CA', 'abbrivation': 'CA', 'continent': 'North America', 'latitude': 56.130366, 'longitude': -106.346771, 'iso3166a3': 'CAN', 'continent_id': 5, }, }, { 'upc': 191018333454, 'vendor_id': 25824, 'product_territory_split': 0.5, 'country_details': { 'id': 1, 'name': 'USA', 'country_code': 'US', 'abbrivation': 'US', 'continent': 'North America', 'latitude': 37.09024, 'longitude': -95.712891, 'iso3166a3': 'USA', 'continent_id': 5, }, }, ], 'pagination': { 'type': 'standard', 'offset': 1, 'limit': 2, 'total_records': 11, }, } ) page_offset = 1 page_limit = 2 monkeypatch.setattr( product_territory_split, 'get_product_territory_split', MagicMock(return_value=expected_paginated_response), ) result = vendor_contracts.get_product_territory_split( 25824, page_offset, page_limit ) assert result.status == 200 def test_delete_product_territory_split(monkeypatch): """Test delete product territory split.""" monkeypatch.setattr( product_territory_split, 'delete_product_territory_split', MagicMock(return_value=response.create_status_ok_response()), ) upc = 191018337490 data = {'country_id': 2, 'oa_user_id': 562, 'vend_contract_id': 2356} result = vendor_contracts.delete_product_territory_split(upc, data) assert result.status == 200 def test_set_product_territory_split( monkeypatch, fixture_product_territory_split_post_request_parameters ): """Test set product territory split.""" monkeypatch.setattr( vendor_contracts, 'set_product_territory_split', MagicMock(return_value=response.create_status_ok_response()), ) vendor_id = 25824 account_type = 'vendor' request_params = fixture_product_territory_split_post_request_parameters result = vendor_contracts.set_product_territory_split( request_params, vendor_id, account_type ) assert result.status == 200 def test_update_product_territory_split( monkeypatch, fixture_product_territory_split_put_request_parameters ): """Test update product territory split.""" monkeypatch.setattr( vendor_contracts, 'update_product_territory_split', MagicMock(return_value=response.create_status_ok_response()), ) vendor_id = 25824 account_type = 'vendor' request_params = fixture_product_territory_split_put_request_parameters result = vendor_contracts.update_product_territory_split( request_params, vendor_id, account_type ) assert result.status == 200 def test_get_vendor_currency(monkeypatch): """Test get vendor currency.""" vendor_id = 1 data = {'id': 1, 'vendor_id': 1, 'currency_id': 1, 'code': 'USD', 'symbol': 'USD'} monkeypatch.setattr( vendor_contract, 'get_vendor_currency', MagicMock(return_value=response.Response(data)), ) result = vendor_contracts.get_vendor_currency(vendor_id) vendor_contract.get_vendor_currency.assert_called_with(vendor_id) assert result.message == data def test_get_booked_contracts_by_period_range(monkeypatch, fixture_booked_contracts): """Test get_booked_contracts with contracts by period range.""" vendor_id = 100 first_period = 99 last_period = 100 expected_items = [ { 'id': 2, 'contract_id': 106, 'period_id': 100, 'vendor_id': vendor_id, 'payment_interval': 'quarter', 'currency_id': 1, 'reserve_rate': 0.1, 'number_of_months_before_payout': 2, 'number_of_installments': 3, 'contract_terms': 'Some Text for Test Contract Term', }, { 'id': 1, 'contract_id': 105, 'period_id': 99, 'vendor_id': vendor_id, 'payment_interval': 'month', 'currency_id': 1, 'reserve_rate': 0.6, 'number_of_months_before_payout': 6, 'number_of_installments': 6, 'contract_terms': 'Some Text for Test Contract Term', }, ] booked_contracts = [item.as_dict() for item in fixture_booked_contracts.message] monkeypatch.setattr( booked_vendor_contract, 'get_booked_contracts_by_period_range', MagicMock(return_value=response.Response(booked_contracts)), ) result = vendor_contracts.get_booked_contracts_by_period_range( vendor_id, first_period, last_period, query.ASC ) assert booked_vendor_contract.get_booked_contracts_by_period_range.called assert result.status == 200 assert result.message == expected_items def test_update_service_type_by_vendor_id(monkeypatch): """Test successful update of vendor contract service type ID and OWS account response.""" vendor_id = 123 service_type_id = 4 vendor_contract_id = 34 mock_active_contract = MagicMock() mock_active_contract.message = {'vendor_contract_id': vendor_contract_id} mock_update_result = MagicMock() mock_update_result.message = 1 mock_ows_response = {'vendor_id': vendor_id, 'vendor_uuid': 'abc-123-uuid'} monkeypatch.setattr( vendor_contracts, 'get_active_vendor_contract', MagicMock(return_value=mock_active_contract), ) monkeypatch.setattr( vendor_contract, 'update_vendor_contract_service_type_id', MagicMock(return_value=mock_update_result), ) monkeypatch.setattr( ows_account, 'get_vendor_uuid_by_vendor_id', MagicMock(return_value=mock_ows_response), ) result = vendor_contracts.update_service_type_by_vendor_id( vendor_id, service_type_id ) assert result == {'vendor_id': vendor_id, 'vendor_uuid': 'abc-123-uuid'} vendor_contracts.get_active_vendor_contract.assert_called_with(vendor_id) vendor_contract.update_vendor_contract_service_type_id.assert_called_with( service_type_id, vendor_id, vendor_contract_id ) ows_account.get_vendor_uuid_by_vendor_id.assert_called_once_with(vendor_id)