import hashlib from unittest import TestCase, mock import boto3 from moto import mock_s3 from slz_appreciationengine_scrapper import config from slz_appreciationengine_scrapper.dsp.entities import DecompressedFile, ReportMeta, S3Path from slz_appreciationengine_scrapper.validation_service import ( DownloadedSizeError, NoFilesError, S3FileNotFoundError, S3FileSizeError, ValidationService, ) @mock_s3 class ValidationServiceSizeTestCase(TestCase): @mock.patch('slz_appreciationengine_scrapper.validation_service.find_schema') @mock.patch('slz_appreciationengine_scrapper.validation_service.load_schema') @mock.patch('jsonschema.validators.validator_for') def setUp(self, load_schema, find_schema, validator_for): self.client = boto3.client('s3') test_id = hashlib.sha1(self.id().encode()).hexdigest() self.configs_bucket_name = 'configs' + test_id self.destination_bucket_name = 'archive' + test_id self.destination_decompressed_bucket_name = 'decompressed' + test_id self.client.create_bucket(Bucket=self.configs_bucket_name) self.client.create_bucket(Bucket=self.destination_bucket_name) self.client.create_bucket(Bucket=self.destination_decompressed_bucket_name) self.service = ValidationService( mock.Mock(), config.Validator( rows_number=2, rate=0, ), { 'config_bucket': 'configs', 'dsp': 'apple', 'report_type': 'amStreams', 'version': 'v1_1', }, self.client, ) def tearDown(self): s3 = boto3.resource('s3') for bucket_name in (self.destination_bucket_name, self.configs_bucket_name): bucket = s3.Bucket(bucket_name) bucket.objects.all().delete() bucket.delete() def test_success(self): body = b'{}' destination_path = S3Path( self.destination_bucket_name, 'apple', 'report.txt', ) self.client.put_object(Body=body, Bucket=destination_path.bucket, Key=destination_path.key) self.service.check_size( ReportMeta( actual_size=len(body), files=[DecompressedFile('report.txt', '/tmp', '/tmp', is_subcontent=True)], expected_size=len(body), destination_path=None, destination_path_quarantine=destination_path, destination_decompressed_path_quarantine=None, destination_corrupted_path=None, ) ) def test_success_not_compressed_source_data(self): body = b'{}' destination_path = S3Path( self.destination_bucket_name, 'apple', 'report.txt.gz', ) destination_decompressed_path = S3Path( self.destination_decompressed_bucket_name, 'apple', 'report.txt', ) self.client.put_object( Body=body, Bucket=destination_decompressed_path.bucket, Key=destination_decompressed_path.key ) self.service.check_size( ReportMeta( actual_size=len(body), files=[DecompressedFile('report.txt', '/tmp', '/tmp', is_subcontent=True)], expected_size=len(body), destination_path=None, destination_path_quarantine=destination_path, destination_decompressed_path_quarantine=destination_decompressed_path, destination_corrupted_path=None, ) ) def test_actual_size_not_eq_expected_size(self): self.assertRaises( DownloadedSizeError, self.service.check_size, ReportMeta( actual_size=10, files=[DecompressedFile('report.txt', '/tmp', '/tmp', is_subcontent=True)], expected_size=11, destination_path=None, destination_path_quarantine=None, destination_decompressed_path_quarantine=None, destination_corrupted_path=None, ) ) def test_s3_file_not_eq_actual_size(self): body = b'{}' destination_path = S3Path( self.destination_bucket_name, 'apple', 'report.txt', ) self.client.put_object(Body=body, Bucket=destination_path.bucket, Key=destination_path.key) self.assertRaises( S3FileSizeError, self.service.check_size, ReportMeta( actual_size=len(body) - 1, files=[DecompressedFile('report.txt', '/tmp', '/tmp', is_subcontent=True)], expected_size=len(body) - 1, destination_path=None, destination_path_quarantine=destination_path, destination_decompressed_path_quarantine=None, destination_corrupted_path=None, ) ) def test_s3_file_not_found(self): body = b'{}' destination_path = S3Path( self.destination_bucket_name, 'apple', 'report.txt', ) self.assertRaises( S3FileNotFoundError, self.service.check_size, ReportMeta( actual_size=len(body) - 1, files=[DecompressedFile('report.txt', '/tmp', '/tmp', is_subcontent=True)], expected_size=len(body) - 1, destination_path=None, destination_path_quarantine=destination_path, destination_decompressed_path_quarantine=None, destination_corrupted_path=None, ) ) def test_no_files_to_validate(self): body = b'{}' destination_path = S3Path( self.destination_bucket_name, 'apple', 'report.txt', ) self.assertRaises( NoFilesError, self.service.check_size, ReportMeta( actual_size=len(body) - 1, files=[], expected_size=len(body) - 1, destination_path=destination_path, destination_path_quarantine=destination_path, destination_decompressed_path_quarantine=None, destination_corrupted_path=None, ) ) @mock_s3 class ValidationServiceSchemaTestCase(TestCase): @mock.patch('slz_appreciationengine_scrapper.validation_service.find_schema') @mock.patch('slz_appreciationengine_scrapper.validation_service.load_schema') @mock.patch('jsonschema.validators.validator_for') def setUp(self, load_schema, find_schema, validator_for): self.client = boto3.client('s3') test_id = hashlib.sha1(self.id().encode()).hexdigest() self.configs_bucket_name = 'configs' + test_id self.destination_bucket_name = 'archive' + test_id self.client.create_bucket(Bucket=self.configs_bucket_name) self.client.create_bucket(Bucket=self.destination_bucket_name) self.service = ValidationService( mock.Mock(), config.Validator( rows_number=2, rate=0, ), { 'config_bucket': 'configs', 'dsp': 'apple', 'report_type': 'amStreams', 'version': 'v1_1', }, self.client, ) self.destination_path = S3Path( self.destination_bucket_name, 'apple', 'report.txt', ) def tearDown(self): s3 = boto3.resource('s3') for bucket_name in (self.destination_bucket_name, self.configs_bucket_name): bucket = s3.Bucket(bucket_name) bucket.objects.all().delete() bucket.delete() @mock.patch('slz_appreciationengine_scrapper.validation_service.validate_stream') def test_valid(self, validate_stream): validate_stream.return_value = (True, {}) self.assertTrue(self.service.check_against_schema(self.destination_path.url)) @mock.patch('slz_appreciationengine_scrapper.validation_service.validate_stream') def test_invalid(self, validate_stream): validate_stream.return_value = (False, {}) excepted_result = (False, {}, None) actual_result = self.service.check_against_schema(self.destination_path.url) self.assertEqual(actual_result, excepted_result) @mock.patch('slz_appreciationengine_scrapper.validation_service.validate_stream') def test_invalid_if_exception(self, validate_stream): validate_stream.side_effect = Exception() is_valid, log, traceback = self.service.check_against_schema(self.destination_path.url) self.assertEqual((is_valid, log), (False, {})) self.assertEqual(type(traceback), str) def get_file_size_mock_side_effect(job): data = { '2019-09-06': 90, '2019-09-13': 0, '2019-09-20': 0, '2019-09-27': 0, } return data[job.report_date] @mock_s3 class ValidationServiceOneRecordTestCase(TestCase): @mock.patch('slz_appreciationengine_scrapper.validation_service.find_schema') def test_check_one_against_schema(self, find_schema): self.client = boto3.client('s3') test_id = hashlib.sha1(self.id().encode()).hexdigest() self.configs_bucket_name = 'configs' + test_id self.destination_bucket_name = 'archive' + test_id self.destination_decompressed_bucket_name = 'decompressed' + test_id self.client.create_bucket(Bucket=self.configs_bucket_name) self.client.create_bucket(Bucket=self.destination_bucket_name) self.client.create_bucket(Bucket=self.destination_decompressed_bucket_name) with mock.patch( 'slz_appreciationengine_scrapper.validation_service.load_schema' ) as load_schema: load_schema.return_value = { "definitions": {}, "$schema": "http://json-schema.org/draft-07/schema#", "$id": "http://example.com/root.json", "type": "object", "title": "test", "required": ["a", "b"], "properties": { "a": { "$id": "#/properties/a", "type": "integer", "default": "", "examples": [0, ], "pattern": "^(\\d+)$" }, "b": { "$id": "#/properties/isrc", "type": "string", "title": "ISRC", "default": "", "examples": ["USMC14704351", "812388024581"], "pattern": "^[A-Z]{2}-?\\w{3}-?\\d{2}-?\\d{5}$" } } } self.service = ValidationService( mock.Mock(), config.Validator( rows_number=2, rate=0, ), { 'config_bucket': 'configs', 'dsp': 'apple', 'report_type': 'amStreams', 'version': 'v1_1', }, self.client, ) result = self.service.check_one_against_schema({'a': 123, 'b': 'USMC14704351'}) self.assertTrue(result) result = self.service.check_one_against_schema({'a': 123, 'b': '///'}) self.assertFalse(result)