"""Factories for testing abacus_file_upload models.""" import datetime import uuid import factory from abacus_common_logic.test_utils.factories import ( BaseModelFactory, BaseModelFactoryWithTimestamps, ) from abacus_file_upload.constants import ( ABACUS_OUTBOX_EVENT_TYPES, ABACUS_OUTBOX_TARGET_TYPES, EVENT_PROCESSING_STATUSES, UPLOAD_STATUSES, UPLOAD_TYPES, ) from abacus_file_upload.models import AbacusOutbox, FileUpload, FileUploadConfig class FileUploadConfigFactory(BaseModelFactoryWithTimestamps): """Factory to create FileUploadConfig models for testing.""" class Meta: """Meta definition for the factory.""" model = FileUploadConfig upload_type = factory.Iterator([UPLOAD_TYPES.ADJUSTMENTS, UPLOAD_TYPES.FLOWTHROUGH]) s3_key_template = '{year}/{month}/{file_key}.{ext}' allowed_file_types = ['csv', 'xlsx'] max_file_size_bytes = 10485760 # 10MB multipart_threshold_bytes = 104857600 # 100MB min_multipart_chunk_size_bytes = 10485760 # 10MB description = factory.Sequence(lambda n: f'Test config {n}') class FileUploadFactory(BaseModelFactoryWithTimestamps): """Factory to create FileUpload models for testing.""" class Meta: """Meta definition for the factory.""" model = FileUpload file_upload_config = factory.SubFactory(FileUploadConfigFactory) file_key = factory.Faker('uuid4') original_file_name = 'test.csv' file_size_bytes = 1024000 file_type = 'csv' mime_type = 'text/csv' s3_bucket = 'test-bucket' s3_key = factory.LazyAttribute( lambda obj: f'2025/01/{obj.file_key}.{obj.file_type}' ) md5sum = '5d41402abc4b2a76b9719d911017c592' upload_status = UPLOAD_STATUSES.INIT multipart_upload_id = None total_parts = 1 upload_metadata = {} error_message = None completed_at = None expires_at = factory.LazyFunction( lambda: ( datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(hours=1) ) ) class AbacusOutboxFactory(BaseModelFactory): """Factory to create AbacusOutbox models for testing.""" class Meta: """Meta definition for the factory.""" model = AbacusOutbox target_type = ABACUS_OUTBOX_TARGET_TYPES.FILE_UPLOAD target_id = 1 event_type = ABACUS_OUTBOX_EVENT_TYPES.FILE_UPLOAD_COMPLETED correlation_id = str(uuid.uuid4()) details = {'upload_type': 'flowthrough'} status = EVENT_PROCESSING_STATUSES.PENDING retry_count = 0 max_retries = 3