"""Test for Legacy product mapping logic tier.""" import datetime from unittest.mock import MagicMock from oto import response from pricing.logic import legacy_product_mapping from pricing.logic import pricing_helper from pricing.logic import product_pricing_override from pricing.models import legacy_database from pricing.models import orchard_pricing_tier as orchard_pricing_tier_model from pricing.models import ows_track from pricing.models import product_orchard_pricing_tier as \ product_orchard_pricing_tier_model from pricing.models import product_pricing_override as \ product_pricing_override_model from pricing.models import track_pricing_override as \ track_pricing_override_model import pytest import requests class MockResponse(): """Mock Response object.""" def __init__(self, status_code, text): """Mock response constructor.""" self.status_code = status_code self.text = text @pytest.fixture def fixture_orchard_pricing_tiers(): """Orchard Pricing tiers fixture.""" orchard_pricing_tiers = [ {'name': 'Mid', 'release_id': 1}, {'name': 'Front', 'release_id': 1}, ] return orchard_pricing_tiers @pytest.fixture def fixture_orchard_pricing_tiers_with_pricing_family_id(): """Orchard Pricing tiers fixture with pf id.""" orchard_pricing_tiers = { 'items': [ { 'name': 'Mid', 'pricing_family_id': 1, 'sort_order': 3, 'is_default': True }, { 'name': 'Front', 'pricing_family_id': 1, 'sort_order': 1, 'is_default': True } ]} return response.Response(orchard_pricing_tiers) @pytest.fixture def fixture_pricing_tiers_legacy_db(): """Orchard Pricing tiers fixture.""" pricing_tiers = [ { 'release_id': 343684, 'pricing_tier': 'Mid', 'pricing_scheme': 'iTunes Film UK' }, { 'release_id': 343684, 'pricing_tier': 'Mid', 'pricing_scheme': 'iTunes Film UK' } ] return pricing_tiers @pytest.fixture def fixture_pricing_tiers_legacy_db_not_the_same(): """Not the same Orchard Pricing tiers fixture.""" pricing_tiers = [ { 'release_id': 343684, 'pricing_tier': 'Mid', 'pricing_scheme': 'iTunes Film UK' }, { 'release_id': 343684, 'pricing_tier': 'Bargain', 'pricing_scheme': 'iTunes Film UK' } ] return pricing_tiers @pytest.fixture def fixture_params(): """Orchard Pricing tiers fixture.""" return { 'product_id': 1, 'pricing_family_id': 1 } @pytest.fixture def fixture_identical_orchard_pricing_tiers(): """Identical orchard pricing tiers fixture.""" pricing_tiers = [ {'pricing_tier': 'Mid', 'release_id': 1}, {'pricing_tier': 'Mid', 'release_id': 1}] return pricing_tiers @pytest.fixture def fixture_not_the_same_orchard_pricing_tiers(): """Non Identical orchard pricing tiers fixture.""" pricing_tiers = [ {'pricing_tier': 'Front', 'release_id': 1}, {'pricing_tier': 'Mid', 'release_id': 1}] return pricing_tiers @pytest.fixture def fixture_orchard_pricing_tier_by_family_id(): """Return an orchard pricing tier.""" return response.Response( {'items': [ { 'name': 'Mid', 'orchard_pricing_tier_id': 1, 'is_default': True }, { 'name': 'Back', 'orchard_pricing_tier_id': 2, 'is_default': False } ]}) @pytest.fixture def fixture_get_by_product_id_and_popt(): """Return response with product orchard pricing tier.""" return response.Response( { 'product_orchard_pricing_tier_id': 2, 'product_id': 1, 'orchard_pricing_tier_id': 3 }) @pytest.fixture def fixture_get_by_product_id_and_popt_not_found(): """Return response with product orchard pricing tier.""" return response.create_not_found_response() @pytest.fixture def fixture_updated_orchard_pricing_tier(): """Return updated orchard pricing tier fixture.""" data = { 'orchard_pricing_tier_id': 1, 'pricing_family_id': 1, 'updated_date': datetime.datetime.now(), 'updated_by': 'Migration Script' } return response.Response(message=data, status=200) @pytest.fixture def fixture_response_404(): """Return response with product orchard pricing tier.""" return response.create_not_found_response() @pytest.fixture def fixture_response_with_deleted_item(): """Return response with deleted product orchard pricing tier.""" return response.Response({ 'orchard_pricing_tier_id': 1, 'product_id': 1 }) @pytest.fixture def fixture_deleted_count_response(): """Response with the deleted count.""" return response.Response({'deleted': 1}) def fixture_product_pricing_override(): """Fixture for single product pricing override.""" return response.Response( message={ 'product_id': 1, 'product_pricing_override_id': 1, 'applies_worldwide': True, 'store_id': 1, 'orchard_pricing_tier_id': 3 }, status=200) @pytest.fixture def fixture_get_legacy_pricing(): """Get legacy pricing.""" return {'pricing': { 'GB': 'Mid', 'FR': 'Front' }, 'chosen_tiers': {}} @pytest.fixture def fixture_product_migrated(): """Get success response.""" return response.Response( status=200, message='Success') @pytest.fixture def fixture_get_product_orchard_pricing_tier(): """Get product orchard pricing tier response.""" return response.Response({ 'product_id': 1, 'pricing_family_id': 2, 'orchard_pricing_tier_id': 1 }) @pytest.fixture def fixture_get_product_orchard_pricing_tier_empty(): """Get product orchard pricing tier empty response.""" return response.Response() @pytest.fixture def fixture_store_pricing_scheme(): """Get store pricing scheme.""" return { 'store_pricing_scheme_id': 1, 'name': 'Itunes Album', 'pricing_family_id': 2, 'store_id': 1, 'use_interval_pricing': False, 'store_pricing_tiers': [ { 'store_pricing_tier_id': 1, 'orchard_pricing_tier_id': 1, 'name': 'Front', 'store_pricing_tier_codes': [ { 'store_pricing_tier_code_id': 1, 'price_code': 100, 'applies_worldwide': False, 'territory_list_include': True, 'territory_set': {'FR', 'GB'} } ] }, { 'store_pricing_tier_id': 2, 'orchard_pricing_tier_id': 2, 'name': 'Mid', 'store_pricing_tier_codes': [ { 'store_pricing_tier_code_id': 1, 'price_code': 200, 'applies_worldwide': False, 'territory_list_include': True, 'territory_set': {'FR', 'GB'} } ] } ] } @pytest.fixture def fixture_get_tracks_by_product_id(): """Get tracks response.""" return response.Response( status=200, message={ 'items': [ {'tuid': 1}, {'tuid': 2} ] }) @pytest.fixture def fixture_default_account_pricing_tier(): """Fixture for single default account pricing tier.""" return response.Response({'orchard_pricing_tier_id': 3}) @pytest.fixture def fixture_default_orchard_pricing_tier(): """Fixture for single default orchard pricing tier.""" return response.Response({'orchard_pricing_tier_id': 3}) @pytest.fixture def fixture_not_found(): """Fixture for not found response.""" return response.create_not_found_response() @pytest.fixture def fixture_get_track_migration_plan(): """Fixture for get_track_migration_plan.""" return response.Response({'orchard_pricing_tier_id': 3}) def test_map_response_array_of_dicts(monkeypatch): """Test the map_response_array_of_dicts function.""" columns = ['upc', 'release_id'] db_response = [(1, 3), (1, 3)] data = legacy_product_mapping.map_response_array_of_dicts( db_response, columns) assert data assert len(data) == len(columns) assert data[0]['upc'] == 1 assert data[0]['release_id'] == 3 def test_check_if_legacy_pricing_tiers_are_the_same( monkeypatch, fixture_identical_orchard_pricing_tiers): """Test the check_if_legacy_pricing_tiers_are_the_same function.""" result = legacy_product_mapping.check_if_legacy_pricing_tiers_are_the_same( fixture_identical_orchard_pricing_tiers) assert result def test_check_if_legacy_pricing_tiers_are_not_the_same( monkeypatch, fixture_not_the_same_orchard_pricing_tiers): """Test if the legacy pricing tiers are not the same.""" expected_value = False result = legacy_product_mapping.check_if_legacy_pricing_tiers_are_the_same( fixture_not_the_same_orchard_pricing_tiers) assert result == expected_value def test_get_matching_orchard_pricing_tier( monkeypatch, fixture_orchard_pricing_tier_by_family_id): """Test get_matching_orchard_pricing_tier function.""" expected_value = None result = legacy_product_mapping. \ get_matching_orchard_pricing_tier( fixture_orchard_pricing_tier_by_family_id.message, 'Front') assert result == expected_value def test_get_matching_orchard_pricing_tier_returns_none( monkeypatch, fixture_orchard_pricing_tier_by_family_id): """Test get_matching_orchard_pricing_tier function.""" result = legacy_product_mapping. \ get_matching_orchard_pricing_tier( fixture_orchard_pricing_tier_by_family_id.message, 'Mid') assert result assert result['name'] == 'Mid' assert result['orchard_pricing_tier_id'] == 1 def test_map_legacy_product_pricing_data_product_already_transferred( monkeypatch, fixture_get_by_product_id_and_popt): """Test if product has been already transferred.""" product_id = 1 pricing_family_id = 1 monkeypatch.setattr( product_orchard_pricing_tier_model, 'get_by_product_id_and_pricing_family_id', MagicMock(return_value=fixture_get_by_product_id_and_popt)) result = legacy_product_mapping.map_legacy_product_pricing_data( pricing_family_id, product_id) assert result.status == 400 def test_map_legacy_product_pricing_data_legacy_pricing_tiers_are_not_the_same( monkeypatch, fixture_params, fixture_response_404, fixture_pricing_tiers_legacy_db_not_the_same): """Test not the same orchard pricing tiers.""" monkeypatch.setattr( product_orchard_pricing_tier_model, 'get_by_product_id_and_pricing_family_id', MagicMock(return_value=fixture_response_404) ) monkeypatch.setattr( legacy_database, 'get_all_product_pricing_tiers', MagicMock(return_value=fixture_pricing_tiers_legacy_db_not_the_same) ) result = legacy_product_mapping.map_legacy_product_pricing_data( fixture_params['pricing_family_id'], fixture_params['product_id']) assert result.status == 400 def test_create_pricing_map(monkeypatch): """Test CREATE pricing map.""" orchard_pricing_tiers = { 'items': [ { 'name': 'Mid', 'pricing_family_id': 1, 'sort_order': 3, 'is_default': True }, { 'name': 'Front', 'pricing_family_id': 1, 'sort_order': 1, 'is_default': True } ]} legacy_pricing = { 'GB': 'Mid', 'FR': 'Front' } result = legacy_product_mapping.create_pricing_map( legacy_pricing, orchard_pricing_tiers) assert result assert result['Mid']['territories'] == {'GB'} assert result['Front']['territories'] == {'FR'} assert result['Mid']['tier'] == orchard_pricing_tiers['items'][0] assert result['Front']['tier'] == orchard_pricing_tiers['items'][1] def test_unmigrate_product( monkeypatch, fixture_response_with_deleted_item, fixture_deleted_count_response, fixture_get_tracks_by_product_id): """Test unmigrate product.""" monkeypatch.setattr( product_orchard_pricing_tier_model, 'delete_by_product_id_and_pricing_family_id', MagicMock(return_value=fixture_response_with_deleted_item)) monkeypatch.setattr( product_pricing_override_model, 'delete_by_product_id', MagicMock(return_value=fixture_deleted_count_response)) monkeypatch.setattr( ows_track, 'get_tracks_by_product_id', MagicMock(return_value=fixture_get_tracks_by_product_id)) monkeypatch.setattr( legacy_product_mapping, 'unmigrate_track', MagicMock(return_value=response.Response())) product_id = 1 result = legacy_product_mapping.unmigrate_product(product_id) assert result assert result.status == 200 def test_unmigrate_track(monkeypatch, fixture_deleted_count_response): """Test unmigrate track.""" monkeypatch.setattr( track_pricing_override_model, 'delete_by_track_id', MagicMock(return_value=fixture_deleted_count_response)) track_id = 2 result = legacy_product_mapping.unmigrate_track(track_id) assert result assert result.status == 200 assert result.message['deleted'] == 1 def test_map_legacy_product_pricing_data_product_has_not_been_transferred_400( monkeypatch, fixture_get_by_product_id_and_popt_not_found): """Test if product has been already transferred.""" product_id = 1 pricing_family_id = 1 monkeypatch.setattr( product_orchard_pricing_tier_model, 'get_by_product_id_and_pricing_family_id', MagicMock(return_value=fixture_get_by_product_id_and_popt_not_found)) result = legacy_product_mapping.map_legacy_product_pricing_data( pricing_family_id, product_id) assert result.status == 400 def test_map_legacy_product_pricing_data_product_has_not_been_transferred( monkeypatch, fixture_get_by_product_id_and_popt_not_found, fixture_orchard_pricing_tiers_with_pricing_family_id, fixture_get_legacy_pricing, fixture_product_migrated): """Test if product has been already transferred.""" product_id = '1' pricing_family_id = '1' monkeypatch.setattr( product_orchard_pricing_tier_model, 'get_by_product_id_and_pricing_family_id', MagicMock(return_value=fixture_get_by_product_id_and_popt_not_found)) monkeypatch.setattr( orchard_pricing_tier_model, 'get_orchard_pricing_tier_by_pricing_family_id', MagicMock( return_value=fixture_orchard_pricing_tiers_with_pricing_family_id) ) monkeypatch.setattr( legacy_product_mapping, 'get_legacy_pricing', MagicMock( return_value=fixture_get_legacy_pricing) ) monkeypatch.setattr( legacy_product_mapping, 'migrate_to_orchard_tier', MagicMock( return_value=fixture_product_migrated) ) result = legacy_product_mapping.map_legacy_product_pricing_data( pricing_family_id, product_id) assert result assert result.status == 200 def test_migrate_to_orchard_tier_did_not_update(monkeypatch): """Test migrate to orchard tier did not update.""" monkeypatch.setattr( product_orchard_pricing_tier_model, 'update_product_with_orchard_pricing_tier', MagicMock( return_value=response.Response(status=404)) ) orchard_tier = { 'orchard_pricing_tier_id': 5} result = legacy_product_mapping.migrate_to_orchard_tier( orchard_tier, {}, 1, 1) assert result.status == 404 def test_migrate_to_orchard_tier( monkeypatch, fixture_updated_orchard_pricing_tier): """Test migrate to orchard tier.""" monkeypatch.setattr( product_orchard_pricing_tier_model, 'update_product_with_orchard_pricing_tier', MagicMock( return_value=fixture_updated_orchard_pricing_tier) ) monkeypatch.setattr( product_pricing_override, 'create_product_pricing_override', MagicMock( return_value=fixture_product_pricing_override) ) orchard_tier = { 'orchard_pricing_tier_id': 1, 'name': 'Front' } pricing_map = {'Front': {'territories': {'FR'}, 'tier': {'name': 'Front', 'sort_order': 1, 'orchard_pricing_tier_id': 1, 'pricing_family_id': 1, 'is_default': True}}, 'Mid': {'territories': {'GB'}, 'tier': {'name': 'Mid', 'sort_order': 3, 'pricing_family_id': 1, 'orchard_pricing_tier_id': 3, 'is_default': True}}} result = legacy_product_mapping.migrate_to_orchard_tier( orchard_tier, pricing_map, 1, 1) assert result assert result.status == 200 def test_migrate_to_orchard_tier_with_error( monkeypatch, fixture_updated_orchard_pricing_tier): """Test migrate to orchard tier returns error.""" monkeypatch.setattr( product_orchard_pricing_tier_model, 'update_product_with_orchard_pricing_tier', MagicMock( return_value=fixture_updated_orchard_pricing_tier) ) monkeypatch.setattr( product_pricing_override, 'create_product_pricing_override', MagicMock( return_value=response.create_error_response(status=500, code=500, message='Error')) ) orchard_tier = { 'orchard_pricing_tier_id': 1, 'name': 'Front' } pricing_map = {'Front': {'territories': {'FR'}, 'tier': {'name': 'Front', 'sort_order': 1, 'orchard_pricing_tier_id': 1, 'pricing_family_id': 1, 'is_default': True}}, 'Mid': {'territories': {'GB'}, 'tier': {'name': 'Mid', 'sort_order': 3, 'pricing_family_id': 1, 'orchard_pricing_tier_id': 3, 'is_default': True}}} result = legacy_product_mapping.migrate_to_orchard_tier( orchard_tier, pricing_map, 1, 1) assert result.errors assert result.status == 500 def test_get_legacy_pricing(monkeypatch): """Test legacy pricing.""" monkeypatch.setattr( legacy_database, 'get_default_pricing_codes', MagicMock( return_value=[{'country_code': 'FR', 'price_code': 3, 'pricing_tier': 'Mid'}])) monkeypatch.setattr( legacy_database, 'get_all_product_territory_pricing', MagicMock( return_value=[{'country_code': 'CA', 'price_code': '202', 'pricing_tier': 'Mid'}])) result = legacy_product_mapping.get_legacy_pricing(1, 1) assert result assert result['chosen_tiers']['Mid'] == 1 assert result['priced_territory_count'] == 1 assert result['pricing']['CA'] == 'Mid' assert result['pricing']['FR'] == 'Mid' def test_count_bulk_migrate_film_products(monkeypatch): """Test legacy pricing.""" monkeypatch.setattr( legacy_database, 'get_all_film_product_ids', MagicMock( return_value=[{'release_id': 123}, {'release_id': 356}])) monkeypatch.setattr( product_orchard_pricing_tier_model, 'get_all_migrated_product_ids', MagicMock( return_value=[123])) result = legacy_product_mapping.count_bulk_migrate_film_products() assert result assert result.message['migrated_product_ids'] == 1 assert result.message['product_ids_to_migrate'] == 1 assert result.message['all_product_ids'] == 2 def test_bulk_migrate_film_products(monkeypatch): """Test legacy pricing.""" monkeypatch.setattr( requests, 'get', MagicMock( return_value=MockResponse(status_code=200, text='Migrated') ) ) monkeypatch.setattr( legacy_database, 'get_all_film_product_ids', MagicMock( return_value=[{'release_id': 123}, {'release_id': 356}])) monkeypatch.setattr( product_orchard_pricing_tier_model, 'get_all_migrated_product_ids', MagicMock( return_value=[123])) result = legacy_product_mapping.bulk_migrate_film_products() assert result assert result.status == 200 assert result.message[0]['product_id'] == 356 assert result.message[0]['result'] == 'Migrated' def test_bulk_migrate_film_products_did_not_migrate(monkeypatch): """Test legacy pricing.""" monkeypatch.setattr( requests, 'get', MagicMock( return_value=MockResponse(status_code=500, text='DB error') ) ) monkeypatch.setattr( legacy_database, 'get_all_film_product_ids', MagicMock( return_value=[{'release_id': 123}, {'release_id': 356}])) monkeypatch.setattr( product_orchard_pricing_tier_model, 'get_all_migrated_product_ids', MagicMock( return_value=[123])) result = legacy_product_mapping.bulk_migrate_film_products() assert result assert result.status == 200 assert result.message[0]['product_id'] == 356 assert result.message[0]['result'] == 'Did not migrate' def test_merge_orchard_pricing_tiers(): """Test that the orchard pricing tiers are correctly merged.""" track_migration_plans = { 123: { 'orchard_pricing_tier_id': 1, 'track_pricing_overrides_to_create': [] }, 456: { 'orchard_pricing_tier_id': 1, 'track_pricing_overrides_to_create': [] }, 789: { 'orchard_pricing_tier_id': 2, 'track_pricing_overrides_to_create': [] } } expected_track_migration_plans = { 123: { 'orchard_pricing_tier_id': 1, 'track_pricing_overrides_to_create': [] }, 456: { 'orchard_pricing_tier_id': 1, 'track_pricing_overrides_to_create': [] }, 789: { 'orchard_pricing_tier_id': 2, 'track_pricing_overrides_to_create': [{ 'track_id': 789, 'sort_order': 1, 'orchard_pricing_tier_id': 2, 'applies_worldwide': True }] }, 'All': { 'orchard_pricing_tier_id': 1 } } legacy_product_mapping.merge_orchard_pricing_tiers( track_migration_plans, 1) assert track_migration_plans == expected_track_migration_plans def test_jit_music_migration_film_product( monkeypatch, fixture_get_product_orchard_pricing_tier, fixture_response_404): """Test jit music migration for a product already migrated.""" monkeypatch.setattr( product_orchard_pricing_tier_model, 'get_by_product_id_and_pricing_family_id', MagicMock(return_value=fixture_get_product_orchard_pricing_tier)) result = legacy_product_mapping.jit_music_migration(1) assert result assert 'status' in result.message assert result.message['status'] == \ 'Not migrating because the product is a Film' def test_jit_music_migration_physical_product( monkeypatch, fixture_get_product_orchard_pricing_tier, fixture_response_404): """Test jit music migration for a product already migrated.""" def get_product_orchard_pricing_tier_side_effect( product_id, pricing_family_id): if pricing_family_id == 1: return fixture_response_404 if pricing_family_id == 4: return fixture_get_product_orchard_pricing_tier return fixture_get_product_orchard_pricing_tier monkeypatch.setattr( product_orchard_pricing_tier_model, 'get_by_product_id_and_pricing_family_id', MagicMock(side_effect=get_product_orchard_pricing_tier_side_effect)) result = legacy_product_mapping.jit_music_migration(1) assert result assert 'status' in result.message assert result.message['status'] == \ 'Not migrating because the product is Physical' def test_jit_music_migration_already_migrated( monkeypatch, fixture_get_product_orchard_pricing_tier, fixture_response_404): """Test jit music migration for a product already migrated.""" def get_product_orchard_pricing_tier_side_effect( product_id, pricing_family_id): if pricing_family_id == 1: return fixture_response_404 if pricing_family_id == 4: return fixture_response_404 return fixture_get_product_orchard_pricing_tier monkeypatch.setattr( product_orchard_pricing_tier_model, 'get_by_product_id_and_pricing_family_id', MagicMock(side_effect=get_product_orchard_pricing_tier_side_effect)) result = legacy_product_mapping.jit_music_migration(1) assert result assert 'status' in result.message assert result.message['status'] == 'Already migrated' def test_jit_music_migration_need_migration( monkeypatch, fixture_response_404, fixture_get_product_orchard_pricing_tier_empty, fixture_product_migrated): """Test jit music migration for a product needing migration.""" def get_product_orchard_pricing_tier_side_effect( product_id, pricing_family_id): if pricing_family_id == 1: return fixture_response_404 return fixture_get_product_orchard_pricing_tier_empty monkeypatch.setattr( product_orchard_pricing_tier_model, 'get_by_product_id_and_pricing_family_id', MagicMock(side_effect=get_product_orchard_pricing_tier_side_effect)) monkeypatch.setattr( legacy_product_mapping, 'execute_music_migration', MagicMock(return_value=fixture_product_migrated)) product_id = 1 result = legacy_product_mapping.jit_music_migration(product_id) legacy_product_mapping.execute_music_migration.assert_called_with( product_id) assert result def test_get_default_orchard_pricing_tier( monkeypatch, fixture_orchard_pricing_tier_by_family_id): """Test that the default orchard pricing tier is returned.""" monkeypatch.setattr( orchard_pricing_tier_model, 'get_orchard_pricing_tier_by_pricing_family_id', MagicMock(return_value=fixture_orchard_pricing_tier_by_family_id)) result = legacy_product_mapping.get_default_orchard_pricing_tier(1) assert result assert result.message assert result.message['name'] == 'Mid' def test_get_orchard_tier_id_for_price_code_and_territories( monkeypatch, fixture_store_pricing_scheme): """Test that the orchard pricing tier id is returned.""" monkeypatch.setattr( legacy_product_mapping, 'get_store_pricing_scheme', MagicMock(return_value=fixture_store_pricing_scheme)) price_code = 100 pricing_family_id = 1 store_id = 1 territories = ['FR'] result = legacy_product_mapping.\ get_orchard_tier_id_for_price_code_and_territories( price_code, pricing_family_id, store_id, territories) assert result == 1 def test_get_orchard_tier_id_for_price_code_and_territories_scheme_not_found( monkeypatch, fixture_store_pricing_scheme): """Test that None is returned when scheme is not found.""" monkeypatch.setattr( legacy_product_mapping, 'get_store_pricing_scheme', MagicMock(return_value=None)) price_code = 100 pricing_family_id = 1 store_id = 1 territories = ['FR'] result = legacy_product_mapping.\ get_orchard_tier_id_for_price_code_and_territories( price_code, pricing_family_id, store_id, territories) assert result is None def test_get_orchard_tier_id_for_price_code_and_territories_no_tier_found( monkeypatch, fixture_store_pricing_scheme): """Test that None is returned when the price code does not correspond.""" monkeypatch.setattr( legacy_product_mapping, 'get_store_pricing_scheme', MagicMock(return_value=fixture_store_pricing_scheme)) price_code = 300 pricing_family_id = 1 store_id = 1 territories = ['FR'] result = legacy_product_mapping.\ get_orchard_tier_id_for_price_code_and_territories( price_code, pricing_family_id, store_id, territories) assert result is None def test_merge_multiple_identical_track_overrides_all_mergeable(): """Test that identical track overrides can be merged.""" track_migration_plans = { 123: { 'track_pricing_overrides_to_create': [ { 'track_id': 123, 'sort_order': 1, 'orchard_pricing_tier_id': 1, 'stores': [1], 'territories': ['JP'], 'applies_worldwide': False, 'territory_list_include': True }, { 'track_id': 123, 'sort_order': 2, 'orchard_pricing_tier_id': 2, 'stores': [], 'territories': ['CA', 'GB'], 'applies_worldwide': False, 'territory_list_include': True } ] }, 456: { 'track_pricing_overrides_to_create': [ { 'track_id': 456, 'sort_order': 3, 'orchard_pricing_tier_id': 1, 'stores': [1], 'territories': ['JP'], 'applies_worldwide': False, 'territory_list_include': True }, { 'track_id': 456, 'sort_order': 4, 'orchard_pricing_tier_id': 2, 'stores': [], 'territories': ['CA', 'GB'], 'applies_worldwide': False, 'territory_list_include': True } ] }, 789: { 'track_pricing_overrides_to_create': [ { 'track_id': 789, 'sort_order': 5, 'orchard_pricing_tier_id': 1, 'stores': [1], 'territories': ['JP'], 'applies_worldwide': False, 'territory_list_include': True }, { 'track_id': 789, 'sort_order': 6, 'orchard_pricing_tier_id': 2, 'stores': [], 'territories': ['CA', 'GB'], 'applies_worldwide': False, 'territory_list_include': True } ] } } product_id = 1 expected_track_migration_plans = { 123: {'track_pricing_overrides_to_create': []}, 456: {'track_pricing_overrides_to_create': []}, 789: {'track_pricing_overrides_to_create': []}, 'All': { 'product_pricing_overrides_to_create': [ { 'product_id': 1, 'pricing_family_id': 3, 'sort_order': 1, 'orchard_pricing_tier_id': 1, 'stores': [1], 'territories': ['JP'], 'applies_worldwide': False, 'territory_list_include': True }, { 'product_id': 1, 'pricing_family_id': 3, 'sort_order': 2, 'orchard_pricing_tier_id': 2, 'stores': [], 'territories': ['CA', 'GB'], 'applies_worldwide': False, 'territory_list_include': True } ] } } legacy_product_mapping.merge_multiple_identical_track_overrides( track_migration_plans, product_id) assert track_migration_plans == expected_track_migration_plans def test_merge_multiple_identical_track_overrides_some_mergeable(): """Test that identical track overrides can be merged.""" track_migration_plans = { 123: { 'track_pricing_overrides_to_create': [ { 'track_id': 123, 'sort_order': 1, 'orchard_pricing_tier_id': 5, 'stores': [], 'territories': ['RU'], 'applies_worldwide': False, 'territory_list_include': True }, { 'track_id': 123, 'sort_order': 2, 'orchard_pricing_tier_id': 1, 'stores': [1], 'territories': ['JP'], 'applies_worldwide': False, 'territory_list_include': True }, { 'track_id': 123, 'sort_order': 3, 'orchard_pricing_tier_id': 2, 'stores': [], 'territories': ['CA', 'GB'], 'applies_worldwide': False, 'territory_list_include': True } ] }, 456: { 'track_pricing_overrides_to_create': [ { 'track_id': 456, 'sort_order': 4, 'orchard_pricing_tier_id': 1, 'stores': [1], 'territories': ['JP'], 'applies_worldwide': False, 'territory_list_include': True }, { 'track_id': 456, 'sort_order': 5, 'orchard_pricing_tier_id': 2, 'stores': [], 'territories': ['CA', 'GB'], 'applies_worldwide': False, 'territory_list_include': True } ] }, 789: { 'track_pricing_overrides_to_create': [ { 'track_id': 789, 'sort_order': 6, 'orchard_pricing_tier_id': 1, 'stores': [1], 'territories': ['JP'], 'applies_worldwide': False, 'territory_list_include': True }, { 'track_id': 789, 'sort_order': 7, 'orchard_pricing_tier_id': 2, 'stores': [], 'territories': ['CA', 'GB'], 'applies_worldwide': False, 'territory_list_include': True }, { 'track_id': 789, 'sort_order': 8, 'orchard_pricing_tier_id': 5, 'stores': [], 'territories': ['RU'], 'applies_worldwide': False, 'territory_list_include': True } ] } } product_id = 1 expected_track_migration_plans = { 123: { 'track_pricing_overrides_to_create': [ { 'track_id': 123, 'sort_order': 1, 'orchard_pricing_tier_id': 5, 'stores': [], 'territories': ['RU'], 'applies_worldwide': False, 'territory_list_include': True } ] }, 456: {'track_pricing_overrides_to_create': []}, 789: { 'track_pricing_overrides_to_create': [ { 'track_id': 789, 'sort_order': 2, 'orchard_pricing_tier_id': 5, 'stores': [], 'territories': ['RU'], 'applies_worldwide': False, 'territory_list_include': True } ] }, 'All': { 'product_pricing_overrides_to_create': [ { 'product_id': 1, 'pricing_family_id': 3, 'sort_order': 1, 'orchard_pricing_tier_id': 1, 'stores': [1], 'territories': ['JP'], 'applies_worldwide': False, 'territory_list_include': True }, { 'product_id': 1, 'pricing_family_id': 3, 'sort_order': 2, 'orchard_pricing_tier_id': 2, 'stores': [], 'territories': ['CA', 'GB'], 'applies_worldwide': False, 'territory_list_include': True } ] } } legacy_product_mapping.merge_multiple_identical_track_overrides( track_migration_plans, product_id) assert track_migration_plans == expected_track_migration_plans def test_merge_multiple_identical_track_overrides_none_mergeable(): """Test that identical track overrides can be merged.""" track_migration_plans = { 123: { 'track_pricing_overrides_to_create': [ { 'track_id': 123, 'sort_order': 1, 'orchard_pricing_tier_id': 5, 'stores': [], 'territories': ['RU'], 'applies_worldwide': False, 'territory_list_include': True } ] }, 456: { 'track_pricing_overrides_to_create': [ { 'track_id': 456, 'sort_order': 2, 'orchard_pricing_tier_id': 5, 'stores': [], 'territories': ['CA'], 'applies_worldwide': False, 'territory_list_include': True } ] }, 789: { 'track_pricing_overrides_to_create': [ { 'track_id': 789, 'sort_order': 3, 'orchard_pricing_tier_id': 5, 'stores': [], 'territories': ['GB'], 'applies_worldwide': False, 'territory_list_include': True } ] } } product_id = 1 expected_track_migration_plans = { 123: { 'track_pricing_overrides_to_create': [ { 'track_id': 123, 'sort_order': 1, 'orchard_pricing_tier_id': 5, 'stores': [], 'territories': ['RU'], 'applies_worldwide': False, 'territory_list_include': True } ] }, 456: { 'track_pricing_overrides_to_create': [ { 'track_id': 456, 'sort_order': 2, 'orchard_pricing_tier_id': 5, 'stores': [], 'territories': ['CA'], 'applies_worldwide': False, 'territory_list_include': True } ] }, 789: { 'track_pricing_overrides_to_create': [ { 'track_id': 789, 'sort_order': 3, 'orchard_pricing_tier_id': 5, 'stores': [], 'territories': ['GB'], 'applies_worldwide': False, 'territory_list_include': True } ] } } legacy_product_mapping.merge_multiple_identical_track_overrides( track_migration_plans, product_id) assert track_migration_plans == expected_track_migration_plans def test_merge_multiple_identical_track_overrides_custom(): """Test that identical track overrides can be merged.""" track_migration_plans = { 123: { 'track_pricing_overrides_to_create': [ { 'track_id': 123, 'sort_order': 1, 'custom_price': 9.99, 'custom_currency_code': 'USD', 'stores': [1], 'territories': ['JP'], 'applies_worldwide': False, 'territory_list_include': True } ] }, 456: { 'track_pricing_overrides_to_create': [ { 'track_id': 456, 'sort_order': 3, 'custom_price': 9.99, 'custom_currency_code': 'USD', 'stores': [1], 'territories': ['JP'], 'applies_worldwide': False, 'territory_list_include': True } ] }, 789: { 'track_pricing_overrides_to_create': [ { 'track_id': 789, 'sort_order': 5, 'custom_price': 9.99, 'custom_currency_code': 'USD', 'stores': [1], 'territories': ['JP'], 'applies_worldwide': False, 'territory_list_include': True } ] } } product_id = 1 expected_track_migration_plans = { 123: {'track_pricing_overrides_to_create': []}, 456: {'track_pricing_overrides_to_create': []}, 789: {'track_pricing_overrides_to_create': []}, 'All': { 'product_pricing_overrides_to_create': [ { 'product_id': 1, 'pricing_family_id': 3, 'sort_order': 1, 'custom_price': 9.99, 'custom_currency_code': 'USD', 'stores': [1], 'territories': ['JP'], 'applies_worldwide': False, 'territory_list_include': True } ] } } legacy_product_mapping.merge_multiple_identical_track_overrides( track_migration_plans, product_id) assert track_migration_plans == expected_track_migration_plans def test_create_migration_plans_for_missing_tracks_default(monkeypatch): """Test that the correct overrides are created.""" monkeypatch.setattr( legacy_product_mapping, 'get_default_orchard_pricing_tier', MagicMock(return_value=response.Response({ 'orchard_pricing_tier_id': 'DEFAULT' }))) monkeypatch.setattr( legacy_product_mapping, 'get_special_jp_track_override', MagicMock(return_value={ 'track_id': 2, 'orchard_pricing_tier_id': 'JAPAN', 'sort_order': 1, 'stores': 1, 'territories': ['JP'], 'applies_worldwide': False, 'territory_list_include': True })) track_rows = [ {'id': 1}, {'id': 2} ] track_migration_plans = { 'All': {'orchard_pricing_tier_id': 'DEFAULT'}, 1: {} } sort_order = 1 expected_track_migration_plans = { 'All': {'orchard_pricing_tier_id': 'DEFAULT'}, 1: {}, 2: { 'track_pricing_overrides_to_create': [ { 'track_id': 2, 'orchard_pricing_tier_id': 'JAPAN', 'sort_order': 1, 'stores': 1, 'territories': ['JP'], 'applies_worldwide': False, 'territory_list_include': True } ] } } legacy_product_mapping.create_migration_plans_for_missing_tracks( track_rows, track_migration_plans, sort_order) assert expected_track_migration_plans == track_migration_plans def test_create_migration_plans_for_missing_tracks_not_default(monkeypatch): """Test that the correct overrides are created.""" monkeypatch.setattr( legacy_product_mapping, 'get_default_orchard_pricing_tier', MagicMock(return_value=response.Response({ 'orchard_pricing_tier_id': 'DEFAULT' }))) monkeypatch.setattr( legacy_product_mapping, 'get_special_jp_track_override', MagicMock(return_value={ 'track_id': 2, 'orchard_pricing_tier_id': 'JAPAN', 'sort_order': 1, 'stores': 1, 'territories': ['JP'], 'applies_worldwide': False, 'territory_list_include': True })) track_rows = [ {'id': 1}, {'id': 2} ] track_migration_plans = { 'All': {'orchard_pricing_tier_id': 'NOT DEFAULT'}, 1: {} } sort_order = 1 expected_track_migration_plans = { 'All': {'orchard_pricing_tier_id': 'NOT DEFAULT'}, 1: {}, 2: { 'track_pricing_overrides_to_create': [ { 'track_id': 2, 'orchard_pricing_tier_id': 'JAPAN', 'sort_order': 1, 'stores': 1, 'territories': ['JP'], 'applies_worldwide': False, 'territory_list_include': True }, { 'track_id': 2, 'sort_order': 2, 'orchard_pricing_tier_id': 'DEFAULT', 'applies_worldwide': True } ] } } legacy_product_mapping.create_migration_plans_for_missing_tracks( track_rows, track_migration_plans, sort_order) assert expected_track_migration_plans == track_migration_plans def test_purge_legacy_overrides(): """Test that the correct overrides are purged.""" overrides = [ { 'id': 1 }, { 'id': 2, 'applies_worldwide': True }, { 'id': 3, 'applies_worldwide': True, 'stores': [] }, { 'id': 4, 'applies_worldwide': False }, { 'id': 5, 'stores': [1] } ] expected_result = [ { 'id': 2, 'applies_worldwide': True }, { 'id': 3, 'applies_worldwide': True, 'stores': [] } ] result = legacy_product_mapping.purge_legacy_overrides(overrides) assert expected_result == result def test_get_product_migration_plan_when_label_default( monkeypatch, fixture_default_account_pricing_tier): """Test that the default account pricing is returned for music product.""" monkeypatch.setattr( legacy_database, 'get_product_pricing_exclude_stores', MagicMock(return_value={})) monkeypatch.setattr( pricing_helper, 'get_default_account_pricing_tier', MagicMock(return_value=fixture_default_account_pricing_tier)) result = legacy_product_mapping.get_product_migration_plan(1234) assert result == fixture_default_account_pricing_tier.message def test_get_product_migration_plan_when_no_label_default( monkeypatch, fixture_not_found, fixture_default_orchard_pricing_tier): """Test that the default orchard pricing is returned for music product.""" monkeypatch.setattr( legacy_database, 'get_product_pricing_exclude_stores', MagicMock(return_value={})) monkeypatch.setattr( pricing_helper, 'get_default_account_pricing_tier', MagicMock(return_value=fixture_not_found)) monkeypatch.setattr( legacy_product_mapping, 'get_default_orchard_pricing_tier', MagicMock(return_value=fixture_default_orchard_pricing_tier)) result = legacy_product_mapping.get_product_migration_plan(1234) assert result == fixture_default_orchard_pricing_tier.message def test_get_track_migration_plan_when_label_default( monkeypatch, fixture_default_account_pricing_tier): """Test that the default account pricing is returned for track.""" monkeypatch.setattr( legacy_database, 'get_track_pricing_exclude_stores', MagicMock(return_value={})) monkeypatch.setattr( pricing_helper, 'get_default_account_pricing_tier', MagicMock(return_value=fixture_default_account_pricing_tier)) get_track_migration_plan_response = { 'All': { 'orchard_pricing_tier_id': fixture_default_account_pricing_tier. message['orchard_pricing_tier_id'], 'product_pricing_overrides_to_create': [] }} result = legacy_product_mapping.get_track_migration_plan(2089534) assert result == get_track_migration_plan_response def test_get_track_migration_plan_when_no_label_default( monkeypatch, fixture_not_found, fixture_default_orchard_pricing_tier): """Test that the default orchard pricing is returned for track.""" monkeypatch.setattr( legacy_database, 'get_track_pricing_exclude_stores', MagicMock(return_value={})) monkeypatch.setattr( pricing_helper, 'get_default_account_pricing_tier', MagicMock(return_value=fixture_not_found)) monkeypatch.setattr( legacy_product_mapping, 'get_default_orchard_pricing_tier', MagicMock(return_value=fixture_default_orchard_pricing_tier)) get_track_migration_plan_response = { 'All': { 'orchard_pricing_tier_id': fixture_default_orchard_pricing_tier. message['orchard_pricing_tier_id'], 'product_pricing_overrides_to_create': [] }} result = legacy_product_mapping.get_track_migration_plan(2089534) assert result == get_track_migration_plan_response