"""Unit tests for CLI functions to run decider, worker and exec processes.""" from datetime import datetime from datetime import timedelta import json from unittest.mock import Mock import flexmock from freezegun import freeze_time import pytest from snowflake_views import cli @pytest.fixture() def boto3_swf_client_mock(monkeypatch): """Boto3 SWF client mock.""" swf_mock_client = Mock() boto3_mock = Mock() monkeypatch.setattr('snowflake_views.cli.boto3', boto3_mock) boto3_mock.client.return_value = swf_mock_client return swf_mock_client @freeze_time('2017-10-01') @pytest.mark.parametrize('execution_number', [0, 1, 2]) def test__is_flow_running( boto3_swf_client_mock, flow_mock, execution_number): """Test when the flow is running.""" # Mocking boto3_swf_client_mock.count_open_workflow_executions.return_value = { 'count': execution_number} workflow_id = 'test_workflow_id' # Test function call result = cli._is_flow_running(flow_mock.domain, workflow_id) # Checks count_workflow_expected_args = dict( domain=flow_mock.domain, startTimeFilter=dict(oldestDate=(datetime.now() - timedelta(days=10))), executionFilter=dict(workflowId=workflow_id) ) (boto3_swf_client_mock.count_open_workflow_executions. assert_called_once_with(**count_workflow_expected_args)) assert result is bool(execution_number) def test_execute_flow_when_is_running(monkeypatch, flow_mock): """Test execute flow function when flow is running.""" # Mocking is_running = Mock() is_running.return_value = True monkeypatch.setattr( 'snowflake_views.cli._is_flow_running', is_running) # Test function call result = cli.execute_flow(flow_mock, '{}') # Checks is_running.assert_called_once_with( flow_mock.domain, flow_mock.workflow_id_mock) assert result is None @pytest.mark.parametrize('flow_timeout', [None, 100, 600]) def test_execute_flow( monkeypatch, boto3_swf_client_mock, flow_mock, flow_timeout): """Test execute flow function when flow is runnin.""" # Mocking is_running = Mock() is_running.return_value = False monkeypatch.setattr( 'snowflake_views.cli._is_flow_running', is_running) expected_result = Mock() boto3_swf_client_mock.start_workflow_execution.return_value = ( expected_result) workflow_id_mock = flow_mock.workflow_id_mock context = '{}' start_workflow_execution_expected_params = dict( domain=flow_mock.domain, workflowId=workflow_id_mock, workflowType=dict( name=flow_mock.name, version=flow_mock.version), taskList=dict(name=flow_mock.name), input=json.dumps(context)) if flow_timeout is None: del flow_mock.timeout else: start_workflow_execution_expected_params.update( executionStartToCloseTimeout=str(flow_timeout)) flow_mock.timeout = flow_timeout # Test function call result = cli.execute_flow(flow_mock, context) # Checks is_running.assert_called_once_with( flow_mock.domain, flow_mock.workflow_id_mock) boto3_swf_client_mock.start_workflow_execution.assert_called_once_with( **start_workflow_execution_expected_params) assert result == expected_result class InfiniteLoopBreakError(Exception): """InfiniteLoopBreakError exception class.""" pass def test_run_decider(monkeypatch, flow_mock): """Test run_decider function.""" # Mocking garcon_decider_mock = Mock() decider_mock = Mock() garcon_decider_mock.DeciderWorker.return_value = decider_mock monkeypatch.setattr( 'snowflake_views.cli.decider', garcon_decider_mock) time_mock = Mock() time_mock.sleep.side_effect = InfiniteLoopBreakError monkeypatch.setattr('snowflake_views.cli.time', time_mock) # Test function call with pytest.raises(InfiniteLoopBreakError): cli.run_decider(flow_mock) # Checks garcon_decider_mock.DeciderWorker.assert_called_once_with(flow_mock) assert decider_mock.run.called def test_run_activity_worker(monkeypatch, flow_mock): """Test run_activity_worker function.""" # Mocking garcon_activity_mock = Mock() monkeypatch.setattr( 'snowflake_views.cli.activity', garcon_activity_mock) activity_worker_mock = Mock() garcon_activity_mock.ActivityWorker.return_value = activity_worker_mock # Test function call cli.run_activity_worker(flow_mock) # Checks garcon_activity_mock.ActivityWorker.assert_called_once_with(flow_mock) assert activity_worker_mock.run.called def test__parse_run_args_with_argumnts(monkeypatch): """Test _parse_run_args with passed arguments.""" # Mocking argparse_mock = Mock() monkeypatch.setattr( 'snowflake_views.cli.argparse', argparse_mock) parser_mock = Mock() argparse_mock.ArgumentParser.return_value = parser_mock args_mock = Mock() # Test function call cli._parse_run_args(args_mock) # Checks parser_mock.parse_args.assert_called_once_with(args_mock) def test__parse_run_args_with_no_argumnts(monkeypatch): """Test _parse_run_args without arguments.""" # Mocking argparse_mock = Mock() monkeypatch.setattr( 'snowflake_views.cli.argparse', argparse_mock) parser_mock = Mock() argparse_mock.ArgumentParser.return_value = parser_mock # Test function call cli._parse_run_args(None) # Checks parser_mock.parse_args.assert_called_once_with() @pytest.fixture() def garcon_commands_mock(monkeypatch): """Mock Garcon commands.""" commands_mock = { command_name: Mock() for command_name, function in cli._COMMANDS.items()} monkeypatch.setattr( 'snowflake_views.cli._COMMANDS', commands_mock) return commands_mock @pytest.mark.parametrize('garcon_command', cli._COMMANDS.keys()) def test_garcon(monkeypatch, garcon_command, garcon_commands_mock): """Test Garcon function.""" # Mocking parse_run_args_mock = Mock() run_args_mock = Mock() parse_run_args_mock.return_value = run_args_mock run_args_mock.cmd = garcon_command run_args_mock.log_level = 'info' run_args_mock.flow = 'flow_name' monkeypatch.setattr( 'snowflake_views.cli._parse_run_args', parse_run_args_mock) get_flow_class_mock = Mock() flow_class_mock = Mock() flow_mock = Mock() flow_class_mock.return_value = flow_mock get_flow_class_mock.return_value = flow_class_mock monkeypatch.setattr( 'snowflake_views.cli.Flow', get_flow_class_mock) # Test function call cli.garcon() # Checks assert garcon_commands_mock[garcon_command].called def test_execute_flows(mock_view_config, flow_mock): """Test that execute_flow is called for each flow defined.""" (flexmock(cli).should_receive('_load_views_config').and_return( mock_view_config)) (flexmock(cli).should_receive('execute_flow').with_args( flow_mock, {'name': 'view_one'}).once()) (flexmock(cli).should_receive('execute_flow').with_args( flow_mock, {'name': 'view_two'}).once()) cli.execute_flows(flow=flow_mock, context='{}')