"""Unit tests for SWF related utils and helpers of Feed Ingestion workflows.""" import datetime from unittest.mock import MagicMock, call from unittest.mock import patch from feed_ingestion.util.aws import swf @patch.object(swf, 'boto3') def test_load_closed_executions(mock_boto3): # Mock the response from the swf_client mock_response = { 'executionInfos': [ { 'execution': { 'workflowId': 'workflow_id_1', 'runId': 'run_id_1', }, 'closeStatus': 'COMPLETED', 'startTimestamp': '2021-01-01T00:00:00Z' }, { 'execution': { 'workflowId': 'workflow_id_2', 'runId': 'run_id_2', }, 'closeStatus': 'COMPLETED', 'startTimestamp': '2021-01-02T00:00:00Z' }, { 'execution': { 'workflowId': 'workflow_id_3', 'runId': 'run_id_3', }, 'closeStatus': 'ACTIVE', 'startTimestamp': '2021-01-02T00:00:00Z' }, ], 'nextPageToken': None } mock_swf_client = mock_boto3.client.return_value mock_swf_client.list_closed_workflow_executions.return_value = ( mock_response) from_date = datetime.datetime(2021, 1, 1) to_date = datetime.datetime(2021, 1, 2) swf_domain = 'test_domain' execution_type = 'test_execution_type' executions = swf.list_closed_swf_executions( from_date, to_date, swf_domain, execution_type) assert len(executions) == 2 assert executions == mock_response['executionInfos'][0:2] @patch.object(swf, 'boto3') def test_load_all_execution_events(mock_boto3): mock_response_1 = { 'events': [ {'eventId': 1, 'eventType': 'WorkflowExecutionStarted'}, {'eventId': 2, 'eventType': 'DecisionTaskScheduled'}, ], 'nextPageToken': 'page2' } mock_response_2 = { 'events': [ {'eventId': 3, 'eventType': 'DecisionTaskStarted'}, ], 'nextPageToken': None } mock_swf_client = mock_boto3.client.return_value mock_swf_client.get_workflow_execution_history.side_effect = [ mock_response_1, mock_response_2] swf_domain = 'test_domain' workflow_id = 'test_workflow_id' run_id = 'test_run_id' events = swf.load_all_execution_events(swf_domain, workflow_id, run_id) assert len(events) == 3 assert events == [ {'eventId': 1, 'eventType': 'WorkflowExecutionStarted'}, {'eventId': 2, 'eventType': 'DecisionTaskScheduled'}, {'eventId': 3, 'eventType': 'DecisionTaskStarted'}, ] assert mock_swf_client.get_workflow_execution_history.call_args_list == [ call(domain='test_domain', execution={'workflowId': 'test_workflow_id', 'runId': 'test_run_id'}, maximumPageSize=1000), call(domain='test_domain', execution={'workflowId': 'test_workflow_id', 'runId': 'test_run_id'}, maximumPageSize=1000, nextPageToken='page2'), ] def test_tasks_from_events(): events = [ {'eventId': 1, 'eventType': 'WorkflowExecutionStarted'}, {'eventId': 2, 'eventType': 'DecisionTaskScheduled'}, {'eventId': 3, 'eventType': 'DecisionTaskStarted'}, {'eventId': 4, 'eventType': 'ActivityTaskScheduled', 'activityTaskScheduledEventAttributes': {'activityType': {'name': 'task1'}}}, {'eventId': 5, 'eventType': 'ActivityTaskStarted', 'activityTaskStartedEventAttributes': {'scheduledEventId': 4}, 'eventTimestamp': datetime.datetime(2021, 1, 1, 0, 0, 0)}, {'eventId': 6, 'eventType': 'ActivityTaskCompleted', 'activityTaskCompletedEventAttributes': {'startedEventId': 5}, 'eventTimestamp': datetime.datetime(2021, 1, 1, 0, 0, 3)} ] tasks = swf.tasks_from_events(events) expected_tasks = [ { 'closed': events[5], 'duration': datetime.timedelta(seconds=3), 'name': 'task1', 'open': events[4], 'scheduled': events[3], } ] assert tasks == expected_tasks @patch('feed_ingestion.util.aws.swf.boto3') @patch('feed_ingestion.util.aws.swf.datetime.date') def test_count_running_workflows_by_type(mock_today, mock_boto3): """Test count_running_workflows_by_type.""" mock_client = MagicMock() mock_client.count_open_workflow_executions.return_value = { 'count': 1 } mock_boto3.client.return_value = mock_client mock_today.today.return_value = datetime.datetime.strptime( '2015-10-01', '%Y-%m-%d') mock_today.side_effect = lambda *args, **kw: datetime.date(*args, **kw) resp = swf.count_running_workflows_by_type( 'test_domain', 'test_type', 3) assert resp == 1 mock_client.count_open_workflow_executions.assert_any_call( domain='test_domain', startTimeFilter={ 'oldestDate': datetime.datetime(2015, 9, 28, 0, 0) }, typeFilter={ 'name': 'test_type' } ) mock_client.count_open_workflow_executions.return_value = { 'count': 0 } resp = swf.count_running_workflows_by_type( 'test_domain', 'test_type', 3) assert resp == 0 @patch('feed_ingestion.util.aws.swf.count_running_workflows_by_type') @patch('feed_ingestion.util.aws.swf.getconf') def test_has_nonconcurrent_workflows_running( mock_getconf, mock_count_running_workflow): """Test has_nonconcurrent_workflows_running.""" mock_getconf.return_value = { 'workflow_types': ['workflow_typeA', 'workflow_typeB'] } flow = MagicMock() flow.name = 'workflow_typeA' flow.domain = 'swf_feed_ingestion' mock_count_running_workflow.return_value = 1 result = swf.has_nonconcurrent_workflows_running(flow) assert result flow.name = 'workflow_typeC' result = swf.has_nonconcurrent_workflows_running(flow) assert not result flow.name = 'workflow_typeA' mock_count_running_workflow.return_value = 0 result = swf.has_nonconcurrent_workflows_running(flow) assert not result @patch('feed_ingestion.util.aws.swf.boto3') @patch('feed_ingestion.util.aws.swf.datetime.date') def test_get_names_of_running_workflows_by_type(mock_today, mock_boto3): """Test get_names_of_running_workflows_by_type.""" mock_client = MagicMock() mock_client.list_open_workflow_executions.return_value = { 'executionInfos': [{ 'execution': { 'workflowId': 'flow_name_feed_ingestion-2020-04-23', 'runId': '123'}}]} mock_boto3.client.return_value = mock_client mock_today.today.return_value = datetime.datetime.strptime( '2015-10-01', '%Y-%m-%d') mock_today.side_effect = lambda *args, **kw: datetime.date(*args, **kw) resp = swf.get_names_of_running_workflows_by_type( 'test_domain', 'test_type', 3) assert resp == ['flow_name_feed_ingestion-2020-04-23'] @patch('feed_ingestion.util.aws.swf.boto3') @patch('feed_ingestion.util.aws.swf.datetime.date') def test_get_names_of_running_workflows_by_type_when_no_executions( mock_today, mock_boto3): """Test get_names_of_running_workflows_by_type.""" mock_client = MagicMock() mock_client.list_open_workflow_executions.return_value = { 'executionInfos': []} mock_boto3.client.return_value = mock_client mock_today.today.return_value = datetime.datetime.strptime( '2015-10-01', '%Y-%m-%d') mock_today.side_effect = lambda *args, **kw: datetime.date(*args, **kw) resp = swf.get_names_of_running_workflows_by_type( 'test_domain', 'test_type', 3) assert resp == [] def test_generate_execution_console_url(): """Test generate_execution_console_url.""" result = swf.generate_execution_console_url( swf_domain='dev_efedorov', workflow_id='insta_reels_feed_ingestion-2022-09-01', run_id='22rRp0KrrMPHan7Il/aUOilNiwE6R7LlRCPJTRQQTIGfc=' ) expected_result = ('https://us-east-1.console.aws.amazon.com/swf/v2/home' '?region=us-east-1#/domains/dev_efedorov/executions' '/insta_reels_feed_ingestion-2022-09-01' '/22rRp0KrrMPHan7Il%2FaUOilNiwE6R7LlRCPJTRQQTIGfc%3D') assert result == expected_result