"""Conftest File for Testing.""" from datetime import datetime from random import choice from string import ascii_uppercase from unittest.mock import MagicMock, NonCallableMagicMock import uuid from flask import g import pytest from python_pdp_sdk.backends.authorization_backend import AuthorizationBackend from sqlalchemy import exc import application from project_manager import api from project_manager.connector import mysql from project_manager.constant import field_const from project_manager.constant import header_const from project_manager.constant import marketing_const from project_manager.constant import validation_const from tests.utils import db from tests.utils import http_utils from tests.utils.project_code import random_project_code @pytest.fixture(scope='session', autouse=True) def register_sqlite_now_function(): """Register a NOW() function on the shared SQLite test connection. SQLite has no built-in NOW(); this shim lets raw SQL that uses NOW() (e.g. the project UPDATE in execute_content_transfer) work in tests without needing to rewrite production queries. """ raw_conn = mysql._project_manager_engine.raw_connection() raw_conn.create_function('NOW', 0, lambda: datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')) header_blank_correlation_id = { 'field': header_const.CORRELATION_ID, 'reason': 'required', 'header': { header_const.CORRELATION_ID: '', header_const.GRASS_ACCOUNT_TYPE: header_const.GRASS_ACCOUNT_TYPE_VENDOR, header_const.GRASS_ACCOUNT_ID: 1, header_const.CONTENT_TYPE: header_const.APPLICATION_JSON}} header_invalid_content_type = { 'field': header_const.CONTENT_TYPE, 'reason': 'pattern', 'header': { header_const.CORRELATION_ID: str(uuid.uuid1()), header_const.GRASS_ACCOUNT_TYPE: header_const.GRASS_ACCOUNT_TYPE_VENDOR, header_const.GRASS_ACCOUNT_ID: 1, header_const.CONTENT_TYPE: 'invalid content type'}} header_invalid_grass_account_type = { 'field': header_const.GRASS_ACCOUNT_TYPE, 'reason': 'pattern', 'header': { header_const.CORRELATION_ID: str(uuid.uuid1()), header_const.GRASS_ACCOUNT_TYPE: 'invalid account', header_const.GRASS_ACCOUNT_ID: 1, header_const.CONTENT_TYPE: header_const.APPLICATION_JSON}} header_blank_grass_account_type = { 'field': header_const.GRASS_ACCOUNT_TYPE, 'reason': 'pattern', 'header': { header_const.CORRELATION_ID: str(uuid.uuid1()), header_const.GRASS_ACCOUNT_TYPE: '', header_const.GRASS_ACCOUNT_ID: 1, header_const.CONTENT_TYPE: header_const.APPLICATION_JSON}} header_blank_grass_account_id = { 'field': header_const.GRASS_ACCOUNT_ID, 'reason': 'pattern', 'header': { header_const.CORRELATION_ID: str(uuid.uuid1()), header_const.GRASS_ACCOUNT_TYPE: header_const.GRASS_ACCOUNT_TYPE_VENDOR, header_const.GRASS_ACCOUNT_ID: '', header_const.CONTENT_TYPE: header_const.APPLICATION_JSON}} json_unknown_field = { 'field': 'additionalProperties', 'reason': 'unexpected', 'body': { field_const.PROJECT_CODE: 'test project code', field_const.PROJECT_NAME: 'test project name', 'unknown_field': 'unknown value'}} json_blank_project_name = { 'field': field_const.PROJECT_NAME, 'reason': 'required', 'body': { field_const.PROJECT_CODE: 'test project code', field_const.PROJECT_NAME: ''}} json_blank_project_code = { 'field': field_const.PROJECT_CODE, 'reason': 'required', 'body': { field_const.PROJECT_CODE: '', field_const.PROJECT_NAME: 'test project name'}} json_missing_project_name = { 'field': field_const.PROJECT_NAME, 'reason': 'required', 'body': { field_const.PROJECT_CODE: 'test project code'}} json_missing_project_code = { 'field': field_const.PROJECT_CODE, 'reason': 'required', 'body': { field_const.PROJECT_NAME: 'test project name'}} json_blank_subaccount_id = { 'field': field_const.SUBACCOUNT_ID, 'reason': 'pattern', 'body': { field_const.PROJECT_NAME: 'test project name', field_const.PROJECT_CODE: 'test project code', field_const.SUBACCOUNT_ID: ''}} json_long_project_code = { 'field': field_const.PROJECT_CODE, 'reason': 'long', 'body': { field_const.PROJECT_NAME: '12abcdef44', field_const.PROJECT_CODE: ''.join( choice(ascii_uppercase) for i in range( validation_const.MAX_PROJECT_CODE_LEN + 1))}} json_long_project_name = { 'field': field_const.PROJECT_NAME, 'reason': 'long', 'body': { field_const.PROJECT_NAME: ''.join( choice(ascii_uppercase) for i in range( validation_const.MAX_PROJECT_NAME_LEN + 1)), field_const.PROJECT_CODE: '12abcdef44'}} json_whitespace_empty_code = { 'field': field_const.PROJECT_CODE, 'reason': 'required', 'body': { field_const.PROJECT_NAME: 'test project name', field_const.PROJECT_CODE: ''}} json_whitespace_space_code = { 'field': field_const.PROJECT_CODE, 'reason': 'required', 'body': { field_const.PROJECT_NAME: 'test project name', field_const.PROJECT_CODE: ' '}} json_whitespace_spaces_code = { 'field': field_const.PROJECT_CODE, 'reason': 'required', 'body': { field_const.PROJECT_NAME: 'test project name', field_const.PROJECT_CODE: ' '}} json_whitespace_newline_code = { 'field': field_const.PROJECT_CODE, 'reason': 'required', 'body': { field_const.PROJECT_NAME: 'test project name', field_const.PROJECT_CODE: '\n'}} json_whitespace_tab_code = { 'field': field_const.PROJECT_CODE, 'reason': 'required', 'body': { field_const.PROJECT_NAME: 'test project name', field_const.PROJECT_CODE: '\t'}} json_whitespace_empty_name = { 'field': field_const.PROJECT_NAME, 'reason': 'required', 'body': { field_const.PROJECT_NAME: '', field_const.PROJECT_CODE: 'test project code'}} json_whitespace_space_name = { 'field': field_const.PROJECT_NAME, 'reason': 'required', 'body': { field_const.PROJECT_NAME: ' ', field_const.PROJECT_CODE: 'test project code'}} json_whitespace_spaces_name = { 'field': field_const.PROJECT_NAME, 'reason': 'required', 'body': { field_const.PROJECT_NAME: ' ', field_const.PROJECT_CODE: 'test project code'}} json_whitespace_newline_name = { 'field': field_const.PROJECT_NAME, 'reason': 'required', 'body': { field_const.PROJECT_NAME: '\n', field_const.PROJECT_CODE: 'test project code'}} json_whitespace_tab_name = { 'field': field_const.PROJECT_NAME, 'reason': 'required', 'body': { field_const.PROJECT_NAME: '\t', field_const.PROJECT_CODE: 'test project code'}} @pytest.fixture def db_fixture(request): """Create and inserts a standard common db fixture. Doesn't create the actual products to leave flexibility for different project configurations to be created Args: request (_pytest.python.SubRequest): a sub request for handling getting a fixture from a test function/fixture. """ db.exit_if_not_test_environment() # db_fixture_teardown() db.create_company_brand_table() db.create_parent_company_table() db.create_project_table() db.create_mkt_priority_project() db.create_release_artist_table() db.create_product_type_table() db.create_distribution_format_table() db.create_distribution_format_media_table() db.create_releases_table() db.create_release_correction_table() db.create_release_approval_queue_table() db.create_subaccount_table() db.create_vendor_table() db.create_artist_info_table() db.insert_company_brand() db.insert_parent_company() db.insert_release_artists() db.insert_product_types() db.insert_distribution_formats() db.insert_releases() db.insert_release_correction() db.insert_release_approval_queue() db.insert_distribution_format_media() db.insert_vendor() db.insert_subaccount() db.insert_mkt_priority_project() def db_fixture_teardown(): db.drop_project_table() db.drop_distribution_format_table() db.drop_release_correction_table() db.drop_release_approval_queue_table() db.drop_releases_table() db.drop_release_artist_table() db.drop_product_type_table() db.drop_distribution_format_media_table() db.drop_vendor_table() db.drop_mkt_priority_project_table() db.drop_artist_info_table() db.drop_company_brand_table() db.drop_parent_company_table() request.addfinalizer(db_fixture_teardown) @pytest.fixture def product_genres_db_fixture(request): """Create table and insert rows needed for product genre queries. Args: request (_pytest.python.SubRequest): a sub request for handling getting a fixture from a test function/fixture. """ with mysql.pm_session_scope() as session: db.exit_if_not_test_environment(session) db.create_product_genre_table(session) db.insert_product_genres(session) request.addfinalizer(db.drop_product_genre_table) @pytest.fixture def product_imprints_db_fixture(request): """Create tables and insert rows needed for product imprint queries. Args: request (_pytest.python.SubRequest): a sub request for handling getting a fixture from a test function/fixture. """ with mysql.pm_session_scope() as session: db.exit_if_not_test_environment(session) db.create_product_imprint_tables(session) db.insert_product_imprint_values(session) db.create_vendor_table() db.insert_vendor() db.create_subaccount_table() db.insert_subaccount() def teardown(): db.drop_product_imprint_tables() db.drop_vendor_table() request.addfinalizer(teardown) @pytest.fixture def product_artist_db_fixture(request): """Create tables and insert rows needed for product imprint queries. Args: request (_pytest.python.SubRequest): a sub request for handling getting a fixture from a test function/fixture. """ with mysql.pm_session_scope() as session: db.exit_if_not_test_environment(session) db.create_product_artist_tables(session) db.insert_product_artist_values(session) request.addfinalizer(db.drop_product_imprint_tables) @pytest.fixture def product_artist_with_deletions_db_fixture(request): """Create tables and insert rows needed for product imprint queries. Args: request (_pytest.python.SubRequest): a sub request for handling getting a fixture from a test function/fixture. """ with mysql.pm_session_scope() as session: db.exit_if_not_test_environment(session) db.create_product_artist_tables(session) db.insert_product_artist_values_with_deletions(session) request.addfinalizer(db.drop_product_imprint_tables) @pytest.fixture def product_subgenres_db_fixture(request): """Create table and insert rows needed for product subgenre queries. Args: request (_pytest.python.SubRequest): a sub request for handling getting a fixture from a test function/fixture. """ with mysql.pm_session_scope() as session: db.exit_if_not_test_environment(session) db.create_product_subgenre_table(session) db.insert_product_subgenres(session) request.addfinalizer(db.drop_product_subgenre_table) @pytest.fixture def available_project_codes_db_fixture(request): """Create table and insert rows needed for available project codes queries. Args: request (_pytest.python.SubRequest): a sub request for handling getting a fixture from a test function/fixture. """ with mysql.pm_session_scope() as session: db.exit_if_not_test_environment(session) db.create_product_artist_tables(session) db.create_subaccount_table() db.create_vendor_table() db.insert_product_artist_values(session) db.insert_vendor() db.insert_subaccount() db.insert_projects() def db_fixture_teardown(): db.drop_product_imprint_tables() db.drop_subaccount_table() db.drop_vendor_table() request.addfinalizer(db_fixture_teardown) @pytest.fixture def project(): """Insert project.""" db.insert_project() @pytest.fixture def deleted_project(): """Insert deleted project.""" db.insert_deleted_project() @pytest.fixture def projects(): """Insert projects.""" db.insert_projects() @pytest.fixture def projects_without_subaccount(): """Insert projects without a subaccount.""" db.insert_projects_without_subaccount() @pytest.fixture def multiple_primary_artists(): """Insert multiple primary artists into the first release.""" db.insert_multiple_primary_artists() @pytest.fixture def product_with_no_primary_artists(): """Insert a release without primary artists.""" db.insert_release_no_primary_artists() @pytest.fixture def release_with_status_label_processing_and_rejected(): """Insert release with status label_processing and release approval entry. Adds a release approval entry of rejected for the label_processing release. """ db.insert_release_label_processing_rejected() @pytest.fixture def release_with_status_in_content_and_correction_submitted(): """Insert release with status in_content and error correction entry. Adds an in_content release and creates a corresponding error correction entry with a status of submitted. """ db.insert_release_in_content_correction_submitted() @pytest.fixture def subaccount(): """Insert subaccount.""" db.insert_subaccount() @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('project_manager.models..authorization_backend', mock_authorization_backend): ... Patching `project_manager.api.authorization_backend` from this fixture would not affect modules that did `from project_manager.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_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 # The test DB can get corrupted if we try to access it in parallel # This makes all flask_executor jobs run in series api.app.config['EXECUTOR_MAX_WORKERS'] = 1 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' yield test_request_context @pytest.fixture def header(): """Return header dict fixture. Returns: dict: header dict """ correlation_id = str(uuid.uuid1()) return http_utils.get_valid_header(correlation_id) @pytest.fixture def header_subaccount(): """Return header dict fixture. Returns: dict: header dict """ correlation_id = str(uuid.uuid1()) return http_utils.get_valid_header_subaccount(correlation_id) def header_user_helper(orchard_user_id): """Return header dict fixture for orchard user. Returns: dict: header dict """ correlation_id = str(uuid.uuid1()) headers = http_utils.get_valid_header_subaccount(correlation_id) headers[header_const.ORCHARD_USER_ID] = orchard_user_id return headers @pytest.fixture def header_user(orchard_user_id): """Return header dict fixture for orchard user. Returns: dict: header dict """ return header_user_helper(orchard_user_id) def header_user_oa_helper(orchard_user_id_oa): """Return header dict fixture for orchard user. Returns: dict: header dict """ return http_utils.get_valid_oa_header(str(uuid.uuid1()), orchard_user_id_oa) @pytest.fixture def header_user_oa(orchard_user_id_oa): """Return header dict fixture for orchard user. Returns: dict: header dict """ return header_user_oa_helper(orchard_user_id_oa) @pytest.fixture(params=[ header_blank_correlation_id, header_invalid_content_type, header_invalid_grass_account_type, header_blank_grass_account_type, header_blank_grass_account_id]) def invalid_headers(request): """Return invalid headers.""" return request.param @pytest.fixture(params=[ json_unknown_field, json_blank_project_name, json_blank_project_code, json_missing_project_name, json_missing_project_code, json_blank_subaccount_id, json_long_project_code, json_long_project_name, json_whitespace_empty_code, json_whitespace_space_code, json_whitespace_spaces_code, json_whitespace_newline_code, json_whitespace_tab_code, json_whitespace_empty_name, json_whitespace_space_name, json_whitespace_spaces_name, json_whitespace_newline_name, json_whitespace_tab_name]) def invalid_json_request(request): """Invalid JSON request.""" return request.param @pytest.fixture def vendor_id(): """Return a vendor id.""" return 7123 @pytest.fixture def subaccount_id(): """Return a subaccount id.""" return 1 @pytest.fixture def orchard_user_id(): """Orchard-User-Id header value.""" return 'alw:123' @pytest.fixture def orchard_user_id_oa(): """Orchard-User-Id header value.""" return 'oa:123' @pytest.fixture def highlight_data(): """Return a dictionary of fake project highlights data.""" return { 'scope': 'public', 'highlight_id': 39752, 'description': 'Kissed Madonna on stage at the VMAs', 'client': 'alw', 'attachment': 'N', 'mkt_program_id': 12, 'entity_id': 1900566, 'entity': 'release', 'subject': 'Project Highlights Subject'} @pytest.fixture def project_id(): """Return a fake project id for testing.""" return 1 @pytest.fixture def mock_owsrequest(): """Mock owsrequest function.""" class MockOwsRequest: def __init__(self, status_code, json): self.status_code = status_code self._json = json def json(self): return self._json return MockOwsRequest @pytest.fixture def project_highlight_ok(highlight_data, project_id): """Return a mock owsrequest spec for an OK project highlight call.""" return { 'service': 'ows-marketing', 'path': '/highlights/project/{}?mkt_program_id={}'.format( project_id, marketing_const.PROGRAM_MARKETING_HIGHLIGHTS), 'status': 200, 'json': {'items': [highlight_data]}} @pytest.fixture def project_highlight_ok_no_products(highlight_data): """Return a mock owsrequest spec for an OK project highlight call.""" return { 'service': 'ows-marketing', 'path': '/highlights/project/{}?mkt_program_id={}'.format( 3, marketing_const.PROGRAM_MARKETING_HIGHLIGHTS), 'status': 200, 'json': {'items': [highlight_data]}} @pytest.fixture def project_highlight_not_found(project_id): """Return an owsrequest spec for a 404 from marketing.""" return { 'service': 'ows-marketing', 'path': '/highlights/project/{}?mkt_program_id={}'.format( project_id, marketing_const.PROGRAM_MARKETING_HIGHLIGHTS), 'status': 404, 'json': {'code': 'not_found_error', 'message': None}} @pytest.fixture def highlight_created(project_id): """Return an owsrequest spec for a successful highlight POST.""" return { 'service': 'ows-marketing', 'path': '/highlights', 'status': 201, 'json': { 'description': 'new project highlight', 'entity': 'project', 'scope': 'public', 'subject': 'Product Marketing Highlight', 'entity_id': project_id, 'highlight_id': 39754, 'attachment': 'N', 'client': 'alw', 'mkt_program_id': 20}} @pytest.fixture def get_valid_post_dict(): """Return valid post request dict. Returns: dict: valid post request dict """ return { field_const.PROJECT_CODE: random_project_code(), field_const.PROJECT_NAME: str(uuid.uuid4()), field_const.ARTIST_ID: 12342, field_const.DESCRIPTION: 'a description'} @pytest.fixture def get_artist_info_ok(): """Return owsrequest spec for a 200 from ows-artist.""" return { 'service': 'ows-artist', 'path': '/artist/123', 'status': 200, 'json': { 'id': 123, 'name': 'Solange Knowles', 'artist_type': 'artist'}} @pytest.fixture def create_artist_info_ok(): """Return owsrequest spec for a 200 from ows-artist.""" return { 'service': 'ows-artist', 'path': '/artist', 'status': 200, 'json': { 'id': 123, 'name': 'Solange Knowles', 'artist_type': 'artist'}} @pytest.fixture def get_artist_info_not_found(): """Return owsrequest spec for a 404 from ows-artist.""" return { 'service': 'ows-artist', 'path': '/artist/123', 'status': 404, 'json': {'code': 'not_found_error', 'message': 'not found message'}} @pytest.fixture def get_artist_info_multiple_artists_ok(): """Return owsrequest spec for a 200 from ows-artist for three artists.""" return [{ 'service': 'ows-artist', 'path': '/artist/4567', 'status': 200, 'json': { 'id': 4567, 'name': 'Solange Knowles', 'artist_type': 'artist'}}, { 'service': 'ows-artist', 'path': '/artist/311', 'status': 200, 'json': { 'id': 311, 'name': 'Solange Knowles', 'artist_type': 'artist'}}, { 'service': 'ows-artist', 'path': '/artist/123', 'status': 200, 'json': { 'id': 123, 'name': 'Solange Knowles', 'artist_type': 'artist'}}] def get_mock_content_review_data(product_id=1001, is_okay=True): """Format content review response data.""" json_data = { 'id': int(product_id), 'ec_and_ar': False, 'release_status': 'transfer_to_content', 'release_approval_status': 'checked_in' } if not is_okay: json_data = { 'code': 'bad_request', 'message': f'Review queue record with id `{product_id}` not found' } return { 'service': 'ows-product-review', 'path': f'/status/{product_id}', 'status': 200 if is_okay else 400, 'json': json_data } @pytest.fixture def get_contentreview_data_ok(): """Return a mock owsrequest spec for an OK project highlight call.""" return get_mock_content_review_data() @pytest.fixture def get_content_review_data_all( get_vendor_ows_response, get_company_brand_ows_response): """Get all responses from ows-product-review for tests.""" return [ get_vendor_ows_response, get_company_brand_ows_response, get_mock_content_review_data(), get_mock_content_review_data(1002, False), get_mock_content_review_data(1003), get_mock_content_review_data(1014, False) ] @pytest.fixture def get_contentreview_data_not_found(): """Return a mock owsrequest spec for an OK project highlight call.""" return get_mock_content_review_data(1002, False) @pytest.fixture def valid_sqs_payload(): """Return dict with valid payload data for sqs connector.""" return { 'project_id': 1, 'job_id': 15} @pytest.fixture def test_lce_database(): """Initialize in-memory database for testing purposes.""" mysql.BaseModel.metadata.create_all(mysql._lce_session_engine) yield mysql.BaseModel.metadata.drop_all(mysql._lce_session_engine) @pytest.fixture def db_exception(): """Return (but not throw) SQLAlchemy exception.""" return exc.SQLAlchemyError('kaboom') @pytest.fixture def mock_vendor_data(): """Mock vendor data.""" return { 'vendor_id': '7123', 'owner': 'odd' } @pytest.fixture def get_vendor_ows_response(mock_vendor_data): """Return owsrequest spec for a 200 from ows-account.""" return { 'service': 'ows-account', 'path': '/vendor/7123', 'status': 200, 'json': mock_vendor_data } @pytest.fixture def track_tables(request): """Create track, track_artist, and track_writer tables with base rows.""" db.create_track_table() db.create_track_artist_table() db.create_track_writer_table() db.insert_tracks() db.insert_track_artists() db.insert_track_writers() def teardown(): db.drop_track_writer_table() db.drop_track_artist_table() db.drop_track_table() request.addfinalizer(teardown) @pytest.fixture def get_company_brand(): """Valid company brand.""" return 'theorchard' @pytest.fixture def get_company_brand_ows_response(get_company_brand): """Return owsrequest spec for a 200 from ows-account.""" return { 'service': 'ows-account', 'path': '/vendor/brand/7123', 'status': 200, 'text': get_company_brand }