import asyncio import contextlib import json import os import tempfile import uuid from contextlib import suppress from pathlib import Path from unittest.mock import AsyncMock, MagicMock # we need to import tmpdir import anyio import pytest from asgi_lifespan import LifespanManager from httpx import ASGITransport, AsyncClient from langflow.api.v2.files import ( delete_all_files, delete_file, delete_files_batch, is_permanent_storage_failure, ) from langflow.api.v2.mcp import get_mcp_file from langflow.main import create_app from langflow.services.auth.utils import get_password_hash from langflow.services.database.models.api_key.model import ApiKey, UnmaskedApiKeyRead from langflow.services.database.models.user.model import User, UserRead from lfx.services.deps import session_scope from sqlalchemy.orm import selectinload from sqlmodel import select from tests.conftest import _delete_transactions_and_vertex_builds @pytest.fixture(name="files_created_api_key") async def files_created_api_key(files_client, files_active_user): # noqa: ARG001 hashed = get_password_hash("random_key") api_key = ApiKey( name="files_created_api_key", user_id=files_active_user.id, api_key="random_key", hashed_api_key=hashed, ) async with session_scope() as session: stmt = select(ApiKey).where(ApiKey.api_key == api_key.api_key) if existing_api_key := (await session.exec(stmt)).first(): existing_api_key = UnmaskedApiKeyRead.model_validate(existing_api_key, from_attributes=True) yield existing_api_key return session.add(api_key) await session.flush() await session.refresh(api_key) api_key = UnmaskedApiKeyRead.model_validate(api_key, from_attributes=True) yield api_key async with session_scope() as session: db_key = await session.get(ApiKey, api_key.id) if db_key: await session.delete(db_key) @pytest.fixture(name="files_active_user") async def files_active_user(files_client): # noqa: ARG001 async with session_scope() as session: user = User( username="files_active_user", password=get_password_hash("testpassword"), is_active=True, is_superuser=False, ) stmt = select(User).where(User.username == user.username) if active_user := (await session.exec(stmt)).first(): user = active_user else: session.add(user) await session.flush() await session.refresh(user) user = UserRead.model_validate(user, from_attributes=True) yield user # Clean up # Now cleanup transactions, vertex_build async with session_scope() as session: user = await session.get(User, user.id, options=[selectinload(User.flows)]) await _delete_transactions_and_vertex_builds(session, user.flows) await session.delete(user) @pytest.fixture def max_file_size_upload_fixture(monkeypatch): monkeypatch.setenv("LANGFLOW_MAX_FILE_SIZE_UPLOAD", "1") yield monkeypatch.undo() @pytest.fixture def max_file_size_upload_10mb_fixture(monkeypatch): monkeypatch.setenv("LANGFLOW_MAX_FILE_SIZE_UPLOAD", "10") yield monkeypatch.undo() @pytest.fixture(name="files_client") async def files_client_fixture( monkeypatch, request, ): # Set the database url to a test database if "noclient" in request.keywords: yield else: def init_app(): db_dir = tempfile.mkdtemp() db_path = Path(db_dir) / "test.db" monkeypatch.setenv("LANGFLOW_DATABASE_URL", f"sqlite:///{db_path}") monkeypatch.setenv("LANGFLOW_AUTO_LOGIN", "false") from lfx.services.manager import get_service_manager get_service_manager().factories.clear() get_service_manager().services.clear() # Clear the services cache app = create_app() return app, db_path app, db_path = await asyncio.to_thread(init_app) async with ( LifespanManager(app, startup_timeout=None, shutdown_timeout=60) as manager, AsyncClient(transport=ASGITransport(app=manager.app), base_url="http://testserver/") as client, ): yield client # app.dependency_overrides.clear() monkeypatch.undo() # clear the temp db with suppress(FileNotFoundError): await anyio.Path(db_path).unlink() async def test_upload_file(files_client, files_created_api_key): headers = {"x-api-key": files_created_api_key.api_key} response = await files_client.post( "api/v2/files", files={"file": ("test.txt", b"test content")}, headers=headers, ) assert response.status_code == 201, f"Expected 201, got {response.status_code}: {response.json()}" response_json = response.json() assert "id" in response_json async def test_should_not_persist_in_my_files_when_upload_is_ephemeral(files_client, files_created_api_key): """Ephemeral uploads save the file to storage but do NOT create a UserFile DB record. This is the expected behavior for chat playground uploads in Desktop, where the file must be servable (for chat history) but should not appear in the user's 'My Files' list. """ headers = {"x-api-key": files_created_api_key.api_key} # Upload with ephemeral=true response = await files_client.post( "api/v2/files", files={"file": ("playground_image.png", b"fake image content")}, params={"ephemeral": "true"}, headers=headers, ) assert response.status_code == 201, f"Expected 201, got {response.status_code}: {response.text}" upload_response = response.json() assert "path" in upload_response # The file must NOT appear in the user's file list list_response = await files_client.get("api/v2/files", headers=headers) assert list_response.status_code == 200 file_names = [f["name"] for f in list_response.json()] assert "playground_image" not in file_names, ( f"Ephemeral file should not appear in My Files, but found: {file_names}" ) # The file is saved in storage and the response includes a valid path file_path = upload_response["path"] assert file_path, "Ephemeral upload should return a non-empty path" # Path format: {user_id}/{stored_file_name} parts = file_path.split("/") assert len(parts) == 2, f"Expected path format 'user_id/filename', got: {file_path}" async def test_should_return_path_with_forward_slashes_when_uploading_file(files_client, files_created_api_key): """Upload response path must use forward slashes on all platforms. On Windows, pathlib.Path serializes with backslashes, but the GET list endpoint returns the raw DB string with forward slashes. If the POST response uses backslashes, the frontend cannot match them with selectedFiles.includes(file.path), leaving checkboxes unchecked. """ headers = {"x-api-key": files_created_api_key.api_key} response = await files_client.post( "api/v2/files", files={"file": ("test_path.txt", b"path test content")}, headers=headers, ) assert response.status_code == 201 upload_path = response.json()["path"] assert "\\" not in upload_path, ( f"Upload response path contains backslashes: '{upload_path}'. " "Path must use forward slashes on all platforms for frontend compatibility." ) # Verify the upload path matches what GET /files returns list_response = await files_client.get("api/v2/files", headers=headers) assert list_response.status_code == 200 listed_paths = [f["path"] for f in list_response.json()] assert upload_path in listed_paths, ( f"Upload path '{upload_path}' not found in listed paths {listed_paths}. " "POST and GET must return identical path strings." ) async def test_download_file(files_client, files_created_api_key): headers = {"x-api-key": files_created_api_key.api_key} # First upload a file response = await files_client.post( "api/v2/files", files={"file": ("test.txt", b"test content")}, headers=headers, ) assert response.status_code == 201 upload_response = response.json() # Then try to download it response = await files_client.get(f"api/v2/files/{upload_response['id']}", headers=headers) assert response.status_code == 200 assert response.content == b"test content" async def test_download_file_not_found(files_client, files_created_api_key): """Test that downloading a non-existent file returns 404 error.""" headers = {"x-api-key": files_created_api_key.api_key} # Try to download a file that doesn't exist fake_file_id = "00000000-0000-0000-0000-000000000000" response = await files_client.get(f"api/v2/files/{fake_file_id}", headers=headers) assert response.status_code == 404 error_response = response.json() assert "File not found" in error_response["detail"] async def test_list_files(files_client, files_created_api_key): headers = {"x-api-key": files_created_api_key.api_key} # First upload a file response = await files_client.post( "api/v2/files", files={"file": ("test.txt", b"test content")}, headers=headers, ) assert response.status_code == 201 # Then list the files response = await files_client.get("api/v2/files", headers=headers) assert response.status_code == 200 files = response.json() assert len(files) == 1 async def test_delete_file(files_client, files_created_api_key): headers = {"x-api-key": files_created_api_key.api_key} response = await files_client.post( "api/v2/files", files={"file": ("test.txt", b"test content")}, headers=headers, ) assert response.status_code == 201 upload_response = response.json() response = await files_client.delete(f"api/v2/files/{upload_response['id']}", headers=headers) assert response.status_code == 200 assert response.json() == {"detail": "File test deleted successfully"} async def test_edit_file(files_client, files_created_api_key): headers = {"x-api-key": files_created_api_key.api_key} # First upload a file response = await files_client.post( "api/v2/files", files={"file": ("test.txt", b"test content")}, headers=headers, ) assert response.status_code == 201 upload_response = response.json() # Then list the files response = await files_client.put(f"api/v2/files/{upload_response['id']}?name=potato.txt", headers=headers) assert response.status_code == 200 file = response.json() assert file["name"] == "potato.txt" async def test_upload_list_delete_and_validate_files(files_client, files_created_api_key): headers = {"x-api-key": files_created_api_key.api_key} # Upload two files response1 = await files_client.post( "api/v2/files", files={"file": ("file1.txt", b"content1")}, headers=headers, ) assert response1.status_code == 201 file1 = response1.json() response2 = await files_client.post( "api/v2/files", files={"file": ("file2.txt", b"content2")}, headers=headers, ) assert response2.status_code == 201 file2 = response2.json() # List files and validate both are present response = await files_client.get("api/v2/files", headers=headers) assert response.status_code == 200 files = response.json() file_names = [f["name"] for f in files] file_ids = [f["id"] for f in files] assert file1["name"] in file_names assert file2["name"] in file_names assert file1["id"] in file_ids assert file2["id"] in file_ids assert len(files) == 2 # Delete one file response = await files_client.delete(f"api/v2/files/{file1['id']}", headers=headers) assert response.status_code == 200 # List files again and validate only the other remains response = await files_client.get("api/v2/files", headers=headers) assert response.status_code == 200 files = response.json() file_names = [f["name"] for f in files] file_ids = [f["id"] for f in files] assert file1["name"] not in file_names assert file1["id"] not in file_ids assert file2["name"] in file_names assert file2["id"] in file_ids assert len(files) == 1 async def test_upload_files_with_same_name_creates_unique_names(files_client, files_created_api_key): """Test that uploading files with the same name creates unique filenames.""" headers = {"x-api-key": files_created_api_key.api_key} # Upload first file response1 = await files_client.post( "api/v2/files", files={"file": ("duplicate.txt", b"content1")}, headers=headers, ) assert response1.status_code == 201 file1 = response1.json() assert file1["name"] == "duplicate" # Upload second file with same name response2 = await files_client.post( "api/v2/files", files={"file": ("duplicate.txt", b"content2")}, headers=headers, ) assert response2.status_code == 201 file2 = response2.json() assert file2["name"] == "duplicate (1)" # Upload third file with same name response3 = await files_client.post( "api/v2/files", files={"file": ("duplicate.txt", b"content3")}, headers=headers, ) assert response3.status_code == 201 file3 = response3.json() assert file3["name"] == "duplicate (2)" # Verify all files can be downloaded with their unique content download1 = await files_client.get(f"api/v2/files/{file1['id']}", headers=headers) assert download1.status_code == 200 assert download1.content == b"content1" download2 = await files_client.get(f"api/v2/files/{file2['id']}", headers=headers) assert download2.status_code == 200 assert download2.content == b"content2" download3 = await files_client.get(f"api/v2/files/{file3['id']}", headers=headers) assert download3.status_code == 200 assert download3.content == b"content3" # List files and verify all three are present with unique names response = await files_client.get("api/v2/files", headers=headers) assert response.status_code == 200 files = response.json() file_names = [f["name"] for f in files] assert "duplicate" in file_names assert "duplicate (1)" in file_names assert "duplicate (2)" in file_names assert len(files) == 3 async def test_upload_files_without_extension_creates_unique_names(files_client, files_created_api_key): """Test that uploading files without extensions also creates unique filenames.""" headers = {"x-api-key": files_created_api_key.api_key} # Upload first file without extension response1 = await files_client.post( "api/v2/files", files={"file": ("noextension", b"content1")}, headers=headers, ) assert response1.status_code == 201 file1 = response1.json() assert file1["name"] == "noextension" # Upload second file with same name response2 = await files_client.post( "api/v2/files", files={"file": ("noextension", b"content2")}, headers=headers, ) assert response2.status_code == 201 file2 = response2.json() assert file2["name"] == "noextension (1)" # Verify both files can be downloaded download1 = await files_client.get(f"api/v2/files/{file1['id']}", headers=headers) assert download1.status_code == 200 assert download1.content == b"content1" download2 = await files_client.get(f"api/v2/files/{file2['id']}", headers=headers) assert download2.status_code == 200 assert download2.content == b"content2" async def test_upload_files_with_different_extensions_same_name(files_client, files_created_api_key): """Test that files with same root name but different extensions create unique names.""" headers = {"x-api-key": files_created_api_key.api_key} # Upload file with .txt extension response1 = await files_client.post( "api/v2/files", files={"file": ("document.txt", b"text content")}, headers=headers, ) assert response1.status_code == 201 file1 = response1.json() assert file1["name"] == "document" # Upload file with .md extension and same root name response2 = await files_client.post( "api/v2/files", files={"file": ("document.md", b"markdown content")}, headers=headers, ) assert response2.status_code == 201 file2 = response2.json() assert file2["name"] == "document (1)" # Upload another .txt file with same root name response3 = await files_client.post( "api/v2/files", files={"file": ("document.txt", b"more text content")}, headers=headers, ) assert response3.status_code == 201 file3 = response3.json() assert file3["name"] == "document (2)" async def test_mcp_servers_file_replacement(files_client, files_created_api_key, files_active_user): """Test that _mcp_servers file gets replaced instead of creating unique names.""" headers = {"x-api-key": files_created_api_key.api_key} mcp_file_ext = await get_mcp_file(files_active_user, extension=True) mcp_file = await get_mcp_file(files_active_user) # Upload first _mcp_servers file response1 = await files_client.post( "api/v2/files", files={"file": (mcp_file_ext, b'{"servers": ["server1"]}')}, headers=headers, ) assert response1.status_code == 201 file1 = response1.json() assert file1["name"] == mcp_file # Upload second _mcp_servers file - should replace the first one response2 = await files_client.post( "api/v2/files", files={"file": (mcp_file_ext, b'{"servers": ["server2"]}')}, headers=headers, ) assert response2.status_code == 201 file2 = response2.json() assert file2["name"] == mcp_file # Note: _mcp_servers files are filtered out from the regular file list # This is expected behavior since they're managed separately response = await files_client.get("api/v2/files", headers=headers) assert response.status_code == 200 files = response.json() mcp_files = [f for f in files if f["name"] == mcp_file] assert len(mcp_files) == 0 # MCP servers files are filtered out from regular list # Verify the second file can be downloaded with the updated content download2 = await files_client.get(f"api/v2/files/{file2['id']}", headers=headers) assert download2.status_code == 200 assert download2.content == b'{"servers": ["server2"]}' # Verify the first file no longer exists (should return 404) download1 = await files_client.get(f"api/v2/files/{file1['id']}", headers=headers) assert download1.status_code == 404 # Verify the file IDs are different (new file replaced old one) assert file1["id"] != file2["id"] async def test_unique_filename_counter_handles_gaps(files_client, files_created_api_key): """Test that the unique filename counter properly handles gaps in sequence.""" headers = {"x-api-key": files_created_api_key.api_key} # Upload original file response1 = await files_client.post( "api/v2/files", files={"file": ("gaptest.txt", b"content1")}, headers=headers, ) assert response1.status_code == 201 file1 = response1.json() assert file1["name"] == "gaptest" # Upload second file (should be gaptest (1)) response2 = await files_client.post( "api/v2/files", files={"file": ("gaptest.txt", b"content2")}, headers=headers, ) assert response2.status_code == 201 file2 = response2.json() assert file2["name"] == "gaptest (1)" # Upload third file (should be gaptest (2)) response3 = await files_client.post( "api/v2/files", files={"file": ("gaptest.txt", b"content3")}, headers=headers, ) assert response3.status_code == 201 file3 = response3.json() assert file3["name"] == "gaptest (2)" # Delete the middle file (gaptest (1)) delete_response = await files_client.delete(f"api/v2/files/{file2['id']}", headers=headers) assert delete_response.status_code == 200 # Upload another file - should be gaptest (3), not filling the gap response4 = await files_client.post( "api/v2/files", files={"file": ("gaptest.txt", b"content4")}, headers=headers, ) assert response4.status_code == 201 file4 = response4.json() assert file4["name"] == "gaptest (3)" # Verify final state response = await files_client.get("api/v2/files", headers=headers) assert response.status_code == 200 files = response.json() file_names = [f["name"] for f in files] assert "gaptest" in file_names assert "gaptest (1)" not in file_names # deleted assert "gaptest (2)" in file_names assert "gaptest (3)" in file_names assert len([name for name in file_names if name.startswith("gaptest")]) == 3 async def test_unique_filename_path_storage(files_client, files_created_api_key): """Test that files with unique names are stored with unique paths.""" headers = {"x-api-key": files_created_api_key.api_key} # Upload two files with same name response1 = await files_client.post( "api/v2/files", files={"file": ("pathtest.txt", b"path content 1")}, headers=headers, ) assert response1.status_code == 201 file1 = response1.json() response2 = await files_client.post( "api/v2/files", files={"file": ("pathtest.txt", b"path content 2")}, headers=headers, ) assert response2.status_code == 201 file2 = response2.json() # Verify both files have different paths and can be downloaded independently assert file1["path"] != file2["path"] download1 = await files_client.get(f"api/v2/files/{file1['id']}", headers=headers) assert download1.status_code == 200 assert download1.content == b"path content 1" download2 = await files_client.get(f"api/v2/files/{file2['id']}", headers=headers) assert download2.status_code == 200 assert download2.content == b"path content 2" # ==================== S3 STORAGE TESTS ==================== @pytest.fixture def aws_credentials(): """Verify AWS credentials are set via environment variables.""" required_vars = ["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"] missing_vars = [var for var in required_vars if not os.environ.get(var)] if missing_vars: pytest.skip(f"Missing required environment variables: {', '.join(missing_vars)}") # Set default region if not provided if not os.environ.get("AWS_DEFAULT_REGION"): os.environ["AWS_DEFAULT_REGION"] = "us-west-2" # No cleanup needed - we're using existing env vars @pytest.fixture(name="s3_files_created_api_key") async def s3_files_created_api_key(s3_files_client, s3_files_active_user): # noqa: ARG001 hashed = get_password_hash("s3_random_key") api_key = ApiKey( name="s3_files_created_api_key", user_id=s3_files_active_user.id, api_key="s3_random_key", # pragma: allowlist secret hashed_api_key=hashed, ) async with session_scope() as session: stmt = select(ApiKey).where(ApiKey.api_key == api_key.api_key) if existing_api_key := (await session.exec(stmt)).first(): existing_api_key = UnmaskedApiKeyRead.model_validate(existing_api_key, from_attributes=True) yield existing_api_key return session.add(api_key) await session.flush() await session.refresh(api_key) api_key = UnmaskedApiKeyRead.model_validate(api_key, from_attributes=True) yield api_key async with session_scope() as session: db_key = await session.get(ApiKey, api_key.id) if db_key: await session.delete(db_key) @pytest.fixture(name="s3_files_active_user") async def s3_files_active_user(s3_files_client): # noqa: ARG001 async with session_scope() as session: user = User( username="s3_files_active_user", password=get_password_hash("testpassword"), is_active=True, is_superuser=False, ) stmt = select(User).where(User.username == user.username) if active_user := (await session.exec(stmt)).first(): user = active_user else: session.add(user) await session.flush() await session.refresh(user) user = UserRead.model_validate(user, from_attributes=True) yield user # Clean up # Now cleanup transactions, vertex_build async with session_scope() as session: user = await session.get(User, user.id, options=[selectinload(User.flows)]) await _delete_transactions_and_vertex_builds(session, user.flows) await session.delete(user) @pytest.fixture(name="s3_files_client") async def s3_files_client_fixture( monkeypatch, request, aws_credentials, # noqa: ARG001 ): """S3 storage client fixture for testing with real S3.""" # Set the database url to a test database if "noclient" in request.keywords: yield else: def init_app(): db_dir = tempfile.mkdtemp() db_path = Path(db_dir) / "test_s3.db" monkeypatch.setenv("LANGFLOW_DATABASE_URL", f"sqlite:///{db_path}") monkeypatch.setenv("LANGFLOW_AUTO_LOGIN", "false") # Configure S3 storage monkeypatch.setenv("LANGFLOW_STORAGE_TYPE", "s3") monkeypatch.setenv( "LANGFLOW_OBJECT_STORAGE_BUCKET_NAME", os.environ.get("LANGFLOW_OBJECT_STORAGE_BUCKET_NAME", "langflow-ci"), ) # Use unique prefix per test run to avoid conflicts test_prefix = f"test-files-api-{uuid.uuid4().hex[:8]}" monkeypatch.setenv("LANGFLOW_OBJECT_STORAGE_PREFIX", test_prefix) tags_json = json.dumps({"env": "test-api", "type": "file-upload"}) monkeypatch.setenv("LANGFLOW_OBJECT_STORAGE_TAGS", tags_json) from langflow.services.manager import service_manager service_manager.factories.clear() service_manager.services.clear() # Clear the services cache app = create_app() return app, db_path, test_prefix app, db_path, test_prefix = await asyncio.to_thread(init_app) async with ( LifespanManager(app, startup_timeout=None, shutdown_timeout=60) as manager, AsyncClient(transport=ASGITransport(app=manager.app), base_url="http://testserver/") as client, ): yield client # Cleanup: Delete all test files from S3 try: import boto3 s3 = boto3.client("s3") bucket_name = os.environ.get("LANGFLOW_OBJECT_STORAGE_BUCKET_NAME", "langflow-ci") # List and delete all objects with our test prefix with contextlib.suppress(Exception): response = s3.list_objects_v2(Bucket=bucket_name, Prefix=test_prefix) if "Contents" in response: for obj in response["Contents"]: s3.delete_object(Bucket=bucket_name, Key=obj["Key"]) except Exception: # noqa: S110 pass # Ignore cleanup errors - outer exception handler monkeypatch.undo() # clear the temp db with suppress(FileNotFoundError): await anyio.Path(db_path).unlink() # Mark all S3 tests as requiring API keys pytestmark_s3 = pytest.mark.api_key_required @pytest.mark.api_key_required class TestS3FileOperations: """Test file operations with S3 storage backend. These tests use actual AWS S3 and verify that file operations work correctly with S3 storage, including the delete bug fix. """ async def test_s3_upload_file(self, s3_files_client, s3_files_created_api_key): """Test uploading a file to S3 storage.""" headers = {"x-api-key": s3_files_created_api_key.api_key} response = await s3_files_client.post( "api/v2/files", files={"file": ("s3_test.txt", b"S3 test content")}, headers=headers, ) assert response.status_code == 201, f"Expected 201, got {response.status_code}: {response.json()}" response_json = response.json() assert "id" in response_json assert response_json["name"] == "s3_test" async def test_s3_upload_and_download_file(self, s3_files_client, s3_files_created_api_key): """Test uploading and downloading a file with S3 storage.""" headers = {"x-api-key": s3_files_created_api_key.api_key} # Upload file response = await s3_files_client.post( "api/v2/files", files={"file": ("s3_download_test.txt", b"S3 download content")}, headers=headers, ) assert response.status_code == 201 upload_response = response.json() # Download file response = await s3_files_client.get(f"api/v2/files/{upload_response['id']}", headers=headers) assert response.status_code == 200 assert response.content == b"S3 download content" async def test_s3_list_files(self, s3_files_client, s3_files_created_api_key): """Test listing files with S3 storage.""" headers = {"x-api-key": s3_files_created_api_key.api_key} # Upload a file response = await s3_files_client.post( "api/v2/files", files={"file": ("s3_list_test.txt", b"S3 list content")}, headers=headers, ) assert response.status_code == 201 # List files response = await s3_files_client.get("api/v2/files", headers=headers) assert response.status_code == 200 files = response.json() assert len(files) >= 1 file_names = [f["name"] for f in files] assert "s3_list_test" in file_names async def test_s3_delete_file(self, s3_files_client, s3_files_created_api_key): """Test deleting a file from S3 storage (verifies delete bug fix).""" headers = {"x-api-key": s3_files_created_api_key.api_key} # Upload a file response = await s3_files_client.post( "api/v2/files", files={"file": ("s3_delete_test.txt", b"S3 delete content")}, headers=headers, ) assert response.status_code == 201 upload_response = response.json() # Delete the file response = await s3_files_client.delete(f"api/v2/files/{upload_response['id']}", headers=headers) assert response.status_code == 200 assert response.json() == {"detail": "File s3_delete_test deleted successfully"} # Verify file is deleted from database response = await s3_files_client.get("api/v2/files", headers=headers) assert response.status_code == 200 files = response.json() file_names = [f["name"] for f in files] assert "s3_delete_test" not in file_names # Verify file is deleted from S3 (should return 404) response = await s3_files_client.get(f"api/v2/files/{upload_response['id']}", headers=headers) assert response.status_code == 404 async def test_s3_upload_list_delete_multiple_files(self, s3_files_client, s3_files_created_api_key): """Test uploading, listing, and deleting multiple files with S3 storage.""" headers = {"x-api-key": s3_files_created_api_key.api_key} # Upload two files response1 = await s3_files_client.post( "api/v2/files", files={"file": ("s3_file1.txt", b"S3 content1")}, headers=headers, ) assert response1.status_code == 201 file1 = response1.json() response2 = await s3_files_client.post( "api/v2/files", files={"file": ("s3_file2.txt", b"S3 content2")}, headers=headers, ) assert response2.status_code == 201 file2 = response2.json() # List files and validate both are present response = await s3_files_client.get("api/v2/files", headers=headers) assert response.status_code == 200 files = response.json() file_names = [f["name"] for f in files] file_ids = [f["id"] for f in files] assert file1["name"] in file_names assert file2["name"] in file_names assert file1["id"] in file_ids assert file2["id"] in file_ids # Delete one file response = await s3_files_client.delete(f"api/v2/files/{file1['id']}", headers=headers) assert response.status_code == 200 # List files again and validate only the other remains response = await s3_files_client.get("api/v2/files", headers=headers) assert response.status_code == 200 files = response.json() file_names = [f["name"] for f in files] file_ids = [f["id"] for f in files] assert file1["name"] not in file_names assert file1["id"] not in file_ids assert file2["name"] in file_names assert file2["id"] in file_ids async def test_s3_upload_binary_file(self, s3_files_client, s3_files_created_api_key): """Test uploading and downloading binary data with S3 storage.""" headers = {"x-api-key": s3_files_created_api_key.api_key} # Create binary data binary_data = bytes(range(256)) # Upload binary file response = await s3_files_client.post( "api/v2/files", files={"file": ("s3_binary.bin", binary_data)}, headers=headers, ) assert response.status_code == 201 upload_response = response.json() # Download and verify binary data response = await s3_files_client.get(f"api/v2/files/{upload_response['id']}", headers=headers) assert response.status_code == 200 assert response.content == binary_data async def test_s3_delete_verifies_s3_cleanup(self, s3_files_client, s3_files_created_api_key): """Test that delete properly cleans up S3 storage (verifies the bug fix).""" headers = {"x-api-key": s3_files_created_api_key.api_key} # Upload a file response = await s3_files_client.post( "api/v2/files", files={"file": ("s3_cleanup_test.txt", b"S3 cleanup content")}, headers=headers, ) assert response.status_code == 201 upload_response = response.json() # Get the user ID from the response path file_path = upload_response["path"] user_id = file_path.split("/")[0] # Delete the file response = await s3_files_client.delete(f"api/v2/files/{upload_response['id']}", headers=headers) assert response.status_code == 200 # Verify file is actually deleted from S3 by checking directly import boto3 s3 = boto3.client("s3") bucket_name = os.environ.get("LANGFLOW_OBJECT_STORAGE_BUCKET_NAME", "langflow-ci") # Extract file name from path file_name = file_path.split("/")[-1] # Build the S3 key using the correct pattern (prefix/user_id/filename) test_prefix = os.environ.get("LANGFLOW_OBJECT_STORAGE_PREFIX") s3_key = f"{test_prefix}/{user_id}/{file_name}" # Try to get the object - should raise NoSuchKey try: s3.head_object(Bucket=bucket_name, Key=s3_key) pytest.fail(f"File {s3_key} should have been deleted from S3 but still exists") except s3.exceptions.NoSuchKey: pass # Expected - file was properly deleted except Exception as e: # Check if it's a 404-related error (different boto3 versions) if "404" not in str(e) and "NoSuchKey" not in str(e): raise class TestStorageFailureHandling: """Test permanent vs transient storage failure handling in delete operations.""" def test_is_permanent_storage_failure_file_not_found_error(self): """Test that FileNotFoundError is recognized as permanent failure.""" error = FileNotFoundError("File not found") assert is_permanent_storage_failure(error) is True def test_is_permanent_storage_failure_s3_no_such_bucket(self): """Test that S3 NoSuchBucket error is recognized as permanent failure.""" # Mock S3 error with NoSuchBucket code class MockS3Error(Exception): def __init__(self): self.response = {"Error": {"Code": "NoSuchBucket", "Message": "Bucket does not exist"}} error = MockS3Error() assert is_permanent_storage_failure(error) is True def test_is_permanent_storage_failure_s3_no_such_key(self): """Test that S3 NoSuchKey error is recognized as permanent failure.""" # Mock S3 error with NoSuchKey code class MockS3Error(Exception): def __init__(self): self.response = {"Error": {"Code": "NoSuchKey", "Message": "Key does not exist"}} error = MockS3Error() assert is_permanent_storage_failure(error) is True def test_is_permanent_storage_failure_transient_error(self): """Test that transient errors (network, timeout) are not permanent failures.""" # Mock transient errors class NetworkError(Exception): pass class CustomTimeoutError(Exception): pass class CustomPermissionError(Exception): pass assert is_permanent_storage_failure(NetworkError("Connection failed")) is False assert is_permanent_storage_failure(CustomTimeoutError("Request timed out")) is False assert is_permanent_storage_failure(CustomPermissionError("Access denied")) is False def test_is_permanent_storage_failure_fallback_string_matching(self): """Test fallback string matching for edge cases.""" # Test with error messages that match permanent patterns class CustomError(Exception): pass assert is_permanent_storage_failure(CustomError("NoSuchBucket error")) is True assert is_permanent_storage_failure(CustomError("NoSuchKey error")) is True assert is_permanent_storage_failure(CustomError("File not found")) is True assert is_permanent_storage_failure(CustomError("FileNotFoundError occurred")) is True async def test_delete_file_with_permanent_failure_deletes_from_db(self): """Test that permanent storage failures still delete from database.""" from langflow.services.database.models.file.model import File as UserFile file_id = uuid.uuid4() user_id = uuid.uuid4() file_name = "test_file.txt" file_path = f"{file_id}.txt" mock_file = UserFile( id=file_id, user_id=user_id, name=file_name, path=file_path, size=100, ) mock_current_user = MagicMock() mock_current_user.id = user_id mock_exec_result = MagicMock() mock_exec_result.first = MagicMock(return_value=mock_file) mock_session = AsyncMock() mock_session.exec = AsyncMock(return_value=mock_exec_result) mock_session.delete = AsyncMock() mock_storage_service = AsyncMock() mock_storage_service.delete_file = AsyncMock(side_effect=FileNotFoundError(f"File {file_path} not found")) result = await delete_file( file_id=file_id, current_user=mock_current_user, session=mock_session, storage_service=mock_storage_service, ) expected_file_name = file_path.split("/")[-1] mock_storage_service.delete_file.assert_called_once_with( flow_id=str(user_id), file_name=expected_file_name, ) mock_session.delete.assert_called_once_with(mock_file) assert result["detail"] == f"File {file_name} deleted successfully" async def test_delete_file_with_transient_failure_keeps_in_db(self): """Test that transient storage failures keep file in database for retry.""" from fastapi import HTTPException from langflow.services.database.models.file.model import File as UserFile file_id = uuid.uuid4() user_id = uuid.uuid4() file_name = "test_file.txt" file_path = f"{file_id}.txt" mock_file = UserFile( id=file_id, user_id=user_id, name=file_name, path=file_path, size=100, ) mock_current_user = MagicMock() mock_current_user.id = user_id mock_exec_result = MagicMock() mock_exec_result.first = MagicMock(return_value=mock_file) mock_session = AsyncMock() mock_session.exec = AsyncMock(return_value=mock_exec_result) mock_session.delete = AsyncMock() mock_storage_service = AsyncMock() mock_storage_service.delete_file = AsyncMock(side_effect=ConnectionError("Network connection failed")) with pytest.raises(HTTPException) as exc_info: await delete_file( file_id=file_id, current_user=mock_current_user, session=mock_session, storage_service=mock_storage_service, ) assert exc_info.value.status_code == 500 assert "Failed to delete file from storage" in exc_info.value.detail expected_file_name = file_path.split("/")[-1] mock_storage_service.delete_file.assert_called_once_with( flow_id=str(user_id), file_name=expected_file_name, ) mock_session.delete.assert_not_called() async def test_batch_delete_with_mixed_failures(self): """Test batch delete with mix of permanent and transient failures.""" from langflow.services.database.models.file.model import File as UserFile user_id = uuid.uuid4() file_ids = [uuid.uuid4() for _ in range(3)] file_names = [f"batch_test_{i}.txt" for i in range(3)] mock_files = [ UserFile( id=file_ids[i], user_id=user_id, name=file_names[i], path=f"{file_ids[i]}.txt", size=100, ) for i in range(3) ] mock_current_user = MagicMock() mock_current_user.id = user_id mock_exec_result = MagicMock() mock_exec_result.all = MagicMock(return_value=mock_files) mock_session = AsyncMock() mock_session.exec = AsyncMock(return_value=mock_exec_result) mock_session.delete = AsyncMock() deleted_file_ids = set() async def mock_delete_file(*, flow_id: str | None = None, file_name: str | None = None) -> None: # noqa: ARG001 if file_name == f"{file_ids[0]}.txt": msg = f"File {file_name} not found" raise FileNotFoundError(msg) if file_name == f"{file_ids[1]}.txt": msg = "Network error" raise ConnectionError(msg) deleted_file_ids.add(file_name) mock_storage_service = AsyncMock() mock_storage_service.delete_file = AsyncMock(side_effect=mock_delete_file) result = await delete_files_batch( file_ids=file_ids, current_user=mock_current_user, session=mock_session, storage_service=mock_storage_service, ) assert ( result["message"] == "2 files deleted successfully, 1 files kept in database due to transient storage errors (can retry)" ) assert mock_storage_service.delete_file.call_count == 3 delete_calls = [call[0][0] for call in mock_session.delete.call_args_list] assert len(delete_calls) == 2 assert mock_files[0] in delete_calls assert mock_files[2] in delete_calls assert mock_files[1] not in delete_calls async def test_delete_all_files_message_all_successful(self): """Test delete_all_files returns correct message when all files deleted successfully.""" from langflow.services.database.models.file.model import File as UserFile user_id = uuid.uuid4() mock_files = [ UserFile( id=uuid.uuid4(), user_id=user_id, name=f"msg_test_{i}.txt", path=f"{uuid.uuid4()}.txt", size=100, ) for i in range(3) ] mock_current_user = MagicMock() mock_current_user.id = user_id mock_exec_result = MagicMock() mock_exec_result.all = MagicMock(return_value=mock_files) mock_session = AsyncMock() mock_session.exec = AsyncMock(return_value=mock_exec_result) mock_session.delete = AsyncMock() mock_storage_service = AsyncMock() mock_storage_service.delete_file = AsyncMock() result = await delete_all_files( current_user=mock_current_user, session=mock_session, storage_service=mock_storage_service, ) assert result["message"] == "All 3 files deleted successfully" async def test_delete_all_files_message_with_transient_failures(self): """Test delete_all_files returns correct message when some files have transient storage failures.""" from langflow.services.database.models.file.model import File as UserFile user_id = uuid.uuid4() mock_files = [ UserFile( id=uuid.uuid4(), user_id=user_id, name=f"transient_msg_test_{i}.txt", path=f"{uuid.uuid4()}.txt", size=100, ) for i in range(3) ] mock_current_user = MagicMock() mock_current_user.id = user_id mock_exec_result = MagicMock() mock_exec_result.all = MagicMock(return_value=mock_files) mock_session = AsyncMock() mock_session.exec = AsyncMock(return_value=mock_exec_result) mock_session.delete = AsyncMock() call_count = 0 async def mock_delete_file(*, flow_id: str | None = None, file_name: str | None = None) -> None: # noqa: ARG001 nonlocal call_count call_count += 1 if call_count == 1: msg = "Network error" raise ConnectionError(msg) mock_storage_service = AsyncMock() mock_storage_service.delete_file = AsyncMock(side_effect=mock_delete_file) result = await delete_all_files( current_user=mock_current_user, session=mock_session, storage_service=mock_storage_service, ) assert result["message"] == "2 files deleted successfully, 1 files failed to delete. See logs for details." async def test_batch_delete_message_all_successful(self): """Test batch delete returns correct message when all files deleted successfully.""" from langflow.services.database.models.file.model import File as UserFile user_id = uuid.uuid4() file_ids = [uuid.uuid4() for _ in range(2)] mock_files = [ UserFile( id=file_ids[i], user_id=user_id, name=f"batch_msg_test_{i}.txt", path=f"{file_ids[i]}.txt", size=100, ) for i in range(2) ] mock_current_user = MagicMock() mock_current_user.id = user_id mock_exec_result = MagicMock() mock_exec_result.all = MagicMock(return_value=mock_files) mock_session = AsyncMock() mock_session.exec = AsyncMock(return_value=mock_exec_result) mock_session.delete = AsyncMock() mock_storage_service = AsyncMock() mock_storage_service.delete_file = AsyncMock() result = await delete_files_batch( file_ids=file_ids, current_user=mock_current_user, session=mock_session, storage_service=mock_storage_service, ) assert result["message"] == "2 files deleted successfully" async def test_batch_delete_message_with_transient_failures(self): """Test batch delete returns correct message when some files have transient storage failures.""" from langflow.services.database.models.file.model import File as UserFile user_id = uuid.uuid4() file_ids = [uuid.uuid4() for _ in range(3)] mock_files = [ UserFile( id=file_ids[i], user_id=user_id, name=f"batch_transient_msg_{i}.txt", path=f"{file_ids[i]}.txt", size=100, ) for i in range(3) ] mock_current_user = MagicMock() mock_current_user.id = user_id mock_exec_result = MagicMock() mock_exec_result.all = MagicMock(return_value=mock_files) mock_session = AsyncMock() mock_session.exec = AsyncMock(return_value=mock_exec_result) mock_session.delete = AsyncMock() call_count = 0 async def mock_delete_file(*, flow_id: str | None = None, file_name: str | None = None) -> None: # noqa: ARG001 nonlocal call_count call_count += 1 if call_count == 1: msg = "Network error" raise ConnectionError(msg) mock_storage_service = AsyncMock() mock_storage_service.delete_file = AsyncMock(side_effect=mock_delete_file) result = await delete_files_batch( file_ids=file_ids, current_user=mock_current_user, session=mock_session, storage_service=mock_storage_service, ) assert ( result["message"] == "2 files deleted successfully, 1 files kept in database due to transient storage errors (can retry)" )