"""conftest. This file gets picked up when running py.test tests: http://pytest.org/latest/writing_plugins.html#conftest """ import datetime from unittest.mock import MagicMock from unittest.mock import Mock from unittest.mock import NonCallableMagicMock from unittest.mock import patch import application from flask import g from flexmock import flexmock from oto import response from oto import status from owsrequest import request from owsrequest.constants import headers from owsrequest import test_utils import pytest from sqlalchemy import exc from product_digital import api from product_digital.constants import account from product_digital.constants import error from product_digital.constants import header as header_constants from product_digital.models import language from product_digital.models import product_events from product_digital.models import project from product_digital.models import release from product_digital.validation import audio_product as product_validation from tests.factories import language as lang_factory DATE_OBJECTS_FOR_RESPONSE_MESSAGE = [ datetime.date(2016, 12, 12), datetime.date(2016, 12, 13), datetime.date(2016, 12, 14)] DATE_STRINGS_FOR_RESPONSE_MESSAGE = ['2016-12-12', '2016-12-13', '2016-12-14'] @pytest.fixture def test_request_context(): """Return flask app test_request_context, including a mock logger. Yields: RequestContext: flask app test_request_context, including a mock logger. """ # noqa class MockLog: def error(self): pass def info(self): pass def warning(self): pass with api.app.test_request_context() as test_request_context: g.log = MockLog g.log = MagicMock() g.ows = NonCallableMagicMock(correlation_id='abc-123-def-ghi') g.request_context = MagicMock() g.request_context.context_type = 'whatever' g.request_context.profile_id = 7123 g.request_context.profile_type = 'vendor' g.request_context.identity_uuid = 'foo' yield test_request_context @pytest.fixture def client(): """Return flask test client. Returns: flask client: flask test client """ return application.app.test_client() @pytest.fixture def mock_authorization_backend(): """Return a mocked AuthorizationBackend, patched onto the api module global.""" mock = MagicMock() with patch("product_digital.api.authorization_backend", mock): yield mock @pytest.fixture def valid_headers(correlation_id): """Generally-required headers for calls to this service. Returns: dict: header values. """ return { header_constants.CORRELATION_ID: correlation_id, header_constants.CONTENT_TYPE: 'application/json'} @pytest.fixture def valid_headers_with_orchard_user_id(valid_headers, orchard_user_id): """Required headers plus Orchard-User-Id. Currently Orchard-User-Id is required for calls that involve ows-track. """ valid_headers.update({header_constants.ORCHARD_USER_ID: orchard_user_id}) return valid_headers @pytest.fixture def valid_oa_headers_with_orchard_user_id(valid_headers, oa_orchard_user_id): """Required headers plus Orchard-User-Id. Currently Orchard-User-Id is required for calls that involve ows-track. """ valid_headers.update({header_constants.ORCHARD_USER_ID: oa_orchard_user_id}) return valid_headers @pytest.fixture def valid_profile_headers(): """Required headers plus Orchard-User-Id. Currently Orchard-User-Id is required for calls that involve ows-track. """ return { header_constants.ORCHARD_USER_ID: 'alw:123', headers.ORCHARD_PROFILE_TYPE: 'vendor', headers.ORCHARD_PROFILE_ID: '7123', headers.ORCHARD_IDENTITY_ID: 'identity' } @pytest.fixture def account_id(): """Fake account id for testing.""" return 12345 @pytest.fixture def vendor_id(): """Fake vendor id for testing.""" return 123456 @pytest.fixture def valid_headers_for_vendor(valid_headers, account_id): """Return valid headers with vendor type and account id. Returns: dict: header dict """ valid_headers.update({ header_constants.GRASS_ACCOUNT_TYPE: account.VENDOR_TYPE, header_constants.GRASS_ACCOUNT_ID: account_id}) return valid_headers @pytest.fixture def valid_headers_for_subaccount(valid_headers, account_id): """Return valid headers with subaccount type and account id. Returns: dict: header dict """ valid_headers.update({ header_constants.GRASS_ACCOUNT_TYPE: account.SUBACCOUNT_TYPE, header_constants.GRASS_ACCOUNT_ID: account_id}) return valid_headers @pytest.fixture def release_artist_data(): """Test data for a release artist instance.""" return { 'release_id': 1234, 'upc': '123456789012', 'role': 'banjoist', 'artist_name': 'Joe Artist'} @pytest.fixture def base_release_data(): """Data used to create a release. Returns: dict: release data """ return { 'artist_id': 178508, 'project_id': 12345, 'delivered_version': 'Live at Wembley', 'release_name': 'Eggensperger And Heydt Yodel The Classics', 'product_code': 'AProduct123', 'upc': '123456789012', 'version': 'Delux', 'product_type_id': 1, 'distribution_format_id': 1, 'release_status': 'label_processing', 'vendor_catalog_number': 'ABC123'} @pytest.fixture def base_product_data(): """Basic product data included in POST request body. Returns: dict: product data """ return { 'format': 'Full Length', 'project_id': 12345, 'product_name': 'Eggensperger And Heydt Yodel The Classics', 'product_code': 'AProduct123', 'upc': '123456789012', 'version': 'Delux'} @pytest.fixture def base_product_data_with_vendor(base_product_data, account_id): """Basic product data plus account info required to save. Returns: dict: product data """ base_product_data.update( {'account_type': 'vendor', 'account_id': account_id}) return base_product_data @pytest.fixture def expected_create_release_response_message(): """Release data returned after release model create function is called. Returns: dict: release data """ return { 'artist_id': 178508, 'display_upc': '123456789012', 'distribution_format_id': 1, 'format': 'Full Length', 'product_code': 'AProduct123', 'product_type_id': 1, 'project_id': 12345, 'release_id': 1, 'release_name': 'Eggensperger And Heydt Yodel The Classics', 'release_status': 'label_processing', 'upc': '123456789012', 'vendor_catalog_number': 'ABC123', 'version': 'Delux', 'not_for_distribution': 'N', } @pytest.fixture def expected_create_product_response_message(): """Product data returned from product.create function in logic. Returns: dict: product data """ return { 'artist_id': 178508, 'delivered_version': None, 'description': None, 'distribution_format_id': 1, 'format': 'Full Length', 'genre_id': None, 'c_line': None, 'imprint': None, 'manufacturer_upc': None, 'meta_language': None, 'p_line': None, 'preorder_date': None, 'product_code': 'AProduct123', 'previewable': None, 'product_id': 1, 'product_name': 'Eggensperger And Heydt Yodel The Classics', 'product_type_id': 1, 'project_id': 12345, 'product_artists': [], 'release_date': None, 'release_status': 'label_processing', 'is_error_correction_and_action_required': False, 'sale_start_date': None, 'special_instructions': None, 'subaccount_id': None, 'subgenre_id': None, 'upc': '123456789012', 'vendor_catalog_number': 'ABC123', 'version': 'Delux', 'not_for_distribution': 'N', } @pytest.fixture def expected_product_payload(base_product_data): """Product data passed to release.create function. This is the payload expected to be passed to the audio_product.create function, called from the logic's create function. Additional properties are added in the logic layer that are then passed to the model. Returns: dict: product payload """ return { 'artist_id': base_product_data.get('artist_id'), 'format': base_product_data.get('format'), 'project_id': base_product_data.get('project_id'), 'product_name': base_product_data.get('product_name'), 'product_code': base_product_data.get('product_code'), 'upc': base_product_data.get('upc'), 'version': base_product_data.get('version'), 'product_type_id': 1, 'distribution_format_id': 1, 'release_status': 'label_processing', 'vendor_catalog_number': 'ABC123', 'not_for_distribution': 'N', } @pytest.fixture def expected_product_payload_without_upc(expected_product_payload): """Product data passed to product.create function in logic. Returns: dict: product payload """ del expected_product_payload['upc'] return expected_product_payload @pytest.fixture def valid_get_header(): """Return valid GET request headers. Returns: (dict): header dict """ return { header_constants.CORRELATION_ID: '1234567890-1234567890', header_constants.GRASS_ACCOUNT_TYPE: 'vendor', header_constants.GRASS_ACCOUNT_ID: '25257'} @pytest.fixture def get_project_response_message(): """Return expected message from project get project method.""" return { 'artist_id': 178508, 'project_name': 'this project', 'project_code': 'ABC123', 'project_id': 12345, 'vendor_id': 9999, 'subaccount_id': 0, } @pytest.fixture def project_model_response(get_project_response_message): """Return expected response from project get project method.""" return response.Response(get_project_response_message) @pytest.fixture def mock_ows_project_get_project(mocker, get_project_response_message): """mock_ows_project_get_project.""" get_specs = [ { 'service': 'ows-project-manager', 'path': '/project/12345', 'status': 200, 'json': get_project_response_message }, { 'service': 'ows-project-manager', 'path': '/project/5678', 'status': 200, 'json': get_project_response_message }, { 'service': 'ows-project-manager', 'path': '/project/5678?include_deletions=True', 'status': 200, 'json': get_project_response_message } ] mocker.patch.object( request, 'get', test_utils.mock_ows_requests(get_specs)) @pytest.fixture def mock_vendor_response(): """Mock vendor response.""" return { 'vendor_id': 9999, 'priority': 3, 'website': 'www.aprilfoolsday.com', 'label_identifier': 'Test', 'owner': 'odd', 'is_owned': 'No', 'is_distributor': 'N', 'assigned_to': 'Conrad Capalbo', 'assigned_reviewer': 101, 'quarterback_label_manager': None, 'date_created': None, 'last_update': '2023-06-26 01:34:30', 'support_contact_email': 'jfowler@theorchard.com', 'contact_email': None, 'sap_vendor_id': None, 'est_releases': 150, 'est_tracks': 1762, 'primary_genre': 20, 'genre': 'Blues', 'country': 'United Kingdom', 'country_id': 3, 'status': 'signed', 'region': None, 'projected_first_year_revenue': '0.000000', 'transfer_pricing_country': None, 'allow_ALW_release_builder': 'N', 'migrated_to_abacus': 0, 'account_name': 'Test Label', 'contact_name': None, 'currency_symbol': '$', 'currency_code': 'USD', 'parent_company': 'theorchard', 'service_tier_uuid': None, 'source': None, 'vendor_uuid': '573d0372-7f2f-48a6-8deb-c9a6558f9549', 'company_brand': 'theorchard', 'service_tier_name': None } @pytest.fixture def mock_vendor_company_brand(): """mock_vendor_company_brand.""" return 'awal' @pytest.fixture def db_exception(): """Return (but not throw) SQLAlchemy exception.""" return exc.SQLAlchemyError('kaboom') @pytest.fixture def product_code_not_available(): """Return product_code not available message.""" return { 'product_code': { 'validator': 'used', 'validator_value': True, 'message': error.ERROR_MESSAGE_PRODUCT_CODE_NOT_AVAILABLE, 'error_code': 'product_code_used', } } @pytest.fixture def upc_not_available(): """Return upc not available message.""" return { 'upc': { 'validator': 'used', 'validator_value': True, 'message': error.ERROR_MESSAGE_UPC_NOT_AVAILABLE, 'error_code': 'upc_used', } } @pytest.fixture def invalid_date_message(): """Return invalid date message.""" return { 'validator': 'dates', 'validator_value': True, 'message': 'invalid date'} @pytest.fixture def early_late_date_message(): """Return invalid date message based on early and late dates.""" return { 'validator': 'dates', 'validator_value': True, 'message': 'must be between 1900-01-01 and 2100-12-31', } @pytest.fixture def valid_general_validation_message(): """Return a successful validation message for product validations.""" return {'valid': True, 'errors': {}} @pytest.fixture def valid_ows_tracks_message(): """Return a successful tracks validation response message.""" return {'valid': True, 'total_tracks': 12, 'valid_tracks': 12, 'errors': []} @pytest.fixture def valid_tracks_validation_message_bit_depth_disabled(): """Return a successful tracks validation response message. @todo: remove on teardown of """ return { 'valid': True, 'total_tracks': 12, 'valid_tracks': 12, 'errors': []} @pytest.fixture def valid_tracks_validation_message(request): """Return a successful tracks validation response message.""" validation_data = { 'valid': True, 'total_tracks': 12, 'valid_tracks': 12, 'errors': []} return validation_data @pytest.fixture def valid_publishing_validation_message(): """Return successful publishing obligation validation response message.""" return { 'valid': True, 'total_tracks': 12, 'valid_tracks': 12, 'errors': []} @pytest.fixture def empty_ows_tracks_message(): """Return a successful tracks validation response message.""" return {'valid': True, 'total_tracks': 0, 'valid_tracks': 0, 'errors': []} @pytest.fixture def empty_tracks_validation_message(): """Return a successful tracks validation response message.""" return { 'valid': False, 'total_tracks': 0, 'valid_tracks': 0, 'errors': []} @pytest.fixture def valid_assets_message(): """Return a successful asset artwork validation response message.""" return { 'assets': [ { 'filename': '191018329778.tif', 'url': 'https://qa-images.theorchard.io/product/cover/123.jpg', 'status': 'finished', 'asset_type': 'image' } ], 'product_id': '1917147', 'upc': 191018329778} @pytest.fixture def invalid_assets_message(): """Return an invalid asset artwork validation response message.""" return { 'assets': [ { 'filename': '191018329778.tif', 'url': 'https://qa-images.theorchard.io/product/cover/123.jpg', 'status': 'something_other_than_finished', 'asset_type': 'image' } ], 'product_id': '1917147', 'upc': 191018329778} @pytest.fixture def not_found_assets_message(): """Return a not found asset artwork validation response message.""" return { 'message': 'Asset not found for provided params.', 'code': 'not_found_error' } @pytest.fixture def empty_assets_message(): """Return an empty asset artwork validation response message.""" return None @pytest.fixture def invalid_product_blocklist_message(): """Return a message from ows-blocklist-manager for an invalid product.""" return { 'validation_error': { 'matched_blacklist_words': ['word'], 'items': [{ 'word': 'word', 'reason': 'some reason', 'alert': None, 'contact': None }] } } @pytest.fixture def track_validation_response(valid_ows_tracks_message): """Return a mock response from ows-track for validation.""" mock_track_response = MagicMock(status_code=200) mock_track_response.json = MagicMock(return_value=valid_ows_tracks_message) return mock_track_response @pytest.fixture def publishing_obligation_validation_response(valid_ows_tracks_message): """Return a mock response from ows-track for validation.""" mock_track_response = MagicMock(status_code=200) mock_track_response.json = MagicMock(return_value=valid_ows_tracks_message) return mock_track_response @pytest.fixture def get_assets_response(valid_assets_message): """Return a mock response from ows-assets.""" mock_assets_response = MagicMock(status_code=200) mock_assets_response.json = MagicMock(return_value=valid_assets_message) return mock_assets_response @pytest.fixture def get_assets_response_invalid(invalid_assets_message): """Return a mock response from ows-assets with invalid artwork.""" mock_assets_response = MagicMock(status_code=200) mock_assets_response.json = MagicMock(return_value=invalid_assets_message) return mock_assets_response @pytest.fixture def validate_artwork_response_invalid(invalid_assets_message): """Return a mock response from ows-assets with invalid artwork.""" mock_assets_response = MagicMock(status_code=404) mock_assets_response.json = MagicMock(return_value=invalid_assets_message) return mock_assets_response @pytest.fixture def validate_product_blocklist_response_invalid(invalid_product_blocklist_message): """Return a mock response from ows-blocklist-manager for valid product.""" mock_response = MagicMock(status_code=400) mock_response.json.return_value = invalid_product_blocklist_message return mock_response @pytest.fixture def empty_success(invalid_assets_message): """Return a mock response with a generic success.""" return MagicMock(status_code=200) @pytest.fixture def successful_validation_message( valid_general_validation_message, valid_tracks_validation_message): """Return a successful validation message for testing.""" return { 'validations': { 'product_basics': valid_general_validation_message, 'scheduling_and_pricing': valid_general_validation_message, 'tracks': valid_tracks_validation_message, 'publishing_obligation': valid_tracks_validation_message, 'artwork': {'errors': [], 'valid': True}}} @pytest.fixture def successful_validation_message_with_post_submission( valid_general_validation_message, valid_tracks_validation_message): """Return a successful validation message for testing.""" return { 'validations': { 'product_basics': valid_general_validation_message, 'scheduling_and_pricing': valid_general_validation_message, 'tracks': valid_tracks_validation_message, 'publishing_obligation': valid_tracks_validation_message, 'product_blocklist': {'errors': [], 'valid': True}, 'artwork': {'errors': [], 'valid': True}, 'artwork_compliance': {'errors': [], 'valid': True}, 'account': {'errors': [], 'valid': True}, 'product_artists': {'errors': [], 'valid': True}, } } @pytest.fixture def error_tracks_message(): """Return a tracks validation message with errors.""" return { 'valid': False, 'total_tracks': 10, 'valid_tracks': 8, 'errors': [ { 'tuid': 1, 'track_name': { 'error_code': 'required', 'error_message': 'Field `track_name` is required'}}, { 'tuid': 1, 'p_info': { 'error_code': 'required', 'error_message': 'Field `p_info` is required'}}]} @pytest.fixture def valid_product_payload(): """Return a payload of valid product data.""" return { 'artist_id': 3030, 'description': 'Unlistenable, to be honest.', 'distribution_format_id': 1, 'format': 'Full Length', 'genre_id': 2828, 'c_line': '2016 DIG For Gold Records', 'imprint': 'DIG For Gold Records', 'manufacturer_upc': '123456789012', 'meta_language': 'SUX', 'preorder_date': '2016-04-22', 'previewable': 'yes', 'product_artists': [ { 'role': 'primary_artist', 'name': 'required primary artist', 'id': 1, 'artist_info_id': None } ], 'product_code': 'AProduct123', 'product_name': 'Eggensperger And Heydt Yodel The Classics', 'product_type_id': 1, 'release_date': '2016-06-23', 'release_status': 'label_processing', 'sale_start_date': '2016-09-07', 'special_instructions': 'Listen at low volume. Better yet, hit mute.', 'subgenre_id': 13, 'vendor_catalog_number': 'ABC123', 'version': 'Delux', 'account_type': 'vendor', 'account_id': 123} @pytest.fixture def correlation_id(): """Provide a correlation_id for testing.""" return 'abc-123' @pytest.fixture def product_id(): """Provide a product_id for testing.""" return '12345' @pytest.fixture def display_upc(): """Return a fake display_upc for testing.""" return '123456789012' @pytest.fixture def project_id(): """Provide a project_id for testing.""" return @pytest.fixture def orchard_user_id(): """Orchard-User-Id header value.""" return 'alw:123' @pytest.fixture def oa_orchard_user_id(): """Orchard-User-Id header value.""" return 'oa:123' @pytest.fixture def release_correction_id(): """Return a fake release correction id for testing.""" return 123 @pytest.fixture def track_pricing_overrides_payload(): """Return a sample track overrides payload from ows-pricing.""" return { 'items': [ { 'territory_list_include': False, 'applies_worldwide': True, 'territories': [], 'custom_currency_code': None, 'updated_date': 1498493891.504341, 'created_date': 1498493828.082951, 'activated': True, 'track_id': 25292698, 'track_pricing_override_id': 1914519, 'custom_price': None, 'orchard_pricing_tier_id': 26, 'start_date': None, 'sort_order': None, 'stores': [], 'price_code': None, 'end_date': None }, { 'territory_list_include': False, 'applies_worldwide': True, 'territories': [], 'custom_currency_code': None, 'updated_date': 1498493891.607976, 'created_date': 1498493828.189113, 'activated': False, 'track_id': 25292700, 'track_pricing_override_id': 1914522, 'custom_price': None, 'orchard_pricing_tier_id': 25, 'start_date': None, 'sort_order': None, 'stores': [], 'price_code': None, 'end_date': None }, { 'territory_list_include': False, 'applies_worldwide': True, 'territories': ['Jumangi', 'Braavos'], 'custom_currency_code': None, 'updated_date': 1498493891.607976, 'created_date': 1498493828.189113, 'activated': False, 'track_id': 25292700, 'track_pricing_override_id': 1914522, 'custom_price': None, 'orchard_pricing_tier_id': 25, 'start_date': None, 'sort_order': None, 'stores': [], 'price_code': None, 'end_date': None } ] } @pytest.fixture def active_track_overrides_payload(): """Return a sample track overrides payload from ows-pricing.""" return { 'items': [ { 'territory_list_include': False, 'applies_worldwide': True, 'territories': [], 'custom_currency_code': None, 'updated_date': 1498493891.504341, 'created_date': 1498493828.082951, 'activated': True, 'track_id': 25292698, 'track_pricing_override_id': 1914519, 'custom_price': None, 'orchard_pricing_tier_id': 26, 'start_date': None, 'sort_order': None, 'stores': [], 'price_code': None, 'end_date': None, 'track_name': 'My Song' } ] } @pytest.fixture def expected_track_pricing_overrides(): """Return expected track overrides payload from ows-pricing.""" return [ { 'territory_list_include': False, 'applies_worldwide': True, 'territories': [], 'custom_currency_code': None, 'updated_date': '2017-06-26', 'created_date': '2017-06-26', 'activated': True, 'track_id': 25292698, 'track_pricing_override_id': 1914519, 'custom_price': None, 'orchard_pricing_tier_id': 26, 'start_date': None, 'sort_order': None, 'stores': [], 'price_code': None, 'end_date': None, 'orchard_pricing_tier_name': 'Mid', 'track_name': 'Best Track' } ] @pytest.fixture def track_ids_for_testing(): """Return list of track ids matching ows_pricing payload.""" return [25292698, 25292700] @pytest.fixture def product_pricing_overrides_payload(): """Return a sample product overrides payload from ows-pricing.""" return { 'items': [ { 'pricing_family_id': 2, 'custom_currency_code': None, 'territories': [], 'product_pricing_override_id': 1523797, 'start_date': 1498521600, 'activated': False, 'product_id': 2003632, 'store_id': None, 'stores': [], 'price_code': None, 'end_date': 1498608000, 'territory_list_include': False, 'updated_date': None, 'created_date': 1498493891.197279, 'resolution': None, 'custom_price': None, 'orchard_pricing_tier_id': 15, 'applies_worldwide': True, 'sort_order': None }, { 'pricing_family_id': 2, 'custom_currency_code': None, 'territories': [], 'product_pricing_override_id': 1523798, 'start_date': 1498694400, 'activated': True, 'product_id': 2003632, 'store_id': None, 'stores': [], 'price_code': None, 'end_date': 1498780800, 'territory_list_include': False, 'updated_date': None, 'created_date': 1498493891.329886, 'resolution': None, 'custom_price': None, 'orchard_pricing_tier_id': 25, 'applies_worldwide': True, 'sort_order': None }, { 'pricing_family_id': 3, 'custom_currency_code': None, 'territories': ['XX'], 'product_pricing_override_id': 1523798, 'start_date': 1498694400, 'activated': True, 'product_id': 2003632, 'store_id': None, 'stores': [], 'price_code': None, 'end_date': 1498780800, 'territory_list_include': False, 'updated_date': None, 'created_date': 1498493891.329886, 'resolution': None, 'custom_price': None, 'orchard_pricing_tier_id': 8, 'applies_worldwide': True, 'sort_order': None }, { 'pricing_family_id': 3, 'custom_currency_code': None, 'territories': [], 'product_pricing_override_id': 1523798, 'start_date': 1498694400, 'activated': True, 'product_id': 2003632, 'store_id': None, 'stores': [123, 456], 'price_code': None, 'end_date': 1498780800, 'territory_list_include': False, 'updated_date': None, 'created_date': 1498493891.329886, 'resolution': None, 'custom_price': None, 'orchard_pricing_tier_id': 8, 'applies_worldwide': True, 'sort_order': None }, ] } @pytest.fixture def product_pricing_active_overrides_payload(): """Return a sample of active product overrides payload from ows-pricing.""" return { 'items': [ { 'pricing_family_id': 2, 'custom_currency_code': None, 'territories': [], 'product_pricing_override_id': 1523798, 'start_date': 1498694400, 'activated': True, 'product_id': 2003632, 'store_id': None, 'stores': [], 'price_code': None, 'end_date': 1498780800, 'territory_list_include': False, 'updated_date': None, 'created_date': 1498493891.329886, 'resolution': None, 'custom_price': None, 'orchard_pricing_tier_id': 25, 'applies_worldwide': True, 'sort_order': None } ] } def _date_convert(timestamp): return datetime.datetime.fromtimestamp(timestamp).strftime('%Y-%m-%d') @pytest.fixture def active_overrides_payload_converted_dates(): """Return a filted, date converted list of product overrides.""" return [ { 'pricing_family_id': 2, 'custom_currency_code': None, 'territories': [], 'product_pricing_override_id': 1523798, 'start_date': _date_convert(1498694400), 'activated': True, 'product_id': 2003632, 'store_id': None, 'stores': [], 'price_code': None, 'end_date': _date_convert(1498780800), 'territory_list_include': False, 'updated_date': None, 'created_date': _date_convert(1498493891.329886), 'resolution': None, 'custom_price': None, 'orchard_pricing_tier_id': 25, 'applies_worldwide': True, 'sort_order': None, 'orchard_pricing_tier_name': 'Mid/Front' } ] @pytest.fixture def generic_error_payload(): """Return generic error payload for mocking services.""" return { 'message': 'Not found error message', 'code': 'not_found_error_code'} @pytest.fixture def product_pricing_tier_payload(): """Return a product pricing tier payload from ows-pricing.""" return { 'product_id': 2003632, 'product_orchard_pricing_tier_id': 1258749, 'pricing_family_id': 2, 'orchard_pricing_tier_id': 15} @pytest.fixture def product_pricing_tier_by_product(product_pricing_tier_payload): """Return product pricing tier message from get_pricing_tier_by_product.""" product_pricing_tier_payload.update({'name': 'Deluxe Five'}) return product_pricing_tier_payload @pytest.fixture def track_pricing_tier_payload(): """Return a track pricing tier payload from ows-pricing.""" return { 'product_id': 2003632, 'product_orchard_pricing_tier_id': 1258749, 'pricing_family_id': 3, 'orchard_pricing_tier_id': 25} @pytest.fixture def track_pricing_tier_by_product(track_pricing_tier_payload): """Return product pricing tier message from get_pricing_tier_by_product.""" track_pricing_tier_payload.update({'name': 'Mid/Front'}) return track_pricing_tier_payload @pytest.fixture def get_tiers_for_albums_message(): """Return a fake pricing tier message for albums.""" return {'items': [ { 'is_default': True, 'sort_order': 1, 'name': 'Deluxe Four', 'orchard_pricing_tier_id': 8, 'pricing_family_id': 2 }, { 'is_default': False, 'sort_order': 2, 'name': 'Deluxe Five', 'orchard_pricing_tier_id': 15, 'pricing_family_id': 2 }]} @pytest.fixture def get_tiers_for_tracks_message(): """Return a fake pricing tier message for tracks.""" return {'items': [ { 'is_default': True, 'sort_order': 1, 'name': 'Mid/Front', 'orchard_pricing_tier_id': 25, 'pricing_family_id': 3 }, { 'is_default': False, 'sort_order': 2, 'name': 'Mid', 'orchard_pricing_tier_id': 26, 'pricing_family_id': 3 }]} @pytest.fixture def get_pricing_from_bff_message(): """Return a fake pricing info message for product from bff endpoint.""" return { 'track_pricing_override': { 'items': [ { 'created_date': 1541068797.883285, 'territories': [], 'price_code': None, 'stores': [], 'end_date': None, 'territory_list_include': False, 'updated_date': None, 'custom_price': None, 'track_id': 29136821, 'sort_order': 1, 'track_pricing_override_id': 981295, 'activated': True, 'orchard_pricing_tier_id': 26, 'applies_worldwide': True, 'start_date': None, 'track_name': 'Track 1', 'custom_currency_code': None } ] }, 'pricing_tier_options': { 'items': [ { 'is_default': False, 'pricing_family_id': 2, 'name': 'Deluxe Four', 'orchard_pricing_tier_id': 8, 'sort_order': 1 }, { 'is_default': False, 'pricing_family_id': 3, 'name': 'Front', 'orchard_pricing_tier_id': 25, 'sort_order': 1 }, { 'is_default': False, 'pricing_family_id': 2, 'name': 'Deluxe Three', 'orchard_pricing_tier_id': 9, 'sort_order': 2 }, { 'is_default': True, 'pricing_family_id': 3, 'name': 'Mid', 'orchard_pricing_tier_id': 26, 'sort_order': 2 }, { 'is_default': False, 'pricing_family_id': 3, 'name': 'Back', 'orchard_pricing_tier_id': 27, 'sort_order': 3 }, { 'is_default': False, 'pricing_family_id': 2, 'name': 'Deluxe Two', 'orchard_pricing_tier_id': 10, 'sort_order': 3 }, { 'is_default': False, 'pricing_family_id': 2, 'name': 'Deluxe One', 'orchard_pricing_tier_id': 11, 'sort_order': 4 }, { 'is_default': False, 'pricing_family_id': 2, 'name': 'Front Plus', 'orchard_pricing_tier_id': 12, 'sort_order': 5 }, { 'is_default': False, 'pricing_family_id': 2, 'name': 'Front 2', 'orchard_pricing_tier_id': 13, 'sort_order': 6 }, { 'is_default': False, 'pricing_family_id': 2, 'name': 'Front 1', 'orchard_pricing_tier_id': 14, 'sort_order': 7 }, { 'is_default': True, 'pricing_family_id': 2, 'name': 'Mid/Front', 'orchard_pricing_tier_id': 15, 'sort_order': 8 }, { 'is_default': False, 'pricing_family_id': 2, 'name': 'Mid', 'orchard_pricing_tier_id': 16, 'sort_order': 9 }, { 'is_default': False, 'pricing_family_id': 2, 'name': 'Back', 'orchard_pricing_tier_id': 17, 'sort_order': 10 }, { 'is_default': False, 'pricing_family_id': 2, 'name': 'Budget 2', 'orchard_pricing_tier_id': 18, 'sort_order': 11 }, { 'is_default': False, 'pricing_family_id': 2, 'name': 'Budget 1', 'orchard_pricing_tier_id': 19, 'sort_order': 12 }, { 'is_default': False, 'pricing_family_id': 2, 'name': 'Mini Album Two', 'orchard_pricing_tier_id': 20, 'sort_order': 13 }, { 'is_default': False, 'pricing_family_id': 2, 'name': 'Mini Album One', 'orchard_pricing_tier_id': 21, 'sort_order': 14 }, { 'is_default': False, 'pricing_family_id': 2, 'name': 'EP (10 Track Max)', 'orchard_pricing_tier_id': 22, 'sort_order': 15 }, { 'is_default': False, 'pricing_family_id': 2, 'name': 'Mini EP (7 Tracks Max)', 'orchard_pricing_tier_id': 23, 'sort_order': 16 }, { 'is_default': False, 'pricing_family_id': 2, 'name': 'Digital 45 (4 Tracks Max)', 'orchard_pricing_tier_id': 24, 'sort_order': 17 } ] }, 'album_pricing_tier': { 'pricing_family_id': 2, 'product_id': 2493970, 'product_orchard_pricing_tier_id': 4051863, 'orchard_pricing_tier_id': 15 }, 'album_pricing_override': { 'items': [] }, 'track_pricing_tier': { 'pricing_family_id': 3, 'product_id': 2493970, 'product_orchard_pricing_tier_id': 4051864, 'orchard_pricing_tier_id': 26 } } def make_response(valid, message=None, status=status.BAD_REQUEST): """Create a response.Response which is either valid or not. Args: valid (bool): if response should be 200 or not message (int|str|dict|list): response message Returns: response.Response: with desired message """ if valid: resp = response.Response(message=message) else: resp = response.create_error_response( code='invalid', message=message, status=status) return resp @pytest.fixture def product_basics(): """Return a dict with product basics data.""" return { 'product_name': 'Test product', 'meta_language': 'en_US', 'genre_id': 123, 'subgenre_id': 1234, 'format': 'Full Length 1', 'imprint': 'DIG For Gold Records', 'c_line': '2016 DIG For Gold Records', } @pytest.fixture def mock_check_project(): """Mock models.project.check_project_ownership with valid response.""" (flexmock(project) .should_receive('check_project_ownership') .and_return(response.Response())) @pytest.fixture def on_create_state(): """Fixture to determine validate_product_on_create state. To use - define with @pytest.mark.parametrize('on_create_state', ['True'] """ return True @pytest.fixture def validate_product_on_create(on_create_state): """Mock validation.audio_product.validate_product_on_create function.""" resp = make_response(on_create_state) (flexmock(product_validation) .should_receive('validate_product_on_create') .and_return(resp)) @pytest.fixture def on_update_state(): """Fixture to determine validate_product_on_create state. To use - define with @pytest.mark.parametrize('on_update_state', ['True'] """ return True @pytest.fixture def validate_product_on_update(on_update_state): """Mock validation.audio_product.validate_product_on_update function.""" resp = make_response(on_update_state) (flexmock(product_validation) .should_receive('validate_product_on_update') .and_return(resp)) @pytest.fixture def get_upc(): """Mock models.release.get_upc.""" upc = 123456789015 resp = make_response(valid=True, message=upc) (flexmock(release) .should_receive('get_upc') .and_return(resp)) @pytest.fixture def copy_product_basics(account_id): """Fxiture with product basics for copy endpoint.""" return { 'product_code': 'NEWAProduct123', 'product_name': 'Another Eggensperger And Heydt Yodel The Classics', 'version': 'Another Delux', 'format': 'New Full Length', 'upc': '012123456789', 'account_type': account.VENDOR_TYPE, 'account_id': str(account_id), } def audio_product_dictionary(dates): """Expected response from audio product model.""" return { 'artist_id': 7923, 'c_line': '2016 This is important info', 'description': 'New description here', 'distribution_format_id': 1, 'format': 'Single', 'genre_id': 13, 'imprint': 'A new and fantastic label', 'manufacturer_upc': '123456789012', 'meta_language': 'SUX', 'preorder_date': dates[0], 'previewable': 'yes', 'product_artists': [ { 'role': 'primary_artist', 'name': 'Jarvis Q. Hinkleberry VII' } ], 'product_code': 'DIGI4LYFE', 'product_id': 4321, 'product_name': '36 Chambers: Enter The Digital Clan', 'product_type_id': 1, 'project_id': 5678, 'release_date': dates[1], 'release_status': 'label_processing', 'special_instructions': 'New instructions', 'subgenre_id': 12, 'sale_start_date': dates[2], 'upc': '123456789012', 'vendor_catalog_number': 'DigProdNumber1', 'version': 'Extremely Explicit Version'} @pytest.fixture def product_model_response_message(): """Return expected message from audio product model.""" return audio_product_dictionary(DATE_OBJECTS_FOR_RESPONSE_MESSAGE) @pytest.fixture def product_model_response(product_model_response_message): """Return a response object with the expected product model response.""" return response.Response(message=product_model_response_message) @pytest.fixture def expected_product_logic_response_message(): """Return expected message from audio product model.""" return audio_product_dictionary(DATE_STRINGS_FOR_RESPONSE_MESSAGE) @pytest.fixture def release_approval_get_message(): """Return a dictionary of release approval queue properties for testing.""" return { 'release_approval_id': 1, 'release_id': 123, 'status': 'rejected'} @pytest.fixture def release_approval_get_response(release_approval_get_message): """Return a mock response from product-workflow showing a rejection.""" release_approval_response = MagicMock(status_code=200) release_approval_response.json = MagicMock( return_value=release_approval_get_message) return release_approval_response @pytest.fixture def get_rejections_message(): """Return a successful rejections response message.""" return { 'items': [ { 'rejection_id': 1, 'table_name': 'releases', 'comments': 'Needs more info about Ken', 'key_id': 0, 'field_name': 'description', 'corrected': 'N' }, { 'rejection_id': 2, 'table_name': 'releases', 'comments': 'Needs to be KEN', 'key_id': 0, 'field_name': 'release_name', 'corrected': 'N' }, { 'rejection_id': 3, 'table_name': 'releases', 'comments': 'blub blub blub', 'key_id': 0, 'field_name': 'general_comments', 'corrected': 'N' } ] } @pytest.fixture def get_status_message(product_id): """Return a successful status response message from ows-product-review.""" return { 'id': product_id, 'ec_and_ar': False, 'release_status': 'transfer_to_content', 'release_approval_status': 'checked_in', } @pytest.fixture def get_rejected_status_message(product_id): """Return a successful status response message from ows-product-review.""" return { 'id': product_id, 'ec_and_ar': False, 'release_status': 'action_required', 'release_approval_status': 'rejected', } @pytest.fixture def product_workflow_rejections_response(get_rejections_message): """Return a response with rejections from ows-workflow.""" rejections_mock_response = MagicMock(status_code=200) rejections_mock_response.json = MagicMock( return_value=get_rejections_message) return rejections_mock_response @pytest.fixture def release_correction_get_message(): """Return a dictionary of release correction properties for testing.""" return { 'release_correction_id': 1, 'status': 'active', 'items': [ { 'release_correction_detail_id': 1, 'table_name': 'releases', 'field_name': 'primary', 'key_id': 288212, 'key_value': 'HEY HEY HEY' }, { 'release_correction_detail_id': 2, 'table_name': 'releases', 'field_name': 'primary', 'key_id': 288213, 'key_value': 'EYH EYH EYH' }, { 'release_correction_detail_id': 3, 'table_name': 'releases', 'field_name': 'primary', 'key_id': 288214, 'key_value': 'HYE HYE HYE' } ] } @pytest.fixture def release_correction_submitted_get_message(): """Return a dictionary of release correction properties for testing.""" return { 'release_correction_id': 1, 'status': 'submitted', 'items': [ { 'release_correction_detail_id': 1, 'table_name': 'releases', 'field_name': 'primary', 'key_id': 288212, 'key_value': 'HEY HEY HEY' }, { 'release_correction_detail_id': 2, 'table_name': 'releases', 'field_name': 'primary', 'key_id': 288213, 'key_value': 'EYH EYH EYH' }, { 'release_correction_detail_id': 3, 'table_name': 'releases', 'field_name': 'primary', 'key_id': 288214, 'key_value': 'HYE HYE HYE' } ] } @pytest.fixture def release_correction_get_response(release_correction_get_message): """Return a mock response from ows-product-workflow.""" release_correction_response = MagicMock(status_code=200) release_correction_response.json = MagicMock( return_value=release_correction_get_message ) return release_correction_response @pytest.fixture def instant_grats_message(): """Return a message with instant grats from ows-track.""" return { 'items': [ {'tuid': '1', 'grats': [ {'date': '2017-09-26T22:27:52.695829', 'store_id': 1, 'created_by': 'oa:25824', 'created_at': '2017-09-26T22:27:52.695829', 'active': 'Y'}, {'date': '2017-09-26T22:27:52.695829', 'store_id': 2, 'created_by': 'oa:25824', 'created_at': '2017-09-26T22:27:52.695829', 'active': 'Y'} ]}, {'tuid': '3', 'grats': [ {'date': '2017-09-26T22:27:52.29', 'store_id': 1, 'created_by': 'oa:25824', 'created_at': '2017-09-26T22:27:52.695829', 'active': 'Y'}, {'date': '2018-09-26T22:27:52.695829', 'store_id': 2, 'created_by': 'oa:25824', 'created_at': '2017-09-26T22:27:52.695829', 'active': 'Y'} ]} ]} @pytest.fixture def instant_grats_with_tracknames_message(): """Return a message with instant grats with track names from ows-track.""" return { 'items': [ {'tuid': '1', 'track_name': 'Track', 'grats': [ {'date': '2017-09-26T22:27:52.695829', 'store_id': 1, 'created_by': 'oa:25824', 'created_at': '2017-09-26T22:27:52.695829', 'active': 'Y'}, {'date': '2017-09-26T22:27:52.695829', 'store_id': 2, 'created_by': 'oa:25824', 'created_at': '2017-09-26T22:27:52.695829', 'active': 'Y'} ]}, {'tuid': '3', 'track_name': 'Track3', 'grats': [ {'date': '2017-09-26T22:27:52.29', 'store_id': 1, 'created_by': 'oa:25824', 'created_at': '2017-09-26T22:27:52.695829', 'active': 'Y'}, {'date': '2018-09-26T22:27:52.695829', 'store_id': 2, 'created_by': 'oa:25824', 'created_at': '2017-09-26T22:27:52.695829', 'active': 'Y'} ]} ]} @pytest.fixture def correction_details_data(): """Return a correction details dict to submit in POST payloads.""" return { 'items': [{ 'table_name': 'releases', 'field_name': 'release_name', 'key_id': 12345, 'key_value': 'new release name'}]} @pytest.fixture def correction_details_data_with_ownership(): """Return a correction details dict to submit in POST payloads.""" return { 'items': [{ 'table_name': 'releases', 'field_name': 'release_name', 'key_id': 12345, 'key_value': 'new release name', 'last_updated_by': 123, 'last_updated_type': 'vendor'}]} @pytest.fixture def post_correction_details_ok_response(): """Return a fake OK response stub posting correction details.""" post_response = MagicMock(status_code=201) post_response.json = MagicMock(return_value={ 'items': [{ 'release_correction_detail_id': 123, 'table_name': 'releases', 'field_name': 'release_name', 'key_id': 12345, 'key_value': 'new release name'}]}) return post_response @pytest.fixture def get_correction_response(product_id, release_correction_id): """Return a get correction response to use in endpoint validation.""" get_response = MagicMock(status_code=200) get_response.json = MagicMock(return_value={ 'release_correction_id': release_correction_id, 'release_id': product_id, 'status': 'active', 'items': []}) return get_response @pytest.fixture def correction_details_data_with_genre_fields(): """Return a correction details with genre to submit in POST payloads.""" return { 'items': [{ 'table_name': 'releases', 'field_name': 'genre_id', 'key_id': 12345, 'key_value': 2, 'last_updated_by': 123, 'last_updated_type': 'vendor'}, {'field_name': 'release_subgenre', 'key_value': [324], 'key_id': 2343165, 'table_name': 'releases', 'last_updated_by': 123, 'last_updated_type': 'vendor' }]} @pytest.fixture def change_genre_request_data(): """Return a request data to submit in /change_genre POST payloads.""" return { 'genre_id': 2, 'subgenre_id': 324} @pytest.fixture def release_approval_create_message(): """Return a dictionary of release approval queue properties for testing.""" return { 'release_approval_id': 1, 'release_id': 123, 'status': 'transfer_to_content'} @pytest.fixture def mkt_priority_dictionary(): """Expected response from mkt priority model.""" return [ { 'id': 1122, 'upc': 829410750950, 'priority': 'a', 'country_id': 0, 'release_id': 92949 }, { 'id': 1155, 'upc': 829410910552, 'priority': 'a', 'country_id': 0, 'release_id': 92949 } ] @pytest.fixture def mkt_priority_model_response_message(mkt_priority_dictionary): """Return expected message from mkt priority model.""" return mkt_priority_dictionary @pytest.fixture def mkt_priority_model_response(mkt_priority_model_response_message): """Return a response object with the expected mkt priority model response.""" return response.Response(message=mkt_priority_model_response_message) @pytest.fixture def expected_mkt_priority_logic_response_message(mkt_priority_dictionary): """Return expected message from mkt priority.""" return {'mkt_priority': mkt_priority_dictionary} @pytest.fixture def expected_mkt_priority_logic_response_message_for_no_data(): """Return expected message from mkt priority.""" return {'mkt_priority': []} @pytest.fixture def mkt_priority_model_responsee_for_no_data(): """Return a response object with the expected mkt priority model response.""" return response.Response(message=[]) @pytest.fixture def base_product_data_to_update_project_subaccount(): """project_id and subaccount_id included in POST request body. Returns: dict: product data """ return { 'project_id': 12345, 'subaccount_id': 1 } @pytest.fixture def account_type(): """Provide an grass_account_type for testing.""" return 'vendor' @pytest.fixture def valid_ows_product_get_response(): """Valid product get response.""" return { 'product_id': 2676431, 'distribution_format_id': 1, 'project_id': 4228930, 'product_type_id': 1, 'upc': 194491190514, 'context_type': 'digital', 'vendor_id': 25517, 'subaccount_id': 0, 'release_date': '2019-10-04', 'status': 'in_content', 'product_name': 'Besame Mucho', 'deletions': 'N' } @pytest.fixture def valid_ows_product_get_response_with_subaccount( valid_ows_product_get_response): """Valid product get response with subaccount.""" valid_ows_product_get_response['subaccount_id'] = 4000 return valid_ows_product_get_response @pytest.fixture def correction_details_data_with_subgenre_fields(): """Return a correction details with subgenre to submit in POST payloads.""" return { 'items': [ {'field_name': 'release_subgenre', 'key_value': [545], 'key_id': 2343165, 'table_name': 'releases', 'last_updated_by': 123, 'last_updated_type': 'vendor' }]} @pytest.fixture def expected_create_product_provided_store_artists_response_message(): """Product data returned from product_provided_store_artists.create function in logic. Returns: dict: product_provided_store_artists data """ return { 'product_provided_store_artists_id': 123, 'release_id': 1, 'spotify': 'Y' } @pytest.fixture(autouse=True) def mock_digital_product_producer(mocker): """Mock the Kafka producer to prevent real Kafka calls in tests.""" return mocker.patch.object(product_events, 'digital_product_producer') @pytest.fixture def mock_languages(): """Fixture data for a fake response from the model layer.""" languages = [ lang_factory.LanguageFactory.build(language_code='eng'), lang_factory.LanguageFactory.build(language_code='foo'), lang_factory.LanguageFactory.build(language_code='bar'), ] return [lang.to_dict() for lang in languages] @pytest.fixture def mock_language_model_response(monkeypatch, mock_languages): """Fake response for mocking the model layer.""" mock_response = response.Response(message={'items': mock_languages}) monkeypatch.setattr(language, 'get_languages', Mock(return_value=mock_response)) return mock_response @pytest.fixture def search_terms_product_data(): """Product data used for metadata search term tests.""" return { 'product_id': 99, 'product_name': 'Blue Album', 'product_artists': [ {'name': 'John Smith', 'role': 'primary_artist'}, {'name': 'Jane Doe', 'role': 'featuring'}, {'name': 'DJ Remix', 'role': 'remixer'}, {'name': 'Bob Jones', 'role': 'producer'}, ], } @pytest.fixture def search_terms_tracks_data(): """Track data used for metadata search term tests.""" return [ {'tuid': '1', 'track_name': 'Song Title'}, ] @pytest.fixture def expected_search_terms_base(): """Expected metadata search terms without any corrections.""" return [ {'metadata_value': 'Blue Album', 'metadata_field': 'product_name', 'dictionary': ['blue album', 'blue', 'album']}, {'metadata_value': 'Song Title', 'metadata_field': 'track_name', 'dictionary': ['song title', 'song', 'title']}, {'metadata_value': 'John Smith', 'metadata_field': 'primary_artist', 'dictionary': ['john smith', 'john', 'smith']}, {'metadata_value': 'Jane Doe', 'metadata_field': 'featuring', 'dictionary': ['jane doe', 'jane', 'doe']}, {'metadata_value': 'DJ Remix', 'metadata_field': 'remixer', 'dictionary': ['dj remix', 'dj', 'remix']}, {'metadata_value': 'Bob Jones', 'metadata_field': 'producer', 'dictionary': ['bob jones', 'bob', 'jones']}, ] @pytest.fixture def basic_product_tracks_response(): """Return a basic response from ows-track for product tracks.""" return MagicMock( status_code=200, json=MagicMock(return_value={ 'items': [{ 'tuid': '123', 'artists': [{ 'type': 'performer', 'name': 'Jarvis Q. Hinkleberry VIII' }] }] }) ) def get_transactional_session_mock(side_effect=[]): """Mock for session from @Neo4jSession(transaction=True).""" enter_mock = MagicMock() result_mock = MagicMock() result_mock.data.side_effect = side_effect result_mock.single.side_effect = side_effect result_mock.peek.side_effect = side_effect transaction_mock = MagicMock() transaction_mock.run.return_value = result_mock enter_mock.begin_transaction.return_value = transaction_mock enter_mock.run.return_value = result_mock session_mock = MagicMock() session_mock.run.return_value = result_mock session_mock.__enter__.return_value = enter_mock return session_mock