# ruff: noqa: PLC2701 DOC402 ANN201 RUF043 import logging from collections.abc import AsyncGenerator, Generator from contextlib import suppress from pathlib import Path from typing import Optional import pytest from minio import Minio from pytest_databases.docker.minio import MinioService from sqlalchemy import Engine, String, create_engine, event from sqlalchemy.exc import InvalidRequestError from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column from advanced_alchemy._listeners import set_async_context, setup_file_object_listeners from advanced_alchemy.base import create_registry from advanced_alchemy.exceptions import ImproperConfigurationError from advanced_alchemy.types.file_object import ( FileObject, FileObjectList, StoredObject, ) from advanced_alchemy.types.file_object.backends.fsspec import FSSpecBackend from advanced_alchemy.types.file_object.backends.obstore import ObstoreBackend from advanced_alchemy.types.file_object.registry import StorageRegistry, storages from advanced_alchemy.types.mutables import MutableList # Setup logger logger = logging.getLogger(__name__) pytestmark = pytest.mark.integration def remove_listeners() -> None: """Remove file object listeners safely to prevent test interactions.""" from sqlalchemy.event import contains from advanced_alchemy._listeners import FileObjectListener # Only try to remove listeners if they're actually registered if contains(Session, "before_flush", FileObjectListener.before_flush): with suppress(InvalidRequestError): event.remove(Session, "before_flush", FileObjectListener.before_flush) if contains(Session, "after_commit", FileObjectListener.after_commit): with suppress(InvalidRequestError): event.remove(Session, "after_commit", FileObjectListener.after_commit) if contains(Session, "after_rollback", FileObjectListener.after_rollback): with suppress(InvalidRequestError): event.remove(Session, "after_rollback", FileObjectListener.after_rollback) # Reset async context flag to ensure clean state set_async_context(False) # --- Fixtures --- orm_registry = create_registry() # --- SQLAlchemy Model Definition --- class Base(DeclarativeBase): metadata = orm_registry.metadata class Document(Base): __tablename__ = "documents" id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] = mapped_column(String(50)) # Single file storage attachment: Mapped[Optional[FileObject]] = mapped_column( StoredObject(backend="local_test_store"), # Use StoredObject wrapper nullable=True, ) # Multiple file storage images: Mapped[Optional[FileObjectList]] = mapped_column( StoredObject(backend="local_test_store", multiple=True), # Use StoredObject wrapper nullable=True, ) @pytest.fixture() def storage_registry(tmp_path: Path) -> "StorageRegistry": """Clears and returns the global storage registry for the module. Returns: StorageRegistry: The global storage registry. """ from obstore.store import LocalStore, MemoryStore if not storages.is_registered("memory"): storages.register_backend(ObstoreBackend(fs=MemoryStore(), key="memory")) if storages.is_registered("local_test_store"): storages.unregister_backend("local_test_store") # Create the storage directory storage_dir = tmp_path / "file_object_test_storage" storage_dir.mkdir(parents=True, exist_ok=True) storages.register_backend( ObstoreBackend( fs=LocalStore(prefix=str(storage_dir)), # pyright: ignore key="local_test_store", ) ) return storages @pytest.fixture() def sync_db_engine(tmp_path: Path) -> Generator[Engine, None, None]: """Provides an SQLite engine scoped to each function.""" db_file = tmp_path / "test_file_object_sync.db" engine = create_engine(f"sqlite:///{db_file}", execution_options={"enable_file_object_listener": True}) yield engine db_file.unlink(missing_ok=True) @pytest.fixture() def async_db_engine(tmp_path: Path) -> Generator[AsyncEngine, None, None]: """Provides an SQLite engine scoped to each function.""" db_file = tmp_path / "test_file_object_async.db" engine = create_async_engine( f"sqlite+aiosqlite:///{db_file}", execution_options={"enable_file_object_listener": True} ) yield engine db_file.unlink(missing_ok=True) @pytest.fixture() def session( sync_db_engine: Engine, storage_registry: "StorageRegistry" ) -> Generator[Session, None, None]: # Depend on sqlalchemy_config to ensure setup runs """Provides a SQLAlchemy session scoped to each function.""" Base.metadata.create_all(sync_db_engine) with Session(sync_db_engine) as db_session: yield db_session Base.metadata.drop_all(sync_db_engine) @pytest.fixture() async def async_session( async_db_engine: AsyncEngine, storage_registry: "StorageRegistry" ) -> AsyncGenerator[AsyncSession, None]: # Depend on sqlalchemy_config to ensure setup runs """Provides a SQLAlchemy session scoped to each function.""" async with async_db_engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) # Create session with flag for listener to identify async operations set_async_context(True) # Create session factory async_session_factory = async_sessionmaker( async_db_engine, expire_on_commit=False, ) # Create session async with async_session_factory() as db_session: # Add flag to session.info dictionary db_session.info["enable_file_object_listener"] = True logger.debug(f"Created async session: {id(db_session)}, with info: {db_session.info}") yield db_session # Reset async context flag set_async_context(False) async with async_db_engine.begin() as conn: await conn.run_sync(Base.metadata.drop_all) await async_db_engine.dispose() @pytest.mark.xdist_group("file_object") async def test_fsspec_s3_basic_operations_async( storage_registry: StorageRegistry, minio_client: "Minio", minio_service: "MinioService", minio_default_bucket_name: str, ) -> None: """Test basic save, get_content, delete via backend and FileObject with prefix.""" remove_listeners() try: import s3fs except ImportError: pytest.skip("s3fs not installed") assert minio_client.bucket_exists(minio_default_bucket_name) _ = minio_client # Create s3fs filesystem instance without bucket info fs = s3fs.S3FileSystem( anon=False, key=minio_service.access_key, secret=minio_service.secret_key, endpoint_url=f"http://{minio_service.endpoint}", client_kwargs={ "verify": False, "use_ssl": False, }, ) # Initialize backend with prefix backend = FSSpecBackend( key="s3_test_store", fs=fs, prefix=minio_default_bucket_name, ) test_content = b"Hello Storage!" # Use relative path, prefix handles the bucket file_path = "test_basic_s3_async.txt" # Create initial FileObject with relative path obj = FileObject(backend=backend, filename="test_basic_s3_async.txt", to_filename=file_path) # Save using backend updated_obj = await backend.save_object_async(obj, test_content) # Assert FileObject updated assert updated_obj is obj # Should update in-place assert obj.path == file_path # Path should remain relative assert obj.filename == "test_basic_s3_async.txt" assert obj.etag is not None assert obj.size == len(test_content) assert obj.backend is backend assert obj.protocol == "s3" # Based on s3fs filesystem # Retrieve content via FileObject method (uses relative obj.path, backend adds prefix) retrieved_content = await obj.get_content_async() assert retrieved_content == test_content # Test sign_async method url_async = await obj.sign_async(expires_in=3600) assert isinstance(url_async, str) assert url_async.startswith("http") # Test for_upload parameter with pytest.raises( NotImplementedError, match=r"Generating signed URLs for upload is generally not supported by fsspec's generic sign method.", ): _ = await obj.sign_async(for_upload=True) # Delete via FileObject method (uses relative obj.path, backend adds prefix) await obj.delete_async() # Verify deletion using relative path with backend (backend adds prefix) with pytest.raises(FileNotFoundError): await backend.get_content_async(file_path) @pytest.mark.xdist_group("file_object") def test_fsspec_s3_basic_operations_sync( storage_registry: StorageRegistry, minio_client: "Minio", minio_service: "MinioService", minio_default_bucket_name: str, ) -> None: """Test basic save, get_content, delete via backend and FileObject with prefix.""" remove_listeners() try: import s3fs except ImportError: pytest.skip("s3fs not installed") assert minio_client.bucket_exists(minio_default_bucket_name) _ = minio_client # Create s3fs filesystem instance without bucket info fs = s3fs.S3FileSystem( anon=False, key=minio_service.access_key, secret=minio_service.secret_key, endpoint_url=f"http://{minio_service.endpoint}", client_kwargs={ "verify": False, "use_ssl": False, }, asynchronous=False, loop=None, ) # Initialize backend with prefix backend = FSSpecBackend( key="s3_test_store", fs=fs, prefix=minio_default_bucket_name, ) test_content = b"Hello Storage!" # Use relative path, prefix handles the bucket file_path = "test_basic_s3_sync.txt" # Create initial FileObject with relative path obj = FileObject(backend=backend, filename="test_basic_s3_sync.txt", to_filename=file_path) # Save using backend updated_obj = backend.save_object(obj, test_content) # Assert FileObject updated assert updated_obj is obj # Should update in-place assert obj.path == file_path # Path should remain relative assert obj.filename == "test_basic_s3_sync.txt" assert obj.etag is not None assert obj.size == len(test_content) assert obj.backend is backend assert obj.protocol == "s3" # Based on s3fs filesystem # Retrieve content via FileObject method (uses relative obj.path, backend adds prefix) retrieved_content = obj.get_content() assert retrieved_content == test_content # Test sign_async method url_async = obj.sign(expires_in=3600) assert isinstance(url_async, str) assert url_async.startswith("http") # Test for_upload parameter with pytest.raises( NotImplementedError, match=r"Generating signed URLs for upload is generally not supported by fsspec's generic sign method.", ): _ = obj.sign(for_upload=True) # Delete via FileObject method (uses relative obj.path, backend adds prefix) obj.delete() # Verify deletion using relative path with backend (backend adds prefix) with pytest.raises(FileNotFoundError): backend.get_content(file_path) @pytest.mark.xdist_group("file_object") async def test_obstore_s3_basic_operations_async( storage_registry: StorageRegistry, minio_client: "Minio", minio_service: "MinioService", minio_default_bucket_name: str, ) -> None: """Test basic save, get_content, delete via backend and FileObject.""" remove_listeners() assert minio_client.bucket_exists(minio_default_bucket_name) _ = minio_client backend = ObstoreBackend( key="s3_test_store", fs=f"s3://{minio_default_bucket_name}/", aws_endpoint=f"http://{minio_service.endpoint}/", aws_access_key_id=minio_service.access_key, aws_secret_access_key=minio_service.secret_key, aws_virtual_hosted_style_request=False, client_options={"allow_http": True}, ) test_content = b"Hello Storage!" file_path = "test_basic_s3_async.txt" # Relative path for the backend # Create initial FileObject obj = FileObject(backend=backend, filename="test_basic_s3_async.txt", to_filename=file_path) # Save using backend updated_obj = await backend.save_object_async(obj, test_content) # Assert FileObject updated assert updated_obj is obj # Should update in-place assert obj.path == file_path assert obj.filename == "test_basic_s3_async.txt" assert obj.etag is not None assert obj.size == len(test_content) assert obj.backend is backend assert obj.protocol == "s3" # Based on LocalFileSystem # Retrieve content via FileObject method retrieved_content = await obj.get_content_async() assert retrieved_content == test_content # Test sign method url = obj.sign(expires_in=3600) assert isinstance(url, str) assert url.startswith("http") # Test sign_async method url_async = await obj.sign_async(expires_in=3600) assert isinstance(url_async, str) assert url_async.startswith("http") url_for_upload_async = await obj.sign_async(for_upload=True) assert isinstance(url_for_upload_async, str) assert url_for_upload_async.startswith("http") # Delete via FileObject method await obj.delete_async() # Verify deletion (expect FileNotFoundError or similar from backend) with pytest.raises(FileNotFoundError): await backend.get_content_async(file_path) @pytest.mark.xdist_group("file_object") def test_obstore_s3_basic_operations_sync( storage_registry: StorageRegistry, minio_client: "Minio", minio_service: "MinioService", minio_default_bucket_name: str, ) -> None: """Test basic save, get_content, delete via backend and FileObject.""" remove_listeners() assert minio_client.bucket_exists(minio_default_bucket_name) _ = minio_client backend = ObstoreBackend( key="s3_test_store", fs=f"s3://{minio_default_bucket_name}/", aws_endpoint=f"http://{minio_service.endpoint}/", aws_access_key_id=minio_service.access_key, aws_secret_access_key=minio_service.secret_key, aws_virtual_hosted_style_request=False, client_options={"allow_http": True}, ) test_content = b"Hello Storage!" file_path = "test_basic_s3_sync.txt" # Relative path for the backend # Create initial FileObject obj = FileObject(backend=backend, filename="test_basic_s3_sync.txt", to_filename=file_path) # Save using backend updated_obj = backend.save_object(obj, test_content) # Assert FileObject updated assert updated_obj is obj # Should update in-place assert obj.path == file_path assert obj.filename == "test_basic_s3_sync.txt" assert obj.etag is not None assert obj.size == len(test_content) assert obj.backend is backend assert obj.protocol == "s3" # Based on LocalFileSystem # Retrieve content via FileObject method retrieved_content = obj.get_content() assert retrieved_content == test_content # Test sign method url = obj.sign(expires_in=3600) assert isinstance(url, str) assert url.startswith("http") # Test sign_async method url_async = obj.sign(expires_in=3600) assert isinstance(url_async, str) assert url_async.startswith("http") # Test for_upload parameter url_for_upload = obj.sign(for_upload=True) assert isinstance(url_for_upload, str) assert url_for_upload.startswith("http") # Delete via FileObject method obj.delete() # Verify deletion (expect FileNotFoundError or similar from backend) with pytest.raises(FileNotFoundError): backend.get_content(file_path) @pytest.mark.xdist_group("file_object") async def test_obstore_basic_operations_async(storage_registry: StorageRegistry) -> None: """Test basic save, get_content, delete via backend and FileObject.""" remove_listeners() backend = storage_registry.get_backend("local_test_store") test_content = b"Hello Storage!" file_path = "test_basic_async.txt" # Relative path for the backend # Create initial FileObject obj = FileObject(backend=backend, filename="test_basic_async.txt", to_filename=file_path) # Save using backend updated_obj = await backend.save_object_async(obj, test_content) # Assert FileObject updated assert updated_obj is obj # Should update in-place assert obj.path == file_path assert obj.filename == "test_basic_async.txt" assert obj.etag is not None assert obj.size == len(test_content) assert obj.backend is backend assert obj.protocol == "file" # Based on LocalFileSystem # Retrieve content via FileObject method retrieved_content = await obj.get_content_async() assert retrieved_content == test_content # Delete via FileObject method await obj.delete_async() # Verify deletion (expect FileNotFoundError or similar from backend) with pytest.raises(FileNotFoundError): await backend.get_content_async(file_path) @pytest.mark.xdist_group("file_object") def test_obstore_basic_operations_sync(storage_registry: StorageRegistry) -> None: """Test basic save, get_content, delete via backend and FileObject.""" remove_listeners() backend = storage_registry.get_backend("local_test_store") test_content = b"Hello Storage!" file_path = "test_basic_sync.txt" # Relative path for the backend # Create initial FileObject obj = FileObject(backend=backend, filename="test_basic_sync.txt", to_filename=file_path) # Save using backend updated_obj = backend.save_object(obj, test_content) # Assert FileObject updated assert updated_obj is obj # Should update in-place assert obj.path == file_path assert obj.filename == "test_basic_sync.txt" assert obj.etag is not None assert obj.size == len(test_content) assert obj.backend is backend assert obj.protocol == "file" # Based on LocalFileSystem # Retrieve content via FileObject method retrieved_content = obj.get_content() assert retrieved_content == test_content # Delete via FileObject method obj.delete() # Verify deletion (expect FileNotFoundError or similar from backend) with pytest.raises(FileNotFoundError): backend.get_content(file_path) @pytest.mark.xdist_group("file_object") async def test_obstore_single_file_async_no_listener( async_session: AsyncSession, storage_registry: StorageRegistry ) -> None: """Test saving and loading a model with a single StoredObject.""" remove_listeners() file_content = b"SQLAlchemy Integration Test" doc_name = "Integration Doc" file_path = "sqlalchemy_single_async.bin" # 1. Prepare FileObject and save via backend initial_obj = FileObject( backend="local_test_store", filename="report.bin", to_filename=file_path, content_type="application/octet-stream", ) updated_obj = await initial_obj.save_async(data=file_content) # 2. Create and save model instance doc = Document(name=doc_name, attachment=updated_obj) async_session.add(doc) await async_session.commit() await async_session.refresh(doc) assert doc.id is not None assert doc.attachment is not None assert isinstance(doc.attachment, FileObject) assert doc.attachment.filename == "sqlalchemy_single_async.bin" assert doc.attachment.path == file_path assert doc.attachment.size == len(file_content) or doc.attachment.size is None assert doc.attachment.content_type == "application/octet-stream" assert doc.attachment.backend.key == "local_test_store" # 3. Retrieve content via loaded FileObject loaded_content = await doc.attachment.get_content_async() assert loaded_content == file_content @pytest.mark.xdist_group("file_object") async def test_obstore_multiple_files_async_no_listener( async_session: AsyncSession, storage_registry: StorageRegistry ) -> None: """Test saving and loading a model with multiple StoredObjects.""" remove_listeners() backend = storage_registry.get_backend("local_test_store") img1_content = b"img_data_1" img2_content = b"img_data_2" doc_name = "Multi Image Doc" img1_path = "img1.jpg" img2_path = "img2.png" # 1. Prepare FileObjects and save via backend obj1 = FileObject(backend=backend, filename="image1.jpg", to_filename=img1_path, content_type="image/jpeg") obj1_updated = await obj1.save_async(img1_content) obj2 = FileObject(backend=backend, filename="image2.png", to_filename=img2_path, content_type="image/png") obj2_updated = await obj2.save_async(img2_content) # 2. Create and save model instance with MutableList img_list = MutableList[FileObject]([obj1_updated, obj2_updated]) doc = Document(name=doc_name, images=img_list) async_session.add(doc) await async_session.commit() await async_session.refresh(doc) assert doc.id is not None assert doc.images is not None assert isinstance(doc.images, MutableList) assert len(doc.images) == 2 # Verify loaded objects loaded_obj1 = doc.images[0] loaded_obj2 = doc.images[1] assert isinstance(loaded_obj1, FileObject) assert loaded_obj1.filename == "img1.jpg" assert loaded_obj1.path == img1_path assert loaded_obj1.size == len(img1_content) or loaded_obj1.size is None assert loaded_obj1.backend and loaded_obj1.backend.driver == backend.driver assert isinstance(loaded_obj2, FileObject) assert loaded_obj2.filename == "img2.png" assert loaded_obj2.path == img2_path assert loaded_obj2.size == len(img2_content) or loaded_obj2.size is None assert loaded_obj2.backend and loaded_obj2.backend.driver == backend.driver # Verify content assert await loaded_obj1.get_content_async() == img1_content assert await loaded_obj2.get_content_async() == img2_content @pytest.mark.xdist_group("file_object") async def test_obstore_update_async_with_listener( async_session: AsyncSession, storage_registry: StorageRegistry ) -> None: """Test listener deletes old file when attribute is updated and session committed.""" # Set async context flag to enable async operations in the listener set_async_context(True) setup_file_object_listeners() backend = storage_registry.get_backend("local_test_store") old_content = b"Old file content" new_content = b"New file content" old_path = "old_file_async.txt" new_path = "new_file_async.txt" # Save initial file and model old_obj = FileObject(backend=backend, filename="old_file_async.txt", to_filename=old_path, content=old_content) # Make sure file is saved to the backend old_obj = await old_obj.save_async() doc = Document(name="DocToUpdate", attachment=old_obj) async_session.add(doc) await async_session.commit() await async_session.refresh(doc) # Verify old file exists assert await backend.get_content_async(old_path) == old_content # Prepare new file new_obj = FileObject(backend=backend, filename="new_file_async.txt", to_filename=new_path, content=new_content) # Update the document with the new file doc.attachment = new_obj async_session.add(doc) await async_session.commit() await async_session.refresh(doc) # Verify new file exists and attachment updated assert await backend.get_content_async(new_path) == new_content assert doc.attachment is not None and doc.attachment.path == new_path # pyright: ignore # Verify the listener deleted the old file with pytest.raises(FileNotFoundError): await backend.get_content_async(old_path) @pytest.mark.xdist_group("file_object") async def test_obstore_delete_async_on_update_clear_with_listener( async_session: AsyncSession, storage_registry: StorageRegistry ) -> None: """Test listener deletes file when attribute is cleared. Note that AsyncSession in SQLAlchemy 2.0 has limitations with event listeners. We will manually handle cleanup of files to ensure proper functionality. """ # Set async context flag to enable async operations in the listener set_async_context(True) setup_file_object_listeners() backend = storage_registry.get_backend("local_test_store") old_content = b"File to clear" old_path = "clear_me_async.log" # Save initial file and model old_obj = FileObject(backend=backend, filename="clear_me_async.log", to_filename=old_path, content=old_content) old_obj = await old_obj.save_async() # Make sure it's saved to the backend doc = Document(name="DocToClear", attachment=old_obj) async_session.add(doc) await async_session.commit() await async_session.refresh(doc) # Verify old file exists assert await backend.get_content_async(old_path) == old_content # Clear the attachment doc.attachment = None async_session.add(doc) await async_session.commit() await async_session.refresh(doc) # Verify attachment is None assert doc.attachment is None # Verify the listener deleted the file with pytest.raises(FileNotFoundError): await backend.get_content_async(old_path) @pytest.mark.xdist_group("file_object") async def test_obstore_delete_async_multiple_removed_with_listener( async_session: AsyncSession, storage_registry: StorageRegistry ) -> None: """Test listener deletes files removed from a multiple list. Note that AsyncSession in SQLAlchemy 2.0 has limitations with event listeners. MutableList tracking doesn't work properly with AsyncSession, so we use direct assignment for updates instead of mutating the list in-place. """ # Set async context flag to enable async operations in the listener set_async_context(True) setup_file_object_listeners() backend = storage_registry.get_backend("local_test_store") content1 = b"img1" content2 = b"img2" path1 = "img1_list_async.jpg" path2 = "img2_list_async.png" # Create file objects and save them obj1 = FileObject(backend=backend, filename="img1_list_async.jpg", to_filename=path1, content=content1) obj1 = await obj1.save_async() obj2 = FileObject(backend=backend, filename="img2_list_async.png", to_filename=path2, content=content2) obj2 = await obj2.save_async() # Create and save model with both images img_list = MutableList[FileObject]([obj1, obj2]) doc = Document(name="ImagesDoc", images=img_list) async_session.add(doc) await async_session.commit() await async_session.refresh(doc) # Verify files exist assert await backend.get_content_async(path1) == content1 assert await backend.get_content_async(path2) == content2 # Verify images are loaded assert doc.images is not None assert len(doc.images) == 2 # With AsyncSession, mutations to MutableList may not be tracked correctly. # Instead of mutating the list in place, we'll create a new list with only obj2 doc.images = MutableList[FileObject]([obj2]) async_session.add(doc) await async_session.commit() await async_session.refresh(doc) # Verify only one image remains assert doc.images is not None assert len(doc.images or []) == 1 assert doc.images[0].path == path2 # pyright: ignore # Verify first file is deleted and second still exists with pytest.raises(FileNotFoundError): await backend.get_content_async(path1) assert await backend.get_content_async(path2) == content2 @pytest.mark.xdist_group("file_object") async def test_file_object_invalid_init(storage_registry: StorageRegistry) -> None: """Test FileObject initialization with invalid parameters.""" backend = storage_registry.get_backend("local_test_store") test_content = b"Test content" test_path = Path("test.txt") # Test both content and source_path provided with pytest.raises(ValueError, match="Cannot provide both 'source_content' and 'source_path'"): FileObject( backend=backend, filename="test.txt", content=test_content, source_path=test_path, ) @pytest.mark.xdist_group("file_object") async def test_file_object_metadata_management(storage_registry: StorageRegistry) -> None: """Test FileObject metadata handling.""" backend = storage_registry.get_backend("local_test_store") initial_metadata = {"category": "test", "tags": ["sample"]} additional_metadata = {"priority": "high", "tags": ["important"]} # Create FileObject with initial metadata obj = FileObject( backend=backend, filename="test.txt", metadata=initial_metadata, ) assert obj.metadata == initial_metadata # Update metadata obj.update_metadata(additional_metadata) expected_metadata = { "category": "test", "tags": ["important"], # New tags override old ones "priority": "high", } assert obj.metadata == expected_metadata @pytest.mark.xdist_group("file_object") async def test_file_object_to_dict(storage_registry: StorageRegistry) -> None: """Test FileObject to_dict method.""" backend = storage_registry.get_backend("local_test_store") obj = FileObject( backend=backend, filename="test.txt", content_type="text/plain", size=100, last_modified=1234567890.0, checksum="abc123", etag="xyz789", version_id="v1", metadata={"category": "test"}, ) # Convert to dict obj_dict = obj.to_dict() assert obj_dict == { "filename": "test.txt", "content_type": "text/plain", "size": 100, "last_modified": 1234567890.0, "checksum": "abc123", "etag": "xyz789", "version_id": "v1", "metadata": {"category": "test"}, "backend": "local_test_store", } @pytest.mark.xdist_group("file_object") async def test_obstore_local_sign_urls(storage_registry: StorageRegistry) -> None: """Test FileObject sign and sign_async methods.""" backend = storage_registry.get_backend("local_test_store") test_content = b"Test content for signing" file_path = "test_sign.txt" # Create and save file obj = FileObject(backend=backend, filename="test.txt", to_filename=file_path) await obj.save_async(data=test_content) # Test sign method with pytest.raises(NotImplementedError, match=r"Error signing path test_sign.txt"): _ = obj.sign(expires_in=3600) # Test sign_async method with pytest.raises(NotImplementedError, match=r"Error signing path test_sign.txt"): _ = await obj.sign_async(expires_in=3600) with pytest.raises( NotImplementedError, match=r"Error signing path test_sign.txt", ): _ = obj.sign(for_upload=True) with pytest.raises( NotImplementedError, match=r"Error signing path test_sign.txt", ): _ = await obj.sign_async(for_upload=True) @pytest.mark.xdist_group("file_object") async def test_file_object_save_with_different_data_types(storage_registry: StorageRegistry) -> None: """Test FileObject save with different data types.""" backend = storage_registry.get_backend("local_test_store") test_content = b"Test content" file_path = "test_data_types.txt" # Test with bytes obj1 = FileObject(backend=backend, filename="test1.txt", to_filename=file_path, content=test_content) obj1.save() assert await obj1.get_content_async() == test_content # Test with Path import tempfile with tempfile.NamedTemporaryFile(mode="wb", delete=False) as f: f.write(test_content) temp_path = Path(f.name) obj2 = FileObject(backend=backend, filename="test2.txt", to_filename=file_path) await obj2.save_async(data=temp_path) assert await obj2.get_content_async() == test_content assert obj2.get_content() == test_content # Cleanup temp_path.unlink() @pytest.mark.xdist_group("file_object") async def test_file_object_pending_data_property(storage_registry: StorageRegistry) -> None: """Test FileObject has_pending_data property.""" backend = storage_registry.get_backend("local_test_store") test_content = b"Test content" test_path = Path("test.txt") # Test with content obj1 = FileObject(backend=backend, filename="test1.txt", content=test_content) assert obj1.has_pending_data # Test with source_path obj2 = FileObject(backend=backend, filename="test2.txt", source_path=test_path) assert obj2.has_pending_data # Test without pending data obj3 = FileObject(backend=backend, filename="test3.txt") assert not obj3.has_pending_data @pytest.mark.xdist_group("file_object") async def test_file_object_delete_methods(storage_registry: StorageRegistry) -> None: """Test FileObject delete and delete_async methods.""" backend = storage_registry.get_backend("local_test_store") test_content = b"Test content to delete" file_path = "test_delete.txt" # Create and save file obj = FileObject(backend=backend, filename="test.txt", to_filename=file_path) await obj.save_async(data=test_content) # Verify file exists assert await backend.get_content_async(file_path) == test_content # Test delete_async await obj.delete_async() with pytest.raises(FileNotFoundError): await backend.get_content_async(file_path) # Create and save file again await obj.save_async(data=test_content) assert await backend.get_content_async(file_path) == test_content # Test delete obj.delete() with pytest.raises(FileNotFoundError): await backend.get_content_async(file_path) @pytest.mark.xdist_group("file_object") async def test_obstore_backend_storage_registry_management(storage_registry: StorageRegistry) -> None: """Test StorageRegistry management methods.""" from obstore.store import MemoryStore from advanced_alchemy.types.file_object.backends.obstore import ObstoreBackend # Test registered_backends initial_backends = storage_registry.registered_backends() assert "local_test_store" in initial_backends assert "memory" in initial_backends # Test unregister_backend storage_registry.unregister_backend("local_test_store") assert "local_test_store" not in storage_registry.registered_backends() with pytest.raises(ImproperConfigurationError): storage_registry.get_backend("local_test_store") # Test clear_backends storage_registry.clear_backends() assert not storage_registry.registered_backends() # Test set_default_backend storage_registry.set_default_backend("advanced_alchemy.types.file_object.backends.obstore.ObstoreBackend") assert storage_registry.default_backend == "advanced_alchemy.types.file_object.backends.obstore.ObstoreBackend" # Test register_backend with string value storage_registry.register_backend("memory://", key="test_backend") assert "test_backend" in storage_registry.registered_backends() assert isinstance(storage_registry.get_backend("test_backend"), ObstoreBackend) # Test register_backend with StorageBackend instance test_backend = ObstoreBackend(fs=MemoryStore(), key="test_backend2") storage_registry.register_backend(test_backend) assert "test_backend2" in storage_registry.registered_backends() assert storage_registry.get_backend("test_backend2") is test_backend # Test error cases with pytest.raises(ImproperConfigurationError, match="key is required when registering a string value"): storage_registry.register_backend("memory://") # type: ignore[arg-type] with pytest.raises(ImproperConfigurationError, match="key is not allowed when registering a StorageBackend"): storage_registry.register_backend(test_backend, key="invalid_key") # type: ignore[arg-type] @pytest.mark.xdist_group("file_object") async def test_obstore_backend_storage_registry_error_handling(storage_registry: StorageRegistry) -> None: """Test StorageRegistry error handling.""" # Test get_backend with non-existent key with pytest.raises(ImproperConfigurationError, match="No storage backend registered with key nonexistent"): storage_registry.get_backend("nonexistent") # Test unregister_backend with non-existent key storage_registry.unregister_backend("nonexistent") # Should not raise an error # Test set_default_backend with invalid backend storage_registry.set_default_backend("invalid.module.path.Backend") @pytest.mark.xdist_group("file_object") async def test_fsspec_backend_basic_operations(storage_registry: StorageRegistry) -> None: """Test basic operations with FSSpec backend.""" try: import fsspec except ImportError: pytest.skip("fsspec not installed") # Create a local filesystem backend fs = fsspec.filesystem("file") backend = FSSpecBackend(fs=fs, key="fsspec_test") test_content = b"Test content" file_path = "test_fsspec.txt" # Test save and get content obj = FileObject(backend=backend, filename="test.txt", to_filename=file_path) await obj.save_async(data=test_content) assert await obj.get_content_async() == test_content # Test delete await obj.delete_async() with pytest.raises(FileNotFoundError): await obj.get_content_async() @pytest.mark.xdist_group("file_object") async def test_fsspec_backend_protocols(storage_registry: StorageRegistry) -> None: """Test FSSpec backend with different protocols.""" try: import fsspec except ImportError: pytest.skip("fsspec not installed") # Test local filesystem fs_local = fsspec.filesystem("file") backend_local = FSSpecBackend(fs=fs_local, key="fsspec_local") assert backend_local.protocol == "file" # Test memory filesystem fs_memory = fsspec.filesystem("memory") backend_memory = FSSpecBackend(fs=fs_memory, key="fsspec_memory") assert backend_memory.protocol == "memory" # Test with protocol string backend_from_string = FSSpecBackend(fs="file", key="fsspec_string") assert backend_from_string.protocol == "file" @pytest.mark.xdist_group("file_object") async def test_fsspec_backend_content_types(storage_registry: StorageRegistry) -> None: """Test FSSpec backend with different content types.""" try: import fsspec except ImportError: pytest.skip("fsspec not installed") fs = fsspec.filesystem("memory") backend = FSSpecBackend(fs=fs, key="fsspec_content") file_path = "test_content.txt" # Test with bytes content_bytes = b"Test bytes" obj_bytes = FileObject(backend=backend, filename="test_bytes.txt", to_filename=file_path) await obj_bytes.save_async(data=content_bytes) assert await obj_bytes.get_content_async() == content_bytes # Test with string content_str = "Test string" obj_str = FileObject(backend=backend, filename="test_str.txt", to_filename=file_path) await obj_str.save_async(data=content_str.encode("utf-8")) assert await obj_str.get_content_async() == content_str.encode("utf-8") @pytest.mark.xdist_group("file_object") async def test_fsspec_backend_multipart_upload(storage_registry: StorageRegistry) -> None: """Test FSSpec backend multipart upload.""" try: import fsspec except ImportError: pytest.skip("fsspec not installed") fs = fsspec.filesystem("memory") backend = FSSpecBackend(fs=fs, key="fsspec_multipart") file_path = "test_multipart.txt" # Create large content for multipart upload large_content = b"x" * (5 * 1024 * 1024 + 1) # 5MB + 1 byte obj = FileObject(backend=backend, filename="test.txt", to_filename=file_path) # Test with multipart upload await obj.save_async( data=large_content, use_multipart=True, chunk_size=1024 * 1024, # 1MB chunks max_concurrency=4, ) assert await obj.get_content_async() == large_content @pytest.mark.xdist_group("file_object") async def test_fsspec_backend_sign_urls(storage_registry: StorageRegistry, tmp_path: Path) -> None: """Test FSSpec backend URL signing.""" try: import fsspec except ImportError: pytest.skip("fsspec not installed") fs = fsspec.filesystem("file") backend = FSSpecBackend(fs=fs, key="fsspec_sign", prefix=str(tmp_path)) file_path = "test_sign.txt" # Create and save test file test_content = b"Test content for signing" obj = FileObject(backend=backend, filename="test.txt", to_filename=file_path) await obj.save_async(data=test_content) # Test sign method with pytest.raises(NotImplementedError, match="Signing URLs not supported by file backend"): _ = obj.sign(expires_in=3600) # Test sign_async method with pytest.raises(NotImplementedError, match="Signing URLs not supported by file backend"): _ = await obj.sign_async(expires_in=3600) # Test for_upload parameter with pytest.raises( NotImplementedError, match=r"Generating signed URLs for upload is generally not supported by fsspec's generic sign method.", ): _ = obj.sign(for_upload=True) @pytest.mark.xdist_group("file_object") def test_file_object_sync_save_and_get_content(storage_registry: StorageRegistry) -> None: """Test FileObject synchronous save and get_content methods.""" backend = storage_registry.get_backend("local_test_store") test_content = b"Test synchronous content" file_path = "test_sync_save.txt" # Create FileObject with content obj = FileObject(backend=backend, filename="test_sync.txt", to_filename=file_path, content=test_content) # Test synchronous save method updated_obj = obj.save() # Verify save worked correctly assert updated_obj is obj # Should update in-place assert obj.path == file_path assert obj.size == len(test_content) or obj.size is None # Test synchronous get_content method retrieved_content = obj.get_content() assert retrieved_content == test_content # Clean up obj.delete() @pytest.mark.xdist_group("file_object") def test_file_object_save_with_source_path(storage_registry: StorageRegistry, tmp_path: Path) -> None: """Test FileObject save with source_path.""" backend = storage_registry.get_backend("local_test_store") test_content = b"Test content from file" file_path = "test_source_path.txt" # Create a temporary file source_file = tmp_path / "source.txt" source_file.write_bytes(test_content) # Create FileObject with source_path obj = FileObject(backend=backend, filename="test_source.txt", to_filename=file_path, source_path=source_file) # Test save method with source_path obj.save() # Verify save worked correctly retrieved_content = obj.get_content() assert retrieved_content == test_content # Clean up obj.delete() @pytest.mark.xdist_group("file_object") def test_file_object_equality_and_hash(storage_registry: StorageRegistry) -> None: """Test FileObject __eq__ and __hash__ methods.""" backend = storage_registry.get_backend("local_test_store") # Create two identical FileObjects obj1 = FileObject(backend=backend, filename="test.txt", to_filename="same_path.txt") obj2 = FileObject(backend=backend, filename="different.txt", to_filename="same_path.txt") # They should be equal because they have the same path and backend assert obj1 == obj2 assert hash(obj1) == hash(obj2) # Create a different FileObject obj3 = FileObject(backend=backend, filename="test.txt", to_filename="different_path.txt") # They should not be equal because they have different paths assert obj1 != obj3 assert hash(obj1) != hash(obj3) # Compare with a non-FileObject assert obj1 != "not a file object" @pytest.mark.xdist_group("file_object") def test_file_object_property_setters(storage_registry: StorageRegistry) -> None: """Test FileObject property setters.""" backend = storage_registry.get_backend("local_test_store") obj = FileObject(backend=backend, filename="test.txt") # Test size property obj.size = 100 assert obj.size == 100 # Test last_modified property timestamp = 1234567890.0 obj.last_modified = timestamp assert obj.last_modified == timestamp # Test checksum property obj.checksum = "abc123" assert obj.checksum == "abc123" # Test etag property obj.etag = "etag123" assert obj.etag == "etag123" # Test version_id property obj.version_id = "v1" assert obj.version_id == "v1" # Test metadata property new_metadata = {"key": "value"} obj.metadata = new_metadata assert obj.metadata == new_metadata @pytest.mark.xdist_group("file_object") def test_file_object_repr(storage_registry: StorageRegistry) -> None: """Test FileObject __repr__ method.""" backend = storage_registry.get_backend("local_test_store") # Create a FileObject with all attributes set obj = FileObject( backend=backend, filename="test.txt", size=100, content_type="text/plain", last_modified=1234567890.0, etag="etag123", version_id="v1", ) # Test __repr__ method repr_str = repr(obj) assert "FileObject" in repr_str assert "filename=test.txt" in repr_str assert "backend=local_test_store" in repr_str assert "size=100" in repr_str assert "content_type=text/plain" in repr_str assert "etag=etag123" in repr_str assert "last_modified=1234567890.0" in repr_str assert "version_id=v1" in repr_str @pytest.mark.xdist_group("file_object") def test_file_object_content_type_guessing(storage_registry: StorageRegistry) -> None: """Test content_type guessing from filename.""" backend = storage_registry.get_backend("local_test_store") # Test common file types file_types = { "test.txt": "text/plain", "image.jpg": "image/jpeg", "doc.pdf": "application/pdf", "data.json": "application/json", "unknown": "application/octet-stream", } for filename, expected_type in file_types.items(): obj = FileObject(backend=backend, filename=filename) assert obj.content_type == expected_type @pytest.mark.xdist_group("file_object") def test_file_object_save_no_data(storage_registry: StorageRegistry) -> None: """Test save method with no data.""" backend = storage_registry.get_backend("local_test_store") # Create a FileObject with no content or source_path obj = FileObject(backend=backend, filename="test.txt") # Saving with no data should raise a TypeError with pytest.raises(TypeError, match="No data provided and no pending content/path found to save."): obj.save() @pytest.mark.xdist_group("file_object") async def test_file_object_save_async_no_data(storage_registry: StorageRegistry) -> None: """Test save_async method with no data.""" backend = storage_registry.get_backend("local_test_store") # Create a FileObject with no content or source_path obj = FileObject(backend=backend, filename="test.txt") # Saving with no data should raise a TypeError with pytest.raises(TypeError, match="No data provided and no pending content/path found to save."): await obj.save_async() @pytest.mark.xdist_group("file_object") def test_obstore_backend_sqlalchemy_single_file_persist_sync( session: Session, storage_registry: StorageRegistry ) -> None: """Test saving and loading a model with a single StoredObject using synchronous SQLAlchemy session.""" remove_listeners() file_content = b"SQLAlchemy Sync Integration Test" doc_name = "Sync Integration Doc" file_path = "sqlalchemy_single_sync.bin" # 1. Prepare FileObject and save via backend initial_obj = FileObject( backend="local_test_store", filename="report.bin", to_filename=file_path, content_type="application/octet-stream", ) updated_obj = initial_obj.save(data=file_content) # 2. Create and save model instance doc = Document(name=doc_name, attachment=updated_obj) session.add(doc) session.commit() session.refresh(doc) assert doc.id is not None assert doc.attachment is not None assert isinstance(doc.attachment, FileObject) assert doc.attachment.filename == "sqlalchemy_single_sync.bin" assert doc.attachment.path == file_path assert doc.attachment.size == len(file_content) or doc.attachment.size is None assert doc.attachment.content_type == "application/octet-stream" assert doc.attachment.backend.key == "local_test_store" # 3. Retrieve content via loaded FileObject loaded_content = doc.attachment.get_content() assert loaded_content == file_content @pytest.mark.xdist_group("file_object") async def test_obstore_backend_listener_sqlalchemy_single_file_persist_async( async_session: AsyncSession, storage_registry: StorageRegistry ) -> None: """Test saving and loading a model with a single StoredObject using synchronous SQLAlchemy session.""" setup_file_object_listeners() set_async_context(True) file_content = b"SQLAlchemy Async Integration Test" doc_name = "Sync Integration Doc" file_path = "sqlalchemy_single_async.bin" # 1. Prepare FileObject and save via backend initial_obj = FileObject( backend="local_test_store", filename="report.bin", to_filename=file_path, content_type="application/octet-stream", content=file_content, ) # 2. Create and save model instance doc = Document(name=doc_name, attachment=initial_obj) async_session.add(doc) await async_session.commit() await async_session.refresh(doc) assert doc.id is not None assert doc.attachment is not None assert isinstance(doc.attachment, FileObject) assert doc.attachment.filename == "sqlalchemy_single_async.bin" assert doc.attachment.path == file_path assert doc.attachment.size == len(file_content) or doc.attachment.size is None assert doc.attachment.content_type == "application/octet-stream" assert doc.attachment.backend.key == "local_test_store" # 3. Retrieve content via loaded FileObject loaded_content = doc.attachment.get_content() assert loaded_content == file_content @pytest.mark.xdist_group("file_object") def test_obstore_backend_sqlalchemy_multiple_files_persist_sync( session: Session, storage_registry: StorageRegistry ) -> None: """Test saving and loading a model with multiple StoredObjects using synchronous SQLAlchemy session.""" remove_listeners() backend = storage_registry.get_backend("local_test_store") img1_content = b"img_data_1_sync" img2_content = b"img_data_2_sync" doc_name = "Multi Image Doc Sync" img1_path = "img1_list_sync.jpg" img2_path = "img2_list_sync.png" # 1. Prepare FileObjects and save via backend obj1 = FileObject( backend=backend, filename="image1_list_sync.jpg", to_filename=img1_path, content_type="image/jpeg" ) obj1_updated = obj1.save(img1_content) obj2 = FileObject(backend=backend, filename="image2_list_sync.png", to_filename=img2_path, content_type="image/png") obj2_updated = obj2.save(img2_content) # 2. Create and save model instance with MutableList img_list = MutableList[FileObject]([obj1_updated, obj2_updated]) doc = Document(name=doc_name, images=img_list) session.add(doc) session.commit() session.refresh(doc) assert doc.id is not None assert doc.images is not None assert isinstance(doc.images, MutableList) assert len(doc.images) == 2 # Verify loaded objects loaded_obj1 = doc.images[0] loaded_obj2 = doc.images[1] assert isinstance(loaded_obj1, FileObject) assert loaded_obj1.filename == "img1_list_sync.jpg" assert loaded_obj1.path == img1_path assert loaded_obj1.size == len(img1_content) or loaded_obj1.size is None assert loaded_obj1.backend and loaded_obj1.backend.driver == backend.driver assert isinstance(loaded_obj2, FileObject) assert loaded_obj2.filename == "img2_list_sync.png" assert loaded_obj2.path == img2_path assert loaded_obj2.size == len(img2_content) or loaded_obj2.size is None assert loaded_obj2.backend and loaded_obj2.backend.driver == backend.driver # Verify content assert loaded_obj1.get_content() == img1_content assert loaded_obj2.get_content() == img2_content @pytest.mark.xdist_group("file_object") def test_obstore_backend_listener_delete_on_update_clear_sync( session: Session, storage_registry: StorageRegistry ) -> None: """Test listener deletes old file when attribute is cleared using synchronous SQLAlchemy session.""" setup_file_object_listeners() backend = storage_registry.get_backend("local_test_store") old_content = b"File to clear sync" old_path = "clear_me_sync.log" # Save initial file and model old_obj = FileObject(backend=backend, filename="clear.log", to_filename=old_path, content=old_content) doc = Document(name="DocToClearSync", attachment=old_obj) session.add(doc) session.commit() session.refresh(doc) # Verify old file exists assert backend.get_content(old_path) == old_content # Clear the attachment doc.attachment = None session.add(doc) session.commit() session.refresh(doc) # Verify attachment is None assert doc.attachment is None # Verify the listener deleted the file from storage with pytest.raises(FileNotFoundError): backend.get_content(old_path) @pytest.mark.flaky(reruns=5) @pytest.mark.xdist_group("file_object") async def test_obstore_backend_listener_delete_on_update_clear_async( async_session: AsyncSession, storage_registry: StorageRegistry ) -> None: """Test listener deletes old file when attribute is cleared using asynchronous SQLAlchemy session.""" setup_file_object_listeners() set_async_context(True) backend = storage_registry.get_backend("local_test_store") old_content = b"File to clear sync" old_path = "clear_me_sync.log" # Save initial file and model old_obj = FileObject(backend=backend, filename="clear.log", to_filename=old_path, content=old_content) doc = Document(name="DocToClearSync", attachment=old_obj) async_session.add(doc) await async_session.commit() await async_session.refresh(doc) # Verify old file exists assert await backend.get_content_async(old_path) == old_content # Clear the attachment doc.attachment = None async_session.add(doc) await async_session.commit() await async_session.refresh(doc) # Verify attachment is None assert doc.attachment is None # Verify the listener deleted the file from storage with pytest.raises(FileNotFoundError): await backend.get_content_async(old_path) @pytest.mark.flaky(reruns=5) @pytest.mark.xdist_group("file_object") def test_obstore_backend_listener_update_file_object_sync(session: Session, storage_registry: StorageRegistry) -> None: """Test listener deletes old file when attribute is updated and session committed using synchronous SQLAlchemy session.""" setup_file_object_listeners() backend = storage_registry.get_backend("local_test_store") old_content = b"Old file content sync" new_content = b"New file content sync" old_path = "old_file_sync_update.txt" new_path = "new_file_sync_update.txt" # Save initial file and model old_obj = FileObject( backend=backend, filename="old_file_sync_update.txt", to_filename=old_path, content=old_content ) doc = Document(name="DocToUpdateSync", attachment=old_obj) session.add(doc) session.commit() session.refresh(doc) # Verify old file exists assert backend.get_content(old_path) == old_content # Update the document's attachment (inline creation) new_obj = FileObject( backend=backend, filename="new_file_sync_update.txt", to_filename=new_path, content=new_content ) doc.attachment = new_obj session.add(doc) # Add again as it's modified session.commit() # Listener should save new_obj and queue deletion of old_obj session.refresh(doc) # Verify new file exists and attachment updated assert backend.get_content(new_path) == new_content assert doc.attachment is not None and doc.attachment.path == new_path # pyright: ignore # Verify the listener deleted the old file from storage with pytest.raises(FileNotFoundError): backend.get_content(old_path) @pytest.mark.flaky(reruns=5) @pytest.mark.xdist_group("file_object") async def test_obstore_backend_listener_update_file_object_async( async_session: AsyncSession, storage_registry: StorageRegistry ) -> None: """Test listener deletes old file when attribute is updated and session committed using asynchronous SQLAlchemy session.""" setup_file_object_listeners() backend = storage_registry.get_backend("local_test_store") old_content = b"Old file content sync" new_content = b"New file content sync" old_path = "old_file_async_update.txt" new_path = "new_file_async_update.txt" # Save initial file and model old_obj = FileObject( backend=backend, filename="old_file_async_update.txt", to_filename=old_path, content=old_content ) doc = Document(name="DocToUpdateSync", attachment=old_obj) async_session.add(doc) await async_session.commit() await async_session.refresh(doc) # Verify old file exists assert backend.get_content(old_path) == old_content # Update the document's attachment (inline creation) new_obj = FileObject( backend=backend, filename="new_file_async_update.txt", to_filename=new_path, content=new_content ) doc.attachment = new_obj async_session.add(doc) # Add again as it's modified await async_session.commit() # Listener should save new_obj and queue deletion of old_obj await async_session.refresh(doc) assert backend.get_content(new_path) == new_content assert doc.attachment is not None and doc.attachment.path == new_path # pyright: ignore # Verify the listener deleted the old file from storage with pytest.raises(FileNotFoundError): backend.get_content(old_path) @pytest.mark.flaky(reruns=5) @pytest.mark.xdist_group("file_object") def test_obstore_backend_listener_delete_multiple_removed_sync( session: Session, storage_registry: StorageRegistry ) -> None: """Test listener deletes files removed from a multiple list using synchronous SQLAlchemy session.""" set_async_context(False) setup_file_object_listeners() backend = storage_registry.get_backend("local_test_store") content1 = b"img1_sync_multi" content2 = b"img2_sync_multi" path1 = "multi_del_1_sync.dat" path2 = "multi_del_2_sync.dat" # Save files obj1 = FileObject(backend=backend, filename=path1, content=content1) obj2 = FileObject(backend=backend, filename=path2, content=content2) # Create model with initial list doc = Document(name="MultiDeleteSyncTest", images=[obj1, obj2]) session.add(doc) session.commit() session.refresh(doc) # Verify all files exist assert backend.get_content(path1) == content1 assert backend.get_content(path2) == content2 # Remove items from the list (triggers MutableList tracking) assert doc.images is not None current_images = list(doc.images) # Create standard list copy removed_item = current_images.pop(1) # Mutate copy assert removed_item.path == obj2.path del current_images[0] # Mutate copy assert len(current_images) == 0 doc.images = MutableList(current_images) # Wrap in MutableList before reassignment session.add(doc) # Commit the session to trigger listener session.commit() session.refresh(doc) assert doc.images == [] # Verify the listener deleted the files with pytest.raises(FileNotFoundError): backend.get_content(path1) with pytest.raises(FileNotFoundError): backend.get_content(path2) @pytest.mark.flaky(reruns=5) @pytest.mark.xdist_group("file_object") async def test_obstore_backend_listener_delete_multiple_removed_async( async_session: AsyncSession, storage_registry: StorageRegistry ) -> None: """Test listener deletes files removed from a multiple list using asynchronous SQLAlchemy session.""" set_async_context(True) setup_file_object_listeners() backend = storage_registry.get_backend("local_test_store") content1 = b"img1_async_multi" content2 = b"img2_async_multi" path1 = "multi_del_1_async.dat" path2 = "multi_del_2_async.dat" # Save files obj1 = FileObject(backend=backend, filename=path1, content=content1) obj2 = FileObject(backend=backend, filename=path2, content=content2) # Create model with initial list doc = Document(name="MultiDeleteAsyncTest", images=[obj1, obj2]) async_session.add(doc) await async_session.commit() await async_session.refresh(doc) # Verify all files exist assert await backend.get_content_async(path1) == content1 assert await backend.get_content_async(path2) == content2 # Remove items from the list (triggers MutableList tracking) assert doc.images is not None current_images = list(doc.images) # Create standard list copy removed_item = current_images.pop(1) # Mutate copy assert removed_item.path == obj2.path del current_images[0] # Mutate copy assert len(current_images) == 0 doc.images = MutableList(current_images) # Wrap in MutableList before reassignment # Commit the session to trigger listener await async_session.commit() await async_session.refresh(doc) # Verify the listener deleted the files with pytest.raises(FileNotFoundError): await backend.get_content_async(path1) with pytest.raises(FileNotFoundError): await backend.get_content_async(path2)