from decimal import Decimal import json import logging import application import boto3 from moto import mock_aws from oto import response from owsfeatures import features as owsfeatures import pytest from flexmock import flexmock from masters_registry import config from masters_registry import features from masters_registry import handlers from masters_registry.connectors import mysql from masters_registry.constant import db_const from masters_registry.constant import field_const from masters_registry.constant import opcode_const from masters_registry.models import bulk_tasks from tests.helpers import db from tests.helpers import patches config.environment = config.TEST_ENVIRONMENT def _is_feature_enabled_mock(feature_name, *args, **kwargs): return owsfeatures.is_feature_enabled(feature_name) @pytest.fixture def feature_engine(feature_engine, monkeypatch): monkeypatch.setattr( features, '_is_feature_enabled', _is_feature_enabled_mock) return feature_engine @pytest.fixture def client(): """Returns flask test client. Returns: flask client: flask test client """ return application.app.test_client() @pytest.fixture def put_ownership_response_object(): return { 'tuid': '12345', 'territories': ['US', 'MX'] } @pytest.fixture def put_ownership_passed_url_vars(): return { 'tuid': '12345', 'territories': ['US', 'MX'] } @pytest.fixture def valid_correlation_id(): """Returns a valid correlation_id string.""" return '8f09a8be-810c-42b2-b529-46d4dfce514a' @pytest.fixture def valid_headers(valid_correlation_id): """Returns valid headers. Returns: dict: header dict """ return { field_const.CORRELATION_ID: valid_correlation_id, field_const.ORCHARD_USER_ID: '1234'} @pytest.fixture def territory_ISO_3166_1_2016_content(): """Return test ows-territories content. Returns: content (bytes): sample content for territories/ISO_3166_1_2016. """ content = json.dumps( { 'items': [{ 'continent': None, 'orch_id': 187, 'standard': 'ISO_3166_1_2016', 'territory_code_a2': 'YE', 'territory_code_a3': 'YEM', 'territory_code_numeric': 887, 'territory_name': 'Yemen' }, { 'continent': None, 'orch_id': 188, 'standard': 'ISO_3166_1_2016', 'territory_code_a2': 'ZM', 'territory_code_a3': 'ZMB', 'territory_code_numeric': 894, 'territory_name': 'Zambia' }, { 'continent': None, 'orch_id': 189, 'standard': 'ISO_3166_1_2016', 'territory_code_a2': 'ZW', 'territory_code_a3': 'ZWE', 'territory_code_numeric': 716, 'territory_name': 'Zimbabwe' }], 'pagination': { 'limit': 1000, 'offset': 0, 'total_records': 3, 'type': 'standard'}}) return content.encode() @pytest.fixture def get_single_isrc_from_db(): return [ { 'track_name': 'Track 1', 'isrc': 'USA560319122', 'tuid': 140030, 'deletions': 'N', 'product_type_id': db_const.MUSIC_PRODUCT_TYPE_ID, 'upc': '123456789120', 'release_status': 'in_content', 'not_for_distribution': 'N', 'track_type': 'music', 'distribution_context_type': db_const.DIGITAL_DISTRIBUTION_FORMAT, }, { 'track_name': 'Track 2', 'isrc': 'USA560319123', 'tuid': 140031, 'deletions': 'N', 'product_type_id': db_const.MUSIC_PRODUCT_TYPE_ID, 'upc': '123456789120', 'release_status': 'in_content', 'not_for_distribution': 'N', 'track_type': 'music', 'distribution_context_type': db_const.DIGITAL_DISTRIBUTION_FORMAT, }, { 'track_name': 'Track 3', 'isrc': 'USA560319124', 'tuid': 140032, 'deletions': 'N', 'product_type_id': db_const.MUSIC_PRODUCT_TYPE_ID, 'upc': '123456789120', 'release_status': 'in_content', 'not_for_distribution': 'N', 'track_type': 'music', 'distribution_context_type': db_const.DIGITAL_DISTRIBUTION_FORMAT, }, { 'track_name': 'Track 4', 'isrc': 'USA560319125', 'tuid': 140033, 'deletions': 'N', 'product_type_id': db_const.MUSIC_PRODUCT_TYPE_ID, 'upc': '123456789120', 'release_status': 'in_content', 'not_for_distribution': 'N', 'track_type': 'music', 'distribution_context_type': db_const.DIGITAL_DISTRIBUTION_FORMAT, } ] @pytest.fixture def get_isrcs_from_db(): return [ { 'track_name': 'Track 1', 'isrc': 'USA560319122', 'tuid': 140030, 'deletions': 'N', 'product_type_id': db_const.MUSIC_PRODUCT_TYPE_ID, 'upc': '123456789120', 'release_status': 'in_content', 'not_for_distribution': 'N', 'track_type': 'music', 'distribution_context_type': db_const.DIGITAL_DISTRIBUTION_FORMAT, }, { 'track_name': 'Track 2', 'isrc': 'USA560319123', 'tuid': 140031, 'deletions': 'N', 'product_type_id': db_const.MUSIC_PRODUCT_TYPE_ID, 'upc': '123456789120', 'release_status': 'in_content', 'not_for_distribution': 'N', 'track_type': 'music', 'distribution_context_type': db_const.DIGITAL_DISTRIBUTION_FORMAT, }, { 'track_name': 'Track 3', 'isrc': 'USA560319124', 'tuid': 140032, 'deletions': 'Y', 'product_type_id': db_const.MUSIC_PRODUCT_TYPE_ID, 'upc': '123456789121', 'release_status': 'in_content', 'not_for_distribution': 'N', 'track_type': 'music', 'distribution_context_type': db_const.DIGITAL_DISTRIBUTION_FORMAT, }, { 'track_name': 'Track 4', 'isrc': 'USA560319125', 'tuid': 140033, 'deletions': 'Y', 'product_type_id': db_const.MUSIC_PRODUCT_TYPE_ID, 'upc': '123456789121', 'release_status': 'in_content', 'not_for_distribution': 'N', 'track_type': 'music', 'distribution_context_type': db_const.DIGITAL_DISTRIBUTION_FORMAT, } ] @pytest.fixture def expected_response(): return [ { 'track_name': 'Track 1', 'isrc': 'USA560319122', 'tuid': 140030, 'deletions': 'N', 'product_type_id': db_const.MUSIC_PRODUCT_TYPE_ID, 'upc': 123456789120, 'release_status': 'in_content', 'not_for_distribution': 'N', 'track_type': 'music', 'distribution_context_type': db_const.DIGITAL_DISTRIBUTION_FORMAT, }, { 'track_name': 'Track 2', 'isrc': 'USA560319123', 'tuid': 140031, 'deletions': 'N', 'product_type_id': db_const.MUSIC_PRODUCT_TYPE_ID, 'upc': 123456789120, 'release_status': 'in_content', 'not_for_distribution': 'N', 'track_type': 'music', 'distribution_context_type': db_const.DIGITAL_DISTRIBUTION_FORMAT, }, { 'track_name': 'Track 3', 'isrc': 'USA560319124', 'tuid': 140032, 'deletions': 'Y', 'product_type_id': db_const.MUSIC_PRODUCT_TYPE_ID, 'upc': 123456789121, 'release_status': 'in_content', 'not_for_distribution': 'N', 'track_type': 'music', 'distribution_context_type': db_const.DIGITAL_DISTRIBUTION_FORMAT, }, { 'track_name': 'Track 4', 'isrc': 'USA560319125', 'tuid': 140033, 'deletions': 'Y', 'product_type_id': db_const.MUSIC_PRODUCT_TYPE_ID, 'upc': 123456789121, 'release_status': 'in_content', 'not_for_distribution': 'N', 'track_type': 'music', 'distribution_context_type': db_const.DIGITAL_DISTRIBUTION_FORMAT, } ] @pytest.fixture def get_deleted_upc_isrcs_from_db(): return [ { 'track_name': 'Track 1', 'isrc': 'USA560319122', 'tuid': 140030, 'deletions': 'Y', 'product_type_id': db_const.MUSIC_PRODUCT_TYPE_ID, 'upc': '123456789120', 'release_status': 'in_content', 'not_for_distribution': 'N', 'track_type': 'music', 'distribution_context_type': db_const.DIGITAL_DISTRIBUTION_FORMAT, } ] @pytest.fixture def get_non_music_upc_isrcs_from_db(): """Returns row from art relations where product_type is not music.""" return [ { 'track_name': 'Track 1', 'isrc': 'USA560319122', 'tuid': 140030, 'deletions': 'Y', 'product_type_id': 2, 'upc': 123456789120, 'release_status': 'in_content', 'not_for_distribution': 'N', 'track_type': 'music', 'distribution_context_type': db_const.DIGITAL_DISTRIBUTION_FORMAT, } ] @pytest.fixture def mock_valid_isrc_info(): return { 'upc': 'UPC22332233', 'isrc': '131131131', 'tuid': 111111, 'track_name': 'Smells like teen spirit', 'deletions': 'N', 'release_status': 'in_content', 'not_for_distribution': 'N', 'track_type': 'music', 'distribution_context_type': db_const.DIGITAL_DISTRIBUTION_FORMAT, } @pytest.fixture def bulk_status_db_fixture(request): """Creates and inserts a standard common db fixture Args: request (_pytest.python.SubRequest): a sub request for handling getting a fixture from a test function/fixture. """ bulk_tasks.BaseModel.metadata.create_all(mysql._bulk_statuses_engine) def db_fixture_teardown(): bulk_tasks.BaseModel.metadata.drop_all(mysql._bulk_statuses_engine) request.addfinalizer(db_fixture_teardown) @pytest.fixture def get_task_not_found(): return response.Response( status=404, errors={ 'message': 'Bulk task does not exist', 'code': 'not_found_error' } ) @pytest.fixture def result_orchard_users_dict(): return { 1: 'Ketan Shah', 3: 'Max Orlov' } @pytest.fixture def db_orchard_users_list(): return [ (1, 'Ketan', 'Shah'), (3, 'Max', 'Orlov') ] @pytest.fixture def db_single_get_orchard_user_names(): return response.Response( message={ 123: 'Test User' } ) @pytest.fixture def ows_features_mock(): """Mock owsfeatures call. Note: this mock doesn't set any feature flags, so you'll need to mock function responsible for check of particular flag. """ (flexmock(handlers.python_owsfeatures.FeatureEngine) .should_receive('load') .and_return(True)) @pytest.fixture def created_conflict(): """Fixture for audit conflict""" conflict = { 'status': opcode_const.CONFLICT_CREATED, 'conflicting_tuids': [999] } return conflict @pytest.fixture def ownership_with_conflict(): """Fixture for ownership record with conflict.""" return { 'isrc': 'USA560319122', 'territories': { 'US': [ {'tuid': 222}, {'tuid': 333} ], 'CA': [{'tuid': 222}], 'RU': {'tuid': 222} }, 'locked_territories': {} } @pytest.fixture def ownership_with_two_conflicts(): """Fixture for ownership record with two conflicts.""" return { 'isrc': 'USA560319122', 'territories': { 'US': [ {'tuid': 222}, {'tuid': 333} ], 'CA': [ {'tuid': 222}, {'tuid': 444} ], 'RU': {'tuid': 222} }, 'locked_territories': {} } @pytest.fixture def ownership_without_conflict(): """Fixture for ownership record with conflict.""" return { 'isrc': 'USA560319122', 'territories': { 'US': [{'tuid': 222}], 'CA': [{'tuid': 222}], 'RU': {'tuid': 222} }, 'locked_territories': {} } @pytest.fixture def ownership_without_conflict_different_tracks(): """Fixture for ownership record without conflict.""" return { 'isrc': 'USA560319122', 'territories': { 'US': [{'tuid': 222}, {'tuid': 333}], 'CA': [{'tuid': 222}], 'RU': {'tuid': 222} }, 'locked_territories': {} } @pytest.fixture def tracks_with_conflict(): """Get tracks with conflict from AR.""" return { 222: { 'track_name': 'Track 1', 'isrc': 'USA560319122', 'tuid': 222, 'upc': '123456789120', 'release_name': 'Super release', 'artist_name': 'Lala Bola', 'vendor_id': 22, 'vendor_name': 'Super Label', 'subaccount_id': None, 'subaccount_name': None }, 333: { 'track_name': 'Track 1', 'isrc': 'USA560319122', 'tuid': 333, 'upc': '123456789120', 'release_name': 'Super release', 'artist_name': 'Lala Bola', 'vendor_id': 33, 'vendor_name': 'Duper Label', 'subaccount_id': None, 'subaccount_name': None } } @pytest.fixture def tracks_with_no_conflict(): """Get tracks with no conflict from AR.""" return { 222: { 'track_name': 'Track 1', 'isrc': 'USA560319122', 'tuid': 222, 'upc': '123456789120', 'release_name': 'Super release', 'artist_name': 'Lala Bola', 'vendor_id': 22, 'vendor_name': 'Super Label', 'subaccount_id': None, 'subaccount_name': None }, 333: { 'track_name': 'Track 1', 'isrc': 'USA560319122', 'tuid': 333, 'upc': '123456789120', 'release_name': 'Super release', 'artist_name': 'Lala Bola', 'vendor_id': 22, 'vendor_name': 'Super Label', 'subaccount_id': None, 'subaccount_name': None } } @pytest.fixture def tracks_with_two_conflicts(): """Get tracks with conflict from AR.""" return { 222: { 'track_name': 'Track 1', 'isrc': 'USA560319122', 'tuid': 222, 'upc': '123456789120', 'release_name': 'Super release', 'artist_name': 'Lala Bola', 'vendor_id': 22, 'vendor_name': 'Super Label', 'subaccount_id': None, 'subaccount_name': None }, 333: { 'track_name': 'Track 1', 'isrc': 'USA560319122', 'tuid': 333, 'upc': '123456789120', 'release_name': 'Super release', 'artist_name': 'Lala Bola', 'vendor_id': 33, 'vendor_name': 'Duper Label', 'subaccount_id': None, 'subaccount_name': None }, 444: { 'track_name': 'Track 1', 'isrc': 'USA560319122', 'tuid': 444, 'upc': '123456789120', 'release_name': 'Super release', 'artist_name': 'Lala Bola', 'vendor_id': 44, 'vendor_name': 'Duper Label', 'subaccount_id': None, 'subaccount_name': None } } @pytest.fixture def tracks_no_conflict_subaccounts(): """Get tracks without conflict from AR.""" return [ { 'track_name': 'Track 1', 'isrc': 'USA560319122', 'tuid': 222, 'upc': '123456789120', 'release_name': 'Super release', 'artist_name': 'Lala Bola', 'vendor_id': 22, 'vendor_name': 'Super Label', 'subaccount_id': 1, 'subaccount_name': 'Super sub' }, { 'track_name': 'Track 1', 'isrc': 'USA560319122', 'tuid': 333, 'upc': '123456789120', 'release_name': 'Super release', 'artist_name': 'Lala Bola', 'vendor_id': 22, 'vendor_name': 'Super Label', 'subaccount_id': 2, 'subaccount_name': 'Duper sub' }, ] @pytest.fixture def mock_mr_session_scope(mocker): """Mock mr_session_scope context manager. Restore it when test finished. Args: mocker (object): pytest-mock Returns: func: mock function that should be called to apply mock """ original_session_scope = mysql.mr_session_scope def wrapped(return_value): """ Mock mr_session_scope context manager with given value. Args: return_value (object): what will be returned from session.execute call Returns: session.execute mock """ session_mock = mocker.Mock() session_mock.execute = mocker.Mock(return_value=return_value) session_scope_mock = mocker.Mock() session_scope_mock.__enter__ = mocker.Mock(return_value=session_mock) session_scope_mock.__exit__ = mocker.Mock() mysql.mr_session_scope = mocker.Mock(return_value=session_scope_mock) return session_mock.execute yield wrapped mysql.mr_session_scope = original_session_scope @pytest.fixture def test_isrc(): """ISRC for testing purposes.""" return 'QA123' @pytest.fixture def test_item(test_isrc): """Test item for DynamoDB masters-active table.""" return { 'isrc': test_isrc, 'locked_territories': {}, 'territories': {'AF': [{'tuid': 1}]}, 'timestamp': Decimal('1489013152408.303'), 'updated_timestamp': Decimal('1515072006.096916') } @pytest.fixture def setup_masters_active_table(): """Create masters-active table for testing with Moto. boto3 produces a lot of debug messages which clutter output on failure. Logging level is decreased for tests that use moto. """ logging.getLogger('boto3').setLevel(logging.WARNING) logging.getLogger('botocore').setLevel(logging.WARNING) def wrapped(dynamo_client): """Wrapper that allows to call fixture from tests.""" dynamo_client.create_table( TableName=config.DYNAMODB_MASTERS_ACTIVE, AttributeDefinitions=[ {'AttributeName': 'isrc', 'AttributeType': 'S'} ], KeySchema=[{'AttributeName': 'isrc', 'KeyType': 'HASH'}], ProvisionedThroughput={ 'ReadCapacityUnits': 123, 'WriteCapacityUnits': 123 }) return wrapped @pytest.fixture def seed_masters_active_table(test_item): """Populate table with test data.""" def wrapped(dynamo_client, item=test_item): """Wrapper that allows to call fixture from tests.""" dynamo_client.put_item( TableName=config.DYNAMODB_MASTERS_ACTIVE, Item=item) return wrapped @pytest.fixture def setup_new_audit_table(): """Create DynamoDB audit table for testing.""" def wrapped(dynamo_client): """Wrapper that allows to call fixture from tests.""" dynamo_client.create_table( TableName=config.DYNAMODB_MASTERS_AUDIT_NEW, AttributeDefinitions=[ {'AttributeName': 'isrc', 'AttributeType': 'S'}, {'AttributeName': 'timestamp', 'AttributeType': 'N'}], KeySchema=[ {'AttributeName': 'isrc', 'KeyType': 'HASH'}, {'AttributeName': 'timestamp', 'KeyType': 'RANGE'} ], ProvisionedThroughput={ 'ReadCapacityUnits': 123, 'WriteCapacityUnits': 123 }) return wrapped @pytest.fixture def setup_bulk_status_db(): """Create MySQL task_status table for testing. Drop it after each test. """ db.create_tables() yield db.drop_tables() @pytest.fixture def seed_bulk_status_db(): """Populate table with test data.""" def wrapped(task_statuses): """Wrapper that allows to call fixture from tests. Args: task_statuses (list(TaskStatus)): List of objects to save """ with mysql.bulk_statuses_session_scope() as session: db.exit_if_not_test_environment(session) session.bulk_save_objects(task_statuses) return wrapped @pytest.fixture def dynamodb_get_item_response_new_format(): """Response containing item from DynamoDB for given ISRC.""" item = { 'Item': { 'territories': { 'AF': [{'tuid': Decimal('11432391')}] }, 'locked_territories': {}, 'isrc': 'QA123', 'updated_timestamp': Decimal('1514457624.334598'), 'timestamp': Decimal('1489014691557.2031') }, 'ResponseMetadata': {} } return item @pytest.fixture def dynamodb_get_item_response_old_format(): """Response containing item from DynamoDB for given ISRC.""" item = { 'Item': { 'territories': { 'AF': {'tuid': Decimal('11684704')} }, 'locked_territories': {}, 'isrc': 'QA123', 'updated_timestamp': Decimal('1514457624.334598'), 'timestamp': Decimal('1489014691557.2031') }, 'ResponseMetadata': {} } return item @mock_aws @pytest.fixture def s3_client(): return boto3.client('s3') @mock_aws @pytest.fixture def s3_report_bucket(): bucket_name = config.REPORTS_BUCKET_NAME return boto3.resource('s3').Bucket(bucket_name) @pytest.fixture def setup_s3_reports_bucket(s3_client, s3_report_bucket): """Create S3 bucket. Drop it after each test. """ bucket_name = config.REPORTS_BUCKET_NAME s3_client.create_bucket(Bucket=bucket_name) yield s3_report_bucket.objects.all().delete() s3_report_bucket.delete() @pytest.fixture def default_task_data(): """Default task dict for TaskStatus.""" task_data = { 'correlation_id': 'test correlation id', 'count': 1, 'create_datetime': patches.TEST_TIME_AS_DATETIME, 'finish_datetime': patches.TEST_TIME_AS_DATETIME, 'id': 1, 'result': {}, 'status': 'DONE', 'type': 'BULK_LOCK', 'user_id': '123', 'user_name': 'Test user' } return task_data