dify/api/tests/unit_tests/services/test_file_service.py

import base64
import hashlib
import os
from collections.abc import Callable, Iterator
from dataclasses import dataclass
from datetime import UTC, datetime
from unittest.mock import patch

import httpx
import pytest
from sqlalchemy import Connection, Engine, event, select
from sqlalchemy.orm import Session, sessionmaker
from werkzeug.exceptions import NotFound

from core.file import remote_fetcher
from enums import DeploymentEdition
from extensions.ext_storage import storage
from extensions.storage.storage_type import StorageType
from graphon.file import helpers as file_helpers
from models.base import TypeBase
from models.enums import CreatorUserRole
from models.model import Account, EndUser, UploadFile
from services.errors.file import (
    BlockedFileExtensionError,
    FileNotExistsError,
    FileTooLargeError,
    UnsupportedFileTypeError,
)
from services.file_service import FileService, FileUploadActor, FileUploadResult
from services.remote_file_service import RemoteFileService
from tests.helpers.unit.model_factories import make_account


def _account() -> Account:
    return make_account(account_id="user_id", name="Test Account", email="test@example.com")


class TestFileServiceCases:
    class TestFileService:
        @pytest.fixture
        def sqlite_session_maker(self, sqlite_engine: Engine) -> sessionmaker[Session]:
            TypeBase.metadata.create_all(sqlite_engine, tables=[TypeBase.metadata.tables[UploadFile.__tablename__]])
            return sessionmaker(bind=sqlite_engine, expire_on_commit=False)

        @pytest.fixture
        def db_session(self, sqlite_session_maker: sessionmaker[Session]) -> Iterator[Session]:
            with sqlite_session_maker() as session:
                yield session

        @pytest.fixture
        def file_service(self, sqlite_session_maker: sessionmaker[Session]) -> FileService:
            return FileService(session_factory=sqlite_session_maker)

        @staticmethod
        def _persist_upload_file(
            session: Session,
            *,
            file_id: str = "file_id",
            tenant_id: str = "tenant_id",
            extension: str = "txt",
            mime_type: str = "text/plain",
            key: str = "key",
        ) -> UploadFile:
            upload_file = UploadFile(
                tenant_id=tenant_id,
                storage_type=StorageType.LOCAL,
                key=key,
                name=f"test.{extension}",
                size=10,
                extension=extension,
                mime_type=mime_type,
                created_by_role=CreatorUserRole.ACCOUNT,
                created_by="user_id",
                created_at=datetime(2024, 1, 1, tzinfo=UTC),
                used=False,
            )
            upload_file.id = file_id
            session.add(upload_file)
            session.commit()
            return upload_file

        def test_init_with_engine(self, sqlite_engine: Engine):
            service = FileService(session_factory=sqlite_engine)
            assert isinstance(service._session_maker, sessionmaker)

        def test_init_with_sessionmaker(self, sqlite_session_maker: sessionmaker[Session]):
            service = FileService(session_factory=sqlite_session_maker)
            assert service._session_maker == sqlite_session_maker

        def test_init_invalid_factory(self):
            with pytest.raises(AssertionError, match="must be a sessionmaker or an Engine."):
                FileService(session_factory="invalid")

        @patch("services.file_service.storage")
        @patch("services.file_service.naive_utc_now")
        @patch("services.file_service.extract_tenant_id")
        @patch("services.file_service.file_helpers.get_signed_file_url")
        def test_upload_file_success(
            self, mock_get_url, mock_tenant_id, mock_now, mock_storage, file_service: FileService, db_session: Session
        ):
            # Setup
            mock_tenant_id.return_value = "tenant_id"
            mock_now.return_value = datetime(2024, 1, 1, tzinfo=UTC)
            mock_get_url.return_value = "http://signed-url"

            user = Account(name="Test Account", email="test@example.com")
            user.id = "user_id"
            content = b"file content"
            filename = "test.jpg"
            mimetype = "image/jpeg"

            # Execute
            result = file_service.upload_file(filename=filename, content=content, mimetype=mimetype, user=user)

            # Assert
            assert isinstance(result, UploadFile)
            assert result.name == filename
            assert result.tenant_id == "tenant_id"
            assert result.size == len(content)
            assert result.extension == "jpg"
            assert result.mime_type == mimetype
            assert result.created_by_role == CreatorUserRole.ACCOUNT
            assert result.created_by == "user_id"
            assert result.hash == hashlib.sha3_256(content).hexdigest()
            assert result.source_url == "http://signed-url"

            mock_storage.save.assert_called_once()
            persisted = db_session.get(UploadFile, result.id)
            assert persisted is not None
            assert persisted.hash == result.hash

        @pytest.mark.parametrize("text", ["ASCII text", "包含多字节 UTF-8 文本 🚀"])
        def test_upload_text_uses_utf8_byte_length(self, text: str, file_service: FileService):
            with patch("services.file_service.storage") as mock_storage:
                result = file_service.upload_text(
                    text=text,
                    text_name="test.txt",
                    user_id="user_id",
                    tenant_id="tenant_id",
                )

            expected_content = text.encode("utf-8")
            assert result.size == len(expected_content)
            mock_storage.save.assert_called_once_with(result.key, expected_content)

        def test_upload_file_uses_explicit_resource_tenant(self, file_service: FileService):
            user = Account(name="Test Account", email="test@example.com")
            user.id = "user-id"

            with (
                patch("services.file_service.storage") as mock_storage,
                patch("services.file_service.extract_tenant_id") as mock_extract_tenant_id,
                patch("services.file_service.file_helpers.get_signed_file_url"),
            ):
                result = file_service.upload_file(
                    filename="test.txt",
                    content=b"test",
                    mimetype="text/plain",
                    user=user,
                    tenant_id="resource-tenant-id",
                )

            assert result.tenant_id == "resource-tenant-id"
            assert mock_storage.save.call_args.args[0].startswith("upload_files/resource-tenant-id/")
            mock_extract_tenant_id.assert_not_called()

        def test_upload_file_invalid_characters(self, file_service):
            with pytest.raises(ValueError, match="Filename contains invalid characters"):
                file_service.upload_file(
                    filename="invalid/file.txt", content=b"", mimetype="text/plain", user=_account()
                )

        def test_upload_file_long_filename(self, file_service: FileService, db_session: Session):
            # Setup
            long_name = "a" * 210 + ".txt"
            user = Account(name="Test Account", email="test@example.com")
            user.id = "user_id"

            with (
                patch("services.file_service.storage"),
                patch("services.file_service.extract_tenant_id") as mock_tenant,
                patch("services.file_service.file_helpers.get_signed_file_url"),
            ):
                mock_tenant.return_value = "tenant"
                result = file_service.upload_file(filename=long_name, content=b"test", mimetype="text/plain", user=user)
                assert len(result.name) <= 205  # 200 + . + extension
                assert result.name.endswith(".txt")
                assert db_session.get(UploadFile, result.id) is not None

        def test_upload_file_blocked_extension(self, file_service, config_overrides: Callable[..., None]):
            config_overrides(inner_UPLOAD_FILE_EXTENSION_BLACKLIST="exe")
            with pytest.raises(BlockedFileExtensionError):
                file_service.upload_file(
                    filename="test.exe", content=b"", mimetype="application/octet-stream", user=_account()
                )

        def test_upload_file_unsupported_type_for_datasets(self, file_service):
            with pytest.raises(UnsupportedFileTypeError):
                file_service.upload_file(
                    filename="test.jpg", content=b"", mimetype="image/jpeg", user=_account(), source="datasets"
                )

        def test_upload_file_too_large(self, file_service, config_overrides: Callable[..., None]):
            # 16MB file for an image with 15MB limit
            content = b"a" * (16 * 1024 * 1024)
            config_overrides(UPLOAD_IMAGE_FILE_SIZE_LIMIT=15)
            with pytest.raises(FileTooLargeError):
                file_service.upload_file(filename="test.jpg", content=content, mimetype="image/jpeg", user=_account())

        def test_upload_file_end_user(self, file_service: FileService, db_session: Session):
            user = EndUser(
                id="end_user_id",
            )

            with (
                patch("services.file_service.storage"),
                patch("services.file_service.extract_tenant_id") as mock_tenant,
                patch("services.file_service.file_helpers.get_signed_file_url"),
            ):
                mock_tenant.return_value = "tenant"
                result = file_service.upload_file(
                    filename="test.txt", content=b"test", mimetype="text/plain", user=user
                )
                assert result.created_by_role == CreatorUserRole.END_USER
                assert db_session.get(UploadFile, result.id) is not None

        def test_is_file_size_within_limit(self, config_overrides: Callable[..., None]):
            config_overrides(
                UPLOAD_IMAGE_FILE_SIZE_LIMIT=10,
                UPLOAD_VIDEO_FILE_SIZE_LIMIT=20,
                UPLOAD_AUDIO_FILE_SIZE_LIMIT=30,
                UPLOAD_FILE_SIZE_LIMIT=5,
            )
            # Image
            assert FileService.is_file_size_within_limit(extension="jpg", file_size=10 * 1024 * 1024) is True
            assert FileService.is_file_size_within_limit(extension="png", file_size=11 * 1024 * 1024) is False

            # Video
            assert FileService.is_file_size_within_limit(extension="mp4", file_size=20 * 1024 * 1024) is True
            assert FileService.is_file_size_within_limit(extension="avi", file_size=21 * 1024 * 1024) is False

            # Audio
            assert FileService.is_file_size_within_limit(extension="mp3", file_size=30 * 1024 * 1024) is True
            assert FileService.is_file_size_within_limit(extension="wav", file_size=31 * 1024 * 1024) is False

            # Default
            assert FileService.is_file_size_within_limit(extension="txt", file_size=5 * 1024 * 1024) is True
            assert FileService.is_file_size_within_limit(extension="pdf", file_size=6 * 1024 * 1024) is False
            assert (
                FileService.is_file_size_within_limit(extension="txt", file_size=0, default_file_size_limit=0) is True
            )
            assert (
                FileService.is_file_size_within_limit(extension="txt", file_size=1, default_file_size_limit=0) is False
            )
            assert (
                FileService.is_file_size_within_limit(
                    extension="pdf",
                    file_size=6 * 1024 * 1024,
                    default_file_size_limit=7,
                )
                is True
            )
            assert (
                FileService.is_file_size_within_limit(
                    extension="pdf",
                    file_size=8 * 1024 * 1024,
                    default_file_size_limit=7,
                )
                is False
            )

            # Media-specific limits are not affected by the knowledge document override.
            assert (
                FileService.is_file_size_within_limit(
                    extension="jpg",
                    file_size=11 * 1024 * 1024,
                    default_file_size_limit=100,
                )
                is False
            )

        def test_file_size_limit(self, config_overrides: Callable[..., None]):
            config_overrides(
                UPLOAD_IMAGE_FILE_SIZE_LIMIT=10,
                UPLOAD_VIDEO_FILE_SIZE_LIMIT=20,
                UPLOAD_AUDIO_FILE_SIZE_LIMIT=30,
                UPLOAD_FILE_SIZE_LIMIT=5,
            )

            assert FileService.file_size_limit(extension="jpg") == 10 * 1024 * 1024
            assert FileService.file_size_limit(extension="mp4") == 20 * 1024 * 1024
            assert FileService.file_size_limit(extension="mp3") == 30 * 1024 * 1024
            assert FileService.file_size_limit(extension="txt") == 5 * 1024 * 1024
            assert FileService.file_size_limit(extension="txt", default_file_size_limit=None) == 5 * 1024 * 1024
            assert FileService.file_size_limit(extension="txt", default_file_size_limit=0) == 0
            assert FileService.file_size_limit(extension="jpg", default_file_size_limit=0) == 10 * 1024 * 1024
            assert FileService.file_size_limit(extension="txt", default_file_size_limit=7) == 7 * 1024 * 1024

        def test_get_file_base64_success(self, file_service: FileService, db_session: Session):
            self._persist_upload_file(db_session, key="test_key")

            with patch("services.file_service.storage") as mock_storage:
                mock_storage.load_once.return_value = b"test content"

                # Execute
                result = file_service.get_file_base64("file_id")

                # Assert
                assert result == base64.b64encode(b"test content").decode()
                mock_storage.load_once.assert_called_once_with("test_key")

        def test_get_file_base64_not_found(self, file_service: FileService):
            with pytest.raises(NotFound, match="File not found"):
                file_service.get_file_base64("non_existent")

        def test_get_file_presigned_url_success(
            self, file_service: FileService, db_session: Session, config_overrides: Callable[..., None]
        ):
            config_overrides(FILES_ACCESS_TIMEOUT=300)
            self._persist_upload_file(
                db_session,
                extension="png",
                mime_type="image/png",
                key="upload_files/tenant_id/icon.png",
            )

            with (
                patch("services.file_service.storage") as mock_storage,
            ):
                mock_storage.generate_presigned_url.return_value = "https://s3.example.com/icon.png?signature=test"

                result = file_service.get_file_presigned_url(file_id="file_id", tenant_id="tenant_id")

            assert result == "https://s3.example.com/icon.png?signature=test"
            mock_storage.generate_presigned_url.assert_called_once_with(
                "upload_files/tenant_id/icon.png",
                expires_in=300,
                content_type="image/png",
            )

        def test_get_file_presigned_url_not_found(self, file_service: FileService):
            with pytest.raises(NotFound, match="File not found"):
                file_service.get_file_presigned_url(file_id="file_id", tenant_id="tenant_id")

        def test_get_icon_url_uses_direct_storage_url_for_cloud_s3(
            self, file_service: FileService, config_overrides: Callable[..., None]
        ):
            config_overrides(DEPLOYMENT_EDITION=DeploymentEdition.CLOUD, STORAGE_TYPE=StorageType.S3)
            with patch.object(file_service, "get_file_presigned_url", return_value="direct-url") as get_presigned_url:
                result = file_service.get_icon_url("file_id", "tenant_id")

            assert result == "direct-url"
            get_presigned_url.assert_called_once_with(file_id="file_id", tenant_id="tenant_id")

        def test_get_icon_url_maps_missing_cloud_file_to_service_error(
            self, file_service: FileService, config_overrides: Callable[..., None]
        ) -> None:
            config_overrides(DEPLOYMENT_EDITION=DeploymentEdition.CLOUD, STORAGE_TYPE=StorageType.S3)
            with (
                patch.object(file_service, "get_file_presigned_url", side_effect=NotFound("File not found")),
                pytest.raises(FileNotExistsError, match="File reference not found"),
            ):
                file_service.get_icon_url("file_id", "tenant_id")

        @pytest.mark.parametrize(
            ("deployment_edition", "storage_type"),
            [
                (DeploymentEdition.COMMUNITY, StorageType.S3),
                (DeploymentEdition.CLOUD, StorageType.LOCAL),
            ],
        )
        def test_get_icon_url_uses_preview_url_outside_cloud_s3(
            self,
            file_service: FileService,
            db_session: Session,
            deployment_edition: DeploymentEdition,
            storage_type: StorageType,
            config_overrides: Callable[..., None],
        ):
            self._persist_upload_file(db_session)
            config_overrides(DEPLOYMENT_EDITION=deployment_edition, STORAGE_TYPE=storage_type)
            with patch("services.file_service.file_helpers.get_signed_file_url", return_value="preview-url") as get_url:
                result = file_service.get_icon_url("file_id", "tenant_id")

            assert result == "preview-url"
            get_url.assert_called_once_with(upload_file_id="file_id")

        @pytest.mark.parametrize(
            ("deployment_edition", "storage_type"),
            [
                (DeploymentEdition.COMMUNITY, StorageType.S3),
                (DeploymentEdition.CLOUD, StorageType.LOCAL),
            ],
        )
        def test_get_icon_url_rejects_missing_file_outside_cloud_s3(
            self,
            file_service: FileService,
            deployment_edition: DeploymentEdition,
            storage_type: StorageType,
            config_overrides: Callable[..., None],
        ) -> None:
            config_overrides(DEPLOYMENT_EDITION=deployment_edition, STORAGE_TYPE=storage_type)

            with pytest.raises(FileNotExistsError, match="File reference not found"):
                file_service.get_icon_url("file_id", "tenant_id")

        def test_get_icon_url_rejects_cross_tenant_file_outside_cloud_s3(
            self,
            file_service: FileService,
            db_session: Session,
            config_overrides: Callable[..., None],
        ) -> None:
            self._persist_upload_file(db_session, tenant_id="other_tenant_id")
            config_overrides(DEPLOYMENT_EDITION=DeploymentEdition.COMMUNITY, STORAGE_TYPE=StorageType.LOCAL)

            with pytest.raises(FileNotExistsError, match="File reference not found"):
                file_service.get_icon_url("file_id", "tenant_id")

        def test_upload_text_success(self, file_service: FileService, db_session: Session):
            # Setup
            text = "sample text"
            text_name = "test.txt"
            user_id = "user_id"
            tenant_id = "tenant_id"

            with patch("services.file_service.storage") as mock_storage:
                # Execute
                result = file_service.upload_text(text, text_name, user_id, tenant_id)

                # Assert
                assert result.name == text_name
                assert result.size == len(text)
                assert result.tenant_id == tenant_id
                assert result.created_by == user_id
                assert result.used is True
                assert result.extension == "txt"
                mock_storage.save.assert_called_once()
                assert db_session.get(UploadFile, result.id) is not None

        def test_upload_text_long_name(self, file_service: FileService, db_session: Session):
            long_name = "a" * 210
            with patch("services.file_service.storage"):
                result = file_service.upload_text("text", long_name, "user", "tenant")
                assert len(result.name) == 200
                assert db_session.get(UploadFile, result.id) is not None

        def test_get_file_preview_success(self, file_service: FileService, db_session: Session):
            self._persist_upload_file(db_session, extension="pdf", mime_type="application/pdf")

            with patch("services.file_service.ExtractProcessor.load_from_upload_file") as mock_extract:
                mock_extract.return_value = "Extracted text content"

                # Execute
                result = file_service.get_file_preview("file_id", "tenant_id")

                # Assert
                assert result == "Extracted text content"

        def test_get_file_preview_not_found(self, file_service: FileService):
            with pytest.raises(NotFound, match="File not found"):
                file_service.get_file_preview("non_existent", "tenant_id")

        def test_get_file_preview_unsupported_type(self, file_service: FileService, db_session: Session):
            self._persist_upload_file(db_session, extension="exe", mime_type="application/octet-stream")
            with pytest.raises(UnsupportedFileTypeError):
                file_service.get_file_preview("file_id", "tenant_id")

        def test_get_file_content_success(self, file_service: FileService, db_session: Session):
            self._persist_upload_file(db_session)

            with patch("services.file_service.storage") as mock_storage:
                mock_storage.load.return_value = b"hello world"
                result = file_service.get_file_content("file_id")
                assert result == "hello world"

        def test_get_file_content_not_found(self, file_service: FileService):
            with pytest.raises(NotFound, match="File not found"):
                file_service.get_file_content("file_id")

        def test_delete_file_success(self, file_service: FileService, db_session: Session):
            self._persist_upload_file(db_session)

            with patch("services.file_service.storage") as mock_storage:
                file_service.delete_file("file_id")
                mock_storage.delete.assert_called_once_with("key")
                db_session.expire_all()
                assert db_session.get(UploadFile, "file_id") is None

        def test_delete_file_not_found(self, file_service: FileService):
            file_service.delete_file("file_id")
            # Should return without doing anything

        def test_get_upload_files_by_ids_empty(self, db_session: Session):
            result = FileService.get_upload_files_by_ids("tenant_id", [], session=db_session)
            assert result == {}

        def test_get_upload_file_by_id_scopes_to_tenant(self, db_session: Session) -> None:
            upload_file = self._persist_upload_file(db_session)

            assert FileService.get_upload_file_by_id("tenant_id", "file_id", session=db_session) == upload_file
            assert FileService.get_upload_file_by_id("other_tenant_id", "file_id", session=db_session) is None

        def test_get_upload_files_by_ids(self, db_session: Session):
            upload_file = self._persist_upload_file(db_session, file_id="550e8400-e29b-41d4-a716-446655440000")
            self._persist_upload_file(
                db_session,
                file_id="550e8400-e29b-41d4-a716-446655440001",
                tenant_id="other-tenant",
            )

            result = FileService.get_upload_files_by_ids(
                "tenant_id",
                ["550e8400-e29b-41d4-a716-446655440000", "550e8400-e29b-41d4-a716-446655440001"],
                session=db_session,
            )
            assert result["550e8400-e29b-41d4-a716-446655440000"] == upload_file

        def test_sanitize_zip_entry_name(self):
            assert FileService._sanitize_zip_entry_name("path/to/file.txt") == "file.txt"
            assert FileService._sanitize_zip_entry_name("../../../etc/passwd") == "passwd"
            assert FileService._sanitize_zip_entry_name("   ") == "file"
            assert FileService._sanitize_zip_entry_name("a\\b") == "a_b"

        def test_dedupe_zip_entry_name(self):
            used = {"a.txt"}
            assert FileService._dedupe_zip_entry_name("b.txt", used) == "b.txt"
            assert FileService._dedupe_zip_entry_name("a.txt", used) == "a (1).txt"
            used.add("a (1).txt")
            assert FileService._dedupe_zip_entry_name("a.txt", used) == "a (2).txt"

        def test_build_upload_files_zip_tempfile(self, db_session: Session):
            upload_file = self._persist_upload_file(db_session)

            with (
                patch("services.file_service.storage") as mock_storage,
                patch("services.file_service.os.remove") as mock_remove,
            ):
                mock_storage.load.return_value = [b"chunk1", b"chunk2"]

                with FileService.build_upload_files_zip_tempfile(upload_files=[upload_file]) as tmp_path:
                    assert os.path.exists(tmp_path)

                mock_remove.assert_called_once()


