"""Edge case tests for abacus_file_upload schemas.""" import pytest from marshmallow import ValidationError from abacus_file_upload.schemas import ( FileUploadConfigPostSchema, InitiateUploadRequestSchema, ) class TestFilenameEdgeCases: """Additional edge case tests for filename validation.""" def test_filename_very_long(self): """Test filename with 255 characters (typical filesystem limit).""" schema = InitiateUploadRequestSchema() long_filename = 'a' * 250 + '.csv' # 254 chars total data = { 'upload_type': 'adjustments', 'filename': long_filename, 'file_size_bytes': 1024000, 'md5sum': '5d41402abc4b2a76b9719d911017c592', } result = schema.load(data) assert result['filename'] == long_filename def test_filename_extremely_long(self): """Test filename exceeding reasonable limits (1000+ chars).""" schema = InitiateUploadRequestSchema() very_long_filename = 'a' * 1000 + '.csv' data = { 'upload_type': 'adjustments', 'filename': very_long_filename, 'file_size_bytes': 1024000, 'md5sum': '5d41402abc4b2a76b9719d911017c592', } # Should either accept or reject based on schema constraints # This documents current behavior result = schema.load(data) assert len(result['filename']) > 1000 def test_filename_with_unicode(self): """Test filename with unicode characters.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': 'test_文件_файл.csv', 'file_size_bytes': 1024000, 'md5sum': '5d41402abc4b2a76b9719d911017c592', } result = schema.load(data) assert result['filename'] == 'test_文件_файл.csv' def test_filename_with_emojis(self): """Test filename with emoji characters.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': 'test_😀_file.csv', 'file_size_bytes': 1024000, 'md5sum': '5d41402abc4b2a76b9719d911017c592', } result = schema.load(data) assert '😀' in result['filename'] def test_filename_hidden_file(self): """Test filename starting with dot (hidden file).""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': '.gitignore', 'file_size_bytes': 1024000, 'md5sum': '5d41402abc4b2a76b9719d911017c592', } result = schema.load(data) assert result['filename'] == '.gitignore' def test_filename_multiple_dots(self): """Test filename with multiple dots.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': 'test.backup.2025.csv', 'file_size_bytes': 1024000, 'md5sum': '5d41402abc4b2a76b9719d911017c592', } result = schema.load(data) assert result['filename'] == 'test.backup.2025.csv' def test_filename_only_extension(self): """Test filename that is only an extension.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': '.csv', 'file_size_bytes': 1024000, 'md5sum': '5d41402abc4b2a76b9719d911017c592', } result = schema.load(data) assert result['filename'] == '.csv' def test_filename_no_extension(self): """Test filename without an extension.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': 'README', 'file_size_bytes': 1024000, 'md5sum': '5d41402abc4b2a76b9719d911017c592', } result = schema.load(data) assert result['filename'] == 'README' def test_filename_leading_trailing_spaces(self): """Test filename with leading/trailing spaces.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': ' test.csv ', 'file_size_bytes': 1024000, 'md5sum': '5d41402abc4b2a76b9719d911017c592', } # May be trimmed or preserved depending on schema result = schema.load(data) assert 'test.csv' in result['filename'] class TestMetadataEdgeCases: """Edge case tests for metadata validation.""" def test_metadata_deeply_nested(self): """Test deeply nested metadata object.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': 'test.csv', 'file_size_bytes': 1024000, 'md5sum': '5d41402abc4b2a76b9719d911017c592', 'metadata': { 'level1': { 'level2': { 'level3': { 'level4': { 'deep_value': 123, } } } } }, } result = schema.load(data) assert ( result['metadata']['level1']['level2']['level3']['level4']['deep_value'] == 123 ) def test_metadata_very_large(self): """Test metadata with many keys.""" schema = InitiateUploadRequestSchema() large_metadata = {f'key_{i}': f'value_{i}' for i in range(100)} data = { 'upload_type': 'adjustments', 'filename': 'test.csv', 'file_size_bytes': 1024000, 'md5sum': '5d41402abc4b2a76b9719d911017c592', 'metadata': large_metadata, } result = schema.load(data) assert len(result['metadata']) == 100 def test_metadata_with_special_characters(self): """Test metadata with special characters in keys/values.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': 'test.csv', 'file_size_bytes': 1024000, 'md5sum': '5d41402abc4b2a76b9719d911017c592', 'metadata': { 'key-with-dashes': 'value', 'key_with_underscores': 'value', 'key.with.dots': 'value', 'unicode_key_文件': 'unicode_value_文件', }, } result = schema.load(data) assert 'unicode_key_文件' in result['metadata'] def test_metadata_empty_dict(self): """Test metadata with empty dictionary.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': 'test.csv', 'file_size_bytes': 1024000, 'md5sum': '5d41402abc4b2a76b9719d911017c592', 'metadata': {}, } result = schema.load(data) assert result['metadata'] == {} def test_metadata_with_null_values(self): """Test metadata with null values.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': 'test.csv', 'file_size_bytes': 1024000, 'md5sum': '5d41402abc4b2a76b9719d911017c592', 'metadata': { 'key1': None, 'key2': 'value', }, } result = schema.load(data) assert result['metadata']['key1'] is None assert result['metadata']['key2'] == 'value' class TestAllowedFileTypesEdgeCases: """Additional edge case tests for allowed_file_types validation.""" def test_allowed_file_types_empty_list(self, sample_config_params): """Test allowed_file_types with empty list (different from None).""" schema = FileUploadConfigPostSchema() sample_config_params['allowed_file_types'] = [] # Empty list should raise ValidationError - use null to allow all types with pytest.raises(ValidationError) as exc_info: schema.load(sample_config_params) assert 'allowed_file_types' in exc_info.value.messages assert 'cannot be empty' in str(exc_info.value.messages['allowed_file_types']) def test_allowed_file_types_case_sensitive(self, sample_config_params): """Test allowed_file_types with different cases.""" schema = FileUploadConfigPostSchema() sample_config_params['allowed_file_types'] = ['CSV', 'xlsx', 'Pdf'] result = schema.load(sample_config_params) # Should preserve case assert 'CSV' in result['allowed_file_types'] assert 'xlsx' in result['allowed_file_types'] assert 'Pdf' in result['allowed_file_types'] def test_allowed_file_types_duplicates(self, sample_config_params): """Test allowed_file_types with duplicate values.""" schema = FileUploadConfigPostSchema() sample_config_params['allowed_file_types'] = ['csv', 'xlsx', 'csv'] result = schema.load(sample_config_params) # May or may not dedupe assert 'csv' in result['allowed_file_types'] def test_allowed_file_types_very_long_list(self, sample_config_params): """Test allowed_file_types with many extensions.""" schema = FileUploadConfigPostSchema() many_types = [f'ext{i}' for i in range(100)] sample_config_params['allowed_file_types'] = many_types result = schema.load(sample_config_params) assert len(result['allowed_file_types']) == 100 def test_allowed_file_types_unusual_extensions(self, sample_config_params): """Test allowed_file_types with unusual but valid extensions.""" schema = FileUploadConfigPostSchema() sample_config_params['allowed_file_types'] = [ 'tar.gz', # Multiple dots 'bak', # Short extension 'backup', # Longer extension 'data', ] result = schema.load(sample_config_params) assert 'tar.gz' in result['allowed_file_types'] class TestS3KeyTemplateEdgeCases: """Edge case tests for s3_key_template validation.""" def test_s3_key_template_minimal(self, sample_config_params): """Test s3_key_template with minimal template.""" schema = FileUploadConfigPostSchema() sample_config_params['s3_key_template'] = '{file_key}' result = schema.load(sample_config_params) assert result['s3_key_template'] == '{file_key}' def test_s3_key_template_no_placeholders(self, sample_config_params): """Test s3_key_template with no placeholders (static path).""" schema = FileUploadConfigPostSchema() sample_config_params['s3_key_template'] = 'uploads/files/' result = schema.load(sample_config_params) assert result['s3_key_template'] == 'uploads/files/' def test_s3_key_template_all_placeholders(self, sample_config_params): """Test s3_key_template with all possible placeholders.""" schema = FileUploadConfigPostSchema() template = '{year}/{month}/{day}/{upload_type}/{file_key}.{ext}' sample_config_params['s3_key_template'] = template result = schema.load(sample_config_params) assert result['s3_key_template'] == template def test_s3_key_template_with_metadata_placeholder(self, sample_config_params): """Test s3_key_template with metadata placeholders.""" schema = FileUploadConfigPostSchema() template = '{year}/{metadata.entity_id}/{file_key}.{ext}' sample_config_params['s3_key_template'] = template result = schema.load(sample_config_params) assert 'metadata.entity_id' in result['s3_key_template'] def test_s3_key_template_very_long(self, sample_config_params): """Test s3_key_template with very long path.""" schema = FileUploadConfigPostSchema() template = '/'.join(['level'] * 20) + '/{file_key}.{ext}' sample_config_params['s3_key_template'] = template result = schema.load(sample_config_params) assert len(result['s3_key_template']) > 100 class TestBoundaryConditions: """Additional boundary condition tests.""" def test_file_size_near_5tb_boundary(self): """Test file size very close to 5TB limit.""" from abacus_file_upload.constants import S3_MAX_FILE_SIZE_BYTES schema = InitiateUploadRequestSchema() # Test just under the limit data = { 'upload_type': 'adjustments', 'filename': 'huge_file.csv', 'file_size_bytes': S3_MAX_FILE_SIZE_BYTES - 1, 'md5sum': '5d41402abc4b2a76b9719d911017c592', } result = schema.load(data) assert result['file_size_bytes'] == S3_MAX_FILE_SIZE_BYTES - 1 def test_multipart_single_byte_over_threshold(self, sample_config_params): """Test multipart threshold at exact boundary.""" from abacus_file_upload.constants import S3_MIN_MULTIPART_THRESHOLD_BYTES schema = FileUploadConfigPostSchema() # Set threshold to minimum sample_config_params['multipart_threshold_bytes'] = ( S3_MIN_MULTIPART_THRESHOLD_BYTES ) result = schema.load(sample_config_params) assert result['multipart_threshold_bytes'] == S3_MIN_MULTIPART_THRESHOLD_BYTES def test_chunk_size_exactly_5mb(self, sample_config_params): """Test min chunk size at exactly 5MB.""" from abacus_file_upload.constants import S3_MIN_CHUNK_SIZE_BYTES schema = FileUploadConfigPostSchema() sample_config_params['min_multipart_chunk_size_bytes'] = S3_MIN_CHUNK_SIZE_BYTES result = schema.load(sample_config_params) assert result['min_multipart_chunk_size_bytes'] == S3_MIN_CHUNK_SIZE_BYTES