"""Test generation of pdf file through jinja render and converting.""" from collections import OrderedDict import copy import datetime import itertools import os import random import json from flexmock import flexmock from oto import response import pytest from salessheets import config from salessheets import features from salessheets.connectors import sqs from salessheets.constants import errors from salessheets.constants import field_const from salessheets.constants import localized_template from salessheets.constants import pdf from salessheets.constants import salessheets_history from salessheets.logic import pdf_generation from salessheets.logic import template_rendering from salessheets.models import history from salessheets.models import ows_assets from salessheets.models import ows_marketing from salessheets.models import ows_pricing from salessheets.models import ows_product from salessheets.models import ows_sales_goals from salessheets.models import salessheets from salessheets.models import template_details from tests.utils import pdf as pdf_test_utils from salessheets.utils import barcode_generator from salessheets.utils import htmltopdf from salessheets.utils import misc from salessheets.validation import validate_pdf TEST_RELEASE_ID = '12345' TEST_MANUFACTURER_UPC = '42' TEST_UPC = '1' TEST_DISPLAY_UPC = '111222' TEST_PROJECT_ID = '34567' TEST_IMAGE_UPC = '123456789012' TEST_EMPTY_IMAGE_UPC = None TEST_EMPTY_RELEASE_ID = None @pytest.fixture def valid_template_details_response_message(): """Valid template details response message.""" return { field_const.LOCALIZED_TEMPLATE_TYPE_ID: 2, field_const.NAME: 'USA', field_const.EMAIL: 'email', field_const.STREET: 'street', field_const.CITY: 'city', field_const.APARTMENT: 'apartment', field_const.STATE: 'state', field_const.POSTAL_CODE: 'postal_code', field_const.COUNTRY: 'country', field_const.TELEPHONE: 'telephone', field_const.REGIONS: [1] } @pytest.fixture def valid_data_from_db(valid_template_details_response_message): """Return editable dict with db data.""" orchard_contact = copy.copy(valid_template_details_response_message) orchard_contact.pop(field_const.NAME) orchard_contact.pop(field_const.REGIONS) orchard_contact.pop(field_const.LOCALIZED_TEMPLATE_TYPE_ID) return { field_const.RELEASE_ID: '1', field_const.HEADER_MAIN_COLOR: pdf.ORCHARD_HEADER_MAIN_COLOR, field_const.SALE_START_DATE: datetime.datetime( day=12, month=12, year=12), field_const.PRODUCT_CODE: '123123', field_const.LABEL: 'some label', field_const.WHOLESALE_PRICE: 'price', field_const.GENRE: 'Rock!', field_const.SUBGENRE: 'Strange rock', field_const.CONTEXT_TYPE: 'physical', field_const.RELEASE_UPC: '1', field_const.UPC: TEST_DISPLAY_UPC, field_const.PRODUCT_NAME: 'Name', field_const.TRACK_LIST: { '1': { 'side': [{ 'track_id': '153', 'cd': '1', 'side': None, 'track_number': 1}]}}, field_const.ARTIST_LIST: ['Cool Artist'], field_const.PROJECT_ID: '12345', field_const.DESCRIPTION: 'Sometext', field_const.VENDOR_OWNER: 'ORCHARD', field_const.VENDOR_NAME: 'some name', field_const.IMAGE_LOGO_ORCHARD: pdf.PDF_IMAGE_LOGO_ORCHARD, field_const.CONFIGURATION: 'CD', field_const.PRICING: '15.99', field_const.BOX_LOT: 30, field_const.EXPORTABLE: 'Y', field_const.EXPLICIT: 'N', field_const.EDITION: 'normal_edition', field_const.BARCODE: '../test_barcode_file', field_const.ORCHARD_CONTACT: orchard_contact, field_const.REGIONS: [1], field_const.TEMPLATE_DETAILS_DATA: ( valid_template_details_response_message), 'localized_orchard_contact': ( 'apartment street\ncity, state postal_code\ncountry\ntelephone') } @pytest.fixture(scope='function') def valid_sqs_message_with_user_context(): """Instance of jsonmessageext that could pass validation by schema.""" msg = { 'Body': '{"product_id": ["1"], "output_type": "single",\ "job_id": "1234214", "localized_template_type_id": 2}' } msg['MessageAttributes'] = { 'Correlation-Id': { 'DataType': 'String', 'StringValue': '1234' }, 'feature_flag_user_context': { 'DataType': 'String', 'StringValue': 'alw:25824' } } return msg @pytest.fixture(scope='function') def valid_sqs_message(): """Emulate valid sqs message without template_type.""" mock_message = { 'Body': '{"product_id": ["1"], "output_type": "single",\ "job_id": "1234214", "localized_template_type_id": 2}' } mock_message['MessageAttributes'] = { 'Correlation-Id': { 'DataType': 'String', 'StringValue': '1234' }, 'feature_flag_user_context': { 'DataType': 'String', 'StringValue': 'oa:1234' } } return mock_message @pytest.fixture def valid_get_album_response_message(): """Valid message for get_album function.""" product_ss_date = datetime.datetime(day=12, month=12, year=12) product_name = 'Name' product_description = 'Sometext' message = { field_const.SALE_START_DATE: product_ss_date, field_const.PRODUCT_NAME: product_name, field_const.PRODUCT_CODE: '123123', field_const.GENRE: 'Rock!', field_const.SUBGENRE: 'Strange rock', field_const.WHOLESALE_PRICE: 'price', field_const.DESCRIPTION: product_description, field_const.UPC: TEST_UPC, field_const.DISPLAY_UPC: TEST_DISPLAY_UPC, field_const.CONTEXT_TYPE: 'physical', field_const.LABEL: 'some label', field_const.VENDOR_NAME: 'some name', field_const.PROJECT_ID: '12345', field_const.MANUFACTURER_UPC: '42', field_const.VENDOR_OWNER: 'ORCHARD', field_const.PRICING: '15.99', field_const.BOX_LOT: 30, field_const.EXPORTABLE: 'Y', field_const.EXPLICIT: 'N', field_const.EDITION: 'normal_edition' } return message @pytest.fixture def valid_get_album_for_digital_message(): """Valid message for get_album function.""" product_ss_date = datetime.datetime(day=12, month=12, year=12) message = { field_const.SALE_START_DATE: product_ss_date, field_const.PRODUCT_NAME: 'Test Name', field_const.PRODUCT_CODE: '123123', field_const.WHOLESALE_PRICE: 'price', field_const.DESCRIPTION: 'Test description', field_const.UPC: TEST_UPC, field_const.CONTEXT_TYPE: 'digital', field_const.LABEL: 'some label', field_const.VENDOR_NAME: 'some name', field_const.PROJECT_ID: '12345', field_const.MANUFACTURER_UPC: '42', field_const.VENDOR_OWNER: 'ORCHARD', field_const.PRICING: '15.99', field_const.BOX_LOT: 30, field_const.EXPORTABLE: 'Y', field_const.EXPLICIT: 'N', } return message @pytest.fixture def test_pdf_data(): """Test pdf data.""" return { field_const.RELEASE_ID: 1, field_const.PROJECT_ID: 1, field_const.TEMPLATE_DETAILS_DATA: { localized_template.STORE_ID: 1, field_const.NAME: localized_template.USA }, field_const.CONTEXT_TYPE: 'physical', field_const.REGIONS: [1] } @pytest.fixture def get_sales_goals_for_project_message(): """Test get_sales_goals_for_product message.""" return [ { 'marketing_highlight': 'This product is good.', 'country_id': 1, 'country_name': 'United States' } ] @pytest.fixture def mock_files(): """List of lists of presumably generated files.""" pdf_rendering_dir = os.path.abspath(config.PDF_RENDERING_DIR) files_list = [ '{dir}/{file}'.format(dir=pdf_rendering_dir, file='1.pdf'), '{dir}/{file}'.format(dir=pdf_rendering_dir, file='1.html'), '{dir}/{file}'.format(dir=pdf_rendering_dir, file='1_header.html'), '{dir}/{file}'.format(dir=pdf_rendering_dir, file='1.pdf'), '{dir}/{file}'.format(dir=pdf_rendering_dir, file='1.html'), '{dir}/{file}'.format(dir=pdf_rendering_dir, file='1_header.html'), '{dir}/{file}'.format(dir=pdf_rendering_dir, file='result.pdf')] return files_list @pytest.fixture(scope='function') def marketing_highlights_response_message(): """Emulate simplest marketing highlights response message.""" test_marketing_highlights = [{ 'scope': 'public', 'description': 'test_highlights', 'mkt_program_id': 20, 'highlight_id': 1234567 }] return test_marketing_highlights @pytest.fixture def tracks_with_side_none(): """Mock tracks response.""" return { 107166: { 'cd': 2, 'side': None, 'product_id': 1, 'track_id': 107166, 'track_number': 1}, 107165: { 'cd': 1, 'side': None, 'product_id': 1, 'track_id': 107165, 'track_number': 1}, } @pytest.fixture def expected_results_with_side_none(): """Mock format_tracks_by_cd response.""" return { 1: {'side': [ { 'product_id': 1, 'track_id': 107165, 'track_number': 1, 'cd': 1, 'side': None}]}, 2: {'side': [ { 'product_id': 1, 'track_id': 107166, 'track_number': 1, 'cd': 2, 'side': None}]} } @pytest.mark.parametrize(('tracks', 'expected_result'), [ # disc numbers and sides both ( { 107166: { 'cd': 2, 'side': 'A', 'product_id': 1, 'track_id': 107166, 'track_number': 1}, 107165: { 'cd': 1, 'side': 'A', 'product_id': 1, 'track_id': 107165, 'track_number': 1}, 107164: { 'cd': 1, 'side': 'B', 'product_id': 1, 'track_id': 107164, 'track_number': 1}, 107167: { 'cd': 1, 'side': 'B', 'product_id': 1, 'track_id': 107167, 'track_number': 2}}, { 1: { 'A': [ { 'product_id': 1, 'track_id': 107165, 'track_number': 1, 'cd': 1, 'side': 'A'}], 'B': [ { 'product_id': 1, 'track_id': 107164, 'track_number': 1, 'cd': 1, 'side': 'B'}, { 'product_id': 1, 'track_id': 107167, 'track_number': 2, 'cd': 1, 'side': 'B'}]}, 2: { 'A': [ { 'product_id': 1, 'track_id': 107166, 'track_number': 1, 'cd': 2, 'side': 'A'}]}} ), # disc numbers but no sides ( { 107166: { 'cd': 2, 'side': None, 'product_id': 1, 'track_id': 107166, 'track_number': 1}, 107165: { 'cd': 1, 'side': None, 'product_id': 1, 'track_id': 107165, 'track_number': 2}, 107164: { 'cd': 1, 'side': None, 'product_id': 1, 'track_id': 107164, 'track_number': 1}}, { 1: {'side': [ { 'product_id': 1, 'track_id': 107164, 'track_number': 1, 'cd': 1, 'side': None}, { 'product_id': 1, 'track_id': 107165, 'track_number': 2, 'cd': 1, 'side': None}]}, 2: {'side': [ { 'product_id': 1, 'track_id': 107166, 'track_number': 1, 'cd': 2, 'side': None}]}} ), # no disc numbers or sides ( { 107166: { 'cd': None, 'side': None, 'product_id': 1, 'track_id': 107166, 'track_number': 4}, 107165: { 'cd': None, 'side': None, 'product_id': 1, 'track_id': 107165, 'track_number': 1}, 107164: { 'cd': None, 'side': None, 'product_id': 1, 'track_id': 107164, 'track_number': 2}, 107167: { 'cd': None, 'side': None, 'product_id': 1, 'track_id': 107167, 'track_number': 3}}, {'cd': { 'side': [ { 'product_id': 1, 'track_id': 107165, 'track_number': 1, 'cd': None, 'side': None}, { 'product_id': 1, 'track_id': 107164, 'track_number': 2, 'cd': None, 'side': None}, { 'product_id': 1, 'track_id': 107167, 'track_number': 3, 'cd': None, 'side': None}, { 'product_id': 1, 'track_id': 107166, 'track_number': 4, 'cd': None, 'side': None}]}} ), # sides but no disc numbers ( { 107166: { 'cd': None, 'side': 'A', 'product_id': 1, 'track_id': 107166, 'track_number': 2}, 107165: { 'cd': None, 'side': 'B', 'product_id': 1, 'track_id': 107165, 'track_number': 2}, 107164: { 'cd': None, 'side': 'A', 'product_id': 1, 'track_id': 107164, 'track_number': 1}, 107167: { 'cd': None, 'side': 'B', 'product_id': 1, 'track_id': 107167, 'track_number': 1}}, {'cd': { 'A': [ { 'product_id': 1, 'track_id': 107164, 'track_number': 1, 'cd': None, 'side': 'A'}, { 'product_id': 1, 'track_id': 107166, 'track_number': 2, 'cd': None, 'side': 'A'}], 'B': [ { 'product_id': 1, 'track_id': 107167, 'track_number': 1, 'cd': None, 'side': 'B'}, { 'product_id': 1, 'track_id': 107165, 'track_number': 2, 'cd': None, 'side': 'B'}]}} ), # ordered dicts ( OrderedDict([(27702910, { 'track_number': 1, 'side': 'A', 'track_id': 27702910, 'artist_name': 'tracks artist 1', 'type': 'performer', 'product_id': 2322108, 'explicit': 'N', 'track_name': 'track 1', 'cd': 1 }), (27702911, { 'track_number': 2, 'side': 'A', 'track_id': 27702911, 'artist_name': 'tracks artist 2', 'type': 'performer', 'product_id': 2322108, 'explicit': 'N', 'track_name': 'track 2', 'cd': 1 }), (27702912, { 'track_number': 1, 'side': 'B', 'track_id': 27702912, 'artist_name': 'tracks artist 3', 'type': 'performer', 'product_id': 2322108, 'explicit': 'N', 'track_name': 'track 3', 'cd': 1 }), (27702913, { 'track_number': 2, 'side': 'B', 'track_id': 27702913, 'artist_name': 'tracks artist 4', 'type': 'performer', 'product_id': 2322108, 'explicit': 'N', 'track_name': 'track 4', 'cd': 1 }), (27702914, { 'track_number': 1, 'side': 'C', 'track_id': 27702914, 'artist_name': 'tracks artist 5', 'type': 'performer', 'product_id': 2322108, 'explicit': 'N', 'track_name': 'track 5', 'cd': 2 }), (27702915, { 'track_number': 2, 'side': 'C', 'track_id': 27702915, 'artist_name': 'tracks artist 6', 'type': 'performer', 'product_id': 2322108, 'explicit': 'N', 'track_name': 'track 6', 'cd': 2 }), (27702916, { 'track_number': 1, 'side': 'D', 'track_id': 27702916, 'artist_name': 'tracks artist 7', 'type': 'performer', 'product_id': 2322108, 'explicit': 'N', 'track_name': 'track 7', 'cd': 2 }), (27702917, { 'track_number': 2, 'side': 'D', 'track_id': 27702917, 'artist_name': 'tracks artist 8', 'type': 'performer', 'product_id': 2322108, 'explicit': 'N', 'track_name': 'track 8', 'cd': 2 })]), OrderedDict([(1, OrderedDict([('A', [{ 'track_id': 27702910, 'product_id': 2322108, 'artist_name': 'tracks artist 1', 'track_name': 'track 1', 'cd': 1, 'type': 'performer', 'side': 'A', 'track_number': 1, 'explicit': 'N' }, { 'track_id': 27702911, 'product_id': 2322108, 'artist_name': 'tracks artist 2', 'track_name': 'track 2', 'cd': 1, 'type': 'performer', 'side': 'A', 'track_number': 2, 'explicit': 'N' }]), ('B', [{ 'track_id': 27702912, 'product_id': 2322108, 'artist_name': 'tracks artist 3', 'track_name': 'track 3', 'cd': 1, 'type': 'performer', 'side': 'B', 'track_number': 1, 'explicit': 'N' }, { 'track_id': 27702913, 'product_id': 2322108, 'artist_name': 'tracks artist 4', 'track_name': 'track 4', 'cd': 1, 'type': 'performer', 'side': 'B', 'track_number': 2, 'explicit': 'N' }])])), (2, OrderedDict([('C', [{ 'track_id': 27702914, 'product_id': 2322108, 'artist_name': 'tracks artist 5', 'track_name': 'track 5', 'cd': 2, 'type': 'performer', 'side': 'C', 'track_number': 1, 'explicit': 'N' }, { 'track_id': 27702915, 'product_id': 2322108, 'artist_name': 'tracks artist 6', 'track_name': 'track 6', 'cd': 2, 'type': 'performer', 'side': 'C', 'track_number': 2, 'explicit': 'N' }]), ('D', [{ 'track_id': 27702916, 'product_id': 2322108, 'artist_name': 'tracks artist 7', 'track_name': 'track 7', 'cd': 2, 'type': 'performer', 'side': 'D', 'track_number': 1, 'explicit': 'N' }, { 'track_id': 27702917, 'product_id': 2322108, 'artist_name': 'tracks artist 8', 'track_name': 'track 8', 'cd': 2, 'type': 'performer', 'side': 'D', 'track_number': 2, 'explicit': 'N' }])]))]) ) ]) def test_format_tracks_by_cd(tracks, expected_result): """Test function that split tracks to dict keyed by cd.""" keys = list(tracks.keys()) random.shuffle(keys) shuffled_tracks = OrderedDict() for key in keys: shuffled_tracks[key] = tracks[key] result = pdf_generation.format_tracks_by_cd(shuffled_tracks) assert result.message == expected_result def test_format_tracks_by_cd_no_side( tracks_with_side_none, expected_results_with_side_none): """Test function that split tracks to dict keyed by cd.""" tracks = tracks_with_side_none keys = list(tracks.keys()) random.shuffle(keys) shuffled_tracks = OrderedDict() for key in keys: shuffled_tracks[key] = tracks[key] result = pdf_generation.format_tracks_by_cd(shuffled_tracks) assert result.message == expected_results_with_side_none def test_format_tracks_by_cd_no_tracks_provided(): """Test function that split tracks to dict if there are no tracks.""" test_track = {} expected_result = {} result = pdf_generation.format_tracks_by_cd(test_track) assert result.message == expected_result def test_get_data_from_db( valid_get_album_response_message, valid_template_details_response_message, valid_data_from_db): """Test of get_data_from_db with valid data.""" release_id = '1' product_type_resp = { 'product_type_id': 1} (flexmock(salessheets) .should_receive('get_product_type_id') .and_return(response.Response(product_type_resp))) (flexmock(pdf_generation) .should_receive('get_album_data_for_product') .with_args(release_id) .and_return(response.Response(message=valid_get_album_response_message)) .once()) tracks = response.Response(message={ '153': { 'track_id': '153', 'track_number': 1, 'side': None, 'cd': '1'}}) (flexmock(salessheets) .should_receive('get_tracks') .with_args(release_id) .and_return(tracks)) artists_list = response.Response(message=['Cool Artist']) (flexmock(salessheets) .should_receive('get_artists') .with_args(release_id) .and_return(artists_list)) (flexmock(barcode_generator) .should_receive('generate_barcode') .with_args(valid_data_from_db[field_const.UPC]) .and_return(response.Response(message={ 'barcode': 'test_barcode_file', 'barcode_type': 'test_type' }))).once() (flexmock(salessheets) .should_receive('get_distribution_format') .with_args(release_id) .and_return(response.Response(message={ 'display_distribution_format': 'CD' }))).once() (flexmock(template_details) .should_receive('get_by_template_type_id') .with_args(2) .and_return(response.Response(valid_template_details_response_message))) result = pdf_generation.get_pdf_data_from_db(release_id, 2) assert result.message == valid_data_from_db def test_get_pdf_data_from_db_if_genre_and_subgenre_are_absent( valid_get_album_response_message, valid_data_from_db, valid_template_details_response_message): """Test get pdf data returns correct response if no genre data provided.""" release_id = '1' product_type_resp = {'product_type_id': 1} genre_and_subgenre_keys = [field_const.GENRE, field_const.SUBGENRE] album_data_message = copy.deepcopy(valid_get_album_response_message) for key in genre_and_subgenre_keys: album_data_message.pop(key) for key in genre_and_subgenre_keys: valid_data_from_db[key] = None (flexmock(salessheets) .should_receive('get_product_type_id') .with_args(release_id) .and_return(response.Response(product_type_resp))) (flexmock(pdf_generation) .should_receive('get_album_data_for_product') .with_args(release_id) .and_return(response.Response( message=album_data_message)) .once()) tracks = response.Response(message={ '153': { 'track_id': '153', 'track_number': 1, 'side': None, 'cd': '1'}}) (flexmock(salessheets) .should_receive('get_tracks') .with_args(release_id) .and_return(tracks)) artists_list = response.Response(message=['Cool Artist']) (flexmock(salessheets) .should_receive('get_artists') .with_args(release_id) .and_return(artists_list)) (flexmock(barcode_generator) .should_receive('generate_barcode') .with_args(valid_data_from_db[field_const.UPC]) .and_return(response.Response(message={ 'barcode': 'test_barcode_file', 'barcode_type': 'test_type' }))).once() (flexmock(salessheets) .should_receive('get_distribution_format') .with_args(release_id) .and_return(response.Response(message={ 'display_distribution_format': 'CD' }))).once() (flexmock(template_details) .should_receive('get_by_template_type_id') .with_args(2) .and_return(response.Response(valid_template_details_response_message))) result = pdf_generation.get_pdf_data_from_db(release_id, 2) assert result.message == valid_data_from_db def test_get_data_from_db_album_exception(): """Test for get_data if get_album had returned error.""" product_id = '1' product_type_resp = { 'product_type_id': 1} (flexmock(salessheets) .should_receive('get_product_type_id') .and_return(response.Response(product_type_resp))) (flexmock(pdf_generation) .should_receive('get_album_data_for_product') .with_args(product_id) .and_return(response.create_fatal_response())).once() result = pdf_generation.get_pdf_data_from_db(product_id, 2) assert result.status == 500 def test_get_data_from_db_no_album_data(): """Test for get_data if get_album had returned empty response.""" product_id = '1' product_type_resp = { 'product_type_id': 1} (flexmock(salessheets) .should_receive('get_product_type_id') .and_return(response.Response(product_type_resp))) (flexmock(pdf_generation) .should_receive('get_album_data_for_product') .with_args(product_id) .and_return(response.create_not_found_response( message=errors.ERROR_MESSAGE_INVALID_PRODUCT_ID))).once() (flexmock(salessheets) .should_receive('get_artists') .and_return(response.Response())).never() result = pdf_generation.get_pdf_data_from_db(product_id, 2) assert result.status == 404 assert result.errors['message'] == errors.ERROR_MESSAGE_INVALID_PRODUCT_ID def test_get_data_from_db_artists_exception(): """Test for get_data if get_artists had returned error.""" product_id = '1' product_type_resp = { 'product_type_id': 1} (flexmock(salessheets) .should_receive('get_product_type_id') .and_return(response.Response(product_type_resp))) (flexmock(pdf_generation) .should_receive('get_album_data_for_product') .with_args(product_id) .and_return(response.Response())).once() (flexmock(salessheets) .should_receive('get_artists') .and_return(response.create_fatal_response())).once() result = pdf_generation.get_pdf_data_from_db(product_id, 2) assert result.status == 500 def test_get_data_from_db_no_artists_data( valid_get_album_response_message, valid_data_from_db, valid_template_details_response_message): """Test for get_data if get_artists had returned not found response.""" product_id = '1' product_type_resp = { 'product_type_id': 1} tracks = response.Response(message={ '153': { 'track_id': '153', 'track_number': 1, 'side': None, 'cd': '1'}}) (flexmock(salessheets) .should_receive('get_product_type_id') .with_args(product_id) .and_return(response.Response(product_type_resp))) (flexmock(pdf_generation) .should_receive('get_album_data_for_product') .with_args(product_id) .and_return(response.Response(valid_get_album_response_message))).once() (flexmock(salessheets) .should_receive('get_artists') .and_return(response.create_not_found_response())).once() valid_data_from_db[field_const.ARTIST_LIST] = [] (flexmock(salessheets) .should_receive('get_tracks') .with_args(product_id) .and_return(tracks)).once() (flexmock(barcode_generator) .should_receive('generate_barcode') .with_args(valid_data_from_db[field_const.UPC]) .and_return(response.Response(message={ 'barcode': 'test_barcode_file', 'barcode_type': 'test_type' }))).once() (flexmock(salessheets) .should_receive('get_distribution_format') .with_args(product_id) .and_return(response.Response(message={ 'display_distribution_format': 'CD' }))).once() (flexmock(template_details) .should_receive('get_by_template_type_id') .with_args(2) .and_return(response.Response(valid_template_details_response_message))) result = pdf_generation.get_pdf_data_from_db(product_id, 2) assert result.status == 200 assert result.message == valid_data_from_db def test_get_data_from_db_tracks_exception(): """Test for get_data if get_tracks had returned error.""" product_id = '1' product_type_resp = { 'product_type_id': 1} (flexmock(salessheets) .should_receive('get_product_type_id') .and_return(response.Response(product_type_resp))) (flexmock(pdf_generation) .should_receive('get_album_data_for_product') .with_args(product_id) .and_return(response.Response())).once() (flexmock(salessheets) .should_receive('get_artists') .and_return(response.Response())).once() (flexmock(salessheets) .should_receive('get_tracks').with_args(product_id) .and_return(response.create_fatal_response())).once() result = pdf_generation.get_pdf_data_from_db(product_id, 2) assert result.status == 500 def test_get_data_from_db_no_tracks_data(valid_get_album_response_message): """Test for get_data if query have got empty tracks_data from db.""" product_id = '1' product_type_resp = { 'product_type_id': 1} (flexmock(salessheets) .should_receive('get_product_type_id') .and_return(response.Response(product_type_resp))) (flexmock(pdf_generation) .should_receive('get_album_data_for_product') .with_args(product_id) .and_return(response.Response( message=valid_get_album_response_message))).once() artists_list = response.Response(message=['Cool Artist']) (flexmock(salessheets) .should_receive('get_artists') .and_return(artists_list)).once() (flexmock(salessheets) .should_receive('get_tracks').with_args(product_id) .and_return(response.create_not_found_response())).once() result = pdf_generation.get_pdf_data_from_db(product_id, 2) assert result.status def test_get_data_didnt_receive_required_params(): """Test for KeyError exception if not enough data took from db.""" product_id = None result = pdf_generation.get_pdf_data_from_db(product_id, 2) assert not result def test_pdf_generation_no_product_id(): """Test of generate_pdf without product_id.""" mock_message = { 'Body': '{"product_id": [], "output_type": "single",\ "job_id": "1234214", "localized_template_type_id": 2}' } mock_message['MessageAttributes'] = { 'Correlation-Id': { 'DataType': 'String', 'StringValue': '1234' }, 'feature_flag_user_context': { 'DataType': 'String', 'StringValue': 'oa:1234' } } result = pdf_generation.generate_pdf(mock_message) assert not result def test_pdf_generation_success_remove_barcode( valid_sqs_message, valid_data_from_db): """Test remove files inside the generate_pdf.""" product_id = '1' job_id = '1234214' mock_message = valid_sqs_message output_file = pdf_generation.generate_bulk_sales_sheets_filename( extension=pdf.PDF_EXTENSION, job_id=job_id) (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.IN_PROGRESS) .and_return(response.Response()) .ordered()) (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.COMPLETED) .and_return(response.Response()) .ordered()) (flexmock(features) .should_receive('check_user_features') .and_return(response.Response({})) .once()) (flexmock(pdf_generation) .should_receive('get_pdf_data_from_db') .with_args('1', 2) .and_return(response.Response(message=valid_data_from_db)) .once()) (flexmock(pdf_generation) .should_receive('get_data_from_microservices') .with_args(valid_data_from_db, False) .and_return(response.Response(message=valid_data_from_db)) .once()) (flexmock(template_rendering) .should_receive('prepare_template_data') .and_return(response.Response(message=valid_data_from_db)) .once()) expected_template_data = {**valid_data_from_db, 'user_features': {}} (flexmock(template_rendering) .should_receive('render_template') .with_args('orch_template.html', expected_template_data) .and_return(response.Response(message='Cool html!')) .once()) (flexmock(template_rendering) .should_receive('render_template') .with_args('orch_header.html', expected_template_data) .and_return(response.Response(message='Cool html!')) .once()) body_template_file = '{dir}/{id}.html'.format( dir=config.PDF_RENDERING_DIR, id=product_id) header_template_file = '{dir}/{id}_header.html'.format( dir=config.PDF_RENDERING_DIR, id=product_id) rendered_pdf_file = '{dir}/{id}.pdf'.format( dir=config.PDF_RENDERING_DIR, id=product_id) (flexmock(htmltopdf) .should_receive('convert_html_to_pdf') .with_args( body_template_file, rendered_pdf_file, { 'header-spacing': pdf.HEADER_SPACING, 'header-html': header_template_file }) .and_return(response.Response()) .once()) (flexmock(pdf_generation) .should_receive('upload_file_to_s3') .with_args(output_file) .and_return(response.Response())) (flexmock(misc) .should_call('remove_file') .times(5)) result = pdf_generation.generate_pdf(mock_message) assert result assert not os.path.isfile(body_template_file) assert not os.path.isfile(header_template_file) assert not os.path.isfile(rendered_pdf_file) def test_pdf_generation_marketing_data( mocker, valid_sqs_message_with_user_context, valid_data_from_db): """Test generate_pdf project marketing moved.""" product_id = '1' job_id = '1234214' output_file = pdf_generation.generate_bulk_sales_sheets_filename( extension=pdf.PDF_EXTENSION, job_id=job_id) key_phrase = 'That should be added anyway' highlights_title = 'Promo and Marketing Highlights' highlights_data = {'marketing_highlights': [ {'description': key_phrase, 'entity': 'project', 'attachment': 'N', 'entity_id': 3531036, 'scope': 'public', 'highlight_id': 46650, 'client': 'alw', 'subject': 'Project Highlights', 'mkt_program_id': 20}]} (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.IN_PROGRESS) .and_return(response.Response()) .ordered()) (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.COMPLETED) .and_return(response.Response()) .ordered()) (flexmock(features) .should_receive('check_user_features') .and_return(response.Response({})) .once()) (flexmock(pdf_generation) .should_receive('get_pdf_data_from_db') .with_args('1', 2) .and_return(response.Response(message=valid_data_from_db)) .once()) (flexmock(pdf_generation) .should_receive('get_data_from_microservices') .with_args(valid_data_from_db, False) .and_return(response.Response(message=highlights_data)) .once()) body_template_file = '{dir}/{id}.html'.format( dir=config.PDF_RENDERING_DIR, id=product_id) header_template_file = '{dir}/{id}_header.html'.format( dir=config.PDF_RENDERING_DIR, id=product_id) rendered_pdf_file = '{dir}/{id}.pdf'.format( dir=config.PDF_RENDERING_DIR, id=product_id) (flexmock(htmltopdf) .should_receive('convert_html_to_pdf') .with_args( body_template_file, rendered_pdf_file, { 'header-spacing': pdf.HEADER_SPACING, 'header-html': header_template_file }) .and_return(response.Response()) .once()) (flexmock(pdf_generation) .should_receive('upload_file_to_s3') .with_args(output_file) .and_return(response.Response())) (flexmock(misc) .should_receive('remove_file') .and_return(None)) result = pdf_generation.generate_pdf(valid_sqs_message_with_user_context) assert result with open(body_template_file) as sheet_body: sheet_text = sheet_body.read() assert key_phrase in sheet_text assert highlights_title not in sheet_text def test_pdf_generation_error_while_get_pdf_data_from_db(valid_sqs_message): """Test of failed create_pdf on getting data step.""" mock_message = valid_sqs_message job_id = '1234214' (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.IN_PROGRESS) .and_return(response.Response()) .ordered() .once()) (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.ERROR) .and_return(response.Response()) .ordered() .once()) (flexmock(features) .should_receive('check_user_features') .and_return(response.Response({})) .once()) (flexmock(pdf_generation) .should_receive('get_pdf_data_from_db') .with_args('1', 2) .and_return(response.create_error_response( code=errors.ERROR_PDF_GENERATION_CODE, message='message')) .once()) result = pdf_generation.generate_pdf(mock_message) assert not result assert result.errors['code'] == errors.ERROR_PDF_GENERATION_CODE def test_pdf_generation_error_while_preparing_template( valid_sqs_message, valid_data_from_db): """Test of failed create_pdf on template preparing step.""" mock_message = valid_sqs_message job_id = '1234214' (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.IN_PROGRESS) .and_return(response.Response()) .ordered() .once()) (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.ERROR) .and_return(response.Response()) .ordered() .once()) (flexmock(features) .should_receive('check_user_features') .and_return(response.Response({})) .once()) (flexmock(pdf_generation) .should_receive('get_pdf_data_from_db') .with_args('1', 2) .and_return(response.Response(message=valid_data_from_db)) .once()) (flexmock(pdf_generation) .should_receive('get_data_from_microservices') .with_args(valid_data_from_db, False) .and_return(response.Response(message=valid_data_from_db)) .once()) (flexmock(template_rendering) .should_receive('prepare_template_data') .and_return(response.create_error_response( code=errors.ERROR_CONVERTING_DATE, message='message'))) result = pdf_generation.generate_pdf(mock_message) assert not result assert result.errors['code'] == errors.ERROR_CONVERTING_DATE def test_pdf_generation_error_while_rendering_template( valid_sqs_message, valid_data_from_db): """Test of failed create_pdf on template rendering step.""" mock_message = valid_sqs_message job_id = '1234214' (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.IN_PROGRESS) .and_return(response.Response()) .ordered() .once()) (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.ERROR) .and_return(response.Response()) .ordered() .once()) (flexmock(features) .should_receive('check_user_features') .and_return(response.Response({})) .once()) (flexmock(pdf_generation) .should_receive('get_pdf_data_from_db') .with_args('1', 2) .and_return(response.Response(message=valid_data_from_db)) .once()) (flexmock(pdf_generation) .should_receive('get_data_from_microservices') .with_args(valid_data_from_db, False) .and_return(response.Response({})) .once()) (flexmock(template_rendering) .should_receive('prepare_template_data') .and_return(response.Response(message=valid_data_from_db))) (flexmock(template_rendering) .should_receive('render_template') .and_return(response.create_error_response( code=errors.ERROR_FINDING_TEMPLATE, message='message'))) result = pdf_generation.generate_pdf(mock_message) assert not result assert result.errors['code'] == errors.ERROR_FINDING_TEMPLATE def test_pdf_creation_error_while_check_user_features(): """Test of failed create_pdf on checking features step.""" product_id = '1' (flexmock(features) .should_receive('check_user_features') .with_args(None) .and_return(response.create_error_response( code='features_error', message='message')) .once()) result = pdf_generation.create_pdf(product_id, 2) assert not result assert result.errors['code'] == 'features_error' def test_pdf_creation_error_while_get_pdf_data_from_db(): """Test of failed create_pdf on getting data step.""" product_id = '1' (flexmock(features) .should_receive('check_user_features') .with_args(None) .and_return(response.Response({})) .once()) (flexmock(pdf_generation) .should_receive('get_pdf_data_from_db') .with_args('1', 2) .and_return(response.create_error_response( code=errors.ERROR_PDF_GENERATION_CODE, message='message')) .once()) result = pdf_generation.create_pdf(product_id, 2) assert not result assert result.errors['code'] == errors.ERROR_PDF_GENERATION_CODE def test_pdf_creation_error_while_preparing_template(valid_data_from_db): """Test of failed create_pdf on template preparing step.""" product_id = '1' (flexmock(features) .should_receive('check_user_features') .with_args(None) .and_return(response.Response({})) .once()) (flexmock(pdf_generation) .should_receive('get_pdf_data_from_db') .with_args('1', 2) .and_return(response.Response(message=valid_data_from_db)) .once()) (flexmock(pdf_generation) .should_receive('get_data_from_microservices') .with_args(valid_data_from_db, False) .and_return(response.Response(message=valid_data_from_db)) .once()) (flexmock(template_rendering) .should_receive('prepare_template_data') .and_return(response.create_error_response( code=errors.ERROR_CONVERTING_DATE, message='message'))) result = pdf_generation.create_pdf(product_id, 2) assert not result assert result.errors['code'] == errors.ERROR_CONVERTING_DATE def test_pdf_creation_error_while_rendering_template(valid_data_from_db): """Test of failed create_pdf on template rendering step.""" product_id = '1' (flexmock(features) .should_receive('check_user_features') .with_args(None) .and_return(response.Response({})) .once()) (flexmock(pdf_generation) .should_receive('get_pdf_data_from_db') .with_args('1', 2) .and_return(response.Response(message=valid_data_from_db)) .once()) (flexmock(pdf_generation) .should_receive('get_data_from_microservices') .with_args(valid_data_from_db, False) .and_return(response.Response({})) .once()) (flexmock(template_rendering) .should_receive('prepare_template_data') .and_return(response.Response(message=valid_data_from_db))) (flexmock(template_rendering) .should_receive('render_template') .and_return(response.create_error_response( code=errors.ERROR_FINDING_TEMPLATE, message='message'))) result = pdf_generation.create_pdf(product_id, 2) assert not result assert result.errors['code'] == errors.ERROR_FINDING_TEMPLATE def test_pdf_generation_error_while_uploading_s3( valid_sqs_message, valid_data_from_db): """Test of failed generate_pdf on s3 uploading step.""" product_id = '1' mock_message = valid_sqs_message job_id = '1234214' output_file = pdf_generation.generate_bulk_sales_sheets_filename( extension=pdf.PDF_EXTENSION, job_id=job_id) (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.IN_PROGRESS) .and_return(response.Response()) .ordered() .once()) (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.ERROR) .and_return(response.Response()) .ordered() .once()) (flexmock(features) .should_receive('check_user_features') .and_return(response.Response({})) .once()) s3_error_msg = 'Some s3 internal error!' (flexmock(pdf_generation) .should_receive('get_pdf_data_from_db') .with_args('1', 2) .and_return(response.Response(message=valid_data_from_db)) .once()) (flexmock(pdf_generation) .should_receive('get_data_from_microservices') .with_args(valid_data_from_db, False) .and_return( response.Response(message=valid_data_from_db)) .once()) (flexmock(template_rendering) .should_receive('prepare_template_data') .and_return(response.Response(message=valid_data_from_db))) (flexmock(template_rendering) .should_receive('render_template') .and_return(response.Response('Cool html!'))) body_template_file = '{dir}/{id}.html'.format( dir=config.PDF_RENDERING_DIR, id=product_id) header_template_file = '{dir}/{id}_header.html'.format( dir=config.PDF_RENDERING_DIR, id=product_id) rendered_pdf_file = '{dir}/{id}.pdf'.format( dir=config.PDF_RENDERING_DIR, id=product_id) (flexmock(htmltopdf) .should_receive('convert_html_to_pdf') .with_args( body_template_file, rendered_pdf_file, { 'header-spacing': pdf.HEADER_SPACING, 'header-html': header_template_file }) .and_return(response.Response()) .once()) (flexmock(pdf_generation) .should_receive('upload_file_to_s3') .with_args(output_file) .and_return(response.create_fatal_response(s3_error_msg))) result = pdf_generation.generate_pdf(mock_message) assert not result assert result.errors['message'] == s3_error_msg @pytest.mark.parametrize( ('gsgfp_status', 'gsgfp_message', 'expected_status', 'expected_message'), [ ( 200, [ { 'marketing_highlight': 'a', 'country_id': None, 'country_name': None }, { 'marketing_highlight': 'b', 'country_id': 1, 'country_name': 'US' }, { 'marketing_highlight': 'c', 'country_id': 2, 'country_name': 'CA' }, { 'marketing_highlight': 'd', 'country_id': 3, 'country_name': 'FR' }, { 'marketing_highlight': 'd', 'store_id': 1, 'store_name': 'spotify' }, ], 200, [{ 'marketing_drivers': { 'value': 'b', 'slogan': None }, 'country': { 'name': 'US', 'country_id': 1 } }, { 'marketing_drivers': { 'value': 'c', 'slogan': None }, 'country': { 'name': 'CA', 'country_id': 2 } }] ), ( 404, None, 200, None ), ( 403, 'OH NOES!', 403, 'OH NOES!' ) ]) def test_get_sales_goals( mocker, gsgfp_status, gsgfp_message, expected_status, expected_message): """Test get_sales_goals.""" mocker.patch.object( ows_sales_goals, 'get_marketing_highlights_for_project', return_value=response.Response( status=gsgfp_status, message=gsgfp_message)) result = pdf_generation.get_sales_goals( 1, 2, [1, 2]) assert result.status == expected_status assert result.message == expected_message def test_get_sales_goals_japan(mocker): """Test get_sales_goals for Japan.""" mock_response = response.Response([{ 'country_id': field_const.JAPAN_COUNTRY_ID, 'country_name': 'JP', 'marketing_highlight': 'a', 'slogan': 'b' }]) mocker.patch.object( ows_sales_goals, 'get_marketing_highlights_for_project', return_value=mock_response) result = pdf_generation.get_sales_goals( 1, 2, [field_const.JAPAN_REGION_ID]) assert result.message == [{ 'country': { 'country_id': field_const.JAPAN_COUNTRY_ID, 'name': 'JP' }, 'marketing_drivers': { 'value': 'a', 'slogan': 'b' } }] @pytest.mark.parametrize( ( 'expected_get_data_from_microservices_status', 'expected_marketing_highlights_status', 'expected_marketing_highlights_message', 'expected_get_large_cover_image_url_status', 'expected_get_large_cover_image_url_message' ), [ ( 200, 200, [{ 'scope': 'public', 'description': 'test_highlights', 'mkt_program_id': 20, 'highlight_id': 1234567 }], 200, 'https://image.url' ), ( 200, 404, None, 200, 'https://image.url' ), ( 200, 200, [{ 'scope': 'public', 'description': 'test_highlights', 'mkt_program_id': 20, 'highlight_id': 1234567 }], 404, None ), ]) def test_get_data_from_microservices( test_pdf_data, expected_get_data_from_microservices_status, expected_marketing_highlights_status, expected_marketing_highlights_message, expected_get_large_cover_image_url_status, expected_get_large_cover_image_url_message, get_sales_goals_for_project_message, mocker): """Test get_data_from_microservices success.""" template_details_data = test_pdf_data[field_const.TEMPLATE_DETAILS_DATA] country_name = template_details_data[field_const.NAME] store_id = template_details_data[localized_template.STORE_ID] (flexmock(ows_assets) .should_receive('get_large_cover_image_url') .with_args(test_pdf_data[field_const.RELEASE_ID]) .and_return(response.Response( message=expected_get_large_cover_image_url_message, status=expected_get_large_cover_image_url_status))).once() (flexmock(ows_marketing) .should_receive('get_marketing_highlights') .and_return(response.Response( message=expected_marketing_highlights_message, status=expected_marketing_highlights_status))) price_code_name = 'store_pricing_tier_name' price_code_value = 'Price US' price_sheet_field = localized_template.REGION_PRICING_MAPPING.get( country_name).sheet_label region_price_sheet_field = '{price_field_name} {price_value}'.format( price_field_name=localized_template.NAME_PRICING_MAPPING.get( country_name), price_value=price_sheet_field ) (flexmock(ows_pricing) .should_receive( 'get_pricing_for_pricing_family_by_product_and_store_ids') .with_args( test_pdf_data[field_const.RELEASE_ID], localized_template.PHYS_PRICING_FAMILY_ID, store_id) .and_return(response.Response( message={price_code_name: price_code_value}, status=200))).once() (flexmock(ows_sales_goals) .should_receive('get_marketing_highlights_for_project') .with_args(test_pdf_data[field_const.PROJECT_ID]) .and_return( response.Response(message=get_sales_goals_for_project_message))) result = pdf_generation.get_data_from_microservices(test_pdf_data) assert result.status == ( expected_get_data_from_microservices_status) assert result.message.get( 'marketing_highlights') == expected_marketing_highlights_message assert result.message.get( 'product_highlights') == expected_marketing_highlights_message assert result.message.get( 'large_cover_image') == expected_get_large_cover_image_url_message result_label = result.message.get( field_const.PRICE).get(localized_template.SHEET_LABEL) result_value = result.message.get( field_const.PRICE).get(localized_template.SHEET_PRICE_VALUE) assert result_label == region_price_sheet_field assert result_value == price_code_value def test_get_data_from_microservices_translations( test_pdf_data, get_sales_goals_for_project_message): """Test get_data_from_microservices when fetching translations.""" template_details_data = test_pdf_data[field_const.TEMPLATE_DETAILS_DATA] country_name = template_details_data[field_const.NAME] store_id = template_details_data[localized_template.STORE_ID] (flexmock(ows_assets) .should_receive('get_large_cover_image_url') .with_args(test_pdf_data[field_const.RELEASE_ID]) .and_return(response.Response( message='https://image.url', status=200))) (flexmock(ows_marketing) .should_receive('get_marketing_highlights') .and_return(response.Response( message=[{ 'scope': 'public', 'description': 'test_highlights', 'mkt_program_id': 20, 'highlight_id': 1234567 }], status=200))) price_code_name = 'store_pricing_tier_name' price_code_value = 'Price US' price_sheet_field = localized_template.REGION_PRICING_MAPPING.get( country_name).sheet_label region_price_sheet_field = '{price_field_name} {price_value}'.format( price_field_name=localized_template.NAME_PRICING_MAPPING.get( country_name), price_value=price_sheet_field ) (flexmock(ows_pricing) .should_receive( 'get_pricing_for_pricing_family_by_product_and_store_ids') .with_args( test_pdf_data[field_const.RELEASE_ID], localized_template.PHYS_PRICING_FAMILY_ID, store_id) .and_return(response.Response( message={price_code_name: price_code_value}, status=200))) (flexmock(ows_sales_goals) .should_receive('get_marketing_highlights_for_project') .with_args(test_pdf_data[field_const.PROJECT_ID]) .and_return( response.Response(message=get_sales_goals_for_project_message))) (flexmock(ows_product) .should_receive('get_japan_phonetic_translations') .with_args(test_pdf_data[field_const.RELEASE_ID]) .and_return( response.Response(message={ 'items': [ { 'field_name': 'release_name', 'phonetic_translation': 'translatedrelease' }, { 'field_name': 'artist_name', 'phonetic_translation': 'translatedartist' } ] }))) result = pdf_generation.get_data_from_microservices(test_pdf_data, True) assert result.message.get('japan_release_name') == 'translatedrelease' assert result.message.get('japan_artist_name') == 'translatedartist' def test_get_data_from_microservices_translations_error( test_pdf_data, get_sales_goals_for_project_message): """Test get_data_from_microservices error fetching translations.""" template_details_data = test_pdf_data[field_const.TEMPLATE_DETAILS_DATA] country_name = template_details_data[field_const.NAME] store_id = template_details_data[localized_template.STORE_ID] (flexmock(ows_assets) .should_receive('get_large_cover_image_url') .with_args(test_pdf_data[field_const.RELEASE_ID]) .and_return(response.Response( message='https://image.url', status=200))) (flexmock(ows_marketing) .should_receive('get_marketing_highlights') .and_return(response.Response( message=[{ 'scope': 'public', 'description': 'test_highlights', 'mkt_program_id': 20, 'highlight_id': 1234567 }], status=200))) price_code_name = 'store_pricing_tier_name' price_code_value = 'Price US' price_sheet_field = localized_template.REGION_PRICING_MAPPING.get( country_name).sheet_label region_price_sheet_field = '{price_field_name} {price_value}'.format( price_field_name=localized_template.NAME_PRICING_MAPPING.get( country_name), price_value=price_sheet_field ) (flexmock(ows_pricing) .should_receive( 'get_pricing_for_pricing_family_by_product_and_store_ids') .with_args( test_pdf_data[field_const.RELEASE_ID], localized_template.PHYS_PRICING_FAMILY_ID, store_id) .and_return(response.Response( message={price_code_name: price_code_value}, status=200))) (flexmock(ows_sales_goals) .should_receive('get_marketing_highlights_for_project') .with_args(test_pdf_data[field_const.PROJECT_ID]) .and_return( response.Response(message=get_sales_goals_for_project_message))) (flexmock(ows_product) .should_receive('get_japan_phonetic_translations') .with_args(test_pdf_data[field_const.RELEASE_ID]) .and_return( response.create_fatal_response())) result = pdf_generation.get_data_from_microservices(test_pdf_data, True) assert not result def test_get_data_from_db_invalid_product_type(): """Test for get_data if validate product data had returned error.""" product_id = '1' product_type_resp = { 'product_type_id': 2} (flexmock(salessheets) .should_receive('get_product_type_id') .and_return(response.Response(product_type_resp))) result = pdf_generation.get_pdf_data_from_db(product_id, 2) assert result.status == 400 assert result.errors['code'] == errors.ERROR_INVALID_PRODUCT_TYPE_CODE def test_validate_product_data_valid(): """Test for validate_product_data with valid result.""" product_data = {'product_type_id': 1} result = pdf_generation.validate_product_data(product_data) assert result def test_validate_product_data_empty(): """Test for validate_product_data with error result due to empty data.""" product_data = {} result = pdf_generation.validate_product_data(product_data) assert not result assert result.status == 404 def test_validate_product_data_invalid_product_type(): """Test for validate_product_data error due to invalid product type.""" product_data = {'product_type_id': 2} result = pdf_generation.validate_product_data(product_data) assert not result assert result.status == 400 def test_generate_pdf_use_orchard_template( valid_sqs_message, valid_data_from_db): """Test generate_pdf use orchard template.""" mock_message = valid_sqs_message job_id = '1234214' output_file = pdf_generation.generate_bulk_sales_sheets_filename( extension=pdf.PDF_EXTENSION, job_id=job_id) (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.IN_PROGRESS) .and_return(response.Response()) .ordered() .once()) (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.COMPLETED) .and_return(response.Response()) .ordered() .once()) (flexmock(features) .should_receive('check_user_features') .and_return(response.Response({})) .once()) (flexmock(pdf_generation) .should_receive('get_pdf_data_from_db') .with_args('1', 2) .and_return( response.Response(message=valid_data_from_db)) .once()) (flexmock(pdf_generation) .should_receive('get_data_from_microservices') .with_args(valid_data_from_db, False) .and_return( response.Response(message=valid_data_from_db)) .once()) (flexmock(template_rendering) .should_receive('prepare_template_data') .and_return(response.Response(message=valid_data_from_db))) (flexmock(template_rendering) .should_receive('render_template') .with_args('orch_template.html', valid_data_from_db) .and_return(response.Response(message='Cool html!')) .once()) (flexmock(template_rendering) .should_receive('render_template') .with_args('orch_header.html', valid_data_from_db) .and_return(response.Response(message='Cool html!')) .once()) (flexmock(htmltopdf) .should_receive('convert_html_to_pdf') .and_return(response.Response()) .once()) (flexmock(pdf_generation) .should_receive('upload_file_to_s3') .with_args(output_file) .and_return(response.Response())) result = pdf_generation.generate_pdf(mock_message) assert result def test_create_pdf_use_orchard_template(valid_data_from_db): """Test create_pdf use orchard template.""" product_id = '1' (flexmock(features) .should_receive('check_user_features') .with_args(None) .and_return(response.Response({})) .once()) (flexmock(pdf_generation) .should_receive('get_pdf_data_from_db') .with_args('1', 2) .and_return( response.Response(message=valid_data_from_db)) .once()) (flexmock(pdf_generation) .should_receive('get_data_from_microservices') .with_args(valid_data_from_db, False) .and_return( response.Response(message=valid_data_from_db)) .once()) (flexmock(template_rendering) .should_receive('prepare_template_data') .and_return(response.Response(valid_data_from_db))) (flexmock(template_rendering) .should_receive('render_template') .with_args('orch_template.html', valid_data_from_db) .and_return(response.Response(message='Cool html!')) .once()) (flexmock(template_rendering) .should_receive('render_template') .with_args('orch_header.html', valid_data_from_db) .and_return(response.Response(message='Cool html!')) .once()) (flexmock(htmltopdf) .should_receive('convert_html_to_pdf') .and_return(response.Response()) .once()) result = pdf_generation.create_pdf(product_id, 2) assert result # The only condition when we expect manufacturer UPC. PHYS_MANUF_UPC_CONDITIONS_OK = ( (field_const.PHYSICAL_CONTEXT_TYPE, TEST_MANUFACTURER_UPC, TEST_MANUFACTURER_UPC),) # Expect product UPC if manufacturer UPC is empty. PHYS_MANUF_UPC_CONDITIONS_EMPTY_UPC = itertools.product( (field_const.DIGITAL_CONTEXT_TYPE, field_const.PHYSICAL_CONTEXT_TYPE), (0, False, ''), (TEST_UPC,), ) # Expect product UPC for digital products. PHYS_MANUF_UPC_CONDITIONS_DIGITAL = itertools.product( field_const.DIGITAL_CONTEXT_TYPE, (0, TEST_MANUFACTURER_UPC,), (TEST_UPC,), ) def test_pdf_generation_success_cleanup_files(mock_files): """Test remove files inside the cleanup files.""" absolute_root_folder = ( os.path.abspath( os.path.join(config.PDF_RENDERING_DIR, os.pardir))) for file in mock_files: with open( os.path.join( absolute_root_folder, file), 'w') as tmp_file: tmp_file.write('alsd;fj;aksdhfja') pdf_generation.cleanup_files(mock_files) for filename in mock_files: assert not os.path.isfile(filename) def test_generate_bulk_sales_sheets_filename(): """Test how bulk file name is being generated.""" job_id = '1' extension = 'pdf' files_timestamp = misc.datetime_to_str( datetime.datetime.today(), pdf.BULK_SALES_SHEETS_TIMESTAMP_FORMAT) output_file_name = ( '{timestamp}_{job_id}{suffix}.{extension}' .format( timestamp=files_timestamp, job_id=job_id, suffix=pdf.BULK_SALES_SHEETS_FILENAME_SUFFIX, extension=extension)) assert ( output_file_name == pdf_generation .generate_bulk_sales_sheets_filename(extension, job_id)) def test_create_bulk_pdf_file(): """Test bulk pdf generation.""" pdf_render_dir = os.path.abspath(config.PDF_RENDERING_DIR) extension = pdf.PDF_EXTENSION output_type = pdf.SINGLE_PDF corresponding_ids = ['1', '2'] job_id = '7654' (pdf_generation .create_output_file(output_type, corresponding_ids, job_id)) timestamp = ( misc.datetime_to_str( datetime.datetime.now(), pdf.BULK_SALES_SHEETS_TIMESTAMP_FORMAT)) filename = ( '{timestamp}_{job_id}{sfx}.{ext}' .format( timestamp=timestamp, job_id=job_id, sfx=pdf.BULK_SALES_SHEETS_FILENAME_SUFFIX, ext=extension)) assert os.path.isfile(os.path.join(pdf_render_dir, filename)) misc.remove_file(os.path.join(pdf_render_dir, filename)) def test_not_create_bulk_pdf_file(): """Test bulk pdf generation interrupt.""" pdf_render_dir = os.path.abspath(config.PDF_RENDERING_DIR) extension = pdf.PDF_EXTENSION output_type = pdf.SINGLE_PDF corresponding_ids = ['1', '2'] with open(os.path.join(pdf_render_dir, '1.pdf'), 'w') as f: f.write('invalid data') job_id = '7654' result = (pdf_generation .create_output_file(output_type, corresponding_ids, job_id)) timestamp = ( misc.datetime_to_str( datetime.datetime.now(), pdf.BULK_SALES_SHEETS_TIMESTAMP_FORMAT)) filename = ( '{timestamp}_{job_id}{sfx}.{ext}' .format( timestamp=timestamp, job_id=job_id, sfx=pdf.BULK_SALES_SHEETS_FILENAME_SUFFIX, ext=extension)) assert not os.path.isfile(os.path.join(pdf_render_dir, filename)) assert not result misc.remove_file(os.path.join(pdf_render_dir, '1.pdf')) def test_create_bulk_zip_file(): """Test bulk zip generation. The zip file is being generated always. There are few cases where zip file is not generated like system crash. """ pdf_render_dir = os.path.abspath(config.PDF_RENDERING_DIR) extension = pdf.ZIP_EXTENSION output_type = pdf.ZIPPED_PDF corresponding_ids = ['1', '2'] job_id = '7654' (pdf_generation .create_output_file(output_type, corresponding_ids, job_id)) timestamp = ( misc.datetime_to_str( datetime.datetime.now(), pdf.BULK_SALES_SHEETS_TIMESTAMP_FORMAT)) filename = ( '{timestamp}_{job_id}{sfx}.{ext}' .format( timestamp=timestamp, job_id=job_id, sfx=pdf.BULK_SALES_SHEETS_FILENAME_SUFFIX, ext=extension)) assert os.path.isfile(os.path.join(pdf_render_dir, filename)) misc.remove_file(os.path.join(pdf_render_dir, filename)) def test_pdf_creation_with_bulk_sales_sheets(valid_data_from_db): """Test create_pdf.""" product_id = '1' valid_data_from_db[field_const.SALE_START_DATE] = ( datetime.datetime(day=12, month=12, year=12)) valid_data_from_db[field_const.DESCRIPTION] = ( valid_data_from_db[field_const.DESCRIPTION][0]) (flexmock(features) .should_receive('check_user_features') .with_args(None) .and_return(response.Response({})) .once()) (flexmock(pdf_generation) .should_receive('get_pdf_data_from_db') .with_args('1', 2) .and_return(response.Response(message=valid_data_from_db)) .once()) (flexmock(pdf_generation) .should_receive('get_data_from_microservices') .with_args(valid_data_from_db, False) .and_return(response.Response(message=valid_data_from_db)) .once()) (flexmock(template_rendering) .should_receive('render_template') .and_return(response.Response(message='Cool html!')) .twice()) body_template_file = '{dir}/{id}.html'.format( dir=config.PDF_RENDERING_DIR, id=product_id) header_template_file = '{dir}/{id}_header.html'.format( dir=config.PDF_RENDERING_DIR, id=product_id) rendered_pdf_file = '{dir}/{id}.pdf'.format( dir=config.PDF_RENDERING_DIR, id=product_id) (flexmock(htmltopdf) .should_receive('convert_html_to_pdf') .with_args( body_template_file, rendered_pdf_file, { 'header-spacing': pdf.HEADER_SPACING, 'header-html': header_template_file }) .and_return(response.Response()) .once()) result = pdf_generation.create_pdf(product_id, 2) assert result def test_pdf_generation_with_bulk_sales_sheets( valid_sqs_message, valid_data_from_db): """Test generate_pdf.""" job_id = '1234214' valid_data_from_db[field_const.SALE_START_DATE] = ( datetime.datetime(day=12, month=12, year=12)) valid_data_from_db[field_const.DESCRIPTION] = ( valid_data_from_db[field_const.DESCRIPTION][0]) valid_sqs_message['MessageAttributes']['Correlation-Id'] = { 'data_type': 'String', 'StringValue': '1234'} (flexmock(validate_pdf) .should_receive('validate_sqs_message') .with_args(valid_sqs_message) .and_return(response.Response())) (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.IN_PROGRESS) .and_return(response.Response()) .ordered()) (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.COMPLETED) .and_return(response.Response()) .ordered()) (flexmock(pdf_generation) .should_receive('create_pdf') .and_return( pdf_test_utils.response_from_create_pdf( '1', valid_data_from_db))) (flexmock(pdf_generation) .should_receive('create_output_file') .with_args( output_type=pdf.SINGLE_PDF, product_ids=['1'], job_id='1234214') .and_return(response.Response('1.pdf'))) (flexmock(pdf_generation) .should_receive('upload_file_to_s3') .with_args('1.pdf') .and_return(response.Response())) result = pdf_generation.generate_pdf(valid_sqs_message) assert result def test_pdf_generation_with_bulk_sales_sheets_and_non_red_vendor( valid_sqs_message, valid_data_from_db): """Test generate_pdf.""" job_id = '1234214' valid_data_from_db[field_const.SALE_START_DATE] = ( datetime.datetime(day=12, month=12, year=12)) valid_data_from_db[field_const.DESCRIPTION] = ( valid_data_from_db[field_const.DESCRIPTION][0]) valid_sqs_message['MessageAttributes']['Correlation-Id'] = { 'data_type': 'String', 'StringValue': '1234'} (flexmock(validate_pdf) .should_receive('validate_sqs_message') .with_args(valid_sqs_message) .and_return(response.Response())) (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.IN_PROGRESS) .and_return(response.Response()) .ordered()) (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.COMPLETED) .and_return(response.Response()) .ordered()) (flexmock(pdf_generation) .should_receive('create_pdf') .and_return( pdf_test_utils.response_from_create_pdf( '1', valid_data_from_db))) (flexmock(pdf_generation) .should_receive('create_output_file') .with_args( output_type=pdf.SINGLE_PDF, product_ids=['1'], job_id='1234214') .and_return(response.Response('1.pdf'))) (flexmock(pdf_generation) .should_receive('upload_file_to_s3') .with_args('1.pdf') .and_return(response.Response())) result = pdf_generation.generate_pdf(valid_sqs_message) assert result def test_pdf_generation_bulk_sales_sheets_multiple_ids( valid_sqs_message, valid_data_from_db): """Test generate_pdf.""" product_ids = ['1'] job_id = '1234214' valid_data_from_db[field_const.SALE_START_DATE] = ( datetime.datetime(day=12, month=12, year=12)) valid_data_from_db[field_const.DESCRIPTION] = ( valid_data_from_db[field_const.DESCRIPTION][0]) valid_sqs_message['MessageAttributes']['Correlation-Id'] = { 'data_type': 'String', 'StringValue': '1234'} (flexmock(validate_pdf) .should_receive('validate_sqs_message') .with_args(valid_sqs_message) .and_return(response.Response())) (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.IN_PROGRESS) .and_return(response.Response()) .ordered()) (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.COMPLETED) .and_return(response.Response()) .ordered()) (flexmock(pdf_generation) .should_receive('create_pdf') .with_args('1', 2, user_id=None) .and_return( pdf_test_utils.response_from_create_pdf( '1', valid_data_from_db))) (flexmock(pdf_generation) .should_receive('create_pdf') .and_return( pdf_test_utils.response_from_create_pdf( '2', valid_data_from_db))) (flexmock(pdf_generation) .should_receive('create_output_file') .with_args(pdf.SINGLE_PDF, product_ids, job_id) .and_return(response.Response('result.pdf')) .once()) (flexmock(pdf_generation) .should_receive('upload_file_to_s3') .with_args('result.pdf') .and_return(response.Response())) result = pdf_generation.generate_pdf(valid_sqs_message) assert result def test_pdf_creation_japan(valid_data_from_db): """Test create_pdf for japan.""" product_id = '1' localized_template_type_id = localized_template.TEMPLATE_DETAILS_IDS['JP'] valid_data_from_db[field_const.SALE_START_DATE] = ( datetime.datetime(day=12, month=12, year=12)) valid_data_from_db[field_const.DESCRIPTION] = ( valid_data_from_db[field_const.DESCRIPTION][0]) (flexmock(features) .should_receive('check_user_features') .with_args(None) .and_return(response.Response({'is_japan_sales_sheets_enabled': True})) .once()) (flexmock(pdf_generation) .should_receive('get_pdf_data_from_db') .with_args('1', localized_template_type_id) .and_return(response.Response(message=valid_data_from_db)) .once()) (flexmock(pdf_generation) .should_receive('get_data_from_microservices') .with_args(valid_data_from_db, True) .and_return(response.Response(message=valid_data_from_db)) .once()) (flexmock(template_rendering) .should_receive('render_template') .and_return(response.Response(message='Cool html!')) .twice()) body_template_file = '{dir}/{id}.html'.format( dir=config.PDF_RENDERING_DIR, id=product_id) header_template_file = '{dir}/{id}_header.html'.format( dir=config.PDF_RENDERING_DIR, id=product_id) rendered_pdf_file = '{dir}/{id}.pdf'.format( dir=config.PDF_RENDERING_DIR, id=product_id) (flexmock(htmltopdf) .should_receive('convert_html_to_pdf') .with_args( body_template_file, rendered_pdf_file, { 'header-spacing': pdf.HEADER_SPACING, 'header-html': header_template_file }) .and_return(response.Response()) .once()) result = pdf_generation.create_pdf(product_id, localized_template_type_id) assert result @pytest.mark.parametrize( 'history_record_result', (response.create_not_found_response(errors.JOB_DOES_NOT_EXIST), response.Response()) ) def test_pdf_generation_bulk_sales_sheets_single_create_pdf_failure( valid_sqs_message, history_record_result, valid_data_from_db): """Test bulk generation fails on any unsuccessful single pdf creation.""" mock_message = { 'Body': '{"product_id": ["3"], "output_type": "single",\ "job_id": "1234214", "localized_template_type_id": 2}' } mock_message['MessageAttributes'] = { 'Correlation-Id': { 'DataType': 'String', 'StringValue': '1234' }, 'feature_flag_user_context': { 'DataType': 'String', 'StringValue': 'oa:1234' } } good_products_ids = ['1'] bad_product_id = '3' job_id = '1234214' bad_pdf_creation_result = response.Response(status=400) mock_message['MessageAttributes']['Correlation-Id'] = { 'DataType': 'String', 'StringValue': '1234'} (flexmock(validate_pdf) .should_receive('validate_sqs_message') .with_args(mock_message) .and_return(response.Response())) (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.IN_PROGRESS) .and_return(response.Response()) .ordered()) (flexmock(pdf_generation) .should_receive('create_pdf') .and_return(bad_pdf_creation_result)) (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.ERROR) .and_return(history_record_result) .ordered()) (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.COMPLETED) .and_return(response.Response()) .never()) result = pdf_generation.generate_pdf(mock_message) if history_record_result: assert result.status == bad_pdf_creation_result.status def test_pdf_generation_bulk_sales_sheets_multiid_and_non_red_vendor( valid_sqs_message, valid_data_from_db): """Test generate_pdf.""" product_ids = ['1'] job_id = '1234214' valid_data_from_db[field_const.SALE_START_DATE] = ( datetime.datetime(day=12, month=12, year=12)) valid_data_from_db[field_const.DESCRIPTION] = ( valid_data_from_db[field_const.DESCRIPTION][0]) valid_sqs_message['MessageAttributes']['Correlation-Id'] = { 'data_type': 'String', 'StringValue': '1234'} (flexmock(validate_pdf) .should_receive('validate_sqs_message') .with_args(valid_sqs_message) .and_return(response.Response())) (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.IN_PROGRESS) .and_return(response.Response()) .ordered()) (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.COMPLETED) .and_return(response.Response()) .ordered()) (flexmock(pdf_generation) .should_receive('create_pdf') .and_return( pdf_test_utils.response_from_create_pdf('1', valid_data_from_db))) (flexmock(pdf_generation) .should_receive('create_output_file') .and_return(response.Response('result.pdf')) .once()) (flexmock(pdf_generation) .should_receive('upload_file_to_s3') .with_args('result.pdf') .and_return(response.Response())) result = pdf_generation.generate_pdf(valid_sqs_message) assert result def test_bulk_pdf_generation_update_job_in_progress_state_failed( valid_sqs_message): """Test generate_pdf. Generation should fail on update_job to in progress state. """ product_ids = ['1'] job_id = '1234214' valid_sqs_message['MessageAttributes']['Correlation-Id'] = { 'data_type': 'String', 'StringValue': '1234'} (flexmock(validate_pdf) .should_receive('validate_sqs_message') .with_args(valid_sqs_message) .and_return(response.Response())) (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.IN_PROGRESS) .and_return(response.create_fatal_response()) .once()) result = pdf_generation.generate_pdf(valid_sqs_message) assert not result assert result.status == 500 def test_bulk_pdf_generation_update_job_error_state_failed( valid_sqs_message, valid_data_from_db): """Test generate_pdf. Generation should fail on update_job to error state. """ job_id = '1234214' valid_sqs_message['MessageAttributes']['Correlation-Id'] = { 'data_type': 'String', 'StringValue': '1234'} (flexmock(validate_pdf) .should_receive('validate_sqs_message') .with_args(valid_sqs_message) .and_return(response.Response())) (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.IN_PROGRESS) .and_return(response.Response()) .ordered()) (flexmock(pdf_generation) .should_receive('create_pdf') .and_return( pdf_test_utils.response_from_create_pdf( '1', valid_data_from_db))) (flexmock(pdf_generation) .should_receive('create_output_file') .with_args( output_type=pdf.SINGLE_PDF, product_ids=['1'], job_id='1234214') .and_return(response.Response('1.pdf'))) (flexmock(pdf_generation) .should_receive('upload_file_to_s3') .with_args('1.pdf') .and_return(response.create_fatal_response())) (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.ERROR) .and_return(response.create_fatal_response()) .once()) result = pdf_generation.generate_pdf(valid_sqs_message) assert not result assert result.status == 500 def test_bulk_pdf_generation_update_job_complete_state_failed( valid_sqs_message, valid_data_from_db): """Test generate_pdf. Generation should fail on update_job to complete state. """ job_id = '1234214' valid_sqs_message['MessageAttributes']['Correlation-Id'] = { 'data_type': 'String', 'StringValue': '1234'} (flexmock(validate_pdf) .should_receive('validate_sqs_message') .with_args(valid_sqs_message) .and_return(response.Response())) (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.IN_PROGRESS) .and_return(response.Response()) .ordered()) (flexmock(pdf_generation) .should_receive('create_pdf') .and_return( pdf_test_utils.response_from_create_pdf( '1', valid_data_from_db))) (flexmock(pdf_generation) .should_receive('create_output_file') .with_args( output_type=pdf.SINGLE_PDF, product_ids=['1'], job_id='1234214') .and_return(response.Response('1.pdf'))) (flexmock(pdf_generation) .should_receive('upload_file_to_s3') .with_args('1.pdf') .and_return(response.Response())) (flexmock(history) .should_receive('update_job') .with_args(job_id, salessheets_history.COMPLETED) .and_return(response.create_fatal_response()) .once()) result = pdf_generation.generate_pdf(valid_sqs_message) assert not result assert result.status == 500 def test_get_pdf_data_from_db_physical_display_distribution_format( valid_get_album_response_message, valid_template_details_response_message): """Test get_pdf_data_from_db adds configuration for physical product.""" product_id = '1' product_type_resp = {'product_type_id': 1} (flexmock(salessheets) .should_receive('get_product_type_id') .and_return(response.Response(product_type_resp))) (flexmock(pdf_generation) .should_receive('get_album_data_for_product').with_args(product_id) .and_return(response.Response( message=valid_get_album_response_message))).once() artists_list = response.Response(message=['Cool Artist']) (flexmock(salessheets) .should_receive('get_artists') .and_return(artists_list)).once() (flexmock(salessheets) .should_receive('get_tracks').with_args(product_id) .and_return(response.Response())).once() (flexmock(salessheets) .should_receive('get_distribution_format') .and_return(response.Response(message={ 'display_distribution_format': 'CD' }))).once() (flexmock(template_details) .should_receive('get_by_template_type_id') .with_args(2) .and_return(response.Response(valid_template_details_response_message))) result = pdf_generation.get_pdf_data_from_db(product_id, 2) assert result assert field_const.CONFIGURATION in result.message assert field_const.DISTRIBUTION_FORMAT not in result.message def test_get_pdf_data_from_db_physical_display_distribution_format_not_found( valid_get_album_response_message, valid_template_details_response_message): """Test get_pdf_data_from_db if display_distribution_format not found.""" product_id = '1' product_type_resp = {'product_type_id': 1} (flexmock(salessheets) .should_receive('get_product_type_id') .and_return(response.Response(product_type_resp))) (flexmock(pdf_generation) .should_receive('get_album_data_for_product').with_args(product_id) .and_return(response.Response( message=valid_get_album_response_message))).once() artists_list = response.Response(message=['Cool Artist']) (flexmock(salessheets) .should_receive('get_artists') .and_return(artists_list)).once() (flexmock(salessheets) .should_receive('get_tracks').with_args(product_id) .and_return(response.Response())).once() (flexmock(salessheets) .should_receive('get_distribution_format') .and_return(response.create_not_found_response())).once() (flexmock(template_details) .should_receive('get_by_template_type_id') .with_args(2) .and_return(response.Response(valid_template_details_response_message))) result = pdf_generation.get_pdf_data_from_db(product_id, 2) assert result assert field_const.CONFIGURATION not in result.message @pytest.mark.parametrize( 'get_context_type_and_status_response_message,' 'get_album_response_message', [ ({field_const.CONTEXT_TYPE: field_const.PHYSICAL_CONTEXT_TYPE, field_const.RELEASE_STATUS: field_const.IN_CONTENT}, valid_get_album_response_message), ({field_const.CONTEXT_TYPE: field_const.PHYSICAL_CONTEXT_TYPE, field_const.RELEASE_STATUS: field_const.LABEL_PROCESSING}, valid_get_album_response_message) ]) def test_get_album_data_for_product_success( get_context_type_and_status_response_message, get_album_response_message): """Test get_album_data_for_product success.""" release_id = '1' (flexmock(salessheets) .should_receive('get_context_type_and_status') .with_args(release_id) .and_return(response.Response( get_context_type_and_status_response_message))).once() (flexmock(salessheets) .should_receive('get_album') .with_args(release_id) .and_return(response.Response(get_album_response_message))).once() result = pdf_generation.get_album_data_for_product(release_id) assert result.message == get_album_response_message def test_get_album_data_for_product_success_for_digital_not_in_content( valid_get_album_response_message): """Test get_album_data_for_product success.""" release_id = '1' get_context_type_and_status_response_msg = { field_const.CONTEXT_TYPE: field_const.DIGITAL_CONTEXT_TYPE, field_const.RELEASE_STATUS: field_const.LABEL_PROCESSING} valid_genre_response_msg = { 'genre': 'test_genre', 'subgenre': 'test_subgenre' } (flexmock(salessheets) .should_receive('get_context_type_and_status') .with_args(release_id) .and_return( response.Response( get_context_type_and_status_response_msg))).once() (flexmock(salessheets) .should_receive('get_album_for_digital') .with_args(release_id) .and_return(response.Response(valid_get_album_response_message))).once() (flexmock(salessheets) .should_receive('get_genre_and_subgenre') .with_args(release_id) .and_return(response.Response(valid_genre_response_msg)) .once()) expected_result_msg = copy.deepcopy(valid_get_album_response_message) expected_result_msg.update(valid_genre_response_msg) result = pdf_generation.get_album_data_for_product(release_id) assert result.message == expected_result_msg @pytest.mark.parametrize( 'get_genre_error_response', [ response.create_not_found_response(), response.create_fatal_response() ] ) def test_get_album_data_for_digital_if_get_genre_error( valid_get_album_response_message, get_genre_error_response): """Test get_album_data_for_product success if get_genre error.""" release_id = '1' get_context_type_and_status_response_msg = { field_const.CONTEXT_TYPE: field_const.DIGITAL_CONTEXT_TYPE, field_const.RELEASE_STATUS: field_const.LABEL_PROCESSING} (flexmock(salessheets) .should_receive('get_context_type_and_status') .with_args(release_id) .and_return( response.Response( get_context_type_and_status_response_msg))).once() (flexmock(salessheets) .should_receive('get_album_for_digital') .with_args(release_id) .and_return(response.Response(valid_get_album_response_message))).once() (flexmock(salessheets) .should_receive('get_genre_and_subgenre') .with_args(release_id) .and_return(get_genre_error_response).once()) result = pdf_generation.get_album_data_for_product(release_id) assert result.message == valid_get_album_response_message @pytest.mark.parametrize( 'status', [ field_const.LABEL_PROCESSING, field_const.IN_CONTENT ] ) def test_get_album_data_for_product_success_for_digital( status, valid_get_album_response_message): """Test get_album_data_for_product success.""" release_id = '1' get_context_type_and_status_response_msg = { field_const.CONTEXT_TYPE: field_const.DIGITAL_CONTEXT_TYPE, field_const.RELEASE_STATUS: status} valid_genre_response_msg = { 'genre': 'test_genre', 'subgenre': 'test_subgenre' } (flexmock(salessheets) .should_receive('get_context_type_and_status') .with_args(release_id) .and_return( response.Response( get_context_type_and_status_response_msg))).once() (flexmock(salessheets) .should_receive('get_album_for_digital') .with_args(release_id) .and_return(response.Response(valid_get_album_response_message))).once() (flexmock(salessheets) .should_receive('get_genre_and_subgenre') .with_args(release_id) .and_return(response.Response(valid_genre_response_msg)) .once()) expected_result_msg = copy.deepcopy(valid_get_album_response_message) expected_result_msg.update(valid_genre_response_msg) result = pdf_generation.get_album_data_for_product(release_id) assert result.message == expected_result_msg @pytest.mark.parametrize( 'get_genre_error_response', [ response.create_not_found_response(), response.create_fatal_response() ] ) def test_get_album_data_for_digital_not_in_content_if_get_genre_error( valid_get_album_response_message, get_genre_error_response): """Test get_album_data_for_product success.""" release_id = '1' get_context_type_and_status_response_msg = { field_const.CONTEXT_TYPE: field_const.DIGITAL_CONTEXT_TYPE, field_const.RELEASE_STATUS: field_const.LABEL_PROCESSING} (flexmock(salessheets) .should_receive('get_context_type_and_status') .with_args(release_id) .and_return( response.Response( get_context_type_and_status_response_msg))).once() (flexmock(salessheets) .should_receive('get_album_for_digital') .with_args(release_id) .and_return(response.Response(valid_get_album_response_message))).once() (flexmock(salessheets) .should_receive('get_genre_and_subgenre') .with_args(release_id) .and_return(get_genre_error_response).once()) result = pdf_generation.get_album_data_for_product(release_id) assert result.message == valid_get_album_response_message @pytest.mark.parametrize( 'get_context_type_and_status_response_message, get_album_function', [ ({field_const.CONTEXT_TYPE: field_const.PHYSICAL_CONTEXT_TYPE, field_const.RELEASE_STATUS: field_const.IN_CONTENT}, 'get_album'), ({field_const.CONTEXT_TYPE: field_const.PHYSICAL_CONTEXT_TYPE, field_const.RELEASE_STATUS: field_const.LABEL_PROCESSING}, 'get_album'), ({field_const.CONTEXT_TYPE: field_const.DIGITAL_CONTEXT_TYPE, field_const.RELEASE_STATUS: field_const.IN_CONTENT}, 'get_album_for_digital'), ({field_const.CONTEXT_TYPE: field_const.DIGITAL_CONTEXT_TYPE, field_const.RELEASE_STATUS: field_const.LABEL_PROCESSING}, 'get_album_for_digital') ]) def test_get_album_data_for_product_not_found_error( get_context_type_and_status_response_message, get_album_function): """Test get_album_data_for_product in case album data is not found.""" release_id = '1' (flexmock(salessheets) .should_receive('get_context_type_and_status') .with_args(release_id) .and_return(response.Response( get_context_type_and_status_response_message))).once() (flexmock(salessheets) .should_receive(get_album_function) .with_args(release_id) .and_return(response.create_not_found_response())).once() result = pdf_generation.get_album_data_for_product(release_id) assert result.status == 404 def test_get_album_data_for_product_context_type_and_status_not_found(): """Test get_album_data in case context type and status is not found.""" release_id = '1' (flexmock(salessheets) .should_receive('get_context_type_and_status') .with_args(release_id) .and_return(response.create_not_found_response())).once() result = pdf_generation.get_album_data_for_product(release_id) assert result.status == 404 def test_get_pdf_data_from_db_contacts_data_has_null_values( valid_get_album_response_message, valid_data_from_db, valid_template_details_response_message): """Test get pdf data updates contact data if localized flag enabled.""" release_id = '1' product_type_resp = {'product_type_id': 1} genre_and_subgenre_keys = [field_const.GENRE, field_const.SUBGENRE] album_data_message = copy.deepcopy(valid_get_album_response_message) for key in genre_and_subgenre_keys: album_data_message.pop(key) for key in genre_and_subgenre_keys: valid_data_from_db[key] = None (flexmock(salessheets) .should_receive('get_product_type_id') .with_args(release_id) .and_return(response.Response(product_type_resp))) (flexmock(pdf_generation) .should_receive('get_album_data_for_product') .with_args(release_id) .and_return(response.Response( message=album_data_message)) .once()) tracks = response.Response(message={ '153': { 'track_id': '153', 'track_number': 1, 'side': None, 'cd': '1'}}) (flexmock(salessheets) .should_receive('get_tracks') .with_args(release_id) .and_return(tracks)) artists_list = response.Response(message=['Cool Artist']) (flexmock(salessheets) .should_receive('get_artists') .with_args(release_id) .and_return(artists_list)) (flexmock(barcode_generator) .should_receive('generate_barcode') .with_args(valid_data_from_db[field_const.UPC]) .and_return(response.Response(message={ 'barcode': 'test_barcode_file', 'barcode_type': 'test_type' }))).once() (flexmock(salessheets) .should_receive('get_distribution_format') .with_args(release_id) .and_return(response.Response(message={ 'display_distribution_format': 'CD' }))).once() (flexmock(template_details) .should_receive('get_by_template_type_id') .with_args(2) .and_return(response.Response(valid_template_details_response_message))) expected_contact_data_string = ( 'apartment street\ncity, state postal_code\ncountry\ntelephone') result = pdf_generation.get_pdf_data_from_db(release_id, 2) assert result.message[localized_template.LOCALIZED_ORCHARD_CONTACT] == ( expected_contact_data_string) @pytest.mark.parametrize( ( 'context_type', 'expected_upc', 'generate_barcode_times_called', 'get_distr_format_times_called'), [ ('physical', '111222', 1, 1), ('digital', '111222', 0, 0) ] ) def test_get_pdf_data_from_db_add_correct_upc( valid_get_album_response_message, context_type, expected_upc, generate_barcode_times_called, get_distr_format_times_called, valid_data_from_db, valid_template_details_response_message): """Test get pdf data updates contact data if localized flag enabled.""" release_id = '1' product_type_resp = {'product_type_id': 1} genre_and_subgenre_keys = [field_const.GENRE, field_const.SUBGENRE] album_data_message = copy.deepcopy(valid_get_album_response_message) for key in genre_and_subgenre_keys: album_data_message.pop(key) album_data_message[field_const.CONTEXT_TYPE] = context_type for key in genre_and_subgenre_keys: valid_data_from_db[key] = None (flexmock(salessheets) .should_receive('get_product_type_id') .with_args(release_id) .and_return(response.Response(product_type_resp))) (flexmock(pdf_generation) .should_receive('get_album_data_for_product') .with_args(release_id) .and_return(response.Response( message=album_data_message)) .once()) tracks = response.Response(message={ '153': { 'track_id': '153', 'track_number': 1, 'cd': '1', 'side': None, 'explicit': 'Y'}}) (flexmock(salessheets) .should_receive('get_tracks') .with_args(release_id) .and_return(tracks)) artists_list = response.Response(message=['Cool Artist']) (flexmock(salessheets) .should_receive('get_artists') .with_args(release_id) .and_return(artists_list)) (flexmock(barcode_generator) .should_receive('generate_barcode') .with_args(valid_data_from_db[field_const.UPC]) .and_return(response.Response(message={ 'barcode': 'test_barcode_file', 'barcode_type': 'test_type' }))).times(generate_barcode_times_called) (flexmock(salessheets) .should_receive('get_distribution_format') .with_args(release_id) .and_return(response.Response(message={ 'display_distribution_format': 'CD' }))).times(get_distr_format_times_called) localized_template_type_id = 1 (flexmock(template_details) .should_receive('get_by_template_type_id') .with_args(localized_template_type_id) .and_return(response.Response(valid_template_details_response_message))) result = pdf_generation.get_pdf_data_from_db( release_id, localized_template_type_id) assert result.message[field_const.UPC] == expected_upc @pytest.mark.parametrize( ( 'context_type', 'explicit_in_track_result', 'generate_barcode_times_called', 'get_distr_format_times_called', 'expected_explicit',), [ ('physical', 'N', 1, 1, 'N'), ('digital', 'N', 0, 0, 'N'), ('digital', 'Y', 0, 0, 'Y') ] ) def test_get_pdf_data_from_db_add_explicit_for_all( valid_get_album_response_message, context_type, expected_explicit, generate_barcode_times_called, get_distr_format_times_called, explicit_in_track_result, valid_data_from_db, valid_template_details_response_message): """Test get pdf data updates contact data if localized flag enabled.""" release_id = '1' product_type_resp = {'product_type_id': 1} genre_and_subgenre_keys = [field_const.GENRE, field_const.SUBGENRE] album_data_message = copy.deepcopy(valid_get_album_response_message) for key in genre_and_subgenre_keys: album_data_message.pop(key) album_data_message[field_const.CONTEXT_TYPE] = context_type for key in genre_and_subgenre_keys: valid_data_from_db[key] = None (flexmock(salessheets) .should_receive('get_product_type_id') .with_args(release_id) .and_return(response.Response(product_type_resp))) (flexmock(pdf_generation) .should_receive('get_album_data_for_product') .with_args(release_id) .and_return(response.Response( message=album_data_message)) .once()) tracks = response.Response(message={ '153': { 'track_id': '153', 'track_number': 1, 'cd': '1', 'side': None, 'explicit': explicit_in_track_result}, }) (flexmock(salessheets) .should_receive('get_tracks') .with_args(release_id) .and_return(tracks)) artists_list = response.Response(message=['Cool Artist']) (flexmock(salessheets) .should_receive('get_artists') .with_args(release_id) .and_return(artists_list)) (flexmock(barcode_generator) .should_receive('generate_barcode') .with_args(valid_data_from_db[field_const.UPC]) .and_return(response.Response(message={ 'barcode': 'test_barcode_file', 'barcode_type': 'test_type' }))).times(generate_barcode_times_called) (flexmock(salessheets) .should_receive('get_distribution_format') .with_args(release_id) .and_return(response.Response(message={ 'display_distribution_format': 'CD' }))).times(get_distr_format_times_called) localized_template_type_id = 1 (flexmock(template_details) .should_receive('get_by_template_type_id') .with_args(localized_template_type_id) .and_return(response.Response(valid_template_details_response_message))) result = pdf_generation.get_pdf_data_from_db( release_id, localized_template_type_id) assert result.message.get(field_const.EXPLICIT) == expected_explicit @pytest.mark.parametrize( 'given_explicit, expected_result', [ ('Y', True), ('N', False), ]) def test_is_digital_product_explicit(given_explicit, expected_result): """Expect to return correct explicit value.""" test_tracks = { 107165: { 'artist_name': 'The Orchard Email Account', 'cd': 1, 'product_id': 1, 'track_id': 107165, 'track_name': 'O Holy Night', 'track_number': 10, 'type': 'performer', 'explicit': 'N'}, 107166: { 'artist_name': 'The Orchard Email Account', 'cd': 1, 'product_id': 1, 'track_id': 107166, 'track_name': 'Happy New Year', 'track_number': 11, 'type': 'performer', 'explicit': given_explicit} } result = pdf_generation.is_digital_product_explicit(test_tracks) assert result is expected_result