_TENANT_ID = "11111111-1111-1111-1111-111111111111"
_ACTOR_ID = "22222222-2222-2222-2222-222222222222"
_REMOTE_URL = "https://example.com/report.txt"


@dataclass
class _UploadHarness:
    files: FileService
    stored: dict[str, bytes]
    active_transactions: set[Connection]
    io_events: list[str]
    database_events: list[str]


class TestFileUploadActorCases:
    @staticmethod
    @pytest.fixture
    def upload_harness(
        sqlite_engine: Engine,
        sqlite_session_factory: sessionmaker[Session],
        monkeypatch: pytest.MonkeyPatch,
    ) -> Iterator[_UploadHarness]:
        harness = _UploadHarness(
            files=FileService(session_factory=sqlite_session_factory),
            stored={},
            active_transactions=set(),
            io_events=[],
            database_events=[],
        )

        def begin(connection: Connection) -> None:
            harness.active_transactions.add(connection)
            harness.database_events.append("begin")

        def end(connection: Connection) -> None:
            harness.active_transactions.remove(connection)
            harness.database_events.append("end")

        def save(key: str, content: bytes) -> None:
            assert not harness.active_transactions
            harness.io_events.append("storage")
            harness.stored[key] = content

        def signed_url(upload_file_id: str) -> str:
            assert not harness.active_transactions
            harness.io_events.append("sign")
            return f"https://files.example.com/{upload_file_id}?sign=test"

        event.listen(sqlite_engine, "begin", begin)
        event.listen(sqlite_engine, "commit", end)
        event.listen(sqlite_engine, "rollback", end)
        monkeypatch.setattr(storage, "save", save)
        monkeypatch.setattr(file_helpers, "get_signed_file_url", signed_url)
        try:
            yield harness
        finally:
            event.remove(sqlite_engine, "begin", begin)
            event.remove(sqlite_engine, "commit", end)
            event.remove(sqlite_engine, "rollback", end)

    @staticmethod
    @pytest.mark.parametrize("role", [CreatorUserRole.ACCOUNT, CreatorUserRole.END_USER])
    @pytest.mark.parametrize("source_url", ["", _REMOTE_URL])
    def test_actor_upload_preserves_owner_creator_and_source_url_without_extra_queries(
        upload_harness: _UploadHarness,
        sqlite_session_factory: sessionmaker[Session],
        role: CreatorUserRole,
        source_url: str,
    ) -> None:
        actor = FileUploadActor(id=_ACTOR_ID, creator_role=role)
        content = "文件内容".encode()

        result = upload_harness.files.upload_file_for_actor(
            actor=actor,
            resource_tenant_id=_TENANT_ID,
            filename="report.TXT",
            content=content,
            mimetype="text/plain",
            source_url=source_url,
        )

        assert isinstance(result, FileUploadResult)
        assert result.name == "report.TXT"
        assert result.size == len(content)
        assert result.extension == "txt"
        assert result.mime_type == "text/plain"
        assert result.created_by == actor.id
        assert isinstance(result.created_at, datetime)
        assert result.tenant_id == _TENANT_ID
        assert result.source_url == (source_url or f"https://files.example.com/{result.id}?sign=test")
        assert upload_harness.database_events == ["begin", "end"]
        assert not upload_harness.active_transactions
        assert upload_harness.io_events == (["storage"] if source_url else ["storage", "sign"])

        with sqlite_session_factory() as session:
            persisted = session.get(UploadFile, result.id)
            assert persisted is not None
            assert persisted.created_by_role == role
            assert persisted.created_by == actor.id
            assert persisted.tenant_id == _TENANT_ID
            assert persisted.source_url == source_url
            assert persisted.hash == hashlib.sha3_256(content).hexdigest()
            assert persisted.key.startswith(f"upload_files/{_TENANT_ID}/")
            assert upload_harness.stored == {persisted.key: content}

    @staticmethod
    @pytest.mark.parametrize("head_status", [httpx.codes.OK, httpx.codes.METHOD_NOT_ALLOWED])
    def test_remote_actor_upload_fetches_before_transaction_and_retains_original_url(
        upload_harness: _UploadHarness,
        sqlite_session_factory: sessionmaker[Session],
        monkeypatch: pytest.MonkeyPatch,
        head_status: int,
    ) -> None:
        actor = FileUploadActor(id=_ACTOR_ID, creator_role=CreatorUserRole.ACCOUNT)
        content = b"remote file content"
        requests: list[tuple[str, dict[str, object]]] = []

        def make_request(method: str, *, url: str, **kwargs: object) -> httpx.Response:
            assert not upload_harness.active_transactions
            assert url == _REMOTE_URL
            requests.append((method, kwargs))
            upload_harness.io_events.append(method)
            return httpx.Response(
                head_status if method == "HEAD" else httpx.codes.OK,
                content=content if method == "GET" else b"",
                headers={"Content-Type": "text/plain", "Content-Length": str(len(content))},
                request=httpx.Request(method, url),
            )

        monkeypatch.setattr(remote_fetcher, "make_request", make_request)

        result = RemoteFileService(files=upload_harness.files).upload_from_url(
            url=_REMOTE_URL, user=actor, tenant_id=_TENANT_ID
        )

        assert result.name == "report.txt"
        assert result.size == len(content)
        assert result.extension == "txt"
        assert result.mime_type == "text/plain"
        assert result.created_by == actor.id
        assert result.url == f"https://files.example.com/{result.id}?sign=test"
        assert requests == [
            ("HEAD", {}),
            ("GET", {} if head_status == httpx.codes.OK else {"timeout": 3, "follow_redirects": True}),
        ]
        assert upload_harness.io_events == ["HEAD", "GET", "storage", "sign"]
        assert upload_harness.database_events == ["begin", "end"]
        assert not upload_harness.active_transactions

        with sqlite_session_factory() as session:
            persisted = session.scalars(select(UploadFile)).one()
            assert persisted.id == result.id
            assert persisted.created_by == actor.id
            assert persisted.created_by_role == CreatorUserRole.ACCOUNT
            assert persisted.tenant_id == _TENANT_ID
            assert persisted.source_url == _REMOTE_URL
            assert persisted.key.startswith(f"upload_files/{_TENANT_ID}/")
            assert upload_harness.stored == {persisted.key: content}

    @staticmethod
    def test_same_actor_can_upload_to_separately_admitted_tenants(
        upload_harness: _UploadHarness,
        sqlite_session_factory: sessionmaker[Session],
    ) -> None:
        actor = FileUploadActor(id=_ACTOR_ID, creator_role=CreatorUserRole.ACCOUNT)
        tenant_ids = (_TENANT_ID, "33333333-3333-3333-3333-333333333333")

        results = [
            upload_harness.files.upload_file_for_actor(
                actor=actor,
                resource_tenant_id=tenant_id,
                filename="report.txt",
                content=tenant_id.encode(),
                mimetype="text/plain",
            )
            for tenant_id in tenant_ids
        ]

        assert [result.tenant_id for result in results] == list(tenant_ids)
        assert len({result.id for result in results}) == 2
        with sqlite_session_factory() as session:
            for result, tenant_id in zip(results, tenant_ids, strict=True):
                persisted = session.get(UploadFile, result.id)
                assert persisted is not None
                assert persisted.tenant_id == tenant_id
                assert persisted.created_by == actor.id
                assert persisted.created_by_role == actor.creator_role
                assert persisted.key.startswith(f"upload_files/{tenant_id}/")
                assert upload_harness.stored[persisted.key] == tenant_id.encode()

    @staticmethod
    @pytest.mark.parametrize("remote", [False, True], ids=["local", "remote"])
    def test_actor_upload_without_target_tenant_fails_before_external_io(
        upload_harness: _UploadHarness,
        monkeypatch: pytest.MonkeyPatch,
        remote: bool,
    ) -> None:
        actor = FileUploadActor(id=_ACTOR_ID, creator_role=CreatorUserRole.ACCOUNT)

        def unexpected_request(*_args: object, **_kwargs: object) -> httpx.Response:
            pytest.fail("Missing target tenant must be rejected before fetching the remote file")

        monkeypatch.setattr(remote_fetcher, "make_request", unexpected_request)

        if remote:
            with pytest.raises(TypeError, match="tenant_id"):
                RemoteFileService(files=upload_harness.files).upload_from_url(url=_REMOTE_URL, user=actor)
        else:
            with pytest.raises(TypeError, match="tenant_id"):
                upload_harness.files.upload_file(
                    filename="report.txt", content=b"file content", mimetype="text/plain", user=actor
                )

        assert upload_harness.io_events == []
        assert upload_harness.database_events == []
        assert upload_harness.stored == {}