"""Tests for requests to handlers with validation by schema.""" import json from flexmock import flexmock from oto import response import pytest from salessheets import features from salessheets.constants import error from salessheets.constants import features as feature_constants from salessheets.constants import field_const from salessheets.constants import models from salessheets.constants import salessheets from salessheets.logic import generation_result from salessheets.logic import salessheets_generation from salessheets.models import history def make_post_duplicate_upc(client, headers, data): """Function that makes request to url.""" url = '/bulk/duplicate-upc' headers['Content-Type'] = 'application/json' return client.post(url, data=json.dumps(data), headers=headers) @pytest.fixture def invalid_post_data(): """Fixture with invalid data for bulk post request.""" return { 'upcs': 'string', 'generation_method': 'not_listed'} def invalid_duplicate_upc_post_data(): """Fixture with invalid data for bulk post request.""" return { 'context': '', 'generation_method': 'single'} def invalid_duplicate_upc_select_template_post_data(): """Return invalid data for bulk post request if select template enabled.""" return { 'upcs': [1234, 4321], 'generation_method': 'single', 'template_type': 'invalid_template'} @pytest.fixture(scope='function') def valid_post_data(): """Fixture with valid data for bulk post request.""" return { 'upcs': [1234, 4321], 'generation_method': 'single'} def valid_post_data_duplicate_upc(): """Fixture with valid data for bulk post request.""" return { 'context': ['1234', '4321'], 'generation_method': 'single'} def valid_post_data_duplicate_upc_select_template(): """Return valid data for bulk post request if select template enabled.""" return { 'context': ['1234', '4321'], 'generation_method': 'single', 'template_type': 'orchard'} def make_get_result_request(client, headers): """Function that makes get request to /bulk/ url.""" url = '/bulk/1234' return client.get(url, headers=headers) def test_get_generation_result_header_validation_failed( client): """Test get data with invalid headers.""" invalid_headers = {field_const.GRASS_ACCOUNT_TYPE: 'wrong_account_type'} res = make_get_result_request(client, invalid_headers) assert ( json.loads(res.data.decode())['code'] == error.HEADER_VALIDATION_ERROR) assert res.status_code == 400 def test_get_generation_result_validation_success( client, valid_headers): """Test get result with valid data.""" (flexmock(generation_result) .should_receive('get_download_url_by_job_id') .and_return(response.Response('http://s0meurl/')) .once()) res = make_get_result_request(client, valid_headers) assert res.status_code == 200 def test_get_generation_result_validation_internal_error( client, valid_headers): """Test get result for valid job and get internal error response.""" (flexmock(generation_result) .should_receive('get_download_url_by_job_id') .and_return(response.create_fatal_response()) .once()) res = make_get_result_request(client, valid_headers) assert res.status_code == 500 def test_get_generation_result_validation_no_such_job( client, valid_headers): """Test get result for valid job and get internal error response.""" (flexmock(history) .should_receive('get_job_by_id') .and_return(response.create_not_found_response( error.JOB_DOES_NOT_EXIST)) .once()) res = make_get_result_request(client, valid_headers) assert ( json.loads(res.data.decode())['message'] == error.JOB_DOES_NOT_EXIST) assert res.status_code == 404 def make_get_history_request(client, headers, query_string): """Function that makes get request to /bulk/history url.""" url = '/bulk/history' return client.get(url, headers=headers, query_string=query_string) def test_get_history_header_validation_failed( client): """Test get data with invalid headers.""" invalid_headers = {field_const.GRASS_ACCOUNT_TYPE: 'wrong_account_type'} query_string_data = {} res = make_get_history_request(client, invalid_headers, query_string_data) assert ( json.loads(res.data.decode())['code'] == error.HEADER_VALIDATION_ERROR) assert res.status_code == 400 def test_get_history_success(client, valid_headers): """Test get result with valid data and different user_ids.""" page_offset = 47 processed_user_id = 'oa:1234' query_params = {field_const.PAGE_OFFSET: page_offset} pagination = { field_const.TYPE: field_const.STANDARD, field_const.OFFSET: page_offset, field_const.LIMIT: field_const.DEFAULT_PAGE_LIMIT, field_const.TOTAL_RECORDS: 0, field_const.SORT_BY: field_const.DEFAULT_SORT_BY, field_const.SORT_ORDER: field_const.DEFAULT_SORT_ORDER} result = { field_const.ITEMS: [], field_const.PAGINATION: pagination} (flexmock(generation_result) .should_receive('get_history_for_user') .with_args( processed_user_id, field_const.DEFAULT_SORT_BY, field_const.DEFAULT_SORT_ORDER, field_const.DEFAULT_PAGE_LIMIT, page_offset) .and_return(response.Response(result)) .once()) res = make_get_history_request(client, valid_headers, query_params) assert res.status_code == 200 assert ( json.loads(res.data.decode()) == result) def test_get_history_validation_success_with_offset( client, valid_headers): """Test get result with valid data.""" page_offset = 47 query_params = {field_const.PAGE_OFFSET: page_offset} pagination = { field_const.TYPE: field_const.STANDARD, field_const.OFFSET: 47, field_const.LIMIT: field_const.DEFAULT_PAGE_LIMIT, field_const.TOTAL_RECORDS: 0, field_const.SORT_BY: field_const.DEFAULT_SORT_BY, field_const.SORT_ORDER: field_const.DEFAULT_SORT_ORDER} result = { field_const.ITEMS: [], field_const.PAGINATION: pagination} (flexmock(history) .should_receive('get_jobs_count') .and_return(response.Response(0)) .once()) res = make_get_history_request(client, valid_headers, query_params) assert res.status_code == 200 assert ( json.loads(res.data.decode()) == result) def test_get_history_validation_success_with_limit( client, valid_headers): """Test get result with valid data.""" page_limit = 1479 query_params = {field_const.PAGE_LIMIT: page_limit} pagination = { field_const.TYPE: field_const.STANDARD, field_const.OFFSET: field_const.DEFAULT_PAGE_OFFSET, field_const.LIMIT: page_limit, field_const.TOTAL_RECORDS: 0, field_const.SORT_BY: field_const.DEFAULT_SORT_BY, field_const.SORT_ORDER: field_const.DEFAULT_SORT_ORDER} result = { field_const.ITEMS: [], field_const.PAGINATION: pagination} (flexmock(history) .should_receive('get_jobs_count') .and_return(response.Response(0)) .once()) res = make_get_history_request(client, valid_headers, query_params) assert res.status_code == 200 assert ( json.loads(res.data.decode()) == result) def test_get_history_validation_internal_error( client, valid_headers): """Test get result for valid job and get internal error response.""" (flexmock(history) .should_receive('get_jobs_count') .and_return(response.create_fatal_response()) .once()) res = make_get_history_request(client, valid_headers, query_string={}) assert res.status_code == 500 @pytest.mark.parametrize( 'sales_sheets_select_template_enabled, invalid_post_data', [ (True, invalid_duplicate_upc_select_template_post_data()), (False, invalid_duplicate_upc_post_data())]) def test_post_bulk_generation_duplicate_upc_body_validation_failed( client, valid_headers, invalid_post_data, sales_sheets_select_template_enabled, feature_engine): """Test post invalid data.""" feature_engine.force_flag( feature_constants.SALES_SHEETS_SELECT_TEMPLATE, sales_sheets_select_template_enabled) res = make_post_duplicate_upc( client, valid_headers, invalid_post_data) assert json.loads( res.data.decode())['code'] == error.BODY_VALIDATION_ERROR assert res.status_code == 400 @pytest.mark.parametrize( 'sales_sheets_select_template_enabled, valid_post_data', [ (True, valid_post_data_duplicate_upc_select_template()), (False, valid_post_data_duplicate_upc())]) def test_post_bulk_generation_duplicate_upc_header_validation_failed( sales_sheets_select_template_enabled, feature_engine, client, valid_post_data): """Test post data with invalid headers.""" feature_engine.force_flag( feature_constants.SALES_SHEETS_SELECT_TEMPLATE, sales_sheets_select_template_enabled) invalid_headers = {field_const.GRASS_ACCOUNT_TYPE: 'wrong_account_type'} res = make_post_duplicate_upc( client, invalid_headers, valid_post_data) assert (json.loads( res.data.decode())['code'] == error.HEADER_VALIDATION_ERROR) assert res.status_code == 400 @pytest.mark.parametrize( 'localized_template_enabled, localized_template_type_id', [ (True, 1), (False, None)]) @pytest.mark.parametrize( 'sales_sheets_select_template_enabled, valid_post_data, ' 'expected_template_type', [ (True, valid_post_data_duplicate_upc_select_template(), 'orchard'), (False, valid_post_data_duplicate_upc(), None)]) def test_post_bulk_generation_duplicate_upc_validation_success( client, valid_headers, sales_sheets_select_template_enabled, feature_engine, expected_template_type, localized_template_type_id, localized_template_enabled, valid_post_data): """Test post valid data.""" feature_engine.force_flag( feature_constants.SALES_SHEETS_SELECT_TEMPLATE, sales_sheets_select_template_enabled) (flexmock(features) .should_receive('is_feature_enabled') .with_args(feature_constants.SALES_SHEETS_SELECT_TEMPLATE) .and_return(sales_sheets_select_template_enabled)) feature_engine.force_flag( feature_constants.LOCALIZED_SALES_SHEETS, localized_template_enabled) (flexmock(features) .should_receive('is_feature_enabled') .with_args(feature_constants.LOCALIZED_SALES_SHEETS) .and_return(localized_template_enabled)) grass_account_type = 'vendor' grass_account_id = '123' valid_headers.update({ field_const.GRASS_ACCOUNT_ID: grass_account_id, field_const.GRASS_ACCOUNT_TYPE: grass_account_type, }) valid_request_post_data = valid_post_data.copy() if localized_template_enabled: valid_request_post_data.update({ salessheets.LOCALIZED_TEMPLATE_TYPE_ID: 1 }) processed_user_id = 'oa:1234' expected_grass_account_type = grass_account_type expected_grass_account_id = grass_account_id (flexmock(salessheets_generation) .should_receive('trigger_salessheets_generation_job_duplicate_upc') .with_args( context=valid_request_post_data[salessheets.CONTEXT], context_type=salessheets.DISPLAY_UPC, generation_method=valid_request_post_data[ salessheets.GENERATION_METHOD], user_id=processed_user_id, correlation_id=valid_headers[field_const.CORRELATION_ID], grass_account_type=expected_grass_account_type, grass_account_id=expected_grass_account_id, template_type=expected_template_type, localized_template_type_id=localized_template_type_id) .and_return(response.Response()) .once()) res = make_post_duplicate_upc( client, valid_headers, valid_request_post_data) assert res.status_code == 200 @pytest.mark.parametrize( 'sales_sheets_select_template_enabled, valid_post_data', [ (True, valid_post_data_duplicate_upc_select_template()), (False, valid_post_data_duplicate_upc())]) def test_post_bulk_generation_duplicate_upc_validation_internal_error( client, valid_headers, valid_post_data, sales_sheets_select_template_enabled, feature_engine): """Test post valid data and get internal error response.""" feature_engine.force_flag( feature_constants.SALES_SHEETS_SELECT_TEMPLATE, sales_sheets_select_template_enabled) (flexmock(salessheets_generation) .should_receive('trigger_salessheets_generation_job_duplicate_upc') .and_return(response.create_fatal_response()) .once()) res = make_post_duplicate_upc( client, valid_headers, valid_post_data) assert res.status_code == 500 @pytest.mark.parametrize( 'sales_sheets_select_template_enabled, valid_post_data', [ (True, valid_post_data_duplicate_upc_select_template()), (False, valid_post_data_duplicate_upc())]) @pytest.mark.parametrize( 'grass_account_type, grass_account_id, status', [ (None, '1234', 400), ('vendor', None, 400), ('vendor', '123', 200)]) def test_post_bulk_generation_duplicate_upc_grass_headers_validation( client, valid_post_data, grass_account_type, grass_account_id, status, valid_headers, sales_sheets_select_template_enabled, feature_engine): """Test post valid data and get internal error response.""" feature_engine.force_flag( feature_constants.SALES_SHEETS_SELECT_TEMPLATE, sales_sheets_select_template_enabled) if grass_account_type: valid_headers.update({ field_const.GRASS_ACCOUNT_TYPE: grass_account_type, }) if grass_account_id: valid_headers.update({ field_const.GRASS_ACCOUNT_ID: grass_account_id, }) if status == 200: (flexmock(salessheets_generation) .should_receive('trigger_salessheets_generation_job_duplicate_upc') .and_return(response.Response()) .once()) res = make_post_duplicate_upc( client, valid_headers, valid_post_data) assert res.status_code == status def make_get_result_by_release_id_request(client, headers, release_id): """Function that makes get request to /single/ url.""" url = '/single/{release_id}'.format(release_id=release_id) return client.get(url, headers=headers) def test_get_result_by_release_id_header_validation_failed( client): """Test get data with invalid headers.""" release_id = 1234 invalid_headers = {field_const.GRASS_ACCOUNT_TYPE: 'wrong_account_type'} res = make_get_result_by_release_id_request( client, invalid_headers, release_id) assert ( json.loads(res.data.decode())['code'] == error.HEADER_VALIDATION_ERROR) assert res.status_code == 400 def test_get_result_by_release_id_validation_success( client, valid_headers): """Test get job by release id with valid data.""" release_id = 1234 (flexmock(generation_result) .should_receive('get_last_successful_job_for_release_id') .and_return(response.Response({ field_const.TIMESTAMP: 'timestamp', field_const.DOWNLOAD_URL: 'url'})) .once()) res = make_get_result_by_release_id_request( client, valid_headers, release_id) assert res.status_code == 200 def test_get_in_progress_status_result_by_release_id_success( client, valid_headers): """Test get job by release id with valid data.""" release_id = 1234 (flexmock(generation_result) .should_receive('get_last_successful_job_for_release_id') .and_return(response.Response({ 'message': salessheets.ERROR_MESSAGE.format( id=release_id, state=salessheets.REQUESTED), 'status': 200})) .once()) res = make_get_result_by_release_id_request( client, valid_headers, release_id) assert res.status_code == 200 def test_get_error_status_result_by_release_id( client, valid_headers): """Test get job by release id with valid data.""" release_id = 1234 (flexmock(generation_result) .should_receive('get_last_successful_job_for_release_id') .and_return(response.create_error_response( code=error.JOB_IS_NOT_COMPLETED_CODE, message=salessheets.ERROR_MESSAGE.format( id=release_id, state=salessheets.STATUS_ERROR))) .once()) res = make_get_result_by_release_id_request( client, valid_headers, release_id) assert res.status_code == 400 def test_get_result_by_release_id_non_int_url_param( client, valid_headers): """Test get job by invalid release id.""" invalid_release_id = 'JAJAJA' res = make_get_result_by_release_id_request( client, valid_headers, invalid_release_id) assert res.status_code == 404 def test_get_result_by_release_id_validation_internal_error( client, valid_headers): """Test get result for release_id and get internal error response.""" release_id = 1234 (flexmock(generation_result) .should_receive('get_last_successful_job_for_release_id') .and_return(response.create_fatal_response()) .once()) res = make_get_result_by_release_id_request( client, valid_headers, release_id) assert res.status_code == 500 def test_get_result_by_release_id_validation_workstation_feature_flag( client, valid_headers): """Test get result for valid release_id and get internal error response.""" release_id = 1234 grass_account_type = 'vendor' grass_account_id = '123' valid_headers.update({ field_const.GRASS_ACCOUNT_ID: grass_account_id, field_const.GRASS_ACCOUNT_TYPE: grass_account_type, }) expected_grass_account_type = grass_account_type expected_grass_account_id = grass_account_id (flexmock(generation_result) .should_receive('get_last_successful_job_for_release_id') .with_args( release_id, expected_grass_account_id, expected_grass_account_type) .and_return(response.Response({ field_const.TIMESTAMP: 'timestamp', field_const.DOWNLOAD_URL: 'url'})) .once()) res = make_get_result_by_release_id_request( client, valid_headers, release_id) assert res.status_code == 200 def make_post_single_generation_by_release_id_request( client, headers, release_id): """Function that makes post request to /single/ url.""" url = '/single/{release_id}'.format(release_id=release_id) return client.post(url, headers=headers) def make_post_single_generation_with_template_type_request( client, headers, release_id, template_type, localized_template_type_id): """Function that makes post request to /single/ url. Args: client: test flask client headers(dict): headers to add to request release_id(int): id of release for which the sales sheet will be generated. template_type(str): template_type: template type to use for salessheet generation, possible values are: 'orchard', 'redessential'. localized_template_type_id (int): id of localized template type, corresponds to the id in salessheets.template_detail table. """ url = ( '/single/{release_id}?template_type={template_type}' '&localized_template_type_id={localized_template_type_id}').format( release_id=release_id, template_type=template_type, localized_template_type_id=localized_template_type_id) return client.post(url, headers=headers) def test_post_single_generation_by_release_id_header_validation_failed( client, feature_engine): """Test post single generation with invalid headers.""" invalid_headers = {field_const.GRASS_ACCOUNT_TYPE: 'wrong_account_type'} release_id = 123 res = make_post_single_generation_by_release_id_request( client, invalid_headers, release_id) assert (json.loads( res.data.decode())['code'] == error.HEADER_VALIDATION_ERROR) assert res.status_code == 400 def test_post_single_generation_by_release_id_validation_success( client, valid_headers, feature_engine): """Test post single generation valid .""" processed_user_id = 'oa:1234' grass_account_type = 'vendor' grass_account_id = '123' valid_headers.update({ field_const.GRASS_ACCOUNT_ID: grass_account_id, field_const.GRASS_ACCOUNT_TYPE: grass_account_type, }) release_id = 1234 expected_grass_account_type = grass_account_type expected_grass_account_id = grass_account_id (flexmock(salessheets_generation) .should_receive('trigger_salessheets_generation_job_duplicate_upc') .with_args( context=[str(release_id)], context_type=models.RELEASE_ID, generation_method=salessheets.SINGLE, user_id=processed_user_id, correlation_id=valid_headers[field_const.CORRELATION_ID], grass_account_type=expected_grass_account_type, grass_account_id=expected_grass_account_id, template_type=None, localized_template_type_id=None) .and_return(response.Response()) .once()) res = make_post_single_generation_by_release_id_request( client, valid_headers, release_id) assert res.status_code == 200 @pytest.mark.parametrize( 'template_type, localized_template_enabled', [ ('orchard', False), ('redessential', True)] ) def test_post_single_generation_with_template_type_success( client, valid_headers, template_type, feature_engine, localized_template_enabled): """Test post single generation valid with template_type arg.""" feature_engine.force_flag( feature_constants.LOCALIZED_SALES_SHEETS, localized_template_enabled) processed_user_id = 'oa:1234' grass_account_type = 'vendor' grass_account_id = '123' localized_template_type_id = '345' valid_headers.update({ field_const.GRASS_ACCOUNT_ID: grass_account_id, field_const.GRASS_ACCOUNT_TYPE: grass_account_type, }) release_id = 1234 expected_grass_account_type = grass_account_type expected_grass_account_id = grass_account_id (flexmock(salessheets_generation) .should_receive('trigger_salessheets_generation_job_duplicate_upc') .with_args( context=[str(release_id)], context_type=models.RELEASE_ID, generation_method=salessheets.SINGLE, user_id=processed_user_id, correlation_id=valid_headers[field_const.CORRELATION_ID], grass_account_type=expected_grass_account_type, grass_account_id=expected_grass_account_id, template_type=template_type, localized_template_type_id=localized_template_type_id) .and_return(response.Response()) .once()) res = make_post_single_generation_with_template_type_request( client, valid_headers, release_id, template_type, localized_template_type_id) assert res.status_code == 200 def test_post_single_generation_by_release_id_internal_error( client, valid_headers, feature_engine): """Test post valid data and get internal error response.""" (flexmock(salessheets_generation) .should_receive('trigger_salessheets_generation_job_duplicate_upc') .and_return(response.create_fatal_response()) .once()) release_id = 1234 res = make_post_single_generation_by_release_id_request( client, valid_headers, release_id) assert res.status_code == 500 def test_post_single_generation_by_release_id_non_int_url_param( client, valid_headers): """Test post single generation invalid release id.""" invalid_release_id = 'JAJAJA' res = make_post_single_generation_by_release_id_request( client, valid_headers, invalid_release_id) assert res.status_code == 404