"""Unit tests for FileUpload model.""" from datetime import datetime, timedelta, timezone from abacus_common_logic.connectors.database import db from abacus_file_upload.constants import UPLOAD_STATUSES from abacus_file_upload.models import FileUpload from abacus_file_upload.tests.utils.factories import ( FileUploadConfigFactory, FileUploadFactory, ) class TestFileUploadModel: """Tests for FileUpload model.""" def test_create_file_upload_with_all_fields(self): """Test creating a FileUpload with all fields.""" # Use timezone-naive datetimes and strip microseconds since NormalizedDateTime truncates them expires_at = datetime.now(timezone.utc).replace( tzinfo=None, microsecond=0 ) + timedelta(hours=2) completed_at = datetime.now(timezone.utc).replace(tzinfo=None, microsecond=0) file_upload_config = FileUploadConfigFactory() upload = FileUploadFactory( file_upload_config=file_upload_config, file_key='550e8400-e29b-41d4-a716-446655440000', original_file_name='test_data.csv', file_size_bytes=2048000, file_type='csv', mime_type='text/csv', s3_bucket='prod-bucket', s3_key='2025/11/550e8400-e29b-41d4-a716-446655440000.csv', md5sum='5d41402abc4b2a76b9719d911017c592', upload_status=UPLOAD_STATUSES.COMPLETE, multipart_upload_id='test-multipart-id', total_parts=5, upload_metadata={'entity_id': 123, 'user_email': 'test@example.com'}, error_message=None, completed_at=completed_at, expires_at=expires_at, ) assert upload.file_upload_id is not None assert upload.file_upload_config_id == file_upload_config.file_upload_config_id assert upload.file_key == '550e8400-e29b-41d4-a716-446655440000' assert upload.original_file_name == 'test_data.csv' assert upload.file_size_bytes == 2048000 assert upload.file_type == 'csv' assert upload.mime_type == 'text/csv' assert upload.s3_bucket == 'prod-bucket' assert upload.s3_key == '2025/11/550e8400-e29b-41d4-a716-446655440000.csv' assert upload.md5sum == '5d41402abc4b2a76b9719d911017c592' assert upload.upload_status == UPLOAD_STATUSES.COMPLETE assert upload.multipart_upload_id == 'test-multipart-id' assert upload.total_parts == 5 assert upload.upload_metadata == { 'entity_id': 123, 'user_email': 'test@example.com', } assert upload.error_message is None assert upload.completed_at == completed_at assert upload.expires_at == expires_at assert upload.created_at is not None assert upload.created_by is not None def test_create_file_upload_with_minimal_fields(self): """Test creating a FileUpload with only required fields.""" file_upload_config = FileUploadConfigFactory() upload = FileUploadFactory( file_upload_config=file_upload_config, file_type=None, mime_type=None, md5sum=None, multipart_upload_id=None, upload_metadata={}, error_message=None, completed_at=None, ) assert upload.file_upload_id is not None assert upload.file_upload_config_id == file_upload_config.file_upload_config_id assert upload.file_key is not None assert upload.original_file_name is not None assert upload.file_size_bytes is not None assert upload.s3_bucket is not None assert upload.s3_key is not None assert upload.upload_status == UPLOAD_STATUSES.INIT assert upload.total_parts == 1 assert upload.file_type is None assert upload.mime_type is None assert upload.md5sum is None assert upload.multipart_upload_id is None def test_file_upload_default_status(self): """Test that default upload status is INIT.""" upload = FileUploadFactory() assert upload.upload_status == UPLOAD_STATUSES.INIT def test_file_upload_default_total_parts(self): """Test that default total_parts is 1.""" upload = FileUploadFactory() assert upload.total_parts == 1 def test_find_by_file_key_found(self): """Test finding a FileUpload by file_key.""" upload = FileUploadFactory(file_key='550e8400-e29b-41d4-a716-446655440000') found_upload = FileUpload.find_by_file_key( '550e8400-e29b-41d4-a716-446655440000' ) assert found_upload is not None assert found_upload.file_upload_id == upload.file_upload_id assert found_upload.file_key == '550e8400-e29b-41d4-a716-446655440000' def test_find_by_file_key_not_found(self): """Test finding a FileUpload by file_key when it doesn't exist.""" found_upload = FileUpload.find_by_file_key('nonexistent-key') assert found_upload is None def test_find_by_file_key_unique(self): """Test that file_key is unique and only one result is returned.""" file_key = '550e8400-e29b-41d4-a716-446655440000' upload = FileUploadFactory(file_key=file_key) found_upload = FileUpload.find_by_file_key(file_key) # Should return a single upload, not a list assert found_upload is not None assert found_upload.file_upload_id == upload.file_upload_id def test_find_by_md5sum_single_result(self): """Test finding FileUploads by md5sum with one match.""" md5sum = '5d41402abc4b2a76b9719d911017c592' upload = FileUploadFactory(md5sum=md5sum) found_uploads = FileUpload.find_by_md5sum(md5sum) assert len(found_uploads) == 1 assert found_uploads[0].file_upload_id == upload.file_upload_id assert found_uploads[0].md5sum == md5sum def test_find_by_md5sum_multiple_results(self): """Test finding FileUploads by md5sum for duplicate detection.""" md5sum = '5d41402abc4b2a76b9719d911017c592' # Reuse same config to avoid unique constraint violation file_upload_config = FileUploadConfigFactory() upload1 = FileUploadFactory( file_upload_config=file_upload_config, md5sum=md5sum ) upload2 = FileUploadFactory( file_upload_config=file_upload_config, md5sum=md5sum ) upload3 = FileUploadFactory( file_upload_config=file_upload_config, md5sum=md5sum ) found_uploads = FileUpload.find_by_md5sum(md5sum) assert len(found_uploads) == 3 found_ids = {u.file_upload_id for u in found_uploads} assert upload1.file_upload_id in found_ids assert upload2.file_upload_id in found_ids assert upload3.file_upload_id in found_ids def test_find_by_md5sum_not_found(self): """Test finding FileUploads by md5sum when none exist.""" found_uploads = FileUpload.find_by_md5sum('nonexistent-md5sum') assert found_uploads == [] def test_find_by_upload_status_init(self): """Test finding FileUploads by upload_status.""" # Reuse same config to avoid unique constraint violation file_upload_config = FileUploadConfigFactory() upload1 = FileUploadFactory( file_upload_config=file_upload_config, upload_status=UPLOAD_STATUSES.INIT ) upload2 = FileUploadFactory( file_upload_config=file_upload_config, upload_status=UPLOAD_STATUSES.INIT ) upload3 = FileUploadFactory( file_upload_config=file_upload_config, upload_status=UPLOAD_STATUSES.COMPLETE, ) found_uploads = FileUpload.find_by_upload_status(UPLOAD_STATUSES.INIT) assert len(found_uploads) == 2 found_ids = {u.file_upload_id for u in found_uploads} assert upload1.file_upload_id in found_ids assert upload2.file_upload_id in found_ids assert upload3.file_upload_id not in found_ids def test_find_by_upload_status_all_statuses(self): """Test finding FileUploads for each upload status.""" # Reuse same config to avoid unique constraint violation file_upload_config = FileUploadConfigFactory() upload_init = FileUploadFactory( file_upload_config=file_upload_config, upload_status=UPLOAD_STATUSES.INIT ) upload_complete = FileUploadFactory( file_upload_config=file_upload_config, upload_status=UPLOAD_STATUSES.COMPLETE, ) upload_error = FileUploadFactory( file_upload_config=file_upload_config, upload_status=UPLOAD_STATUSES.ERROR ) upload_cancelled = FileUploadFactory( file_upload_config=file_upload_config, upload_status=UPLOAD_STATUSES.CANCELLED, ) # Test each status init_uploads = FileUpload.find_by_upload_status(UPLOAD_STATUSES.INIT) assert len(init_uploads) == 1 assert init_uploads[0].file_upload_id == upload_init.file_upload_id complete_uploads = FileUpload.find_by_upload_status(UPLOAD_STATUSES.COMPLETE) assert len(complete_uploads) == 1 assert complete_uploads[0].file_upload_id == upload_complete.file_upload_id error_uploads = FileUpload.find_by_upload_status(UPLOAD_STATUSES.ERROR) assert len(error_uploads) == 1 assert error_uploads[0].file_upload_id == upload_error.file_upload_id cancelled_uploads = FileUpload.find_by_upload_status(UPLOAD_STATUSES.CANCELLED) assert len(cancelled_uploads) == 1 assert cancelled_uploads[0].file_upload_id == upload_cancelled.file_upload_id def test_find_by_upload_status_not_found(self): """Test finding FileUploads by upload_status when none match.""" file_upload_config = FileUploadConfigFactory() FileUploadFactory( file_upload_config=file_upload_config, upload_status=UPLOAD_STATUSES.INIT ) found_uploads = FileUpload.find_by_upload_status(UPLOAD_STATUSES.COMPLETE) assert found_uploads == [] def test_default_order(self): """Test that default ordering is by created_at descending.""" # Create uploads with different created_at times, reuse config file_upload_config = FileUploadConfigFactory() upload1 = FileUploadFactory(file_upload_config=file_upload_config) upload1.created_at = datetime(2025, 1, 1, 10, 0, 0, tzinfo=timezone.utc) db.session.commit() upload2 = FileUploadFactory(file_upload_config=file_upload_config) upload2.created_at = datetime(2025, 1, 1, 12, 0, 0, tzinfo=timezone.utc) db.session.commit() upload3 = FileUploadFactory(file_upload_config=file_upload_config) upload3.created_at = datetime(2025, 1, 1, 11, 0, 0, tzinfo=timezone.utc) db.session.commit() uploads = FileUpload.query.order_by(FileUpload.default_order()).all() # Should be ordered by created_at descending (newest first) assert uploads[0].file_upload_id == upload2.file_upload_id assert uploads[1].file_upload_id == upload3.file_upload_id assert uploads[2].file_upload_id == upload1.file_upload_id def test_file_upload_relationship_to_config(self): """Test relationship between FileUpload and FileUploadConfig.""" upload = FileUploadFactory() config = upload.file_upload_config assert config is not None assert config.file_upload_config_id == upload.file_upload_config_id def test_file_upload_json_metadata(self): """Test that upload_metadata JSON field works correctly.""" metadata = { 'entity_id': 123, 'user_email': 'test@example.com', 'department': 'finance', 'tags': ['urgent', 'q4-2025'], } upload = FileUploadFactory(upload_metadata=metadata) assert upload.upload_metadata == metadata assert isinstance(upload.upload_metadata, dict) assert upload.upload_metadata['entity_id'] == 123 assert 'urgent' in upload.upload_metadata['tags'] def test_update_upload_status_to_complete(self): """Test updating upload status to complete.""" upload = FileUploadFactory(upload_status=UPLOAD_STATUSES.INIT) # Update to complete - use timezone-naive datetime without microseconds completed_at = datetime.now(timezone.utc).replace(tzinfo=None, microsecond=0) upload.upload_status = UPLOAD_STATUSES.COMPLETE upload.completed_at = completed_at db.session.commit() updated_upload = FileUpload.query.get(upload.file_upload_id) assert updated_upload.upload_status == UPLOAD_STATUSES.COMPLETE assert updated_upload.completed_at == completed_at def test_update_upload_status_to_error(self): """Test updating upload status to error with error message.""" upload = FileUploadFactory(upload_status=UPLOAD_STATUSES.INIT) # Update to error upload.upload_status = UPLOAD_STATUSES.ERROR upload.error_message = 'File upload failed due to network error' db.session.commit() updated_upload = FileUpload.query.get(upload.file_upload_id) assert updated_upload.upload_status == UPLOAD_STATUSES.ERROR assert updated_upload.error_message == 'File upload failed due to network error' def test_update_upload_status_to_cancelled(self): """Test updating upload status to cancelled.""" upload = FileUploadFactory(upload_status=UPLOAD_STATUSES.INIT) # Update to cancelled upload.upload_status = UPLOAD_STATUSES.CANCELLED db.session.commit() updated_upload = FileUpload.query.get(upload.file_upload_id) assert updated_upload.upload_status == UPLOAD_STATUSES.CANCELLED def test_multipart_upload_fields(self): """Test multipart upload specific fields.""" upload = FileUploadFactory( multipart_upload_id='test-multipart-upload-id-12345', total_parts=100, ) assert upload.multipart_upload_id == 'test-multipart-upload-id-12345' assert upload.total_parts == 100 def test_soft_delete_file_upload(self): """Test soft deleting a FileUpload.""" upload = FileUploadFactory() # Soft delete upload.deleted_at = datetime.now(timezone.utc) upload.deleted_by = 'test_user@example.com' db.session.commit() # Should still exist in database but have deleted_at set found_upload = FileUpload.query.filter_by( file_upload_id=upload.file_upload_id ).first() assert found_upload is not None assert found_upload.deleted_at is not None assert found_upload.deleted_by == 'test_user@example.com' def test_file_upload_timestamps(self): """Test that timestamp fields are properly set.""" upload = FileUploadFactory() assert upload.created_at is not None assert isinstance(upload.created_at, datetime) assert upload.created_by is not None def test_file_upload_expires_at(self): """Test expires_at field for presigned URL expiration.""" # Use timezone-naive datetime without microseconds expires_at = datetime.now(timezone.utc).replace( tzinfo=None, microsecond=0 ) + timedelta(hours=1) upload = FileUploadFactory(expires_at=expires_at) assert upload.expires_at is not None assert isinstance(upload.expires_at, datetime) assert upload.expires_at == expires_at def test_find_by_file_key_excludes_soft_deleted(self): """Test that find_by_file_key excludes soft-deleted records.""" file_key = '550e8400-e29b-41d4-a716-446655440000' upload = FileUploadFactory(file_key=file_key) # Verify it's found before soft delete found_upload = FileUpload.find_by_file_key(file_key) assert found_upload is not None assert found_upload.file_upload_id == upload.file_upload_id # Soft delete the upload upload.deleted_at = datetime.now(timezone.utc) upload.deleted_by = 'test_user@example.com' db.session.commit() # Should not be found after soft delete found_upload = FileUpload.find_by_file_key(file_key) assert found_upload is None def test_find_by_md5sum_excludes_soft_deleted(self): """Test that find_by_md5sum excludes soft-deleted records.""" md5sum = '5d41402abc4b2a76b9719d911017c592' file_upload_config = FileUploadConfigFactory() # Create multiple uploads with same md5sum upload1 = FileUploadFactory( file_upload_config=file_upload_config, md5sum=md5sum ) upload2 = FileUploadFactory( file_upload_config=file_upload_config, md5sum=md5sum ) upload3 = FileUploadFactory( file_upload_config=file_upload_config, md5sum=md5sum ) # All three should be found found_uploads = FileUpload.find_by_md5sum(md5sum) assert len(found_uploads) == 3 # Soft delete one upload upload2.deleted_at = datetime.now(timezone.utc) upload2.deleted_by = 'test_user@example.com' db.session.commit() # Should only find two now (upload1 and upload3) found_uploads = FileUpload.find_by_md5sum(md5sum) assert len(found_uploads) == 2 found_ids = {u.file_upload_id for u in found_uploads} assert upload1.file_upload_id in found_ids assert upload2.file_upload_id not in found_ids assert upload3.file_upload_id in found_ids def test_find_by_upload_status_excludes_soft_deleted(self): """Test that find_by_upload_status excludes soft-deleted records.""" file_upload_config = FileUploadConfigFactory() # Create multiple uploads with same status upload1 = FileUploadFactory( file_upload_config=file_upload_config, upload_status=UPLOAD_STATUSES.INIT ) upload2 = FileUploadFactory( file_upload_config=file_upload_config, upload_status=UPLOAD_STATUSES.INIT ) upload3 = FileUploadFactory( file_upload_config=file_upload_config, upload_status=UPLOAD_STATUSES.INIT ) # All three should be found found_uploads = FileUpload.find_by_upload_status(UPLOAD_STATUSES.INIT) assert len(found_uploads) == 3 # Soft delete one upload upload1.deleted_at = datetime.now(timezone.utc) upload1.deleted_by = 'test_user@example.com' db.session.commit() # Should only find two now (upload2 and upload3) found_uploads = FileUpload.find_by_upload_status(UPLOAD_STATUSES.INIT) assert len(found_uploads) == 2 found_ids = {u.file_upload_id for u in found_uploads} assert upload1.file_upload_id not in found_ids assert upload2.file_upload_id in found_ids assert upload3.file_upload_id in found_ids def test_find_by_md5sum_all_soft_deleted_returns_empty(self): """Test that find_by_md5sum returns empty list when all matches are soft-deleted.""" md5sum = '5d41402abc4b2a76b9719d911017c592' file_upload_config = FileUploadConfigFactory() upload1 = FileUploadFactory( file_upload_config=file_upload_config, md5sum=md5sum ) upload2 = FileUploadFactory( file_upload_config=file_upload_config, md5sum=md5sum ) # Soft delete both uploads for upload in [upload1, upload2]: upload.deleted_at = datetime.now(timezone.utc) upload.deleted_by = 'test_user@example.com' db.session.commit() # Should return empty list found_uploads = FileUpload.find_by_md5sum(md5sum) assert found_uploads == [] def test_find_by_upload_status_all_soft_deleted_returns_empty(self): """Test that find_by_upload_status returns empty list when all matches are soft-deleted.""" file_upload_config = FileUploadConfigFactory() upload1 = FileUploadFactory( file_upload_config=file_upload_config, upload_status=UPLOAD_STATUSES.COMPLETE, ) upload2 = FileUploadFactory( file_upload_config=file_upload_config, upload_status=UPLOAD_STATUSES.COMPLETE, ) # Soft delete both uploads for upload in [upload1, upload2]: upload.deleted_at = datetime.now(timezone.utc) upload.deleted_by = 'test_user@example.com' db.session.commit() # Should return empty list found_uploads = FileUpload.find_by_upload_status(UPLOAD_STATUSES.COMPLETE) assert found_uploads == [] class TestFileUploadValidation: """Tests for FileUpload validation and edge cases.""" def test_file_type_without_leading_dot(self): """Test that file_type is stored without leading dot and lowercase.""" upload = FileUploadFactory(file_type='csv') assert upload.file_type == 'csv' assert not upload.file_type.startswith('.') def test_various_file_types(self): """Test various file types.""" file_types = ['csv', 'xlsx', 'pdf', 'txt', 'json', 'xml'] file_upload_config = FileUploadConfigFactory() for file_type in file_types: upload = FileUploadFactory( file_upload_config=file_upload_config, file_type=file_type ) assert upload.file_type == file_type def test_various_mime_types(self): """Test various MIME types.""" mime_types = [ 'text/csv', 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', 'application/pdf', 'text/plain', 'application/json', ] file_upload_config = FileUploadConfigFactory() for mime_type in mime_types: upload = FileUploadFactory( file_upload_config=file_upload_config, mime_type=mime_type ) assert upload.mime_type == mime_type def test_large_file_size(self): """Test creating upload with large file size.""" # Test with 5GB file five_gb = 5 * 1024 * 1024 * 1024 upload = FileUploadFactory(file_size_bytes=five_gb) assert upload.file_size_bytes == five_gb def test_small_file_size(self): """Test creating upload with small file size.""" # Test with 1KB file upload = FileUploadFactory(file_size_bytes=1024) assert upload.file_size_bytes == 1024 def test_s3_key_max_length(self): """Test S3 key with long path.""" # S3 keys can be up to 1024 characters long_s3_key = 'uploads/' + '/'.join(['level'] * 100) + '/file.csv' upload = FileUploadFactory(s3_key=long_s3_key[:1024]) assert len(upload.s3_key) <= 1024 def test_md5sum_format(self): """Test MD5 sum is 32 character hex string.""" md5sum = '5d41402abc4b2a76b9719d911017c592' upload = FileUploadFactory(md5sum=md5sum) assert upload.md5sum == md5sum assert len(upload.md5sum) == 32 # Verify it's a valid hex string int(upload.md5sum, 16) def test_original_filename_with_special_characters(self): """Test original filename with special characters.""" filenames = [ 'test file with spaces.csv', 'test-file-with-dashes.csv', 'test_file_with_underscores.csv', 'test (1).csv', 'test[brackets].csv', ] file_upload_config = FileUploadConfigFactory() for filename in filenames: upload = FileUploadFactory( file_upload_config=file_upload_config, original_file_name=filename ) assert upload.original_file_name == filename def test_error_message_text_field(self): """Test that error_message can store long error text.""" long_error = 'Error: ' + 'x' * 10000 # Long error message upload = FileUploadFactory( upload_status=UPLOAD_STATUSES.ERROR, error_message=long_error ) assert upload.error_message == long_error assert len(upload.error_message) > 10000 def test_multiple_uploads_same_config(self): """Test multiple uploads can reference the same config.""" file_upload_config = FileUploadConfigFactory() upload1 = FileUploadFactory(file_upload_config=file_upload_config) upload2 = FileUploadFactory(file_upload_config=file_upload_config) upload3 = FileUploadFactory(file_upload_config=file_upload_config) assert upload1.file_upload_config_id == file_upload_config.file_upload_config_id assert upload2.file_upload_config_id == file_upload_config.file_upload_config_id assert upload3.file_upload_config_id == file_upload_config.file_upload_config_id # Verify relationship uploads = file_upload_config.file_uploads.all() assert len(uploads) == 3