"""Tests for abacus_file_upload.schemas.file_upload module.""" import pytest from marshmallow import ValidationError from abacus_file_upload.constants import S3_MAX_FILE_SIZE_BYTES, UPLOAD_STATUSES from abacus_file_upload.schemas.file_upload import ( FileUploadDetailSchema, FileUploadStatusPutSchema, InitiateUploadRequestSchema, InitiateUploadResponseSchema, ) class TestInitiateUploadRequestSchema: """Tests for InitiateUploadRequestSchema.""" def test_valid_request(self): """Test valid upload request.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': 'test.csv', 'file_size_bytes': 1024000, 'mime_type': 'text/csv', 'md5sum': '5d41402abc4b2a76b9719d911017c592', 'metadata': {'entity_id': 123}, } result = schema.load(data) assert result['upload_type'] == 'adjustments' assert result['filename'] == 'test.csv' assert result['file_size_bytes'] == 1024000 assert result['md5sum'] == '5d41402abc4b2a76b9719d911017c592' def test_missing_required_fields(self): """Test missing required fields.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', # Missing filename, file_size_bytes, md5sum } with pytest.raises(ValidationError) as exc_info: schema.load(data) assert 'filename' in exc_info.value.messages assert 'file_size_bytes' in exc_info.value.messages assert 'md5sum' in exc_info.value.messages def test_optional_fields(self): """Test optional fields can be omitted.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': 'test.csv', 'file_size_bytes': 1024000, 'md5sum': '5d41402abc4b2a76b9719d911017c592', # mime_type and metadata are optional } result = schema.load(data) assert 'mime_type' not in result or result['mime_type'] is None assert 'metadata' not in result or result['metadata'] is None class TestMD5Validation: """Tests for MD5 hash validation.""" def test_valid_md5(self): """Test valid MD5 hash.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': 'test.csv', 'file_size_bytes': 1024000, 'md5sum': '5d41402abc4b2a76b9719d911017c592', # Valid MD5 } result = schema.load(data) assert result['md5sum'] == '5d41402abc4b2a76b9719d911017c592' def test_md5_lowercase(self): """Test MD5 hash with lowercase hex.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': 'test.csv', 'file_size_bytes': 1024000, 'md5sum': 'abcdef1234567890abcdef1234567890', } result = schema.load(data) assert result['md5sum'] == 'abcdef1234567890abcdef1234567890' def test_md5_uppercase(self): """Test MD5 hash with uppercase hex.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': 'test.csv', 'file_size_bytes': 1024000, 'md5sum': 'ABCDEF1234567890ABCDEF1234567890', } result = schema.load(data) assert result['md5sum'] == 'ABCDEF1234567890ABCDEF1234567890' def test_md5_mixed_case(self): """Test MD5 hash with mixed case hex.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': 'test.csv', 'file_size_bytes': 1024000, 'md5sum': 'AbCdEf1234567890aBcDeF1234567890', } result = schema.load(data) assert result['md5sum'] == 'AbCdEf1234567890aBcDeF1234567890' def test_md5_too_short(self): """Test MD5 hash that's too short.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': 'test.csv', 'file_size_bytes': 1024000, 'md5sum': '5d41402abc4b2a76b9719d91101', # Only 27 chars } with pytest.raises(ValidationError) as exc_info: schema.load(data) assert 'md5sum' in str(exc_info.value) assert '32 hexadecimal' in str(exc_info.value) def test_md5_too_long(self): """Test MD5 hash that's too long.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': 'test.csv', 'file_size_bytes': 1024000, 'md5sum': '5d41402abc4b2a76b9719d911017c592abc', # 35 chars } with pytest.raises(ValidationError) as exc_info: schema.load(data) assert 'md5sum' in str(exc_info.value) assert '32 hexadecimal' in str(exc_info.value) def test_md5_invalid_characters(self): """Test MD5 hash with invalid (non-hex) characters.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': 'test.csv', 'file_size_bytes': 1024000, 'md5sum': '5d41402abc4b2a76b9719d911017c5XZ', # Contains X and Z } with pytest.raises(ValidationError) as exc_info: schema.load(data) assert 'md5sum' in str(exc_info.value) assert '32 hexadecimal' in str(exc_info.value) def test_md5_empty(self): """Test empty MD5 hash.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': 'test.csv', 'file_size_bytes': 1024000, 'md5sum': '', } with pytest.raises(ValidationError) as exc_info: schema.load(data) # Should fail at field-level validation (NonemptyString) or our validator assert 'md5sum' in str(exc_info.value) class TestFileSizeValidation: """Tests for file_size_bytes validation.""" def test_valid_file_size(self): """Test valid file size.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': 'test.csv', 'file_size_bytes': 10485760, # 10MB 'md5sum': '5d41402abc4b2a76b9719d911017c592', } result = schema.load(data) assert result['file_size_bytes'] == 10485760 def test_file_size_one_byte(self): """Test minimum valid file size (1 byte).""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': 'test.csv', 'file_size_bytes': 1, 'md5sum': '5d41402abc4b2a76b9719d911017c592', } result = schema.load(data) assert result['file_size_bytes'] == 1 def test_file_size_zero(self): """Test zero file size (invalid).""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': 'test.csv', 'file_size_bytes': 0, 'md5sum': '5d41402abc4b2a76b9719d911017c592', } with pytest.raises(ValidationError) as exc_info: schema.load(data) assert 'file_size_bytes' in str(exc_info.value) assert 'greater than 0' in str(exc_info.value) def test_file_size_negative(self): """Test negative file size (invalid).""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': 'test.csv', 'file_size_bytes': -1000, 'md5sum': '5d41402abc4b2a76b9719d911017c592', } with pytest.raises(ValidationError) as exc_info: schema.load(data) assert 'file_size_bytes' in str(exc_info.value) assert 'greater than 0' in str(exc_info.value) def test_file_size_at_maximum(self): """Test file size at S3 maximum (5TB).""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': 'test.csv', 'file_size_bytes': S3_MAX_FILE_SIZE_BYTES, 'md5sum': '5d41402abc4b2a76b9719d911017c592', } result = schema.load(data) assert result['file_size_bytes'] == S3_MAX_FILE_SIZE_BYTES def test_file_size_above_maximum(self): """Test file size above S3 maximum (5TB).""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': 'test.csv', 'file_size_bytes': S3_MAX_FILE_SIZE_BYTES + 1, 'md5sum': '5d41402abc4b2a76b9719d911017c592', } with pytest.raises(ValidationError) as exc_info: schema.load(data) assert 'file_size_bytes' in str(exc_info.value) assert '5 TB' in str(exc_info.value) or '5.00 TB' in str(exc_info.value) class TestFilenameValidation: """Tests for filename validation.""" def test_valid_filename(self): """Test valid filename.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': 'test-file_2025.csv', 'file_size_bytes': 1024000, 'md5sum': '5d41402abc4b2a76b9719d911017c592', } result = schema.load(data) assert result['filename'] == 'test-file_2025.csv' def test_filename_with_spaces(self): """Test filename with spaces.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': 'test file.csv', 'file_size_bytes': 1024000, 'md5sum': '5d41402abc4b2a76b9719d911017c592', } result = schema.load(data) assert result['filename'] == 'test file.csv' def test_filename_empty(self): """Test empty filename.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': '', 'file_size_bytes': 1024000, 'md5sum': '5d41402abc4b2a76b9719d911017c592', } with pytest.raises(ValidationError) as exc_info: schema.load(data) # Should fail at field-level validation (NonemptyString) or our validator assert 'filename' in str(exc_info.value) def test_filename_whitespace_only(self): """Test filename with only whitespace.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': ' ', 'file_size_bytes': 1024000, 'md5sum': '5d41402abc4b2a76b9719d911017c592', } with pytest.raises(ValidationError) as exc_info: schema.load(data) assert 'filename' in str(exc_info.value) assert 'empty' in str(exc_info.value).lower() @pytest.mark.parametrize( 'invalid_char', ['\\', '/', ':', '*', '?', '"', '<', '>', '|', '\x00'], ) def test_filename_invalid_characters(self, invalid_char): """Test filename with invalid characters.""" schema = InitiateUploadRequestSchema() data = { 'upload_type': 'adjustments', 'filename': f'test{invalid_char}file.csv', 'file_size_bytes': 1024000, 'md5sum': '5d41402abc4b2a76b9719d911017c592', } with pytest.raises(ValidationError) as exc_info: schema.load(data) assert 'filename' in str(exc_info.value) assert 'invalid character' in str(exc_info.value).lower() class TestFileUploadDetailSchema: """Tests for FileUploadDetailSchema (response schema).""" def test_dump_file_upload(self, mock_file_upload): """Test dumping a FileUpload object.""" schema = FileUploadDetailSchema() result = schema.dump(mock_file_upload) assert result['file_upload_id'] == 1 assert result['file_key'] == '550e8400-e29b-41d4-a716-446655440000' assert result['original_file_name'] == 'test.csv' assert result['file_size_bytes'] == 1024000 assert result['upload_status'] == 'init' class TestInitiateUploadResponseSchema: """Tests for InitiateUploadResponseSchema.""" def test_single_part_upload_response(self): """Test response schema for single-part upload.""" schema = InitiateUploadResponseSchema() data = { 'file_key': '550e8400-e29b-41d4-a716-446655440000', 'is_multipart': False, 'expires_at': '2025-11-18T12:00:00Z', 'chunk_size_bytes': None, 'upload_url': 'https://s3.amazonaws.com/bucket/key?presigned', 'complete_url': None, 'parts': None, 'required_headers': {'Content-MD5': 'XUFAKrxLKna5cZ2REBfFkg=='}, } result = schema.dump(data) assert result['file_key'] == '550e8400-e29b-41d4-a716-446655440000' assert result['is_multipart'] is False assert result['upload_url'] is not None assert result['required_headers'] is not None assert 'Content-MD5' in result['required_headers'] def test_multipart_upload_response(self): """Test response schema for multipart upload.""" schema = InitiateUploadResponseSchema() data = { 'file_key': '550e8400-e29b-41d4-a716-446655440000', 'is_multipart': True, 'expires_at': '2025-11-18T12:00:00Z', 'chunk_size_bytes': 10485760, 'upload_url': None, 'complete_url': 'https://api.example.com/complete', 'parts': [ {'part_number': 1, 'upload_url': 'https://s3.amazonaws.com/part1'}, {'part_number': 2, 'upload_url': 'https://s3.amazonaws.com/part2'}, ], } result = schema.dump(data) assert result['file_key'] == '550e8400-e29b-41d4-a716-446655440000' assert result['is_multipart'] is True assert result['chunk_size_bytes'] == 10485760 assert len(result['parts']) == 2 class TestFileUploadStatusPutSchema: """Tests for FileUploadStatusPutSchema.""" def test_file_upload_status_put_schema(self): """Test dumping a file_upload_status PUT request object.""" result = FileUploadStatusPutSchema().dump( {'upload_status': UPLOAD_STATUSES.COMPLETE} ) assert result == {'upload_status': UPLOAD_STATUSES.COMPLETE} def test_unknown_fields(self): """Test dumping a file_upload_status PUT request object for unknown fields.""" mock_data = {'md5sum': 21312312, 'upload_status': UPLOAD_STATUSES.SCANNING} with pytest.raises(ValidationError) as exc_info: FileUploadStatusPutSchema().load(mock_data) assert exc_info.value.args[0] == {'md5sum': ['Unknown field.']} def test_invalid_file_upload_status(self): """Test dumping a file_upload_status PUT request object for invalid file upload status.""" with pytest.raises(ValidationError) as exc_info: FileUploadStatusPutSchema().load({'upload_status': 'test'}) assert ( 'Must be one of init, scanning, complete, error, cancelled, quarantined' in str(exc_info.value) )