"""Unit tests for base_flow.py.""" import datetime import os from unittest.mock import call, Mock, patch from freezegun import freeze_time import pytest from activity_detector.flows import base_flow @pytest.mark.parametrize('env', [None, 'dev']) def test_get_domain_dev(monkeypatch, env): """Test get_domain for dev env.""" # Mocking os_mock = Mock() os_mock.getenv.side_effect = [env, 'dev'] monkeypatch.setattr( 'activity_detector.flows.base_flow.os', os_mock) # Test function call domain = base_flow.get_domain() # Checks assert domain == 'dev' os_mock.getenv.assert_has_calls([ call('Environment'), call('DEV_SWF_DOMAIN', 'dev')]) @pytest.mark.parametrize('env', ['qa', 'prod']) def test_get_domain_not_dev(monkeypatch, env): """Test get_domain for not dev env.""" # Mocking os_mock = Mock() os_mock.getenv.side_effect = [env, 'dev'] monkeypatch.setattr( 'activity_detector.flows.base_flow.os', os_mock) # Test function call domain = base_flow.get_domain() # Checks assert domain == '{}_{}'.format(env, 'activity_detector') os_mock.getenv.assert_called_once_with('Environment') def test_base_flow(monkeypatch, flow_mock): """Test BaseFlow class __init__ method.""" # Mocking os_mock = Mock() os_mock.getenv.return_value = None monkeypatch.setattr( 'activity_detector.flows.base_flow.os', os_mock) # Test class method flow = base_flow.BaseFlow( flow_name=flow_mock.flow_name, version=flow_mock.flow_version, swf_domain=flow_mock.flow_domain) # Checks assert flow.domain == flow_mock.flow_domain assert flow.name == flow_mock.flow_name assert flow.version == flow_mock.flow_version assert flow.sentry_client is None def test_base_flow_default_domain(monkeypatch, flow_mock): """Test BaseFlow class __init__ method with default domain.""" # Mocking os_mock = Mock() os_mock.getenv.return_value = None monkeypatch.setattr( 'activity_detector.flows.base_flow.os', os_mock) get_domain_mock = Mock() domain_mock = get_domain_mock.return_value monkeypatch.setattr( 'activity_detector.flows.base_flow.get_domain', get_domain_mock) # Test class method flow = base_flow.BaseFlow( flow_name=flow_mock.flow_name, version=flow_mock.flow_version) # Checks assert flow.domain == domain_mock assert flow.name == flow_mock.flow_name assert flow.version == flow_mock.flow_version assert flow.sentry_client is None @patch.dict(os.environ, {'AWS_REGION': 'us-east-1', 'SENTRY_DSN': 'https://example.com'}) def test_base_flow_sentry_client(monkeypatch, flow_mock): """Test BaseFlow class __init__ method, check sentry client.""" # Mocking sentry_sdk_mock = Mock() sentry_client = sentry_sdk_mock monkeypatch.setattr( 'activity_detector.flows.base_flow.sentry_sdk', sentry_sdk_mock) # Test class method flow = base_flow.BaseFlow( flow_name=flow_mock.flow_name, version=flow_mock.flow_version, swf_domain=flow_mock.flow_domain) # Checks assert flow.domain == flow_mock.flow_domain assert flow.name == flow_mock.flow_name assert flow.version == flow_mock.flow_version assert flow.sentry_client == sentry_client @patch.dict(os.environ, {'AWS_REGION': 'us-east-1', 'SENTRY_DSN': 'https://example.com'}) def test_base_flow_ddtracer(monkeypatch): """Test BaseFlow class __init__ method, check Datadog Tracer works.""" # Mocking sentry_sdk_mock = Mock() monkeypatch.setattr( 'activity_detector.flows.base_flow.sentry_sdk', sentry_sdk_mock) ddtracer_mock = Mock() ddtracer_mock.enabled = True monkeypatch.setattr('activity_detector.flows.base_flow.ddtracer', ddtracer_mock) # Mock flow_name and swf_domain flow_name = 'release_approval' swf_domain = 'qa_activity_detector' # Test class method with patch('activity_detector.flows.base_flow.ddtracer.trace') as trace_mock: # Create a mock span object span_mock = Mock() # Set up the behavior of trace_mock trace_mock.return_value.__enter__.return_value = span_mock flow = base_flow.BaseFlow( flow_name=flow_name, version='1.0', swf_domain=swf_domain) # Checks assert flow.domain == swf_domain assert flow.name == flow_name assert flow.version == '1.0' assert ddtracer_mock.trace.called # Additional assertions within the context manager block trace_mock.assert_called_with('flow_init') trace_mock.assert_called_once() span_mock.set_tag.assert_any_call('flow_name', flow_name) span_mock.set_tag.assert_any_call('swf_domain', swf_domain) @patch.dict(os.environ, {'AWS_REGION': 'us-east-1', 'SENTRY_DSN': 'https://example.com'}) def test_base_flow_ddtracer_not_setup(monkeypatch): """Test BaseFlow class __init__ method, check Datadog Tracer.""" # Mocking sentry_sdk_mock = Mock() monkeypatch.setattr( 'activity_detector.flows.base_flow.sentry_sdk', sentry_sdk_mock) # Test with ddtracer disabled ddtracer_mock = Mock() ddtracer_mock.enabled = False with patch('activity_detector.flows.base_flow.ddtracer', ddtracer_mock): flow = base_flow.BaseFlow( flow_name='release_approval', version='1.0', swf_domain='qa_activity_detector') # Checks assert flow.version == '1.0' assert not ddtracer_mock.enabled assert not ddtracer_mock.trace.called @freeze_time('2017-10-01') def test_base_flow_workflow_id(monkeypatch, flow_mock): """Test BaseFlow calls workflow_id method.""" # Mocking os_mock = Mock() os_mock.getenv.return_value = None monkeypatch.setattr( 'activity_detector.flows.base_flow.os', os_mock) date = datetime.datetime.today().strftime('%Y-%m-%d') expected_workflow_id = '{flow_name}-{date}'.format( flow_name=flow_mock.flow_name, date=date) # Test class method flow = base_flow.BaseFlow( flow_name=flow_mock.flow_name, version=flow_mock.flow_version, swf_domain=flow_mock.flow_domain) workflow_id = flow.workflow_id({}) # Checks assert workflow_id == expected_workflow_id def test_base_flow_workflow_id_with_context_date(monkeypatch, flow_mock): """Test BaseFlow calls workflow_id method passed context_date.""" # Mocking os_mock = Mock() os_mock.getenv.return_value = None monkeypatch.setattr( 'activity_detector.flows.base_flow.os', os_mock) expected_workflow_id = '{flow_name}-{date}'.format( flow_name=flow_mock.flow_name, date='2017-10-01') # Test class method flow = base_flow.BaseFlow( flow_name=flow_mock.flow_name, version=flow_mock.flow_version, swf_domain=flow_mock.flow_domain) workflow_id = flow.workflow_id({'context_date': '2017-10-01'}) # Checks assert workflow_id == expected_workflow_id