"""Tests for the tasks of the s32sf flow.""" import json from unittest.mock import call from unittest.mock import MagicMock from garcon_contrib.dynamo_feed_status import garcon_feed_status import pytest from snowflake_etl.conf import config from snowflake_etl.flows.s32sf import tasks @pytest.fixture def mock_activity(mocker): """Convenient single location for mocking garcon.Activity.""" return mocker.patch('garcon.activity.Activity') @pytest.fixture def mock_configuration(): """Mock test configuration for snowflake.""" return { 'user': 'SNOWFLAKE_USER', 'password': 'SNOWFLAKE_PASSWORD', 'account': 'SNOWFLAKE_ACCOUNT', 'role': 'SNOWFLAKE_ROLE', 'warehouse': 'SNOWFLAKE_WAREHOUSE', 'db': 'SNOWFLAKE_DATABASE', 'schema': 'production' } @pytest.fixture def mock_sf_params(): """Mock test configuration for snowflake.""" return { 'role': 'SNOWFLAKE_ROLE', 'warehouse': 'SNOWFLAKE_WAREHOUSE', 'db': 'SNOWFLAKE_DATABASE', 'schema': 'production' } @pytest.fixture def mock_sf_credentials(): """Mock test configuration for snowflake.""" return { 'user': 'SNOWFLAKE_USER', 'password': 'SNOWFLAKE_PASSWORD', 'account': 'SNOWFLAKE_ACCOUNT', } def test_bootstrap(mock_activity, mock_sf_params): """Test bootstrap.""" tasks.bootstrap( activity=mock_activity, sfdb_params=mock_sf_params, table='dim_artist', s3_path_data='nopath', s3_path_schema='nopath', load_strategy='snapshot') def test_bootstrap_bad_load_strategy(mock_activity, mock_sf_params): """Test bootstrap with bad load strategy.""" with pytest.raises(AssertionError): tasks.bootstrap( activity=mock_activity, sfdb_params=mock_sf_params, table='dim_artist', s3_path_data='nopath', s3_path_schema='nopath', load_strategy='nostrategy') @pytest.fixture def mysql_schema_json(): """MySQL-style table schema export with two columns.""" return '''{"table": "vendor_contract", "schema": "art_relations", "columns":[ {"numeric_scale": 0, "column_name": "id", "character_octet_length": null, "column_default": null, "numeric_precision": 10, "numeric_precision_radix": null, "data_type": "int", "is_nullable": "NO", "datetime_precision": null, "character_maximum_length": null}, {"numeric_scale": null, "column_name": "cont_start", "character_octet_length": null, "column_default": null, "numeric_precision": null, "numeric_precision_radix": null, "data_type": "date", "is_nullable": "YES", "datetime_precision": null, "character_maximum_length": null}, {"numeric_scale": null, "column_name": "cont_version", "character_octet_length": 150, "column_default": null, "numeric_precision": null, "numeric_precision_radix": null, "data_type": "varchar", "is_nullable": "YES", "datetime_precision": null, "character_maximum_length": 50}]}''' @pytest.fixture def snowflake_tbl_desc(): """Fixture of Snowflakes `desc table ` query. (name, type, kind, null, default, primary key, unique key, check, expresssion, comment) """ return (('id', 'NUMBER(38,0)', 'COLUMN', 'N', 'NULL'), ('name', 'VARCHAR(2000)', 'COLUMN', 'Y', 'NULL')) def test_target_schema( mocker, mock_activity, snowflake_tbl_desc, mock_configuration): """Test comparison of Redshift and Snowflake schema.""" with mocker.patch( 'garcon_contrib.snowflake.garcon_snowflake.execute_with_py_conn', return_value=dict(results=snowflake_tbl_desc)): ctx = tasks.target_schema( activity=mock_activity, sfdb_params=mock_configuration, table='dim_artist') schema = ctx.get('target_schema') assert schema == ('id NUMBER(38,0) NOT NULL', 'name VARCHAR(2000)') def test_create_destination_table_from_mysql_source( mocker, mock_activity, mysql_schema_json, mock_sf_params, mock_sf_credentials, mock_configuration): """Test create of table from MySQL source schema if it already exist.""" mocker.patch.dict(config.SF_CREDENTIALS, mock_sf_credentials) with mocker.patch( 'garcon_contrib.snowflake.garcon_snowflake.table_exists', return_value=False): execute_mock = MagicMock() with mocker.patch( 'garcon_contrib.snowflake.garcon_snowflake.execute_with_py_conn', new=execute_mock): tasks.create_destination_table( activity=mock_activity, sfdb_params=mock_sf_params, table='vendor_contract', db_type='mysql', source_schema=json.loads(mysql_schema_json)) execute_mock.assert_called_with( 'CREATE TABLE ' 'SNOWFLAKE_DATABASE.production.vendor_contract ' '(ID NUMBER(12,0) NOT NULL, CONT_START DATE, ' 'CONT_VERSION VARCHAR(50))', sf_config=mock_configuration) def test_validate_compatibility_valid_mysql(mock_activity, mysql_schema_json): """Test validation of schema compatibility (valid).""" tasks.validate_compatibility( activity=mock_activity, db_type='mysql', source_schema=json.loads(mysql_schema_json), target_schema=( 'ID NUMBER(12,0) NOT NULL', 'CONT_START DATE', 'CONT_VERSION VARCHAR(50)')) def test_validate_compatibility_invalid_mysql( mock_activity, mysql_schema_json): """Test validation of schema compatibility (invalid).""" with pytest.raises(AssertionError): tasks.validate_compatibility( activity=mock_activity, db_type='mysql', source_schema=json.loads(mysql_schema_json), target_schema=( 'ID NUMBER(38,0) NOT NULL', 'cont_st DATE', 'CONT_VERSION VARCHAR(50)')) def test_load_staging_table_mysql( mock_activity, mocker, mock_sf_params, mock_sf_credentials, mock_configuration): """Test load staging table.""" mocker.patch.dict(config.SF_CREDENTIALS, mock_sf_credentials) execute_mock = MagicMock() with mocker.patch( 'garcon_contrib.snowflake.garcon_snowflake.execute_with_py_conn', new=execute_mock): cfg_mock = MagicMock( side_effect=[ {'aws': {'access_key': 'nokey', 'access_secret': 'nosecret'}}]) with mocker.patch( 'snowflake_etl.flows.s32sf.tasks.cfg.getconf', new=cfg_mock): tasks.load_staging_table( mock_activity, mock_sf_params, 'vendor_contract', 'no_s3_path_data', None, db_type='mysql') truncate_call = call( 'TRUNCATE TABLE ' 'SNOWFLAKE_DATABASE.production.stg_vendor_contract', sf_config=mock_configuration) copy_call = call( 'COPY INTO SNOWFLAKE_DATABASE.production.stg_vendor_contract ' "FROM no_s3_path_data CREDENTIALS=(\n AWS_KEY_ID='nokey'\n " "AWS_SECRET_KEY='nosecret')\nFILE_FORMAT = " "(FIELD_DELIMITER='\\t' RECORD_DELIMITER='\\n' " "COMPRESSION='GZIP' NULL_IF=('NULL', '0000-00-00', " "'2009-11-00', '2012-02-00', '2010-11-00',\n '2008-11-00', " "'2011-03-00', '0000-00-00 00:00:00', '2014-00-00')" '\nEMPTY_FIELD_AS_NULL=FALSE TRIM_SPACE=TRUE\n)\n', sf_config=mock_configuration) calls = [truncate_call, copy_call] execute_mock.assert_has_calls(calls, any_order=False) def test_swap_snowflake_tables_snapshot( mock_activity, mocker, mock_sf_params, mock_sf_credentials, mock_configuration): """Test swap staging and destination Snowflake tables for snapshot.""" mocker.patch.dict(config.SF_CREDENTIALS, mock_sf_credentials) execute_mock = MagicMock() with mocker.patch( 'garcon_contrib.snowflake.garcon_snowflake.execute_with_py_conn', new=execute_mock): tasks.swap_snowflake_tables( mock_activity, mock_sf_params, 'dim_artist', 'snapshot') execute_mock.assert_called_with( 'ALTER TABLE SNOWFLAKE_DATABASE.production.dim_artist ' 'SWAP WITH SNOWFLAKE_DATABASE.production.stg_dim_artist\n', sf_config=mock_configuration) mock_activity.assert_has_calls([call.logger.info( 'Swapping destination table SNOWFLAKE_DATABASE.production.' 'dim_artist with staging table SNOWFLAKE_DATABASE.production.' 'stg_dim_artist\n>> Executing query ALTER TABLE ' 'SNOWFLAKE_DATABASE.production.dim_artist SWAP WITH ' 'SNOWFLAKE_DATABASE.production.stg_dim_artist\n')]) def test_swap_snowflake_tables_incremental( mock_activity, mocker, mock_configuration): """Test swap staging and destination Snowflake tables for incremental.""" execute_mock = MagicMock() with mocker.patch( 'garcon_contrib.snowflake.garcon_snowflake.execute_with_py_conn', new=execute_mock): tasks.swap_snowflake_tables( mock_activity, mock_configuration, 'fact_analytics', 'incremental') assert execute_mock.call_count == 0 mock_activity.assert_has_calls([call.logger.info( 'Swap staging and permanent SF tables: skipping for strategy: ' 'incremental')]) def test_drop_staging_table_incremental( mock_activity, mocker, mock_configuration): """Test drop staging table (was skipped if incremental strategy).""" execute_mock = MagicMock() with mocker.patch( 'garcon_contrib.snowflake.garcon_snowflake.execute_with_py_conn', new=execute_mock): tasks.drop_staging_table( mock_activity, mock_configuration, 'fact_analytics', 'incremental') assert execute_mock.call_count == 0 mock_activity.assert_has_calls([ call.logger.info( 'drop_staging_table: skipping for strategy: incremental')]) def test_drop_staging_table_snapshot( mock_activity, mocker, mock_sf_params, mock_sf_credentials, mock_configuration): """Test drop staging table (was executed if snapshot strategy).""" mocker.patch.dict(config.SF_CREDENTIALS, mock_sf_credentials) execute_mock = MagicMock() with mocker.patch( 'garcon_contrib.snowflake.garcon_snowflake.execute_with_py_conn', new=execute_mock): tasks.drop_staging_table( mock_activity, mock_sf_params, 'dim_artist', 'snapshot') execute_mock.assert_called_with( 'DROP TABLE SNOWFLAKE_DATABASE.production.stg_dim_artist', sf_config=mock_configuration) mock_activity.assert_has_calls([ call.logger.info( 'dropping old staging table: SNOWFLAKE_DATABASE.production.' 'dim_artist\n>> Executing query DROP TABLE SNOWFLAKE_' 'DATABASE.production.stg_dim_artist')]) def test_insert_into_destination_table_incremental( mock_activity, mocker, mock_sf_params, mock_sf_credentials, mock_configuration): """Test insert (copy) from staging to destination.""" mocker.patch.dict(config.SF_CREDENTIALS, mock_sf_credentials) execute_mock = MagicMock() with mocker.patch( 'garcon_contrib.snowflake.garcon_snowflake.execute_with_py_conn', new=execute_mock): tasks.insert_into_destination_table( mock_activity, mock_sf_params, 'fact_analytics', 'incremental') execute_mock.assert_called_with( 'INSERT INTO SNOWFLAKE_DATABASE.production.fact_analytics\n ' 'SELECT * FROM SNOWFLAKE_DATABASE.production.stg_fact_analytics\n', sf_config=mock_configuration) def test_insert_into_destination_table_snapshot( mock_activity, mocker, mock_configuration): """Test insert from staging to destination not executed for a snapshot.""" execute_mock = MagicMock() with mocker.patch( 'garcon_contrib.snowflake.garcon_snowflake.execute_with_py_conn', new=execute_mock): tasks.insert_into_destination_table( mock_activity, mock_configuration, 'dim_artist', 'snapshot') assert execute_mock.call_count == 0 def test_ingest_set_ingestion_status(mock_activity, monkeypatch): """Test feed status setting into DynamoDB.""" set_status_mock = MagicMock() monkeypatch.setattr( garcon_feed_status, 'set_overall_status', set_status_mock) tasks.set_ingestion_status( mock_activity, '2016-10-07', 'testing feed', garcon_feed_status.STATUS_INGESTED) set_status_mock.assert_called_with( 'testing feed', '2016-10-07', garcon_feed_status.STATUS_INGESTED) def test_ingest_set_ingestion_status_without_date(mock_activity, monkeypatch): """Test if feed status setting into DynamoDB is skipped.""" set_status_mock = MagicMock() monkeypatch.setattr( garcon_feed_status, 'set_overall_status', set_status_mock) tasks.set_ingestion_status( mock_activity, None, 'testing feed', garcon_feed_status.STATUS_INGESTED) assert not set_status_mock.called def test_ingest_set_ingestion_status_without_name(mock_activity, monkeypatch): """Test if feed status setting into DynamoDB is skipped.""" set_status_mock = MagicMock() monkeypatch.setattr( garcon_feed_status, 'set_overall_status', set_status_mock) tasks.set_ingestion_status( mock_activity, '2016-10-07', None, garcon_feed_status.STATUS_INGESTED) assert not set_status_mock.called