"""Tests for orders model.""" import datetime from opensearchpy import OpenSearchException import pytest from src.connectors.opensearch import os_client from src.constants import fields from src.constants import http_statuses from src.constants import opensearch as os_consts from src.models import jobs from tests.test_utils import db @pytest.mark.parametrize( 'value, expected_value', ( ('v', '"v"'), (True, 'true'), (1, 1), ) ) def test_get_query(value, expected_value): """Test _get_query function.""" field = 'f' assert jobs._get_query(field, value) == ( os_consts.SOURCE_COMMAND_TEMPLATE.format( field=field, value=expected_value)) def test_get_body(): """Test _get_body function.""" source = 'test source' queries = {'foo': 'bar'} assert jobs._get_body(source, queries) == { 'script': {'source': source}, 'query': queries } @pytest.mark.parametrize( 'commands', ({'c_k': 'c_v'}, {'c_k1': 'c_v1', 'c_k2': 'c_v2'})) @pytest.mark.parametrize( 'conditions, expected_query', ( ( [[{'term': {'t_k': 't_v'}}]], {'bool': {'should': [ {'bool': {'must': [ {'term': {'t_k': 't_v'}}]}}]}}, ), ( [[{'term': {'t_k': 't_v'}}]], {'bool': {'should': [ {'bool': {'must': [ {'term': {'t_k': 't_v'}}]}}]}}, ), ( [[{'term': {'t_k1': 't_v1'}}, {'term': {'t_k2': 't_v2'}}]], {'bool': {'should': [{'bool': {'must': [ {'term': {'t_k1': 't_v1'}}, {'term': {'t_k2': 't_v2'}}]}}]}}, ), ( [ [{'term': {'t1_k1': 't1_v1'}}, {'term': {'t1_k2': 't1_v2'}}], [{'term': {'t2_k1': 't2_v1'}}, {'term': {'t2_k2': 't2_v2'}}], ], {'bool': {'should': [ {'bool': {'must': [ {'term': {'t1_k1': 't1_v1'}}, {'term': {'t1_k2': 't1_v2'}}]}}, {'bool': {'must': [ {'term': {'t2_k1': 't2_v1'}}, {'term': {'t2_k2': 't2_v2'}}]}} ]}}, ), ) ) def test_update_by_query(commands, conditions, expected_query, mocker): """Test _update_by_query function.""" update_result = 'test result' body_result = 'test body' query_result = 'test query' mocked_client = mocker.patch('src.models.jobs.os_client') mocked_update = mocked_client.update_by_query mocked_update.return_value = update_result mocked_get_body = mocker.patch('src.models.jobs._get_body') mocked_get_body.return_value = body_result mocked_get_query = mocker.patch('src.models.jobs._get_query') mocked_get_query.return_value = query_result kwargs = dict(foo='bar') assert jobs._update_by_query( commands, conditions, **kwargs) == update_result assert mocked_update.call_count == 1 assert mocked_update.call_args[0] == () assert mocked_update.call_args[1] == { 'index': os_consts.OS_INDEX_VO_DETAIL, 'body': body_result, **kwargs, } query_count = len(commands) assert mocked_get_query.call_count == query_count assert {a[0][0]: a[0][1] for a in mocked_get_query.call_args_list} == commands assert mocked_get_body.call_count == 1 assert mocked_get_body.call_args[1] == {} assert mocked_get_body.call_args[0] == ( query_result * query_count, expected_query) @pytest.mark.parametrize('term_conditions', (None, [])) @pytest.mark.parametrize('commands', (None, {})) def test_update_by_query_raises(commands, term_conditions, mocker): """Test _update_by_query function raises exception on empty args.""" mocked_client = mocker.patch('src.models.jobs.os_client') mocked_update = mocked_client.update_by_query mocked_get_body = mocker.patch('src.models.jobs._get_body') mocked_get_query = mocker.patch('src.models.jobs._get_query') with pytest.raises(ValueError): jobs._update_by_query(commands, term_conditions) assert mocked_update.call_count == 0 assert mocked_get_body.call_count == 0 assert mocked_get_query.call_count == 0 @pytest.mark.parametrize( 'record, expected_result', ( ( { fields.ENCODING_QUEUE_DETAIL_ID: 101, fields.STORE_ID: 10, fields.ORDER_ID: 1000, fields.UPC: 10000, }, '101' ), ( { fields.ENCODING_QUEUE_DETAIL_ID: 102, fields.STORE_ID: 10, fields.ORDER_ID: 1000, fields.UPC: 10000, }, '102' ), ) ) def test_get_id(record, expected_result): """Test _get_id function returns encoding_queue_detail_id as a string.""" assert jobs._get_id(record) == expected_result @pytest.mark.parametrize( 'record', ( {}, {fields.ORDER_ID: 100}, {fields.ORDER_ID: 100, fields.STORE_ID: 1}, {fields.STORE_ID: 1, fields.UPC: 1000}, {fields.ORDER_ID: 100, fields.UPC: 1000}, {fields.ORDER_ID: 100, fields.UPC: 1000, fields.STORE_ID: 'str'}, ) ) def test_get_id_raises(record): """Test _get_id function raises on missing/invalid record keys.""" with pytest.raises((KeyError, ValueError)): jobs._get_id(record) @pytest.mark.parametrize( 'input_record, expected_result', ( # All fields. ( { fields.DMS_MASTER_MASTER_ID: 1, fields.ENCODING_STARTED: 2, fields.ENCODING_ENDED: 3, fields.DELIVERY_STARTED: 4, fields.DELIVERY_ENDED: 5, fields.CREATED_AT: 6, fields.META_UPDATE: False, 'test': 7 }, { fields.STORE_ID: 1, fields.ENCODING_STARTED: '2', fields.ENCODING_ENDED: '3', fields.DELIVERY_STARTED: '4', fields.DELIVERY_ENDED: '5', fields.CREATED_AT: '6', fields.META_UPDATE: False, 'test': 7, }, ), # All fields with metadata update, that leads to is_duplicate = False. ( { fields.DMS_MASTER_MASTER_ID: 1, fields.ENCODING_STARTED: 2, fields.ENCODING_ENDED: 3, fields.DELIVERY_STARTED: 4, fields.DELIVERY_ENDED: 5, fields.CREATED_AT: 6, fields.META_UPDATE: True, 'test': 7 }, { fields.STORE_ID: 1, fields.ENCODING_STARTED: '2', fields.ENCODING_ENDED: '3', fields.DELIVERY_STARTED: '4', fields.DELIVERY_ENDED: '5', fields.CREATED_AT: '6', fields.META_UPDATE: True, fields.IS_DUPLICATE: False, 'test': 7, }, ), # Without an optional created_at field. ( { fields.DMS_MASTER_MASTER_ID: 1, fields.ENCODING_STARTED: 2, fields.ENCODING_ENDED: 3, fields.DELIVERY_STARTED: 4, fields.DELIVERY_ENDED: 5, fields.META_UPDATE: False, 'test': 7 }, { fields.STORE_ID: 1, fields.ENCODING_STARTED: '2', fields.ENCODING_ENDED: '3', fields.DELIVERY_STARTED: '4', fields.DELIVERY_ENDED: '5', fields.META_UPDATE: False, 'test': 7, }, ) ) ) def test_extract_from_db_record(input_record, expected_result, mocker): """Test _extract_from_db_record function.""" mocked_timezone = mocker.patch('src.models.jobs.timezone') mocked_convert_to_utc_str = mocked_timezone.convert_to_utc_str mocked_convert_to_utc_str.side_effect = lambda x: str(x) result = jobs._extract_from_db_record(input_record) assert result == expected_result @pytest.mark.parametrize( 'data, selected_fields, expected_result', ( ({'f1': 1, 'f2': 2, 'f3': 3}, {'f1', 'f3'}, {'f1': 1, 'f3': 3}), ({'f1': 1}, {'f1', 'f3'}, {'f1': 1}), ({'f2': 1}, {'f1', 'f3'}, {}), ({}, {'f1', 'f3'}, {}), ({'f1': 1}, {}, {}), ) ) def test_get_only_necessary_fields(data, selected_fields, expected_result): """Test _get_only_necessary_fields function.""" assert jobs._get_only_necessary_fields(data, selected_fields) == ( expected_result) def test_update_priority(mocker): """Test update_priority function.""" order_id = 100 priority = 1 mocked_update = mocker.patch('src.models.jobs._update_by_query') jobs.update_priority(order_id, priority) assert mocked_update.call_count == 1 assert mocked_update.call_args[0] == () assert mocked_update.call_args[1] == { 'commands': {fields.PRIORITY: priority}, 'conditions': [[{'term': {fields.ORDER_ID: order_id}}]], 'conflicts': 'proceed', 'wait_for_completion': False, } @pytest.mark.parametrize( 'duplicates, non_duplicates, expected_duplicates_condition, ' 'expected_non_duplicates_condition', ( ( # Record content does not matter. list(range(2)), list(range(2, 5)), [ { 'ids': { 'type': os_consts.DOCUMENT_TYPE, 'values': [0, 1], # based on mocked _get_id function. } }, ], [ { 'ids': { 'type': os_consts.DOCUMENT_TYPE, 'values': [2, 3, 4], } }, ], ), ) ) def test_set_duplicates_bulk( duplicates, non_duplicates, expected_duplicates_condition, expected_non_duplicates_condition, mocker): """Test setting is_duplicate flag operation.""" mocked_update_by_query = mocker.patch('src.models.jobs._update_by_query') mocked_get_id = mocker.patch('src.models.jobs._get_id') mocked_get_id.side_effect = list(range(len(duplicates + non_duplicates))) jobs.set_duplicates_bulk( duplicates=duplicates, non_duplicates=non_duplicates) assert mocked_get_id.call_args_list == [ ((r_id,), {},) for r_id in duplicates + non_duplicates ] assert mocked_update_by_query.call_count == 2 assert mocked_update_by_query.call_args_list == [ ((), { 'commands': {fields.IS_DUPLICATE: True}, 'conditions': [expected_duplicates_condition], 'conflicts': 'proceed', }), ((), { 'commands': {fields.IS_DUPLICATE: False}, 'conditions': [expected_non_duplicates_condition], 'conflicts': 'proceed', }), ] def test_set_duplicates_bulk_empty_records(mocker): """Test setting is_duplicate flag operation on empty input.""" mocked_update_by_query = mocker.patch('src.models.jobs._update_by_query') mocked_get_id = mocker.patch('src.models.jobs._get_id') jobs.set_duplicates_bulk(duplicates=[], non_duplicates=[]) assert mocked_get_id.call_count == 0 assert mocked_update_by_query.call_count == 0 @pytest.mark.parametrize( 'selected_fields', (os_consts.VO_DETAIL_ADD, os_consts.VO_DETAIL_UPDATE) ) def test_prepare_records(selected_fields, mocker): """Test preparing records for ES insertion.""" records = [mocker.Mock(1), mocker.Mock(2)] expected_extract_args_list = [((r,), {}) for r in records] extract_return_values = [{'id': 1}, {'id': 2}] expected_get_id_args_list = [((r,), {}) for r in extract_return_values] get_id_return_values = [d['id'] for d in extract_return_values] expected_get_fields_args_list = [ ((kw, selected_fields), {}) for kw in extract_return_values] get_fields_return_values = [{'f': v} for v in get_id_return_values] expected_result = [ (doc_id, body) for doc_id, body in zip( get_id_return_values, get_fields_return_values)] mocked_extract_from_db_record = mocker.patch( 'src.models.jobs._extract_from_db_record') mocked_extract_from_db_record.side_effect = extract_return_values mocked_get_id = mocker.patch( 'src.models.jobs._get_id') mocked_get_id.side_effect = get_id_return_values mocked_get_only_necessary_fields = mocker.patch( 'src.models.jobs._get_only_necessary_fields') mocked_get_only_necessary_fields.side_effect = get_fields_return_values assert jobs._prepare_records(records, selected_fields) == expected_result assert mocked_extract_from_db_record.call_args_list == ( expected_extract_args_list) assert mocked_get_id.call_args_list == expected_get_id_args_list assert mocked_get_only_necessary_fields.call_args_list == ( expected_get_fields_args_list) def test_create_bulk(mocker): """Test bulk ES records creation.""" records = mocker.Mock() records_tuples = [('id-1', {'f1': 'v1'}), ('id-2', {'f1': 'v2'})] expected_actions_list = [ { '_op_type': 'index', '_index': os_consts.OS_INDEX_VO_DETAIL, '_type': os_consts.DOCUMENT_TYPE, '_id': 'id-1', '_source': {'f1': 'v1'}, }, { '_op_type': 'index', '_index': os_consts.OS_INDEX_VO_DETAIL, '_type': os_consts.DOCUMENT_TYPE, '_id': 'id-2', '_source': {'f1': 'v2'}, }, ] mocked_prepare_records = mocker.patch('src.models.jobs._prepare_records') expected_prepare_records_args = mocker.call( records, os_consts.VO_DETAIL_ADD) mocked_prepare_records.side_effect = [records_tuples] mocked_helpers_bulk = mocker.patch('src.models.jobs.os_helpers.bulk') mocked_helpers_bulk.return_value = (2, 0) jobs.create_bulk(records) assert mocked_prepare_records.call_count == 1 assert mocked_prepare_records.call_args == expected_prepare_records_args assert mocked_helpers_bulk.call_count == 1 assert mocked_helpers_bulk.call_args[0] == () assert mocked_helpers_bulk.call_args[1]['client'] == os_client assert list(mocked_helpers_bulk.call_args[1]['actions']) == ( expected_actions_list) assert mocked_helpers_bulk.call_args[1]['max_retries'] == ( os_consts.MAX_BULK_RETRIES) def test_update_bulk(mocker): """Test bulk ES records update.""" records = [{'f1': 'r0_v1', 'f2': 'r0_v2'}, {'f1': 'r1_v1', 'f2': 'r1_v2'}] records_tuples = [('id-1', {'f1': 'v1'}), ('id-2', {'f1': 'v2'})] successful_updates = 1 errors = [ # Stripped down version of the full error dict. {'update': { '_id': records_tuples[0][0], 'status': http_statuses.NOT_FOUND}}, ] expected_actions_list = [ { '_op_type': 'update', '_index': os_consts.OS_INDEX_VO_DETAIL, '_type': os_consts.DOCUMENT_TYPE, '_id': 'id-1', '_source': {'doc': {'f1': 'v1'}}, }, { '_op_type': 'update', '_index': os_consts.OS_INDEX_VO_DETAIL, '_type': os_consts.DOCUMENT_TYPE, '_id': 'id-2', '_source': {'doc': {'f1': 'v2'}}, }, ] mocked_prepare_records = mocker.patch('src.models.jobs._prepare_records') expected_prepare_records_args = mocker.call( records, os_consts.VO_DETAIL_UPDATE) mocked_prepare_records.side_effect = [records_tuples] mocked_helpers_bulk = mocker.patch('src.models.jobs.os_helpers.bulk') mocked_helpers_bulk.return_value = successful_updates, errors assert jobs.update_bulk(records) == (successful_updates, [records[0]]) assert mocked_prepare_records.call_count == 1 assert mocked_prepare_records.call_args == expected_prepare_records_args assert mocked_helpers_bulk.call_count == 1 assert mocked_helpers_bulk.call_args[0] == () assert mocked_helpers_bulk.call_args[1]['client'] == os_client assert list(mocked_helpers_bulk.call_args[1]['actions']) == ( expected_actions_list) assert mocked_helpers_bulk.call_args[1]['raise_on_error'] is False assert mocked_helpers_bulk.call_args[1][ 'max_retries'] == os_consts.MAX_BULK_RETRIES def test_update_bulk_raises(mocker): """Test bulk ES records update raises on unexpected error.""" records = [{'f1': 'r0_v1', 'f2': 'r0_v2'}, {'f1': 'r1_v1', 'f2': 'r1_v2'}] records_tuples = [('id-1', {'f1': 'v1'}), ('id-2', {'f1': 'v2'})] errors = [ # Stripped down version of the full error dict. {'update': { '_id': records_tuples[0][0], 'status': http_statuses.NOT_FOUND}}, {'update': { '_id': records_tuples[0][1], 'status': 418}} ] expected_actions_list = [ { '_op_type': 'update', '_index': os_consts.OS_INDEX_VO_DETAIL, '_type': os_consts.DOCUMENT_TYPE, '_id': doc_id, '_source': {'doc': prepared_record}, } for doc_id, prepared_record in records_tuples] mocked_prepare_records = mocker.patch('src.models.jobs._prepare_records') expected_prepare_records_args = mocker.call( records, os_consts.VO_DETAIL_UPDATE) mocked_prepare_records.side_effect = [records_tuples] mocked_helpers_bulk = mocker.patch('src.models.jobs.os_helpers.bulk') mocked_helpers_bulk.return_value = 0, errors with pytest.raises( OpenSearchException, match=str(errors[1]['update'])): jobs.update_bulk(records) assert mocked_prepare_records.call_count == 1 assert mocked_prepare_records.call_args == expected_prepare_records_args assert mocked_helpers_bulk.call_count == 1 assert mocked_helpers_bulk.call_args[0] == () assert mocked_helpers_bulk.call_args[1]['client'] == os_client assert list(mocked_helpers_bulk.call_args[1]['actions']) == ( expected_actions_list) assert mocked_helpers_bulk.call_args[1]['raise_on_error'] is False assert mocked_helpers_bulk.call_args[1][ 'max_retries'] == os_consts.MAX_BULK_RETRIES def test_set_store_spec(mocker): """Test set_store_spec function.""" order_type = 't' store_id = 2 encoding = 'E' delivery = 'D' mocked_update = mocker.patch('src.models.jobs._update_by_query') jobs.set_store_spec(order_type, store_id, encoding, delivery) assert mocked_update.call_count == 1 assert mocked_update.call_args[0] == () assert mocked_update.call_args[1] == { 'commands': {fields.ENCODING: encoding, fields.DELIVERY: delivery}, 'conditions': [ [ {'term': {fields.STORE_ID: store_id}}, {'term': {fields.ORDER_TYPE: order_type}} ] ], 'conflicts': 'proceed', 'wait_for_completion': False, } @pytest.mark.parametrize( 'input_ids, db_query_result_mock, ' 'expected_query, expected_result', ( ( (), ((),), '', {}, ), ( (10000, 20000), ( # Note: using a subset of fields in the mock db query result. ( (fields.DELIVERY_STARTED, datetime.datetime( 2018, 1, 1, 0, 1, 0)), (fields.ENCODING_QUEUE_DETAIL_ID, 10000), (fields.ENCODING_QUEUE_ID, 100), ), ( (fields.DELIVERY_STARTED, datetime.datetime( 2018, 1, 1, 0, 2, 0)), (fields.ENCODING_QUEUE_DETAIL_ID, 20000), (fields.ENCODING_QUEUE_ID, 200), ), ), """SELECT delivery_started, delivery_ended, encoding_started, encoding_ended, encoding_queue_detail_id, encoding_queue_id, error_log, status, dms_master_master_id, upc FROM encoding_queue_detail WHERE encoding_queue_detail_id IN :{:s};""".format( fields.SQL_ID_LIST), { 10000: { 'delivery_started': datetime.datetime(2018, 1, 1, 0, 1), fields.ENCODING_QUEUE_ID: 100, fields.ENCODING_QUEUE_DETAIL_ID: 10000, }, 20000: { 'delivery_started': datetime.datetime(2018, 1, 1, 0, 2), fields.ENCODING_QUEUE_ID: 200, fields.ENCODING_QUEUE_DETAIL_ID: 20000, }, }, ), )) def test_get_encoding_queue_detail_bulk( input_ids, db_query_result_mock, expected_query, expected_result, patch_session): """Test bulk queries for dms_delivery_spec records.""" mocked_session = patch_session( 'jobs', 'direct_delivery', db_query_result_mock) mocked_execute = mocked_session.execute result = jobs.get_encoding_queue_detail_bulk(input_ids) assert result == expected_result if not input_ids: assert mocked_execute.call_count == 0 return assert mocked_execute.call_count == 1 expected_query_text = db.remove_repeating_spaces(expected_query) actual_query_text = db.remove_repeating_spaces( mocked_execute.call_args[0][0]) assert actual_query_text == expected_query_text assert mocked_execute.call_args[0][1] == {fields.SQL_ID_LIST: list(set(input_ids))}