"""Unit tests for WorkstaionTimedRelease logic.""" from datetime import date, datetime from unittest.mock import ANY, MagicMock, call import botocore import pytest from dateutil.tz import tzutc from oto import response from oto import status as http_status from tests.unit.testutils import db from timed_release.config import ( TRIGGER_SFN_WS_UPDATE_METADATA_ARN, WS_SCHEDULER_GROUP, WS_SCHEDULER_ROLE_ARN, WS_SEND_NOTIFICATION_LAMBDA_ARN, ) from timed_release.connectors import scheduler from timed_release.constants import error from timed_release.constants import timed_release as timed_release_consts from timed_release.logic import timed_release_ws from timed_release.models import ows_store as ows_store_model from timed_release.models import scheduled_update as scheduled_update_model from timed_release.models import timed_release as timed_release_model @pytest.mark.parametrize( 'supported_resp, unsupported_resp, oa_supported_exists, ' 'oa_unsupported_exists, expected_status, expected_message, expected_calls', [ # Case 1: Supported and Unsupported not found -> sales_date_time None ( response.Response( status=http_status.NOT_FOUND ), response.Response( status=http_status.NOT_FOUND ), False, False, http_status.OK, {'product_id': 1, 'sales_date_time': None}, { 'supported_ws': 1, 'unsupported_ws': 1, 'oa_supported': 1, 'oa_unsupported': 1 } ), # Case 2: Supported not found -> OK if unsupported exists ( response.Response( status=http_status.NOT_FOUND ), response.Response( status=http_status.OK, message={'sales_date_time': '2026-06-09T07:00:00Z'} ), False, False, http_status.OK, {'product_id': 1, 'sales_date_time': '2026-06-09T07:00:00Z'}, { 'supported_ws': 1, 'unsupported_ws': 1, 'oa_supported': 1, 'oa_unsupported': 1 } ), # Case 3: Multiple supported sales_date_times -> 400 BAD_REQUEST ( response.Response( status=http_status.OK, message={ 'timed_releases': [ {'sales_date_time': '2025-04-08T05:00:00Z'}, {'sales_date_time': '2025-05-09T01:00:00Z'}, ] } ), None, False, False, http_status.BAD_REQUEST, error.ERROR_MESSAGE_MULTIPLE_SALES_DATE_TIME, { 'supported_ws': 1, 'unsupported_ws': 0, 'oa_supported': 1, 'oa_unsupported': 1 } ), # Case 4: Unsupported not found -> OK if supported set consistent ( response.Response( status=http_status.OK, message={ 'timed_releases': [ {'sales_date_time': '2025-04-08T05:00:00Z'}, {'sales_date_time': '2025-04-08T05:00:00Z'}, ] } ), response.Response( status=http_status.NOT_FOUND ), False, False, http_status.OK, {'product_id': 1, 'sales_date_time': '2025-04-08T05:00:00Z'}, { 'supported_ws': 1, 'unsupported_ws': 1, 'oa_supported': 1, 'oa_unsupported': 1 } ), # Case 5: Different supported vs unsupported sales_date_times -> # 400 BAD_REQUEST ( response.Response( status=http_status.OK, message={ 'timed_releases': [ {'sales_date_time': '2025-04-08T05:00:00Z'}, {'sales_date_time': '2025-04-08T05:00:00Z'}, ] } ), response.Response( status=http_status.OK, message={'sales_date_time': '2026-06-09T07:00:00Z'} ), False, False, http_status.BAD_REQUEST, error.ERROR_MESSAGE_DIFFERENT_SALES_DATE_TIME, { 'supported_ws': 1, 'unsupported_ws': 1, 'oa_supported': 1, 'oa_unsupported': 1 } ), # Case 6: Success (consistent sales_date_time) ( response.Response( status=http_status.OK, message={ 'timed_releases': [ {'sales_date_time': '2025-04-08T05:00:00Z'}, {'sales_date_time': '2025-04-08T05:00:00Z'}, ] } ), response.Response( status=http_status.OK, message={'sales_date_time': '2025-04-08T05:00:00Z'} ), False, False, http_status.OK, {'product_id': 1, 'sales_date_time': '2025-04-08T05:00:00Z'}, { 'supported_ws': 1, 'unsupported_ws': 1, 'oa_supported': 1, 'oa_unsupported': 1 } ), # Case 7: OA Supported timed release exists -> # 400 BAD_REQUEST (early return) ( None, None, True, False, http_status.BAD_REQUEST, error.ERROR_MESSAGE_SUPPORTED_TIMED_RELEASE_SET_IN_OA, { 'supported_ws': 0, 'unsupported_ws': 0, 'oa_supported': 1, 'oa_unsupported': 0 } ), # Case 8: OA Unsupported timed release exists -> # 400 BAD_REQUEST (early return) ( None, None, False, True, http_status.BAD_REQUEST, error.ERROR_MESSAGE_UNSUPPORTED_TIMED_RELEASE_SET_IN_OA, { 'supported_ws': 0, 'unsupported_ws': 0, 'oa_supported': 1, 'oa_unsupported': 1 } ) ] ) def test_get_combined_timed_release( monkeypatch, supported_resp, unsupported_resp, oa_supported_exists, oa_unsupported_exists, expected_status, expected_message, expected_calls, db_session_fixture, ): """Test for _get_combined_timed_release for all use cases.""" mock_oa_supported = MagicMock(return_value=oa_supported_exists) mock_oa_unsupported = MagicMock(return_value=oa_unsupported_exists) monkeypatch.setattr( timed_release_model, 'check_if_supported_stores_timed_release_oa_exists', mock_oa_supported ) monkeypatch.setattr( scheduled_update_model, 'check_if_unsupported_stores_schedule_oa_exists', mock_oa_unsupported ) mock_supported_ws = MagicMock(return_value=supported_resp) mock_unsupported_ws = MagicMock(return_value=unsupported_resp) monkeypatch.setattr( timed_release_model, 'get_supported_stores_timed_release_ws', mock_supported_ws ) monkeypatch.setattr( scheduled_update_model, 'get_unsupported_stores_timed_release_ws', mock_unsupported_ws ) result = timed_release_ws._get_combined_timed_release( 1, session=db_session_fixture() ) if ( expected_status == http_status.NOT_FOUND and result == http_status.NOT_FOUND ): assert result == http_status.NOT_FOUND else: assert result.status == expected_status assert result.message == expected_message assert mock_oa_supported.call_count == expected_calls['oa_supported'] assert mock_oa_unsupported.call_count == expected_calls['oa_unsupported'] assert mock_supported_ws.call_count == expected_calls['supported_ws'] assert mock_unsupported_ws.call_count == expected_calls['unsupported_ws'] @pytest.mark.parametrize( 'data, ws_tr_response, stores_response, supported_response, ' 'unsupported_response, event_schedule_response, warning_schedule_response' ' , expected_status, expected_message, expected_calls', [ # Case 1: Same sales_date_time as workstation -> # return status OK and same datetime message ( {'sales_date_time': '2025-04-08T05:00:00Z'}, response.Response( status=http_status.OK, message={ 'product_id': 12, 'sales_date_time': '2025-04-08T05:00:00Z' } ), None, None, None, None, None, http_status.OK, timed_release_consts.MESSAGE_SAME_TIMED_RELEASE_SET_IN_WORKSTATION, { 'get_stores_called': False, 'upsert_supported_tr_ws_called': False, 'upsert_unsupported_schedule_ws_called': False, 'upsert_event_schedule_called': False, 'upsert_warning_schedule_called': False } ), # Case 2: _get_combined_timed_release returns error -> return error ( {'sales_date_time': '2025-04-08T05:00:00Z'}, response.Response( status=http_status.BAD_REQUEST, message='Invalid data' ), None, None, None, None, None, http_status.BAD_REQUEST, 'Invalid data', { 'get_stores_called': False, 'upsert_supported_tr_ws_called': False, 'upsert_unsupported_schedule_ws_called': False, 'upsert_event_schedule_called': False, 'upsert_warning_schedule_called': False } ), # Case 3: get_stores fails -> should return get_stores response ( {'sales_date_time': '2025-04-08T05:00:00Z'}, response.Response( status=http_status.OK, message={ 'product_id': 12, 'sales_date_time': '2025-04-05T01:00:00Z' } ), response.Response( status=http_status.INTERNAL_ERROR, message='ows-store internal error' ), None, None, None, None, http_status.INTERNAL_ERROR, 'ows-store internal error', { 'get_stores_called': True, 'upsert_supported_tr_ws_called': False, 'upsert_unsupported_schedule_ws_called': False, 'upsert_event_schedule_called': False, 'upsert_warning_schedule_called': False } ), # Case 4: Success ( {'sales_date_time': '2025-04-08T05:00:00Z'}, response.Response( status=http_status.OK, message={ 'product_id': 12, 'sales_date_time': '2025-04-05T01:00:00Z' } ), response.Response( status=http_status.OK, message=[ { 'id': 1, 'name': 'First Store', 'supports_timed_release': True, 'status': 'active' }, { 'id': 2, 'name': 'Second store', 'supports_timed_release': False, 'status': 'active' } ] ), response.Response( status=http_status.OK, message={ 'supported_releases': [ { 'sales_date_time': '2025-04-08T05:00:00Z', 'store_id': 1, 'timing': 'timed', 'source': 'workstation' } ] } ), response.Response( status=http_status.OK, message={ 'unsupported_releases': { 'product_id': 124, 'schedule_datetime': '2025-04-08T05:00:00Z', 'schedule_time_offset': '00:30', 'process_at_datetime': '2025-04-08T04:30:00Z', 'delivery_store_ids': [2], 'scheduled_by': 'user-123', 'update': {'sale_start_date': '2025-04-08'}, 'source': 'workstation', } } ), response.Response(status=http_status.OK), response.Response(status=http_status.OK), http_status.OK, [ { 'supported_releases': [ { 'sales_date_time': '2025-04-08T05:00:00Z', 'store_id': 1, 'timing': 'timed', 'source': 'workstation' } ] }, { 'unsupported_releases': { 'product_id': 124, 'schedule_datetime': '2025-04-08T05:00:00Z', 'schedule_time_offset': '00:30', 'process_at_datetime': '2025-04-08T04:30:00Z', 'delivery_store_ids': [2], 'scheduled_by': 'user-123', 'update': {'sale_start_date': '2025-04-08'}, 'source': 'workstation', } } ], { 'get_stores_called': True, 'upsert_supported_tr_ws_called': True, 'upsert_unsupported_schedule_ws_called': True, 'upsert_event_schedule_called': True, 'upsert_warning_schedule_called': True } ), # Case 5: supported stores internal error ( {'sales_date_time': '2025-04-08T05:00:00Z'}, response.Response( status=http_status.OK, message={ 'product_id': 12, 'sales_date_time': '2025-04-05T01:00:00Z' } ), response.Response( status=http_status.OK, message=[ { 'id': 1, 'name': 'First Store', 'supports_timed_release': True, 'status': 'active' }, { 'id': 2, 'name': 'Second store', 'supports_timed_release': False, 'status': 'active' } ] ), response.Response( status=http_status.INTERNAL_ERROR, message='supported stores internal error' ), None, None, None, http_status.INTERNAL_ERROR, 'supported stores internal error', { 'get_stores_called': True, 'upsert_supported_tr_ws_called': True, 'upsert_unsupported_schedule_ws_called': False, 'upsert_event_schedule_called': False, 'upsert_warning_schedule_called': False } ), # Case 6: unsupported stores internal error ( {'sales_date_time': '2025-04-08T05:00:00Z'}, response.Response( status=http_status.OK, message={ 'product_id': 12, 'sales_date_time': '2025-04-05T01:00:00Z' } ), response.Response( status=http_status.OK, message=[ { 'id': 1, 'name': 'First Store', 'supports_timed_release': True, 'status': 'active' }, { 'id': 2, 'name': 'Second store', 'supports_timed_release': False, 'status': 'active' } ] ), response.Response( status=http_status.OK, message={ 'supported_releases': [ { 'sales_date_time': '2025-04-08T05:00:00Z', 'store_id': 1, 'timing': 'timed', 'source': 'workstation' } ] } ), response.Response( status=http_status.INTERNAL_ERROR, message='unsupported stores internal error' ), None, None, http_status.INTERNAL_ERROR, 'unsupported stores internal error', { 'get_stores_called': True, 'upsert_supported_tr_ws_called': True, 'upsert_unsupported_schedule_ws_called': True, 'upsert_event_schedule_called': False, 'upsert_warning_schedule_called': False } ), # Case 7: upsert event schedule error ( {'sales_date_time': '2025-04-08T05:00:00Z'}, response.Response( status=http_status.OK, message={ 'product_id': 12, 'sales_date_time': '2025-04-05T01:00:00Z' } ), response.Response( status=http_status.OK, message=[ { 'id': 1, 'name': 'First Store', 'supports_timed_release': True, 'status': 'active' }, { 'id': 2, 'name': 'Second store', 'supports_timed_release': False, 'status': 'active' } ] ), response.Response( status=http_status.OK, message={ 'supported_releases': [ { 'sales_date_time': '2025-04-08T05:00:00Z', 'store_id': 1, 'timing': 'timed', 'source': 'workstation' } ] } ), response.Response( status=http_status.OK, message={ 'unsupported_releases': { 'product_id': 124, 'schedule_datetime': '2025-04-08T05:00:00Z', 'schedule_time_offset': '00:30', 'process_at_datetime': '2025-04-08T04:30:00Z', 'delivery_store_ids': [2], 'scheduled_by': 'user-123', 'update': {'sale_start_date': '2025-04-08'}, 'source': 'workstation', } } ), botocore.exceptions.ClientError( {'Error': {'Code': 'NoAccessError', 'Message': 'No Access'}}, 'CreateSchedule' ), None, http_status.INTERNAL_ERROR, error.ERROR_MESSAGE_UPSERTING_WS_SCHEDULE_FAILED, { 'get_stores_called': True, 'upsert_supported_tr_ws_called': True, 'upsert_unsupported_schedule_ws_called': True, 'upsert_event_schedule_called': True, 'upsert_warning_schedule_called': False } ), # Case 8: upsert warning schedule error ( {'sales_date_time': '2025-04-08T05:00:00Z'}, response.Response( status=http_status.OK, message={ 'product_id': 12, 'sales_date_time': '2025-04-05T01:00:00Z' } ), response.Response( status=http_status.OK, message=[ { 'id': 1, 'name': 'First Store', 'supports_timed_release': True, 'status': 'active' }, { 'id': 2, 'name': 'Second store', 'supports_timed_release': False, 'status': 'active' } ] ), response.Response( status=http_status.OK, message={ 'supported_releases': [ { 'sales_date_time': '2025-04-08T05:00:00Z', 'store_id': 1, 'timing': 'timed', 'source': 'workstation' } ] } ), response.Response( status=http_status.OK, message={ 'unsupported_releases': { 'product_id': 124, 'schedule_datetime': '2025-04-08T05:00:00Z', 'schedule_time_offset': '00:30', 'process_at_datetime': '2025-04-08T04:30:00Z', 'delivery_store_ids': [2], 'scheduled_by': 'user-123', 'update': {'sale_start_date': '2025-04-08'}, 'source': 'workstation', } } ), response.Response(status=http_status.OK), botocore.exceptions.ClientError( {'Error': {'Code': 'NoAccessError', 'Message': 'No Access'}}, 'CreateSchedule' ), http_status.INTERNAL_ERROR, error.ERROR_MESSAGE_UPSERTING_WS_SCHEDULE_FAILED, { 'get_stores_called': True, 'upsert_supported_tr_ws_called': True, 'upsert_unsupported_schedule_ws_called': True, 'upsert_event_schedule_called': True, 'upsert_warning_schedule_called': True } ), # Case 9: Success with user_timezone field ( { 'sales_date_time': '2025-04-08T05:00:00Z', 'user_timezone': 'America/Los_Angeles' }, response.Response( status=http_status.OK, message={ 'product_id': 12, 'sales_date_time': '2025-04-05T01:00:00Z' } ), response.Response( status=http_status.OK, message=[ { 'id': 1, 'name': 'First Store', 'supports_timed_release': True, 'status': 'active' }, { 'id': 2, 'name': 'Second store', 'supports_timed_release': False, 'status': 'active' } ] ), response.Response( status=http_status.OK, message={ 'supported_releases': [ { 'sales_date_time': '2025-04-08T05:00:00Z', 'store_id': 1, 'timing': 'timed', 'source': 'workstation' } ] } ), response.Response( status=http_status.OK, message={ 'unsupported_releases': { 'product_id': 124, 'schedule_datetime': '2025-04-08T05:00:00Z', 'schedule_time_offset': '00:30', 'process_at_datetime': '2025-04-08T04:30:00Z', 'delivery_store_ids': [2], 'scheduled_by': 'user-123', 'update': {'sale_start_date': '2025-04-08'}, 'source': 'workstation', } } ), response.Response(status=http_status.OK), response.Response(status=http_status.OK), http_status.OK, [ { 'supported_releases': [ { 'sales_date_time': '2025-04-08T05:00:00Z', 'store_id': 1, 'timing': 'timed', 'source': 'workstation' } ] }, { 'unsupported_releases': { 'product_id': 124, 'schedule_datetime': '2025-04-08T05:00:00Z', 'schedule_time_offset': '00:30', 'process_at_datetime': '2025-04-08T04:30:00Z', 'delivery_store_ids': [2], 'scheduled_by': 'user-123', 'update': {'sale_start_date': '2025-04-08'}, 'source': 'workstation', } } ], { 'get_stores_called': True, 'upsert_supported_tr_ws_called': True, 'upsert_unsupported_schedule_ws_called': True, 'upsert_event_schedule_called': True, 'upsert_warning_schedule_called': True } ) ] ) @db.test_schema def test_upsert_timed_release_and_schedule( monkeypatch, data, ws_tr_response, stores_response, supported_response, unsupported_response, event_schedule_response, warning_schedule_response, expected_status, expected_message, expected_calls, db_session_fixture ): """Test upsert_timed_release_and_schedule for all use cases.""" product_id = 12 identity_id = 'identity-id' request_headers = { 'Orchard-Identity-Id': 1234, 'Orchard-Profile-Id': 123, 'Orchard-Profile-Type': 'OrchAdminProfile' } supported_upsert_data = [{ 'product_id': 12, 'timing': 'timed', 'store_id': 1, 'sales_date_time': '2025-04-08T05:00:00Z', 'source': 'workstation', 'user_timezone': data.get('user_timezone') }] unsupported_upsert_data = { 'delivery_store_ids': [2], 'schedule_datetime': datetime(2025, 4, 8, 5, 0, tzinfo=tzutc()), 'process_at_datetime': datetime(2025, 4, 8, 4, 30, tzinfo=tzutc()), 'process_at_offset': '00:30', 'update': { 'sale_start_date': date(2025, 4, 8) }, 'user_timezone': data.get('user_timezone') } monkeypatch.setattr( timed_release_ws, '_get_combined_timed_release', MagicMock(return_value=ws_tr_response), ) monkeypatch.setattr( ows_store_model, 'get_stores', MagicMock(return_value=stores_response), ) monkeypatch.setattr( timed_release_model, 'upsert_supported_stores_timed_release_ws', MagicMock(return_value=supported_response), ) monkeypatch.setattr( scheduled_update_model, 'upsert_unsupported_stores_schedule_ws', MagicMock(return_value=unsupported_response), ) mock_upsert_event_schedule = MagicMock(side_effect=[ event_schedule_response, warning_schedule_response ]) monkeypatch.setattr( scheduler, 'upsert_event_schedule', mock_upsert_event_schedule ) monkeypatch.setattr( scheduler, 'delete_event_schedule', MagicMock() ) result = timed_release_ws.upsert_timed_release_and_schedule( product_id, data, identity_id, request_headers, session=db_session_fixture() ) timed_release_ws._get_combined_timed_release.assert_called_once_with( product_id, ANY ) if expected_calls['get_stores_called']: ows_store_model.get_stores.assert_called_once_with(ANY) else: assert not ows_store_model.get_stores.called if expected_calls['upsert_supported_tr_ws_called']: ( timed_release_model.upsert_supported_stores_timed_release_ws. assert_called_once_with( product_id, supported_upsert_data, identity_id, ANY) ) else: assert not ( timed_release_model.upsert_supported_stores_timed_release_ws.called ) if expected_calls['upsert_unsupported_schedule_ws_called']: ( scheduled_update_model.upsert_unsupported_stores_schedule_ws. assert_called_once_with( product_id, unsupported_upsert_data, identity_id, ANY) ) else: assert not ( scheduled_update_model.upsert_unsupported_stores_schedule_ws.called ) if expected_calls['upsert_event_schedule_called']: assert mock_upsert_event_schedule.call_args_list[0] == call( schedule_name=12, scheduler_group=WS_SCHEDULER_GROUP, scheduler_role_arn=WS_SCHEDULER_ROLE_ARN, schedule_datetime='2025-04-08T04:30:00Z', description='Metadata update for Product: 12', target_arn=TRIGGER_SFN_WS_UPDATE_METADATA_ARN, payload={ 'productId': 12, 'eventType': timed_release_consts.UPSERT_SCHEDULE, 'salesDateTime': data['sales_date_time'], 'source': timed_release_consts.WORKSTATION } ) else: assert not mock_upsert_event_schedule.called if expected_calls['upsert_warning_schedule_called']: assert mock_upsert_event_schedule.call_args_list[1] == call( schedule_name='12-warning', scheduler_group=WS_SCHEDULER_GROUP, scheduler_role_arn=WS_SCHEDULER_ROLE_ARN, schedule_datetime='2025-04-08T02:30:00Z', description='Pre-metadata update warning for Product: 12', target_arn=WS_SEND_NOTIFICATION_LAMBDA_ARN, payload={ 'productId': 12, 'eventProcessOn': '2025-04-08T04:30:00Z', 'eventType': timed_release_consts.WARNING_EMAIL, 'salesDateTime': data['sales_date_time'], 'source': timed_release_consts.WORKSTATION } ) if isinstance(warning_schedule_response, botocore.exceptions.ClientError): scheduler.delete_event_schedule.assert_called_once_with( 12, WS_SCHEDULER_GROUP) assert result.status == expected_status if expected_message: if ( result.message is None and result.status == http_status.INTERNAL_ERROR ): assert result.errors['message'] == expected_message else: assert result.message == expected_message @pytest.mark.parametrize( ( 'supported_resp', 'unsupported_resp', 'oa_supported_exists', 'oa_unsupported_exists', 'expected_status', 'expected_message', 'expected_calls' ), [ # Case 1: Supported and Unsupported not found -> sales_date_time None ( response.Response( status=http_status.NOT_FOUND ), response.Response( status=http_status.NOT_FOUND ), False, False, http_status.OK, { 'product_id': 1, 'sales_date_time': None, 'user_timezone': None, 'is_oa_set_timed_release': False }, { 'supported_ws': 1, 'unsupported_ws': 1, 'oa_supported': 1, 'oa_unsupported': 1 } ), # Case 2: Supported not found -> OK, if unsupported exists with UTC ( response.Response( status=http_status.NOT_FOUND ), response.Response( status=http_status.OK, message={ 'sales_date_time': '2026-06-09T07:00:00Z', 'user_timezone': 'UTC' } ), False, False, http_status.OK, { 'product_id': 1, 'sales_date_time': '2026-06-09T07:00:00Z', 'user_timezone': 'UTC', 'is_oa_set_timed_release': False }, { 'supported_ws': 1, 'unsupported_ws': 1, 'oa_supported': 1, 'oa_unsupported': 1 } ), # Case 3: Multiple supported sales_date_times -> 400 BAD_REQUEST ( response.Response( status=http_status.OK, message={ 'timed_releases': [ { 'sales_date_time': '2025-04-08T05:00:00Z', 'user_timezone': 'UTC' }, { 'sales_date_time': '2025-05-09T01:00:00Z', 'user_timezone': 'UTC' }, ] } ), None, False, False, http_status.BAD_REQUEST, error.ERROR_MESSAGE_MULTIPLE_SALES_DATE_TIME, { 'supported_ws': 1, 'unsupported_ws': 0, 'oa_supported': 1, 'oa_unsupported': 1 } ), # Case 4: Unsupported not found -> OK, supported set consistent (UTC) ( response.Response( status=http_status.OK, message={ 'timed_releases': [ { 'sales_date_time': '2025-04-08T05:00:00Z', 'user_timezone': 'UTC' }, { 'sales_date_time': '2025-04-08T05:00:00Z', 'user_timezone': 'UTC' }, ] } ), response.Response( status=http_status.NOT_FOUND ), False, False, http_status.OK, { 'product_id': 1, 'sales_date_time': '2025-04-08T05:00:00Z', 'user_timezone': 'UTC', 'is_oa_set_timed_release': False }, { 'supported_ws': 1, 'unsupported_ws': 1, 'oa_supported': 1, 'oa_unsupported': 1 } ), # Case 5: Different supported vs unsupported sales_date_times -> # 400 BAD_REQUEST ( response.Response( status=http_status.OK, message={ 'timed_releases': [ { 'sales_date_time': '2025-04-08T05:00:00Z', 'user_timezone': 'UTC' }, { 'sales_date_time': '2025-04-08T05:00:00Z', 'user_timezone': 'UTC' }, ] } ), response.Response( status=http_status.OK, message={ 'sales_date_time': '2026-06-09T07:00:00Z', 'user_timezone': 'UTC' } ), False, False, http_status.BAD_REQUEST, error.ERROR_MESSAGE_DIFFERENT_SALES_DATE_TIME, { 'supported_ws': 1, 'unsupported_ws': 1, 'oa_supported': 1, 'oa_unsupported': 1 } ), # Case 6: Success (consistent sales_date_time with UTC default) ( response.Response( status=http_status.OK, message={ 'timed_releases': [ { 'sales_date_time': '2025-04-08T05:00:00Z', 'user_timezone': 'UTC' }, { 'sales_date_time': '2025-04-08T05:00:00Z', 'user_timezone': 'UTC' }, ] } ), response.Response( status=http_status.OK, message={ 'sales_date_time': '2025-04-08T05:00:00Z', 'user_timezone': 'UTC' } ), False, False, http_status.OK, { 'product_id': 1, 'sales_date_time': '2025-04-08T05:00:00Z', 'user_timezone': 'UTC', 'is_oa_set_timed_release': False }, { 'supported_ws': 1, 'unsupported_ws': 1, 'oa_supported': 1, 'oa_unsupported': 1 } ), # Case 7: OA Supported timed release exists -> # return 200 with data indicating oa set supported # stores timed release and error message ( None, None, True, False, http_status.OK, { 'product_id': 1, 'sales_date_time': None, 'is_oa_set_timed_release': True, 'message': ( error.ERROR_MESSAGE_SUPPORTED_TIMED_RELEASE_SET_IN_OA ) }, { 'supported_ws': 0, 'unsupported_ws': 0, 'oa_supported': 1, 'oa_unsupported': 0 } ), # Case 8: OA Unsupported timed release exists -> # return 200 with data indicating oa set unsupported # stores timed release and error message ( None, None, False, True, http_status.OK, { 'product_id': 1, 'sales_date_time': None, 'is_oa_set_timed_release': True, 'message': ( error.ERROR_MESSAGE_UNSUPPORTED_TIMED_RELEASE_SET_IN_OA ) }, { 'supported_ws': 0, 'unsupported_ws': 0, 'oa_supported': 1, 'oa_unsupported': 1 } ), # Case 9: user_timezone set to America/New_York (supported stores) ( response.Response( status=http_status.OK, message={ 'timed_releases': [ { 'sales_date_time': '2025-04-08T05:00:00Z', 'user_timezone': 'America/New_York' }, { 'sales_date_time': '2025-04-08T05:00:00Z', 'user_timezone': 'America/New_York' }, ] } ), response.Response( status=http_status.NOT_FOUND ), False, False, http_status.OK, { 'product_id': 1, 'sales_date_time': '2025-04-08T05:00:00Z', 'user_timezone': 'America/New_York', 'is_oa_set_timed_release': False }, { 'supported_ws': 1, 'unsupported_ws': 1, 'oa_supported': 1, 'oa_unsupported': 1 } ), # Case 10: user_timezone set to UTC (unsupported stores) ( response.Response( status=http_status.NOT_FOUND ), response.Response( status=http_status.OK, message={ 'sales_date_time': '2026-06-09T07:00:00Z', 'user_timezone': 'UTC' } ), False, False, http_status.OK, { 'product_id': 1, 'sales_date_time': '2026-06-09T07:00:00Z', 'user_timezone': 'UTC', 'is_oa_set_timed_release': False }, { 'supported_ws': 1, 'unsupported_ws': 1, 'oa_supported': 1, 'oa_unsupported': 1 } ), # Case 11: user_timezone consistent across supported and unsupported ( response.Response( status=http_status.OK, message={ 'timed_releases': [ { 'sales_date_time': '2025-04-08T05:00:00Z', 'user_timezone': 'America/Los_Angeles' }, ] } ), response.Response( status=http_status.OK, message={ 'sales_date_time': '2025-04-08T05:00:00Z', 'user_timezone': 'America/Los_Angeles' } ), False, False, http_status.OK, { 'product_id': 1, 'sales_date_time': '2025-04-08T05:00:00Z', 'user_timezone': 'America/Los_Angeles', 'is_oa_set_timed_release': False }, { 'supported_ws': 1, 'unsupported_ws': 1, 'oa_supported': 1, 'oa_unsupported': 1 } ), # Case 12: Multiple user_timezone values -> ERROR ( response.Response( status=http_status.OK, message={ 'timed_releases': [ { 'sales_date_time': '2025-04-08T05:00:00Z', 'user_timezone': 'America/New_York' }, ] } ), response.Response( status=http_status.OK, message={ 'sales_date_time': '2025-04-08T05:00:00Z', 'user_timezone': 'America/Los_Angeles' } ), False, False, http_status.BAD_REQUEST, error.ERROR_MESSAGE_MULTIPLE_USER_TIMEZONE, { 'supported_ws': 1, 'unsupported_ws': 1, 'oa_supported': 1, 'oa_unsupported': 1 } ) ] ) def test_get_product_timed_release_for_ws( monkeypatch, supported_resp, unsupported_resp, oa_supported_exists, oa_unsupported_exists, expected_status, expected_message, expected_calls, db_session_fixture, ): """Test for _get_combined_timed_release for all use cases.""" mock_oa_supported = MagicMock(return_value=oa_supported_exists) mock_oa_unsupported = MagicMock(return_value=oa_unsupported_exists) monkeypatch.setattr( timed_release_model, 'check_if_supported_stores_timed_release_oa_exists', mock_oa_supported ) monkeypatch.setattr( scheduled_update_model, 'check_if_unsupported_stores_schedule_oa_exists', mock_oa_unsupported ) mock_supported_ws = MagicMock(return_value=supported_resp) mock_unsupported_ws = MagicMock(return_value=unsupported_resp) monkeypatch.setattr( timed_release_model, 'get_supported_stores_timed_release_ws', mock_supported_ws ) monkeypatch.setattr( scheduled_update_model, 'get_unsupported_stores_timed_release_ws', mock_unsupported_ws ) result = timed_release_ws.get_product_timed_release_for_ws(1) if ( expected_status == http_status.NOT_FOUND and result == http_status.NOT_FOUND ): assert result == http_status.NOT_FOUND else: assert result.status == expected_status assert result.message == expected_message assert mock_oa_supported.call_count == expected_calls['oa_supported'] assert mock_oa_unsupported.call_count == expected_calls['oa_unsupported'] assert mock_supported_ws.call_count == expected_calls['supported_ws'] assert mock_unsupported_ws.call_count == expected_calls['unsupported_ws'] @pytest.mark.parametrize( ( 'timed_release_expected_status', 'scheduled_update_ws_response', 'scheduled_update_expected_status', 'delete_event_schedule_side_effect', 'upsert_event_schedule_response', 'expected_status' ), [ # Case 1: Both deleted successfully → 200 ( http_status.OK, response.Response( status=http_status.OK, message={ 'product_id': 2, 'schedule_datetime': '2030-04-03T10:00:00Z', 'schedule_time_offset': '00:30', 'process_at_datetime': '2030-04-03T09:30:00Z' } ), http_status.OK, [ response.Response(status=http_status.OK), response.Response(status=http_status.OK) ], None, http_status.OK ), # Case 2: Timed release deleted successfully, # scheduled update not found → 200 ( http_status.OK, response.Response(status=http_status.NOT_FOUND), http_status.NOT_FOUND, [ None, None ], None, http_status.OK ), # Case 3: Timed release not found, # scheduled update deleted successfully → 200 ( http_status.NOT_FOUND, response.Response( status=http_status.OK, message={ 'product_id': 2, 'schedule_datetime': '2030-04-03T10:00:00Z', 'schedule_time_offset': '00:30', 'process_at_datetime': '2030-04-03T09:30:00Z' } ), http_status.OK, [ response.Response(status=http_status.OK), response.Response(status=http_status.OK) ], None, http_status.OK ), # Case 4: Both not found → 404 ( http_status.NOT_FOUND, response.Response(status=http_status.NOT_FOUND), http_status.NOT_FOUND, [ None, None ], None, http_status.NOT_FOUND ), # Case 5: Timed release delete fails (500) → # skip scheduled update and return 500 ( http_status.INTERNAL_ERROR, response.Response(status=http_status.NOT_FOUND), http_status.NOT_FOUND, [ None, None ], None, http_status.INTERNAL_ERROR ), # Case 6: Scheduled update delete fails (500) → # after timed release success, return 500 ( http_status.OK, response.Response( status=http_status.OK, message={ 'product_id': 2, 'schedule_datetime': '2030-04-03T10:00:00Z', 'schedule_time_offset': '00:30', 'process_at_datetime': '2030-04-03T09:30:00Z' } ), http_status.INTERNAL_ERROR, [ None, None ], None, http_status.INTERNAL_ERROR ), # Case 7: Delete event schedule failed -> return 500 ( http_status.OK, response.Response( status=http_status.OK, message={ 'product_id': 2, 'schedule_datetime': '2030-04-03T10:00:00Z', 'schedule_time_offset': '00:30', 'process_at_datetime': '2030-04-03T09:30:00Z' } ), http_status.OK, [ botocore.exceptions.ClientError( {'Error': {'Code': 'error', 'Message': 'error'}}, 'DeleteSchedule' ), None, ], None, http_status.INTERNAL_ERROR ), # Case 8: Delete warning schedule failed -> return 500 ( http_status.OK, response.Response( status=http_status.OK, message={ 'product_id': 2, 'schedule_datetime': '2030-04-03T10:00:00Z', 'schedule_time_offset': '00:30', 'process_at_datetime': '2030-04-03T09:30:00Z' } ), http_status.OK, [ response.Response(status=http_status.OK), botocore.exceptions.ClientError( {'Error': {'Code': 'error', 'Message': 'error'}}, 'DeleteSchedule' ) ], response.Response(status=http_status.OK), http_status.INTERNAL_ERROR ) ] ) @db.test_schema def test_delete_timed_releases_and_scheduled_updates_ws( mocker, timed_release_expected_status, scheduled_update_ws_response, scheduled_update_expected_status, delete_event_schedule_side_effect, upsert_event_schedule_response, expected_status ): """Test delete_timed_releases_and_scheduled_updates_ws all use cases.""" product_id = 2 mocker.patch.object( timed_release_model, 'delete_timed_release_ws', return_value=response.Response(status=timed_release_expected_status) ) mocker.patch.object( scheduled_update_model, 'get_scheduled_update_ws', return_value=scheduled_update_ws_response ) mocker.patch.object( scheduled_update_model, 'delete_scheduled_update_ws', return_value=response.Response(status=scheduled_update_expected_status) ) mock_delete_event_schedule = MagicMock( side_effect=delete_event_schedule_side_effect ) mocker.patch.object( scheduler, 'delete_event_schedule', mock_delete_event_schedule ) if upsert_event_schedule_response: mock_upsert_event_schedule = mocker.patch.object( scheduler, 'upsert_event_schedule', return_value=upsert_event_schedule_response ) result = timed_release_ws.delete_timed_releases_and_scheduled_updates_ws( product_id) timed_release_model.delete_timed_release_ws.assert_called_once_with( product_id, ANY) if timed_release_expected_status == http_status.INTERNAL_ERROR: scheduled_update_model.get_scheduled_update_ws.assert_not_called() scheduled_update_model.delete_scheduled_update_ws.assert_not_called() else: ( scheduled_update_model.get_scheduled_update_ws. assert_called_once_with(product_id, ANY) ) if scheduled_update_ws_response.status == http_status.OK: ( scheduled_update_model.delete_scheduled_update_ws. assert_called_once_with(product_id, ANY) ) if scheduled_update_expected_status == http_status.OK: assert mock_delete_event_schedule.call_args_list[0] == call( 2, WS_SCHEDULER_GROUP) if ( delete_event_schedule_side_effect[1] is not None or not isinstance( delete_event_schedule_side_effect[0], botocore.exceptions.ClientError ) ): assert mock_delete_event_schedule.call_args_list[1] == call( '2-warning', WS_SCHEDULER_GROUP) if isinstance( delete_event_schedule_side_effect[1], botocore.exceptions.ClientError ): mock_upsert_event_schedule.assert_called_once_with( schedule_name=2, scheduler_group=WS_SCHEDULER_GROUP, scheduler_role_arn=WS_SCHEDULER_ROLE_ARN, schedule_datetime='2030-04-03T09:30:00Z', description='Metadata update for Product: 2', target_arn=TRIGGER_SFN_WS_UPDATE_METADATA_ARN, payload={ 'productId': 2, 'eventType': timed_release_consts.UPSERT_SCHEDULE } ) else: assert not mock_delete_event_schedule.called assert result.status == expected_status