"""Test Replacement Generated.""" from unittest.mock import patch from moto import mock_s3 # noqa from moto import mock_dynamodb2 # noqa import boto3 from reports.period import Period from reports.replacement_generator import ReplacementGenerator from reports.report import Report class MockConfig: """Config for testing.""" ENVIRONMENT = 'test' DYNAMO_TABLE = 'foo' S3_REPORT_BUCKET = 'bar' AWS_REGION = 'us-east-1' @mock_s3 @patch('reports.replacement_generator.config', new=MockConfig()) def test_replacement_generator_get_s3_listings(): """Test creating a Replacement Generator object.""" resource = boto3.resource('s3') resource.create_bucket(Bucket=MockConfig.S3_REPORT_BUCKET) period = Period(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') mock_file_1 = resource.Object(MockConfig.S3_REPORT_BUCKET, s3_object_key) mock_file_1.put(Body=b'boo') 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' @mock_dynamodb2 @patch('reports.replacement_generator.config', new=MockConfig()) def test_replacement_generator_dynamo_record_is_missing(): """Test dynamo_record_is_missing.""" client = boto3.client('dynamodb', region_name=MockConfig.AWS_REGION) p_key = 'user_id_type' s_key = 'user-params' client.create_table( TableName=MockConfig.DYNAMO_TABLE, KeySchema=[ {'AttributeName': p_key, 'KeyType': 'HASH'}, {'AttributeName': s_key, 'KeyType': 'RANGE'}], AttributeDefinitions=[ {'AttributeName': p_key, 'AttributeType': 'S'}, {'AttributeName': s_key, 'AttributeType': 'S'}]) account_key = '123L' s3_path_template = 'schematized_files/1_label_all_all_month/user_type_id={key}/1_0' # noqa period = Period(1, 'month', 'label') report = Report( period, MockConfig.S3_REPORT_BUCKET, s3_path_template.format(key=account_key)) item = { p_key: {'S': report.user_id_type}, s_key: {'S': report.user_params}, 'file_type': {'S': report.file_type}} client.put_item(TableName=MockConfig.DYNAMO_TABLE, Item=item) table = boto3.resource( 'dynamodb', region_name=MockConfig.AWS_REGION ).Table(MockConfig.DYNAMO_TABLE) 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