import hashlib from unittest import TestCase, mock import boto3 from moto import mock_s3 from slz_config import DSPSettings, DSPSpecificConfig from slz_downloader import config from slz_downloader.dsp.entities import DecompressedFile, ReportMeta, S3Path from slz_downloader.dsp.spotify import SpotifyClient from slz_downloader.entities import Job from slz_downloader.validation_service import ( CalculatedFileSizeError, DownloadedSizeError, NoFilesError, S3FileNotFoundError, S3FileSizeError, ValidationService, ) @mock_s3 class ValidationServiceSizeTestCase(TestCase): @mock.patch('slz_downloader.validation_service.find_schema') @mock.patch('slz_downloader.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, threshold_min_file_size=0, enable_historical_validation=False, ), { '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_downloader.validation_service.find_schema') @mock.patch('slz_downloader.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, threshold_min_file_size=0, enable_historical_validation=False, ), { '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_downloader.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_downloader.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_downloader.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] class ValidateCalcFilesizeTestCase(TestCase): @mock.patch('slz_downloader.validation_service.find_schema') @mock.patch('slz_downloader.validation_service.load_schema') @mock.patch('jsonschema.validators.validator_for') def setUp(self, load_schema, find_schema, validator_for): self.service = ValidationService( mock.Mock(), mock.Mock(), { 'config_bucket': 'configs', "dsp": "spotify", "extension": "json", "subtype": None, "is_reprocessing": False, "config_key": "batch-run-config.json", "uow_id": "spotify-20191004-sme-streams-v2", "unit_of_work_id": 0, "licensor": "sme", "report_type": "streams", "report_date": "2019-10-04", "version": "v2", "contexts": ["AD", "AE", "AF", "AG"] }, mock.Mock(), ) self.context = "AD" self.content_status_service = mock.Mock() self.content_status_service.get_unit_of_work.return_value = Job( uow_id='spotify-20190906-sme-streams-v2', unit_of_work_id=0, dsp='spotify', report_type='streams', subtype=None, version='v2', report_date='2019-09-06', licensor='sme', extension='json', config_bucket='configs', context='AD', context_params=None, job_id=None ) self.content_status_service.get_file_size.return_value = 0 self.dsp_specific_settings = DSPSettings.parse( { "uow_active_days_limit": 14, "implementations": [{ "rule": "*/*", "label": "spotify" }], "validation_thresholds": { "low": 10, "high": 20 }, "default_priority_hrs": 24 } ) def test_get_correct_historical_data(self): dsp_specific_config = DSPSettings.empty() dsp_specific_config.validation_thresholds = {'low': 10, 'high': 20} self.service.check_calc_size( mock.Mock(), self.content_status_service, dsp_specific_config, SpotifyClient(mock.Mock()), self.context ) self.content_status_service.get_file_size.assert_has_calls( [ mock.call( Job( uow_id='spotify-20190927-sme-streams-v2', unit_of_work_id=0, dsp='spotify', report_type='streams', subtype=None, version='v2', report_date='2019-09-27', licensor='sme', extension='json', config_bucket='configs', context='AD', context_params=None, job_id=None ), ), mock.call( Job( uow_id='spotify-20190920-sme-streams-v2', unit_of_work_id=0, dsp='spotify', report_type='streams', subtype=None, version='v2', report_date='2019-09-20', licensor='sme', extension='json', config_bucket='configs', context='AD', context_params=None, job_id=None ), ), mock.call( Job( uow_id='spotify-20190913-sme-streams-v2', unit_of_work_id=0, dsp='spotify', report_type='streams', subtype=None, version='v2', report_date='2019-09-13', licensor='sme', extension='json', config_bucket='configs', context='AD', context_params=None, job_id=None ), ), mock.call( Job( uow_id='spotify-20190906-sme-streams-v2', unit_of_work_id=0, dsp='spotify', report_type='streams', subtype=None, version='v2', report_date='2019-09-06', licensor='sme', extension='json', config_bucket='configs', context='AD', context_params=None, job_id=None ), ), ] ) def test_get_correct_raising_exception(self): self.content_status_service.get_file_size = mock.Mock( side_effect=get_file_size_mock_side_effect ) report_meta = mock.Mock() report_meta.actual_size = 200 self.assertRaises( CalculatedFileSizeError, self.service.check_calc_size, report_meta, self.content_status_service, self.dsp_specific_settings, SpotifyClient(mock.Mock()), self.context ) def test_successful_validation(self): self.content_status_service.get_file_size = mock.Mock( side_effect=get_file_size_mock_side_effect ) report_meta = mock.Mock() report_meta.actual_size = 100 result = self.service.check_calc_size( report_meta, self.content_status_service, self.dsp_specific_settings, SpotifyClient(mock.Mock()), self.context ) self.assertIsNone(result) @mock_s3 class ValidationServiceOneRecordTestCase(TestCase): @mock.patch('slz_downloader.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_downloader.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, threshold_min_file_size=0, enable_historical_validation=False, ), { '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)