"""Global test configuration and set of fixtures.""" import datetime import os from unittest.mock import MagicMock, Mock import pytest from activity_detector import base_config base_config.ENVIRONMENT = base_config.ENVIRONMENT_TEST # Set a AWS default region during tests so that boto.client calls don't fail os.environ['AWS_DEFAULT_REGION'] = 'us-east-1' @pytest.fixture() def flow_mock(): """Flow object fixture.""" flow = Mock( flow_domain='flow_domain', flow_name='flow_name', flow_version='flow_version') flow.workflow_id_mock = 'test_workflow_id' flow.workflow_id.return_value = flow.workflow_id_mock return flow @pytest.fixture(autouse=True) def sf_config_mock(): """Fixture returning the dict with Snowflake connection params.""" return { 'account': 'test_acc', 'role': 'test_role', 'host': 'test_host', 'warehouse': 'test_wh', 'port': 10, 'user': 'test_user', 'password': 'test_pass', 'db': 'test_db', 'schema': 'test_schema' } @pytest.fixture def date_params(): """Return dictionary of date parameters.""" return { 'start_date': '2018-01-01', 'end_date': '2018-01-07', 'growth_start_date': '2017-12-25', 'growth_end_date': '2018-01-01' } @pytest.fixture() def mock_snowflake(monkeypatch): """Mock for Snowflake database client.""" snowflake_mock = MagicMock() snowflake_mock.sessionmaker.return_value = snowflake_mock.session monkeypatch.setattr( 'snowflake_connector.snowflake_conn.sqlalchemy', snowflake_mock) monkeypatch.setattr( 'snowflake_connector.snowflake_conn._sf_default_sessionmaker', snowflake_mock.sessionmaker) sf_config = MagicMock() sf_config.return_value = {} monkeypatch.setattr( 'snowflake_connector.snowflake_conn._get_sf_config', sf_config ) monkeypatch.setattr( 'sqlalchemy.orm.session.Session.execute', snowflake_mock.execute) # set up mock for fetching data session_execute_mock = snowflake_mock.session_execute session_execute_mock.cursor.fetchall = snowflake_mock.fetchall session_execute_mock.cursor.fetchone = snowflake_mock.fetchone snowflake_mock.fetchall.return_value = ((), ) snowflake_mock.fetchone.return_value = None snowflake_mock.session.execute.return_value = session_execute_mock return snowflake_mock @pytest.fixture def mock_schedule(): """Mock schedule.""" return Mock() @pytest.fixture def mock_context(): """Mock context.""" return { 'name': 'test-name', 'date': datetime.datetime.today().strftime('%Y-%m-%d')} @pytest.fixture def mock_context_with_label_ids(): """Mock context.""" return { 'name': 'test-name', 'date': datetime.datetime.today().strftime('%Y-%m-%d'), 'label_ids': '123,456'} @pytest.fixture(autouse=True) def mock_swf_base(monkeypatch): """Mock all connections to SWF.""" mm = MagicMock(return_value=None) monkeypatch.setattr( 'boto.swf.layer2.SWFBase.__init__', mm)