"""Tests for abacus_file_upload schemas.""" import pytest from marshmallow import ValidationError from abacus_file_upload.constants import ( S3_MAX_CHUNK_SIZE_BYTES, S3_MAX_FILE_SIZE_BYTES, S3_MAX_MULTIPART_THRESHOLD_BYTES, S3_MIN_CHUNK_SIZE_BYTES, S3_MIN_MULTIPART_THRESHOLD_BYTES, ) from abacus_file_upload.schemas import ( FileUploadConfigDetailSchema, FileUploadConfigPostSchema, FileUploadConfigPutSchema, ) class TestFileUploadConfigSchemas: """Tests for FileUploadConfig schemas.""" def test_post_schema_valid(self, sample_config_params): """Test FileUploadConfigPostSchema with valid data.""" schema = FileUploadConfigPostSchema() result = schema.load(sample_config_params) assert result['upload_type'] == 'adjustments' assert result['s3_key_template'] == '{year}/{month}/{file_key}.{ext}' assert result['allowed_file_types'] == ['csv', 'xlsx'] def test_post_schema_missing_required(self, sample_config_params): """Test FileUploadConfigPostSchema with missing required field.""" schema = FileUploadConfigPostSchema() del sample_config_params['upload_type'] with pytest.raises(ValidationError) as exc_info: schema.load(sample_config_params) assert 'upload_type' in exc_info.value.messages def test_post_schema_invalid_upload_type(self, sample_config_params): """Test FileUploadConfigPostSchema with invalid upload_type.""" schema = FileUploadConfigPostSchema() sample_config_params['upload_type'] = 'test' with pytest.raises(ValidationError) as exc_info: schema.load(sample_config_params) assert str(exc_info.value.messages) == str( {'upload_type': ['Must be one of adjustments, flowthrough']} ) def test_put_schema_partial_update(self): """Test FileUploadConfigPutSchema for partial update.""" schema = FileUploadConfigPutSchema() data = {'max_file_size_bytes': 20971520} # Only updating one field result = schema.load(data) assert result['max_file_size_bytes'] == 20971520 assert 'upload_type' not in result # Not required in PUT def test_detail_schema_dump(self, mock_file_upload_config): """Test FileUploadConfigDetailSchema dump.""" schema = FileUploadConfigDetailSchema() result = schema.dump(mock_file_upload_config) assert result['file_upload_config_id'] == 1 assert result['upload_type'] == 'adjustments' assert result['allowed_file_types'] == ['csv', 'xlsx'] assert result['event_name'] == 'test_event' class TestMinChunkSizeValidation: """Tests for min_multipart_chunk_size_bytes validation.""" def test_min_chunk_size_below_minimum(self, sample_config_params): """Test min_multipart_chunk_size_bytes below S3 minimum (5MB).""" schema = FileUploadConfigPostSchema() sample_config_params['min_multipart_chunk_size_bytes'] = ( S3_MIN_CHUNK_SIZE_BYTES - 1 ) with pytest.raises(ValidationError) as exc_info: schema.load(sample_config_params) assert 'min_multipart_chunk_size_bytes' in str(exc_info.value) assert '5 MB' in str(exc_info.value) or '5.00 MB' in str(exc_info.value) def test_min_chunk_size_at_minimum(self, sample_config_params): """Test min_multipart_chunk_size_bytes at S3 minimum (5MB).""" 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 def test_min_chunk_size_above_maximum(self, sample_config_params): """Test min_multipart_chunk_size_bytes above S3 maximum (5GB).""" schema = FileUploadConfigPostSchema() sample_config_params['min_multipart_chunk_size_bytes'] = ( S3_MAX_CHUNK_SIZE_BYTES + 1 ) with pytest.raises(ValidationError) as exc_info: schema.load(sample_config_params) assert 'min_multipart_chunk_size_bytes' in str(exc_info.value) assert '5 GB' in str(exc_info.value) or '5.00 GB' in str(exc_info.value) def test_min_chunk_size_at_maximum(self, sample_config_params): """Test min_multipart_chunk_size_bytes at S3 maximum (5GB).""" schema = FileUploadConfigPostSchema() sample_config_params['min_multipart_chunk_size_bytes'] = S3_MAX_CHUNK_SIZE_BYTES result = schema.load(sample_config_params) assert result['min_multipart_chunk_size_bytes'] == S3_MAX_CHUNK_SIZE_BYTES class TestMaxFileSizeValidation: """Tests for max_file_size_bytes validation.""" def test_max_file_size_negative(self, sample_config_params): """Test max_file_size_bytes with negative value.""" schema = FileUploadConfigPostSchema() sample_config_params['max_file_size_bytes'] = -1 with pytest.raises(ValidationError) as exc_info: schema.load(sample_config_params) assert 'max_file_size_bytes' in str(exc_info.value) assert '>= 0' in str(exc_info.value) def test_max_file_size_zero(self, sample_config_params): """Test max_file_size_bytes with zero (disables uploads).""" schema = FileUploadConfigPostSchema() sample_config_params['max_file_size_bytes'] = 0 result = schema.load(sample_config_params) assert result['max_file_size_bytes'] == 0 def test_max_file_size_above_maximum(self, sample_config_params): """Test max_file_size_bytes above S3 maximum (5TB).""" schema = FileUploadConfigPostSchema() sample_config_params['max_file_size_bytes'] = S3_MAX_FILE_SIZE_BYTES + 1 with pytest.raises(ValidationError) as exc_info: schema.load(sample_config_params) assert 'max_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) def test_max_file_size_at_maximum(self, sample_config_params): """Test max_file_size_bytes at S3 maximum (5TB).""" schema = FileUploadConfigPostSchema() sample_config_params['max_file_size_bytes'] = S3_MAX_FILE_SIZE_BYTES result = schema.load(sample_config_params) assert result['max_file_size_bytes'] == S3_MAX_FILE_SIZE_BYTES class TestMultipartThresholdValidation: """Tests for multipart_threshold_bytes validation.""" def test_multipart_threshold_below_minimum(self, sample_config_params): """Test multipart_threshold_bytes below S3 minimum (5MB).""" schema = FileUploadConfigPostSchema() sample_config_params['multipart_threshold_bytes'] = ( S3_MIN_MULTIPART_THRESHOLD_BYTES - 1 ) with pytest.raises(ValidationError) as exc_info: schema.load(sample_config_params) assert 'multipart_threshold_bytes' in str(exc_info.value) assert '5 MB' in str(exc_info.value) or '5.00 MB' in str(exc_info.value) def test_multipart_threshold_at_minimum(self, sample_config_params): """Test multipart_threshold_bytes at S3 minimum (5MB).""" schema = FileUploadConfigPostSchema() 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_multipart_threshold_above_maximum(self, sample_config_params): """Test multipart_threshold_bytes above S3 maximum (5GB).""" schema = FileUploadConfigPostSchema() sample_config_params['multipart_threshold_bytes'] = ( S3_MAX_MULTIPART_THRESHOLD_BYTES + 1 ) with pytest.raises(ValidationError) as exc_info: schema.load(sample_config_params) assert 'multipart_threshold_bytes' in str(exc_info.value) assert '5 GB' in str(exc_info.value) or '5.00 GB' in str(exc_info.value) def test_multipart_threshold_at_maximum(self, sample_config_params): """Test multipart_threshold_bytes at S3 maximum (5GB).""" schema = FileUploadConfigPostSchema() sample_config_params['multipart_threshold_bytes'] = ( S3_MAX_MULTIPART_THRESHOLD_BYTES ) result = schema.load(sample_config_params) assert result['multipart_threshold_bytes'] == S3_MAX_MULTIPART_THRESHOLD_BYTES class TestFileTypesValidation: """Tests for allowed_file_types validation.""" def test_file_types_with_leading_dot(self, sample_config_params): """Test allowed_file_types with leading dot.""" schema = FileUploadConfigPostSchema() sample_config_params['allowed_file_types'] = ['.csv', 'xlsx'] with pytest.raises(ValidationError) as exc_info: schema.load(sample_config_params) assert 'allowed_file_types' in str(exc_info.value) assert 'leading dot' in str(exc_info.value).lower() def test_file_types_with_empty_string(self, sample_config_params): """Test allowed_file_types with empty string.""" schema = FileUploadConfigPostSchema() sample_config_params['allowed_file_types'] = ['csv', ''] with pytest.raises(ValidationError) as exc_info: schema.load(sample_config_params) assert 'allowed_file_types' in str(exc_info.value) assert 'non-empty' in str(exc_info.value).lower() def test_file_types_valid(self, sample_config_params): """Test allowed_file_types with valid values.""" schema = FileUploadConfigPostSchema() sample_config_params['allowed_file_types'] = ['csv', 'xlsx', 'pdf', 'json'] result = schema.load(sample_config_params) assert result['allowed_file_types'] == ['csv', 'xlsx', 'pdf', 'json'] def test_file_types_none(self, sample_config_params): """Test allowed_file_types with None (allows all types).""" schema = FileUploadConfigPostSchema() sample_config_params['allowed_file_types'] = None result = schema.load(sample_config_params) assert result['allowed_file_types'] is None class TestPutSchemaValidation: """Tests for PUT schema validation (same rules apply).""" def test_put_schema_invalid_max_file_size(self): """Test PUT schema with invalid max_file_size_bytes.""" schema = FileUploadConfigPutSchema() data = {'max_file_size_bytes': -100} with pytest.raises(ValidationError) as exc_info: schema.load(data) assert 'max_file_size_bytes' in str(exc_info.value) def test_put_schema_invalid_threshold(self): """Test PUT schema with invalid multipart_threshold_bytes.""" schema = FileUploadConfigPutSchema() data = {'multipart_threshold_bytes': 1000} # Less than 5MB with pytest.raises(ValidationError) as exc_info: schema.load(data) assert 'multipart_threshold_bytes' in str(exc_info.value) def test_put_schema_invalid_file_types(self): """Test PUT schema with invalid file types.""" schema = FileUploadConfigPutSchema() data = {'allowed_file_types': ['.pdf', 'csv']} with pytest.raises(ValidationError) as exc_info: schema.load(data) assert 'allowed_file_types' in str(exc_info.value)