"""conftest. This file gets picked up when running py.test tests: http://pytest.org/latest/writing_plugins.html#conftest """ from unittest.mock import MagicMock import application import boto3 from moto import mock_sqs from oto import response from owslogger import flask_logger from owsrequest import request from owsrequest import test_utils import pytest import pandas as pd from python_pdp_sdk import AuthorizationBackend from product.constants import header from tests.testutils import db @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() -> AuthorizationBackend: """Return a MagicMock spec'd as AuthorizationBackend. This fixture intentionally does NOT patch any module attribute. The caller is responsible for patching at the usage/import site of the code under test, e.g.: with patch('product.models.audience.authorization_backend', mock_authorization_backend): ... Patching `product.api.authorization_backend` from this fixture would not affect modules that did `from product.api import authorization_backend`, since those imports hold their own reference to the original object — so patching is left to the test author who knows the actual import path of the code being exercised. """ return MagicMock(spec=AuthorizationBackend) @pytest.fixture def test_upc(): """Return a UPC for testing.""" return '123456123456' @pytest.fixture def vw_products(): """Insert products for vw_product table.""" db.insert_vw_product() @pytest.fixture def test_product(): """Return a product_id for testing.""" return 1 @pytest.fixture def test_valid_product(test_product): """Return a valid product for testing.""" return { 'product_id': test_product, 'upc': 888831283041, 'display_upc': '888831283041', 'subaccount_id': 1, 'distribution_format_id': 2, 'product_type_id': 1, 'project_id': 22, 'vendor_id': 1, 'context_type': 'physical', 'release_date': '2017-09-09', 'status': 'label_processing', 'product_name': 'product test 1', 'deletions': 'N', 'not_for_distribution': 'N' } @pytest.fixture def test_valid_digital_product(test_product): """Return a valid digital product for testing.""" return { 'product_id': 32123, 'upc': 888831283041, 'subaccount_id': 1, 'distribution_format_id': 1, 'product_type_id': 1, 'project_id': 22, 'vendor_id': 55, 'context_type': 'digital', 'release_date': '2017-09-09', 'status': 'label_processing', 'product_name': 'product test 1', 'deletions': 'N', 'not_for_distribution': 'N' } @pytest.fixture def sqs(): """Return an boto3 sqs client.""" with mock_sqs(): yield boto3.client('sqs', region_name='us-east-1') @pytest.fixture def test_deleted_product(): """Return deleted product.""" return { 'context_type': 'physical', 'deletions': 'Y', 'display_upc': '888831283041', 'distribution_format_id': 2, 'not_for_distribution': 'N', 'product_id': 4, 'product_name': None, 'product_type_id': 1, 'project_id': 22, 'release_date': None, 'status': 'label_processing', 'subaccount_id': 1, 'upc': 888831283041, 'vendor_id': 1 } @pytest.fixture def test_invalid_product(): """Return an invalid product for testing.""" return { 'product_id': 8, 'upc': 100, 'display_upc': '100', 'subaccount_id': 1, 'distribution_format_id': 2, 'product_type_id': 1, 'project_id': 22, 'vendor_id': 1, 'context_type': 'physical', 'release_date': '2017-09-09', 'status': 'label_processing', 'product_name': 'product test 2', 'deletions': 'N' } @pytest.fixture def test_products_by_upcs(): """Return products for testing.""" return [ { 'product_id': 1, 'upc': 888831283041, 'display_upc': '888831283041', 'subaccount_id': 1, 'vendor_id': 1, 'context_type': 'physical', 'release_date': '2017-09-09', 'product_type': 'Music', 'product_name': 'product test 1', 'deletions': 'N', 'distribution_format_id': 2, 'product_type_id': 1, 'status': 'label_processing', 'project_id': 22, 'not_for_distribution': None }, { 'product_id': 2, 'upc': 888831283042, 'display_upc': '888831283042', 'subaccount_id': 0, 'vendor_id': 1, 'context_type': 'digital', 'release_date': '2017-09-17', 'product_type': 'Music', 'product_name': 'product test 2', 'deletions': 'N', 'distribution_format_id': 1, 'product_type_id': 1, 'status': 'label_processing', 'project_id': 33, 'not_for_distribution': None } ] @pytest.fixture def test_products_payload(): """Return test products payload.""" return [ { 'product_id': 123, 'upc': 123456781222, 'display_upc': '123456781222', 'product_type': 'Film', 'product_name': 'Title 1', 'context_type': 'null', 'vendor_id': 123, 'subaccount_id': 123, 'release_date': '01-01-2017' }, { 'product_id': 345, 'upc': 123456000000, 'display_upc': '123456000000', 'product_type': 'Film', 'product_name': 'Title 2', 'context_type': 'null', 'vendor_id': 123, 'subaccount_id': 123, 'release_date': '01-01-2017' } ] @pytest.fixture def test_vendor(): """Return a vendor_id for testing.""" return 1234 @pytest.fixture def test_subaccount(): """Return a subaccount_id for testing.""" return 5 @pytest.fixture def invalid_context_type(): """Return an invalid context type for testing.""" return 'purple' @pytest.fixture def test_account_type(): """Return a subaccount_id for testing.""" return 'vendor' @pytest.fixture def profile_id(): """Return profile id""" return 1 @pytest.fixture def profile_type_label(): """Return LabelProfile type""" return "LabelProfile" @pytest.fixture def valid_headers(test_account_type): """Return valid account headers.""" return { header.GRASS_ACCOUNT_TYPE: test_account_type, header.GRASS_ACCOUNT_ID: 1} @pytest.fixture def not_valid_headers(test_account_type, test_vendor): """Return invalid account headers.""" return { header.GRASS_ACCOUNT_TYPE: test_account_type, header.GRASS_ACCOUNT_ID: test_vendor} @pytest.fixture def fixture_language(): """Return a language_id for testing.""" return 12 @pytest.fixture def fixture_valid_localization(): """Return a valid product for testing.""" return { 'product_name': 'film name in de', 'short_synopsis': 'short desc in de' } @pytest.fixture def fixture_music_localization(): """Return a valid localization for audio testing.""" return { 'product_name': 'release name in de', 'performer': { 1: 'performer1 in de', 2: 'performer2 in de' } } @pytest.fixture def fixture_all_itunes_languages(): """Return a valid all_itunes_languages response.""" return { 'pagination': { 'type': 'none', 'total_records': 2 }, 'items': [ { 'language_code': 'af', 'language_id': 1, 'language': 'Afrikaans' }, { 'language_code': 'ar', 'language_id': 2, 'language': 'Arabic' }, { 'language_code': 'cmn-Hans', 'language_id': 6, 'language': 'Chinese (Simplified)' }, { 'language_code': 'cmn-Hant', 'language_id': 7, 'language': 'Chinese (Traditional)' } ] } @pytest.fixture def fixture_all_languages(): """Return a valid set of languages from the language table.""" return { 'items': [ { 'code': 'AFR', 'name': 'Afrikaans', 'iso_code': 'AF' }, { 'code': 'TEL', 'name': 'Telugu', 'iso_code': 'TE' }, { 'code': 'CHI', 'name': 'Chinese', 'iso_code': 'ZH' }, { 'code': 'cmn-Hant', 'name': 'Chinese (Traditional)', 'iso_code': 'ZH' } ] } class OwsRequestMockBuilder: """Utility to build mocking setup for owsrequest.""" def __init__(self, mocker): """Initializer.""" self.specs = {method: [] for method in ['get', 'post', 'head', 'put']} self.mocker = mocker def add(self, spec_with_method): """Add a spec to the list of available methods.""" method = spec_with_method.pop('method') self.specs[method].append(spec_with_method) return self def apply(self): """Set up actual mock with added specs.""" for method in ['get', 'post', 'head', 'put']: specs = self.specs[method] requests = test_utils.mock_ows_requests(specs) self.mocker.patch.object(request, method, requests) @pytest.fixture def ows_request_mocker(mocker): """Fixture for mocking requests.""" return OwsRequestMockBuilder(mocker) class MockLogAdaptor: """Mocking Class for Logger. This class replaces the ows logger class, so it's easier to use and test in our codebase. """ def error(self, message, *args, **kwargs): """Log errors. Args: message (str): the message to log. args (list): additional arguments. kwargs (dict): additional named arguments. """ pass def warning(self, message, *args, **kwargs): """Log Warnings. Args: message (str): the message to log. args (list): additional arguments. kwargs (dict): additional named arguments. """ pass def info(self, message, *args, **kwargs): """Log information. Args: message (str): the message to log. args (list): additional arguments. kwargs (dict): additional named arguments. """ pass @pytest.fixture def context(): """Fixture app context.""" with application.app.app_context() as context: context.g.ows = flask_logger.Ows() context.g.ows.log = MockLogAdaptor() context.g.ows.correlation_id = 'correlation-id' context.g.request_context = MagicMock() context.g.log = MagicMock() yield context @pytest.fixture def fixture_subaccount_ownership_success(): """Return subaccount ownership success reponse.""" return response.Response(status=200) @pytest.fixture def fixture_subaccount_ownership_failure(): """Return subaccount ownership failure reponse.""" return response.Response(status=403) @pytest.fixture def test_feature_control(): """Return a feature control for testing.""" return 'Test feature' @pytest.fixture def fixture_subaccount_info_response(): """Return subaccount info success response.""" response_message = { 'subaccount_id': 5, 'vendor_id': 1234, 'subaccount_name': 'Test subaccount', } return response.Response(status=200, message=response_message) @pytest.fixture def fixture_subaccount_info_failure(): """Return subaccount info failure response.""" return response.Response(status=403) @pytest.fixture def fixture_enabled_features_response(): """Return enabled features success response.""" response_message = {'items': [ {'feature_id': 1, 'feature_name': 'Test feature'}] } return response.Response(status=200, message=response_message) @pytest.fixture def fixture_enabled_features_failure(): """Return enabled features failure response.""" return response.Response(status=403) @pytest.fixture def mock_hfa_eligible_digital_physical_release_rows(): """Returns a list of dicts representing HFA eligible digital/physical release query results.""" return [ { "track_id": 101, "track_isrc": "USABC1234567", "track_name": "Song A", "release_name": "Album A", "upc": "123456789012", "vendor_status": "signed", "release_status": "in_content", "track_third_party_publishing": None, "release_territory_carveouts": "2,3", "subaccount_territory_carveouts": "2,3", "ingestion_completed": 1, "vendor_id": 3, "subaccount_id": 1001, "trans_type": "digital", }, { "track_id": 102, "track_isrc": "USDEF7654321", "track_name": "Song B", "release_name": "Album B", "upc": "987654321098", "vendor_status": "signed", "release_status": "in_content", "track_third_party_publishing": "Yes", "release_territory_carveouts": "4,5", "subaccount_territory_carveouts": "4,5,6", "ingestion_completed": 1, "vendor_id": 3, "subaccount_id": None, "trans_type": "physical", }, { "track_id": 103, "track_isrc": "USGHI7654321", "track_name": "Song C", "release_name": "Album C", "upc": "135789012346", "vendor_status": "signed", "release_status": "in_content", "track_third_party_publishing": "Yes", "release_territory_carveouts": "4,5", "subaccount_territory_carveouts": "4,5,6", "ingestion_completed": 1, "vendor_id": 1, "subaccount_id": 1002, "trans_type": "digital", }, ] @pytest.fixture def mock_hfa_eligible_digital_physical_release_df(mock_hfa_eligible_digital_physical_release_rows): """Returns a HFA eligible digital/physical release DataFrame.""" return pd.DataFrame(mock_hfa_eligible_digital_physical_release_rows) @pytest.fixture def mock_hfa_eligible_ringtone_rows(): """Returns a list of dicts representing HFA eligible ringtone query results.""" return [ { "track_id": 501, "track_isrc": "USXYZ1234567", "upc": "112233445566", "vendor_id": 3, "hfa_configuration_code": "RT", }, { "track_id": 502, "track_isrc": "USXYZ7654321", "upc": "665544332211", "vendor_id": 43, "hfa_configuration_code": "RT", }, ] @pytest.fixture def mock_hfa_eligible_ringtone_df(mock_hfa_eligible_ringtone_rows): """Returns a HFA eligible ringtone DataFrame.""" return pd.DataFrame(mock_hfa_eligible_ringtone_rows) @pytest.fixture def mock_oa_vendor_active_contract_rows(): """Returns a list of dicts representing OA vendor active contract query results.""" return [ { "vendor_id": 1, "oms_type": None, "territory_carve_out": None, "physical_track_publishing_type": None, "ringtone_publishing_type": None, }, { "vendor_id": 2, "oms_type": "both", "territory_carve_out": "1", "physical_track_publishing_type": "both", "ringtone_publishing_type": "both", }, { "vendor_id": 3, "oms_type": "orchard", "territory_carve_out": "145", "physical_track_publishing_type": "orchard", "ringtone_publishing_type": "orchard", }, { "vendor_id": 4, "oms_type": "label", "territory_carve_out": "145", "physical_track_publishing_type": "label", "ringtone_publishing_type": "label", }, { "vendor_id": 5, "oms_type": "none", "territory_carve_out": "145", "physical_track_publishing_type": "none", "ringtone_publishing_type": "none", }, ] @pytest.fixture def mock_oa_vendor_active_contract_df(mock_oa_vendor_active_contract_rows): """Returns a OA vendor active contract DataFrame.""" return pd.DataFrame(mock_oa_vendor_active_contract_rows) @pytest.fixture def mock_distribution_format_hfa_rows(): """Returns a list of dicts representing distribution_format_hfa rows.""" return [ {"hfa_configuration_code": "CD", "context_type": "physical"}, {"hfa_configuration_code": "LP", "context_type": "physical"}, {"hfa_configuration_code": "CS", "context_type": "physical"}, {"hfa_configuration_code": "SP", "context_type": "digital"}, ] @pytest.fixture def mock_distribution_format_hfa_df(mock_distribution_format_hfa_rows): """Returns a distribution_format_hfa DataFrame.""" return pd.DataFrame(mock_distribution_format_hfa_rows) @pytest.fixture def mock_get_all_active_contract_mechanical_deductions_response(): """Returns a list of dicts representing abacus active contract with mechanical deductions.""" return [ { "account_id": 1, "contract_id": 1, "term_type": "product", "attachments": [ "135789012346", ], "mechanical_type": ["digital", "physical"], }, { "account_id": 1, "contract_id": 2, "term_type": "label", "attachments": [ "1", ], "mechanical_type": ["digital", "physical"], }, { "account_id": 1, "contract_id": 3, "term_type": "track", "attachments": [ "USGHI7654321", ], "mechanical_type": ["digital", "physical"], }, { "account_id": 6, "contract_id": 6, "term_type": "product", "attachments": ["000000000000", "11111111111"], "mechanical_type": ["digital", "physical"], }, ] @pytest.fixture def mock_hfa_eligible_tracks(): """Returns a list of dicts representing HFA eligible tracks for processing query results.""" return [ { "track_id": 101, "track_isrc": "USABC1234567", "upc": "123456789012", "release_id": 201, "vendor_id": 3, "hfa_configuration_code": "CD", "trans_type": "digital", }, { "track_id": 102, "track_isrc": "USDEF7654321", "upc": "987654321098", "release_id": 202, "vendor_id": 3, "hfa_configuration_code": "LP", "trans_type": "physical", }, { "track_id": 501, "track_isrc": "USXYZ1234567", "upc": "112233445566", "release_id": None, "vendor_id": 3, "hfa_configuration_code": "RT", "trans_type": "digital", } ]