"""Test Replacement Generated.""" from unittest.mock import patch class MockConfig: """Config for testing.""" ENVIRONMENT = 'test' DYNAMO_TABLE = 'foo' S3_REPORT_BUCKET = 'bar' AWS_REGION = 'us-east-1' PRINT_AFTER_MODULO = 250 @patch('reports.replacement_generator.config', new=MockConfig()) def test_replacement_generator_get_s3_listings(s3, dynamodb): """Test creating a Replacement Generator object.""" from reports.period_reports import PeriodReports from reports.replacement_generator import ReplacementGenerator s3.create_bucket(Bucket=MockConfig.S3_REPORT_BUCKET) period = PeriodReports(1, 'month', 'label') account_key = '123L' s3_object_key = '{prefix}user_id_type={key}/{filename}'.format( prefix=period.prefix, key=account_key, filename='bar.baz' ) s3.put_object(Bucket=MockConfig.S3_REPORT_BUCKET, Body=b'boo', Key=s3_object_key) actual = ReplacementGenerator.get_s3_listings(period) assert len(actual) == 1 actual_report = actual[0] assert actual_report.period_ids == '1' expected_s3_path = 's3://{bucket}/{s3_object_key}'.format( bucket=MockConfig.S3_REPORT_BUCKET, s3_object_key=s3_object_key ) assert actual_report.s3_path == expected_s3_path assert actual_report.user_id_type == account_key assert actual_report.user_params == '1__all__AVRO__en_US' @patch('reports.replacement_generator.config', new=MockConfig()) def test_replacement_generator_dynamo_record_is_missing(s3, dynamodb): """Test dynamo_record_is_missing.""" from reports.period_reports import PeriodReports from reports.replacement_generator import ReplacementGenerator from reports.report import Report p_key = 'user_id_type' s_key = 'user_params' dynamodb.create_table( TableName=MockConfig.DYNAMO_TABLE, BillingMode='PAY_PER_REQUEST', KeySchema=[ {'AttributeName': p_key, 'KeyType': 'HASH'}, {'AttributeName': s_key, 'KeyType': 'RANGE'}], AttributeDefinitions=[ {'AttributeName': p_key, 'AttributeType': 'S'}, {'AttributeName': s_key, 'AttributeType': 'S'}] ) s3_path_template = ( 'schematized_files/' '1_label_all_all_month/' 'user_type_id={key}/1_0' ) period = PeriodReports(1, 'month', 'label') report = Report( period, MockConfig.S3_REPORT_BUCKET, s3_path_template.format(key='123L') ) table = dynamodb.Table(MockConfig.DYNAMO_TABLE) table.put_item(Item={ p_key: report.user_id_type, s_key: report.user_params, 'file_type': report.file_type }) is_missing = ReplacementGenerator.dynamo_record_is_missing(report, table) assert not is_missing fake_report = Report( period, MockConfig.S3_REPORT_BUCKET, s3_path_template.format(key='321L')) result = ReplacementGenerator.dynamo_record_is_missing(fake_report, table) assert result