composiohq__composio/python/tests/test_files.py

"""Tests for file helper functionality in composio/core/models/_files.py.

These tests ensure that the FileHelper class correctly handles JSON schemas
that use anyOf, oneOf, allOf, or $ref instead of direct 'type' properties.
"""

import tempfile
from pathlib import Path
from unittest.mock import Mock, patch, MagicMock

import pytest
import requests

from composio.client.types import Tool, tool_list_response
from composio.core.models._files import (
    FileDownloadable,
    FileHelper,
    FileUploadable,
    upload,
    _is_url,
    _get_extension_from_mimetype,
    _generate_timestamped_filename,
    _truncate_filename,
    _fetch_file_from_url,
    _upload_bytes_to_s3,
    _sanitize_url_for_logging,
    _MAX_FILENAME_LENGTH,
)
from composio.core.models.base import allow_tracking
from composio.exceptions import (
    BlockedInternalUrlError,
    ErrorDownloadingFile,
    ErrorUploadingFile,
    ResponseTooLargeError,
    SensitiveFilePathBlockedError,
    UnsafePathComponentError,
)


@pytest.fixture(autouse=True)
def disable_telemetry():
    """Disable telemetry for all tests to prevent thread issues."""
    token = allow_tracking.set(False)
    yield
    allow_tracking.reset(token)


@pytest.fixture
def mock_client():
    """Create a mock HTTP client."""
    return Mock()


@pytest.fixture
def file_helper(mock_client, monkeypatch, tmp_path):
    """Create a FileHelper instance with a mock client.

    Sandboxes ``COMPOSIO_CACHE_DIR`` so the default-outdir download path
    (``ensure_cache_directory()``) never touches the real home directory,
    even though ``FileDownloadable.download`` itself is mocked in most of
    these tests.
    """
    monkeypatch.setenv("COMPOSIO_CACHE_DIR", str(tmp_path / ".composio"))
    return FileHelper(client=mock_client)


@pytest.fixture
def mock_tool():
    """Create a mock tool for testing."""
    return Tool(
        name="Test Tool",
        slug="TEST_TOOL",
        description="Test tool",
        input_parameters={
            "properties": {
                "query": {"type": "string"},
            },
            "type": "object",
        },
        output_parameters={
            "properties": {
                "data": {"type": "object", "properties": {}},
            },
            "type": "object",
        },
        available_versions=["v1.0.0"],
        version="v1.0.0",
        scopes=[],
        toolkit=tool_list_response.ItemToolkit(
            name="Test Toolkit", slug="test_toolkit", logo=""
        ),
        deprecated=tool_list_response.ItemDeprecated(
            available_versions=["v1.0.0"],
            displayName="Test Tool",
            version="v1.0.0",
            toolkit=tool_list_response.ItemDeprecatedToolkit(logo=""),
            is_deprecated=False,
        ),
        is_deprecated=False,
        no_auth=False,
        tags=[],
    )


class TestFileHelperSchemaHandling:
    """Test cases for handling schemas without direct 'type' property.

    Regression tests for PLEN-766: KeyError: 'type' when schemas use
    anyOf, oneOf, allOf, or $ref instead of direct type properties.
    """

    def test_substitute_file_uploads_with_oneof_schema(self, file_helper, mock_tool):
        """Test that oneOf schemas don't cause KeyError in upload path."""
        schema_with_oneof = {
            "properties": {
                "input": {
                    "oneOf": [
                        {"type": "string"},
                        {"type": "object", "properties": {"file": {"type": "string"}}},
                    ]
                }
            }
        }
        request = {"input": {"file": "test.txt"}}

        # Should not raise KeyError
        result = file_helper._substitute_file_uploads_recursively(
            tool=mock_tool,
            schema=schema_with_oneof,
            request=request.copy(),
        )
        assert result == {"input": {"file": "test.txt"}}

    def test_substitute_file_uploads_with_anyof_schema(self, file_helper, mock_tool):
        """Test that anyOf schemas don't cause KeyError in upload path."""
        schema_with_anyof = {
            "properties": {
                "data": {
                    "anyOf": [
                        {"type": "string"},
                        {"type": "object", "properties": {"value": {"type": "string"}}},
                    ]
                }
            }
        }
        request = {"data": {"value": "test"}}

        # Should not raise KeyError
        result = file_helper._substitute_file_uploads_recursively(
            tool=mock_tool,
            schema=schema_with_anyof,
            request=request.copy(),
        )
        assert result == {"data": {"value": "test"}}

    def test_substitute_file_uploads_with_allof_schema(self, file_helper, mock_tool):
        """Test that allOf schemas don't cause KeyError in upload path."""
        schema_with_allof = {
            "properties": {
                "config": {
                    "allOf": [
                        {"properties": {"name": {"type": "string"}}},
                        {"properties": {"value": {"type": "string"}}},
                    ]
                }
            }
        }
        request = {"config": {"name": "test", "value": "123"}}

        # Should not raise KeyError
        result = file_helper._substitute_file_uploads_recursively(
            tool=mock_tool,
            schema=schema_with_allof,
            request=request.copy(),
        )
        assert result == {"config": {"name": "test", "value": "123"}}

    def test_substitute_file_uploads_with_ref_schema(self, file_helper, mock_tool):
        """Test that $ref schemas don't cause KeyError in upload path."""
        schema_with_ref = {
            "properties": {"reference": {"$ref": "#/definitions/SomeType"}}
        }
        request = {"reference": {"nested": "value"}}

        # Should not raise KeyError
        result = file_helper._substitute_file_uploads_recursively(
            tool=mock_tool,
            schema=schema_with_ref,
            request=request.copy(),
        )
        assert result == {"reference": {"nested": "value"}}

    def test_substitute_file_downloads_with_oneof_schema(self, file_helper, mock_tool):
        """Test that oneOf schemas don't cause KeyError in download path."""
        schema_with_oneof = {
            "properties": {
                "result": {
                    "oneOf": [
                        {"type": "string"},
                        {"type": "object", "properties": {"data": {"type": "string"}}},
                    ]
                }
            }
        }
        response = {"result": {"data": "some value"}}

        # Should not raise KeyError
        result = file_helper._substitute_file_downloads_recursively(
            tool=mock_tool,
            schema=schema_with_oneof,
            request=response.copy(),
        )
        assert result == {"result": {"data": "some value"}}

    def test_substitute_file_downloads_with_anyof_schema(self, file_helper, mock_tool):
        """Test that anyOf schemas don't cause KeyError in download path."""
        schema_with_anyof = {
            "properties": {
                "output": {
                    "anyOf": [
                        {"type": "string"},
                        {
                            "type": "object",
                            "properties": {"nested": {"type": "string"}},
                        },
                    ]
                }
            }
        }
        response = {"output": {"nested": "value"}}

        # Should not raise KeyError
        result = file_helper._substitute_file_downloads_recursively(
            tool=mock_tool,
            schema=schema_with_anyof,
            request=response.copy(),
        )
        assert result == {"output": {"nested": "value"}}

    def test_substitute_file_downloads_with_allof_schema(self, file_helper, mock_tool):
        """Test that allOf schemas don't cause KeyError in download path."""
        schema_with_allof = {
            "properties": {
                "response": {
                    "allOf": [
                        {"properties": {"status": {"type": "string"}}},
                        {"properties": {"message": {"type": "string"}}},
                    ]
                }
            }
        }
        response = {"response": {"status": "ok", "message": "success"}}

        # Should not raise KeyError
        result = file_helper._substitute_file_downloads_recursively(
            tool=mock_tool,
            schema=schema_with_allof,
            request=response.copy(),
        )
        assert result == {"response": {"status": "ok", "message": "success"}}

    def test_substitute_file_downloads_with_ref_schema(self, file_helper, mock_tool):
        """Test that $ref schemas don't cause KeyError in download path."""
        schema_with_ref = {
            "properties": {"data": {"$ref": "#/definitions/ResponseType"}}
        }
        response = {"data": {"field": "value"}}

        # Should not raise KeyError
        result = file_helper._substitute_file_downloads_recursively(
            tool=mock_tool,
            schema=schema_with_ref,
            request=response.copy(),
        )
        assert result == {"data": {"field": "value"}}

    def test_substitute_file_uploads_with_normal_type_still_works(
        self, file_helper, mock_tool
    ):
        """Test that normal schemas with 'type' property still work correctly."""
        schema_with_type = {
            "properties": {
                "nested": {
                    "type": "object",
                    "properties": {
                        "value": {"type": "string"},
                    },
                }
            }
        }
        request = {"nested": {"value": "test"}}

        # Should process normally and recurse into nested object
        result = file_helper._substitute_file_uploads_recursively(
            tool=mock_tool,
            schema=schema_with_type,
            request=request.copy(),
        )
        assert result == {"nested": {"value": "test"}}

    def test_substitute_file_downloads_with_normal_type_still_works(
        self, file_helper, mock_tool
    ):
        """Test that normal schemas with 'type' property still work correctly."""
        schema_with_type = {
            "properties": {
                "data": {
                    "type": "object",
                    "properties": {
                        "result": {"type": "string"},
                    },
                }
            }
        }
        response = {"data": {"result": "success"}}

        # Should process normally and recurse into nested object
        result = file_helper._substitute_file_downloads_recursively(
            tool=mock_tool,
            schema=schema_with_type,
            request=response.copy(),
        )
        assert result == {"data": {"result": "success"}}

    def test_substitute_with_empty_properties(self, file_helper, mock_tool):
        """Test handling of schemas with empty properties."""
        schema_empty = {"properties": {}}
        request = {"unknown": {"nested": "value"}}

        # Should not raise any errors
        result = file_helper._substitute_file_uploads_recursively(
            tool=mock_tool,
            schema=schema_empty,
            request=request.copy(),
        )
        assert result == {"unknown": {"nested": "value"}}

    def test_substitute_with_no_properties_key(self, file_helper, mock_tool):
        """Test handling of schemas without properties key."""
        schema_no_props = {"type": "object"}
        request = {"data": {"value": "test"}}

        # Should return request unchanged
        result = file_helper._substitute_file_uploads_recursively(
            tool=mock_tool,
            schema=schema_no_props,
            request=request.copy(),
        )
        assert result == {"data": {"value": "test"}}


class TestFileHelperRefDefsResolution:
    """File flags reachable only through a $ref/$defs indirection.

    Regression coverage for
    https://github.com/ComposioHQ/composio/issues/3506. References are inlined
    once at the public-method boundary (``dereference_json_schema``), so the
    walkers themselves stay reference-agnostic.
    """

    @patch("composio.core.models._files.FileDownloadable.download")
    def test_download_resolves_ref_defs(self, mock_download, file_helper, mock_tool):
        """GMAIL_GET_ATTACHMENT shape: file_downloadable two $ref hops deep."""
        mock_download.return_value = "/tmp/invoice.pdf"
        mock_tool.output_parameters = {
            "type": "object",
            "$defs": {
                "FileDownloadable": {
                    "type": "object",
                    "properties": {
                        "name": {"type": "string"},
                        "mimetype": {"type": "string"},
                        "s3url": {"type": "string"},
                    },
                    "required": ["name", "mimetype", "s3url"],
                    "file_downloadable": True,
                },
                "GetAttachmentResponse": {
                    "type": "object",
                    "properties": {
                        "file": {"$ref": "#/$defs/FileDownloadable"},
                        "display_url": {"type": "string"},
                    },
                    "required": ["file"],
                },
            },
            "properties": {
                "data": {"$ref": "#/$defs/GetAttachmentResponse"},
                "successful": {"type": "boolean"},
            },
            "required": ["data", "successful"],
        }
        response = {
            "data": {
                "file": {
                    "name": "invoice.pdf",
                    "mimetype": "application/pdf",
                    "s3url": "https://s3.example.com/invoice.pdf",
                },
                "display_url": "https://mail.google.com/...",
            },
            "successful": True,
        }

        result = file_helper.substitute_file_downloads(
            tool=mock_tool,
            response=response,
        )

        assert result["data"]["file"] == "/tmp/invoice.pdf"
        assert result["data"]["display_url"] == "https://mail.google.com/..."
        assert result["successful"] is True
        mock_download.assert_called_once()

    @patch("composio.core.models._files.FileUploadable.from_path")
    def test_upload_resolves_ref_defs(self, mock_from_path, file_helper, mock_tool):
        """$ref/$defs input schema exposes a nested file_uploadable leaf."""
        upload = MagicMock()
        upload.model_dump.return_value = {
            "name": "invoice.pdf",
            "mimetype": "application/pdf",
            "s3key": "uploads/invoice.pdf",
        }
        mock_from_path.return_value = upload
        mock_tool.input_parameters = {
            "type": "object",
            "$defs": {
                "FileUploadable": {"type": "string", "file_uploadable": True},
                "AttachmentInput": {
                    "type": "object",
                    "properties": {
                        "file": {"$ref": "#/$defs/FileUploadable"},
                        "name": {"type": "string"},
                    },
                    "required": ["file"],
                },
            },
            "properties": {"attachment": {"$ref": "#/$defs/AttachmentInput"}},
            "required": ["attachment"],
        }
        request = {"attachment": {"file": "/local/invoice.pdf", "name": "invoice.pdf"}}

        result = file_helper.substitute_file_uploads(
            tool=mock_tool,
            request=request,
        )

        assert result["attachment"]["file"] == {
            "name": "invoice.pdf",
            "mimetype": "application/pdf",
            "s3key": "uploads/invoice.pdf",
        }
        assert result["attachment"]["name"] == "invoice.pdf"
        mock_from_path.assert_called_once()

    def test_process_file_uploadable_schema_resolves_ref_defs(self, file_helper):
        """The LLM-facing input transform inlines $ref and rewrites the leaf."""
        schema = {
            "type": "object",
            "$defs": {
                "FileUploadable": {"type": "string", "file_uploadable": True},
                "AttachmentInput": {
                    "type": "object",
                    "properties": {"file": {"$ref": "#/$defs/FileUploadable"}},
                },
            },
            "properties": {"attachment": {"$ref": "#/$defs/AttachmentInput"}},
        }

        result = file_helper.process_file_uploadable_schema(schema)

        leaf = result["properties"]["attachment"]["properties"]["file"]
        assert leaf["type"] == "string"
        assert leaf["format"] == "path"
        assert leaf["file_uploadable"] is True
        # $defs is inlined away; no dangling references remain.
        assert "$defs" not in result
        assert "$ref" not in result["properties"]["attachment"]

    @patch("composio.core.models._files.FileDownloadable.download")
    def test_dangling_ref_download_degrades_gracefully(
        self, mock_download, file_helper, mock_tool
    ):
        """A $ref with no $defs block (issue #3307) must not raise (sentinel)."""
        mock_tool.output_parameters = {
            "type": "object",
            "properties": {
                "data": {"$ref": "#/$defs/Missing"},
                "successful": {"type": "boolean"},
            },
        }
        response = {"data": {"value": "passthrough"}, "successful": True}

        # No $defs block at all: strict resolution would raise and abort the
        # tool call. Sentinel mode keeps execution working — the unresolved
        # branch is simply not treated as a file.
        result = file_helper.substitute_file_downloads(
            tool=mock_tool,
            response=response,
        )

        assert result == {"data": {"value": "passthrough"}, "successful": True}
        mock_download.assert_not_called()

    @patch("composio.core.models._files.FileDownloadable.download")
    def test_recursive_ref_schema_does_not_recurse_infinitely(
        self, mock_download, file_helper, mock_tool
    ):
        """A self-referential $ref schema must not blow the stack.

        A per-node lazy resolver without threaded cycle tracking infinite-loops
        here; the centralized walker breaks the cycle with a sentinel.
        """
        mock_download.return_value = "/tmp/node.bin"
        mock_tool.output_parameters = {
            "type": "object",
            "$defs": {
                "Node": {
                    "type": "object",
                    "properties": {
                        "file": {
                            "type": "object",
                            "properties": {"s3url": {"type": "string"}},
                            "file_downloadable": True,
                        },
                        "child": {"$ref": "#/$defs/Node"},
                    },
                },
            },
            "properties": {"root": {"$ref": "#/$defs/Node"}},
        }
        response = {
            "root": {
                "file": {
                    "name": "n.bin",
                    "mimetype": "application/octet-stream",
                    "s3url": "https://s3/x",
                },
                "child": {"value": "leaf"},
            }
        }

        # Must complete without RecursionError.
        result = file_helper.substitute_file_downloads(
            tool=mock_tool,
            response=response,
        )

        assert result["root"]["file"] == "/tmp/node.bin"
        assert result["root"]["child"] == {"value": "leaf"}


class TestFileUploadableInUnionTypes:
    """Test cases for file_uploadable detection in anyOf, oneOf, and allOf schemas."""

    def test_has_file_property_direct(self, file_helper):
        """Test _has_file_property detects direct file_uploadable."""
        schema = {"type": "string", "file_uploadable": True}
        assert file_helper._has_file_property(schema, "file_uploadable") is True

    def test_has_file_property_in_anyof(self, file_helper):
        """Test _has_file_property detects file_uploadable in anyOf."""
        schema = {
            "anyOf": [
                {"type": "string", "file_uploadable": True},
                {"type": "null"},
            ]
        }
        assert file_helper._has_file_property(schema, "file_uploadable") is True

    def test_has_file_property_in_oneof(self, file_helper):
        """Test _has_file_property detects file_uploadable in oneOf."""
        schema = {
            "oneOf": [
                {"type": "string", "file_uploadable": True},
                {"type": "string", "description": "URL"},
            ]
        }
        assert file_helper._has_file_property(schema, "file_uploadable") is True

    def test_has_file_property_in_allof(self, file_helper):
        """Test _has_file_property detects file_uploadable in allOf."""
        schema = {
            "allOf": [
                {"type": "string", "file_uploadable": True},
                {"minLength": 1},
            ]
        }
        assert file_helper._has_file_property(schema, "file_uploadable") is True

    def test_has_file_property_nested_in_anyof(self, file_helper):
        """Test _has_file_property detects file_uploadable nested in anyOf object."""
        schema = {
            "anyOf": [
                {
                    "type": "object",
                    "properties": {
                        "attachment": {"type": "string", "file_uploadable": True}
                    },
                },
                {"type": "string"},
            ]
        }
        assert file_helper._has_file_property(schema, "file_uploadable") is True

    def test_has_file_property_not_present(self, file_helper):
        """Test _has_file_property returns False when not present."""
        schema = {
            "anyOf": [
                {"type": "string"},
                {"type": "null"},
            ]
        }
        assert file_helper._has_file_property(schema, "file_uploadable") is False

    def test_has_file_downloadable_in_anyof(self, file_helper):
        """Test _has_file_property detects file_downloadable in anyOf."""
        schema = {
            "anyOf": [
                {"type": "object", "file_downloadable": True},
                {"type": "null"},
            ]
        }
        assert file_helper._has_file_property(schema, "file_downloadable") is True

    def test_transform_schema_direct_file_uploadable(self, file_helper):
        """Test _transform_schema_for_file_upload transforms direct file_uploadable."""
        schema = {
            "type": "string",
            "file_uploadable": True,
            "description": "Upload a file",
        }
        result = file_helper._transform_schema_for_file_upload(schema)
        assert result["format"] == "path"
        assert result["type"] == "string"
        assert result["file_uploadable"] is True

    def test_transform_schema_file_uploadable_in_anyof(self, file_helper):
        """Test _transform_schema_for_file_upload transforms file_uploadable in anyOf."""
        schema = {
            "anyOf": [
                {"type": "string", "file_uploadable": True, "description": "File path"},
                {"type": "null"},
            ]
        }
        result = file_helper._transform_schema_for_file_upload(schema)
        assert result["anyOf"][0]["format"] == "path"
        assert result["anyOf"][0]["file_uploadable"] is True
        assert "format" not in result["anyOf"][1]

    def test_transform_schema_file_uploadable_in_oneof(self, file_helper):
        """Test _transform_schema_for_file_upload transforms file_uploadable in oneOf."""
        schema = {
            "oneOf": [
                {"type": "string", "file_uploadable": True},
                {"type": "string", "description": "URL reference"},
            ]
        }
        result = file_helper._transform_schema_for_file_upload(schema)
        assert result["oneOf"][0]["format"] == "path"
        assert result["oneOf"][0]["file_uploadable"] is True
        assert "format" not in result["oneOf"][1]

    def test_transform_schema_file_uploadable_in_allof(self, file_helper):
        """Test _transform_schema_for_file_upload transforms file_uploadable in allOf."""
        schema = {
            "allOf": [
                {"type": "string", "file_uploadable": True},
                {"minLength": 1},
            ]
        }
        result = file_helper._transform_schema_for_file_upload(schema)
        assert result["allOf"][0]["format"] == "path"
        assert result["allOf"][0]["file_uploadable"] is True

    def test_transform_schema_nested_file_uploadable_in_anyof(self, file_helper):
        """Test transform handles nested file_uploadable inside anyOf objects."""
        schema = {
            "anyOf": [
                {
                    "type": "object",
                    "properties": {
                        "attachment": {"type": "string", "file_uploadable": True}
                    },
                },
                {"type": "string"},
            ]
        }
        result = file_helper._transform_schema_for_file_upload(schema)
        assert result["anyOf"][0]["properties"]["attachment"]["format"] == "path"
        assert result["anyOf"][0]["properties"]["attachment"]["file_uploadable"] is True

    def test_transform_schema_array_items_with_anyof(self, file_helper):
        """Test transform handles array items with anyOf containing file_uploadable."""
        schema = {
            "type": "array",
            "items": {
                "anyOf": [
                    {"type": "string", "file_uploadable": True},
                    {"type": "null"},
                ]
            },
        }
        result = file_helper._transform_schema_for_file_upload(schema)
        assert result["items"]["anyOf"][0]["format"] == "path"
        assert result["items"]["anyOf"][0]["file_uploadable"] is True

    def test_process_file_uploadable_schema_with_anyof(self, file_helper):
        """Test process_file_uploadable_schema handles anyOf properties."""
        schema = {
            "type": "object",
            "properties": {
                "fileInput": {
                    "anyOf": [
                        {"type": "string", "file_uploadable": True},
                        {"type": "null"},
                    ]
                },
                "text": {"type": "string"},
            },
        }
        result = file_helper.process_file_uploadable_schema(schema)
        assert result["properties"]["fileInput"]["anyOf"][0]["format"] == "path"
        assert "format" not in result["properties"]["text"]


class TestFileUploadSubstitutionWithUnionTypes:
    """Test cases for file upload substitution with anyOf, oneOf, and allOf schemas."""

    def test_substitute_upload_with_file_uploadable_in_anyof(
        self, file_helper, mock_tool, mock_client
    ):
        """Test that file_uploadable in anyOf triggers file upload."""
        mock_tool.input_parameters = {
            "type": "object",
            "properties": {
                "fileInput": {
                    "anyOf": [
                        {"type": "string", "file_uploadable": True},
                        {"type": "null"},
                    ]
                }
            },
        }

        # Mock the file upload
        mock_client.post.return_value = Mock(
            key="s3key/file.txt", new_presigned_url="https://s3.example.com/upload"
        )

        with (
            pytest.raises(Exception),
        ):  # Will fail because file doesn't exist, but proves detection works
            file_helper._substitute_file_uploads_recursively(
                tool=mock_tool,
                schema=mock_tool.input_parameters,
                request={"fileInput": "/path/to/file.txt"},
            )

    def test_substitute_upload_null_value_in_anyof(self, file_helper, mock_tool):
        """Test that null values in anyOf with file_uploadable are handled."""
        mock_tool.input_parameters = {
            "type": "object",
            "properties": {
                "fileInput": {
                    "anyOf": [
                        {"type": "string", "file_uploadable": True},
                        {"type": "null"},
                    ]
                }
            },
        }

        result = file_helper._substitute_file_uploads_recursively(
            tool=mock_tool,
            schema=mock_tool.input_parameters,
            request={"fileInput": None},
        )
        # None/empty values should be removed
        assert "fileInput" not in result

    def test_substitute_upload_preserves_null_optional_object_with_nested_file(
        self, file_helper, mock_tool
    ):
        """A null container is not itself a file-uploadable leaf."""
        mock_tool.input_parameters = {
            "type": "object",
            "properties": {
                "options": {
                    "type": "object",
                    "properties": {
                        "attachment": {
                            "type": "object",
                            "file_uploadable": True,
                        }
                    },
                }
            },
        }
        request = {"options": None}

        with patch.object(FileUploadable, "from_path") as from_path:
            result = file_helper.substitute_file_uploads(
                tool=mock_tool,
                request=request,
            )

        assert result is request
        assert result == {"options": None}
        from_path.assert_not_called()

    def test_drop_empty_file_uploads_omits_empty_strings_without_uploading(
        self, file_helper, mock_tool
    ):
        """Disabled auto-upload omits empty strings but preserves explicit nulls."""
        file_uploadable = {
            "type": "object",
            "file_uploadable": True,
            "title": "FileUploadable",
            "properties": {
                "name": {"type": "string"},
                "mimetype": {"type": "string"},
                "s3key": {"type": "string"},
            },
            "required": ["name", "mimetype", "s3key"],
        }
        mock_tool.input_parameters = {
            "type": "object",
            "properties": {
                "subject": {"type": "string"},
                "attachment": {
                    "anyOf": [
                        file_uploadable,
                        {"type": "array", "items": file_uploadable},
                        {"type": "null"},
                    ],
                    "default": None,
                },
                "extra": {
                    "anyOf": [
                        {"type": "array", "items": file_uploadable},
                        {"type": "null"},
                    ]
                },
                "nested": {
                    "type": "object",
                    "properties": {"file": {"$ref": "#/$defs/F"}},
                },
                "opaque": {"type": "object", "additionalProperties": True},
                "thread_id": {"type": "string"},
            },
            "$defs": {"F": file_uploadable},
        }
        staged = {"name": "a.txt", "mimetype": "text/plain", "s3key": "k"}
        request = {
            "subject": "Test",
            "attachment": "",
            "extra": [None, "", staged, "/tmp/keep.txt"],
            "nested": {"file": None},
            "opaque": {"preserve_identity": True},
            "thread_id": "",
        }
        original_request = {
            "subject": "Test",
            "attachment": "",
            "extra": [None, "", staged, "/tmp/keep.txt"],
            "nested": {"file": None},
            "opaque": {"preserve_identity": True},
            "thread_id": "",
        }

        with patch.object(FileUploadable, "from_path") as from_path:
            result = file_helper.drop_empty_file_uploads(
                tool=mock_tool, request=request
            )

        from_path.assert_not_called()
        assert result is not request
        assert result == {
            "subject": "Test",
            "extra": [None, staged, "/tmp/keep.txt"],
            "nested": {"file": None},
            "opaque": {"preserve_identity": True},
            # non-file empty strings are not the walker's business
            "thread_id": "",
        }
        assert result["extra"] is not request["extra"]
        assert result["extra"][1] is request["extra"][2]
        assert result["nested"] is not request["nested"]
        assert result["opaque"] is request["opaque"]
        assert request == original_request

    def test_drop_empty_file_uploads_skips_dereference_for_non_file_schema(
        self, file_helper, mock_tool
    ):
        """Default execution does not dereference schemas without file inputs."""
        mock_tool.input_parameters = {
            "type": "object",
            "properties": {
                "filters": {
                    "type": "object",
                    "properties": {"query": {"type": "string"}},
                }
            },
        }
        request = {"filters": {"query": "open"}}

        with patch(
            "composio.core.models._files.dereference_json_schema"
        ) as dereference:
            result = file_helper.drop_empty_file_uploads(
                tool=mock_tool, request=request
            )

        assert result is request
        dereference.assert_not_called()

    @pytest.mark.parametrize(
        ("definitions_key", "ref"),
        [
            ("$defs", "#/$defs/FileUploadable"),
            ("definitions", "#/definitions/FileUploadable"),
        ],
    )
    def test_drop_empty_file_uploads_resolves_referenced_file_schema(
        self, file_helper, mock_tool, definitions_key, ref
    ):
        """The raw cheap gate still admits modern and legacy referenced schemas."""
        mock_tool.input_parameters = {
            "type": "object",
            "properties": {"attachment": {"$ref": ref}},
            definitions_key: {
                "FileUploadable": {
                    "type": "object",
                    "file_uploadable": True,
                }
            },
        }

        result = file_helper.drop_empty_file_uploads(
            tool=mock_tool, request={"attachment": ""}
        )

        assert result == {}

    @pytest.mark.parametrize(
        "attachment_schema",
        [
            {
                "type": "array",
                "items": {"type": "object", "file_uploadable": True},
            },
            {
                "items": {"type": "object", "file_uploadable": True},
            },
            {
                "anyOf": [
                    {
                        "type": "array",
                        "items": {"type": "object", "file_uploadable": True},
                    },
                    {"type": "null"},
                ]
            },
            {
                "anyOf": [
                    {
                        "items": {"type": "object", "file_uploadable": True},
                    },
                    {"type": "null"},
                ]
            },
        ],
    )
    @pytest.mark.parametrize(
        ("value", "expected_without_upload"),
        [("", {}), (None, {"attachment": None})],
    )
    def test_empty_values_follow_mode_for_array_only_file_schema(
        self, file_helper, mock_tool, attachment_schema, value, expected_without_upload
    ):
        """Array-only file inputs omit empty strings and upload-mode nulls."""
        mock_tool.input_parameters = {
            "type": "object",
            "properties": {"attachment": attachment_schema},
        }

        dropped = file_helper.drop_empty_file_uploads(
            tool=mock_tool, request={"attachment": value}
        )
        with patch.object(FileUploadable, "from_path") as from_path:
            uploaded = file_helper.substitute_file_uploads(
                tool=mock_tool, request={"attachment": value}
            )

        assert dropped == expected_without_upload
        assert uploaded == {}
        from_path.assert_not_called()

    @pytest.mark.parametrize("array_type", ["explicit", "inferred"])
    def test_empty_string_is_preserved_when_non_file_string_variant_matches(
        self, file_helper, mock_tool, array_type
    ):
        """A composed string branch takes precedence over array-file cleanup."""
        array_file_schema = {
            "items": {
                "type": "object",
                "file_uploadable": True,
            },
        }
        if array_type == "explicit":
            array_file_schema["type"] = "array"

        mock_tool.input_parameters = {
            "type": "object",
            "properties": {
                "attachment": {
                    "anyOf": [
                        array_file_schema,
                        {"type": "string"},
                    ]
                }
            },
        }

        dropped = file_helper.drop_empty_file_uploads(
            tool=mock_tool, request={"attachment": ""}
        )
        with patch.object(FileUploadable, "from_path") as from_path:
            uploaded = file_helper.substitute_file_uploads(
                tool=mock_tool, request={"attachment": ""}
            )

        assert dropped == {"attachment": ""}
        assert uploaded == {"attachment": ""}
        from_path.assert_not_called()

    def test_substitute_upload_empty_string_in_anyof(self, file_helper, mock_tool):
        """Test that empty string values in anyOf with file_uploadable are handled."""
        mock_tool.input_parameters = {
            "type": "object",
            "properties": {
                "fileInput": {
                    "anyOf": [
                        {"type": "string", "file_uploadable": True},
                        {"type": "null"},
                    ]
                }
            },
        }

        result = file_helper._substitute_file_uploads_recursively(
            tool=mock_tool,
            schema=mock_tool.input_parameters,
            request={"fileInput": ""},
        )
        # Empty values should be removed
        assert "fileInput" not in result

    def test_substitute_upload_nested_in_anyof_object(self, file_helper, mock_tool):
        """Test file upload for nested file_uploadable inside anyOf object."""
        mock_tool.input_parameters = {
            "type": "object",
            "properties": {
                "content": {
                    "anyOf": [
                        {
                            "type": "object",
                            "properties": {
                                "attachment": {
                                    "type": "string",
                                    "file_uploadable": True,
                                }
                            },
                        },
                        {"type": "string"},
                    ]
                }
            },
        }

        # Should try to process the nested file_uploadable
        with pytest.raises(Exception):  # Will fail because file doesn't exist
            file_helper._substitute_file_uploads_recursively(
                tool=mock_tool,
                schema=mock_tool.input_parameters,
                request={"content": {"attachment": "/path/to/file.pdf"}},
            )

    @patch("composio.core.models._files.FileUploadable.from_path")
    def test_substitute_upload_anyof_single_or_array_prefers_array_for_list_value(
        self, mock_from_path, file_helper, mock_tool
    ):
        """List runtime values should use the array branch and upload each item."""
        mock_tool.input_parameters = {
            "type": "object",
            "properties": {
                "attachment": {
                    "anyOf": [
                        {"type": "string", "file_uploadable": True},
                        {
                            "type": "array",
                            "items": {"type": "string", "file_uploadable": True},
                        },
                    ]
                }
            },
        }

        upload_1 = MagicMock()
        upload_1.model_dump.return_value = {
            "name": "a.txt",
            "mimetype": "text/plain",
            "s3key": "key-a",
        }
        upload_2 = MagicMock()
        upload_2.model_dump.return_value = {
            "name": "b.txt",
            "mimetype": "text/plain",
            "s3key": "key-b",
        }
        mock_from_path.side_effect = [upload_1, upload_2]

        request = {"attachment": ["/tmp/a.txt", "/tmp/b.txt"]}
        result = file_helper._substitute_file_uploads_recursively(
            tool=mock_tool,
            schema=mock_tool.input_parameters,
            request=request,
        )

        assert result is request
        assert result["attachment"] == [
            {"name": "a.txt", "mimetype": "text/plain", "s3key": "key-a"},
            {"name": "b.txt", "mimetype": "text/plain", "s3key": "key-b"},
        ]
        assert [call.kwargs["file"] for call in mock_from_path.call_args_list] == [
            "/tmp/a.txt",
            "/tmp/b.txt",
        ]

    @patch("composio.core.models._files.FileUploadable.from_path")
    def test_substitute_upload_anyof_single_or_array_keeps_string_behavior(
        self, mock_from_path, file_helper, mock_tool
    ):
        """String runtime values should still use the single-file branch."""
        mock_tool.input_parameters = {
            "type": "object",
            "properties": {
                "attachment": {
                    "anyOf": [
                        {"type": "string", "file_uploadable": True},
                        {
                            "type": "array",
                            "items": {"type": "string", "file_uploadable": True},
                        },
                    ]
                }
            },
        }

        upload = MagicMock()
        upload.model_dump.return_value = {
            "name": "a.txt",
            "mimetype": "text/plain",
            "s3key": "key-a",
        }
        mock_from_path.return_value = upload

        result = file_helper._substitute_file_uploads_recursively(
            tool=mock_tool,
            schema=mock_tool.input_parameters,
            request={"attachment": "/tmp/a.txt"},
        )

        assert result["attachment"] == {
            "name": "a.txt",
            "mimetype": "text/plain",
            "s3key": "key-a",
        }
        mock_from_path.assert_called_once()
        assert mock_from_path.call_args.kwargs["file"] == "/tmp/a.txt"

    @patch("composio.core.models._files.FileUploadable.from_path")
    def test_substitute_upload_array_items_with_anyof_file_branch(
        self, mock_from_path, file_helper, mock_tool
    ):
        """Array item schemas can contain file-uploadable union branches."""
        mock_tool.input_parameters = {
            "type": "object",
            "properties": {
                "attachments": {
                    "type": "array",
                    "items": {
                        "anyOf": [
                            {"type": "string", "file_uploadable": True},
                            {"type": "null"},
                        ]
                    },
                }
            },
        }

        upload_1 = MagicMock()
        upload_1.model_dump.return_value = {"s3key": "key-a"}
        upload_2 = MagicMock()
        upload_2.model_dump.return_value = {"s3key": "key-b"}
        mock_from_path.side_effect = [upload_1, upload_2]

        result = file_helper._substitute_file_uploads_recursively(
            tool=mock_tool,
            schema=mock_tool.input_parameters,
            request={"attachments": ["/tmp/a.txt", None, "", "/tmp/b.txt"]},
        )

        assert result["attachments"] == [{"s3key": "key-a"}, {"s3key": "key-b"}]
        assert [call.kwargs["file"] for call in mock_from_path.call_args_list] == [
            "/tmp/a.txt",
            "/tmp/b.txt",
        ]

    @patch("composio.core.models._files.FileUploadable.from_path")
    def test_substitute_upload_array_drops_all_null_and_empty_items(
        self, mock_from_path, file_helper, mock_tool
    ):
        """Array of file-uploadable items collapses to [] when every item is null/empty."""
        mock_tool.input_parameters = {
            "type": "object",
            "properties": {
                "attachments": {
                    "type": "array",
                    "items": {"type": "string", "file_uploadable": True},
                }
            },
        }

        result = file_helper._substitute_file_uploads_recursively(
            tool=mock_tool,
            schema=mock_tool.input_parameters,
            request={"attachments": [None, "", None]},
        )

        assert result["attachments"] == []
        mock_from_path.assert_not_called()


class TestFileDownloadSubstitutionWithUnionTypes:
    """Test cases for file download substitution with anyOf, oneOf, and allOf schemas."""

    def test_substitute_download_with_file_downloadable_in_anyof(
        self, file_helper, mock_tool
    ):
        """Test that file_downloadable in anyOf triggers file download detection."""
        mock_tool.output_parameters = {
            "type": "object",
            "properties": {
                "fileOutput": {
                    "anyOf": [
                        {
                            "type": "object",
                            "file_downloadable": True,
                            "properties": {
                                "s3url": {"type": "string"},
                                "mimetype": {"type": "string"},
                                "name": {"type": "string"},
                            },
                        },
                        {"type": "null"},
                    ]
                }
            },
        }

        # Should try to download the file (will fail but proves detection)
        with pytest.raises(Exception):
            file_helper._substitute_file_downloads_recursively(
                tool=mock_tool,
                schema=mock_tool.output_parameters,
                request={
                    "fileOutput": {
                        "s3url": "https://s3.example.com/file.txt",
                        "mimetype": "text/plain",
                        "name": "file.txt",
                    }
                },
            )

    def test_substitute_download_null_value_in_anyof(self, file_helper, mock_tool):
        """Test that null values in anyOf with file_downloadable are handled."""
        mock_tool.output_parameters = {
            "type": "object",
            "properties": {
                "fileOutput": {
                    "anyOf": [
                        {"type": "object", "file_downloadable": True},
                        {"type": "null"},
                    ]
                }
            },
        }

        result = file_helper._substitute_file_downloads_recursively(
            tool=mock_tool,
            schema=mock_tool.output_parameters,
            request={"fileOutput": None},
        )
        assert result["fileOutput"] is None

    def test_substitute_download_nested_in_anyof_object(self, file_helper, mock_tool):
        """Test file download for nested file_downloadable inside anyOf object."""
        mock_tool.output_parameters = {
            "type": "object",
            "properties": {
                "response": {
                    "anyOf": [
                        {
                            "type": "object",
                            "properties": {
                                "attachment": {
                                    "type": "object",
                                    "file_downloadable": True,
                                    "properties": {
                                        "s3url": {"type": "string"},
                                        "mimetype": {"type": "string"},
                                        "name": {"type": "string"},
                                    },
                                }
                            },
                        },
                        {"type": "string"},
                    ]
                }
            },
        }

        # Should try to download the nested file
        with pytest.raises(Exception):
            file_helper._substitute_file_downloads_recursively(
                tool=mock_tool,
                schema=mock_tool.output_parameters,
                request={
                    "response": {
                        "attachment": {
                            "s3url": "https://s3.example.com/doc.pdf",
                            "mimetype": "application/pdf",
                            "name": "document.pdf",
                        }
                    }
                },
            )

    def test_substitute_download_with_file_downloadable_in_oneof(
        self, file_helper, mock_tool
    ):
        """Test that file_downloadable in oneOf triggers file download detection."""
        mock_tool.output_parameters = {
            "type": "object",
            "properties": {
                "result": {
                    "oneOf": [
                        {
                            "type": "object",
                            "file_downloadable": True,
                            "properties": {
                                "s3url": {"type": "string"},
                                "mimetype": {"type": "string"},
                                "name": {"type": "string"},
                            },
                        },
                        {"type": "string"},
                    ]
                }
            },
        }

        with pytest.raises(Exception):
            file_helper._substitute_file_downloads_recursively(
                tool=mock_tool,
                schema=mock_tool.output_parameters,
                request={
                    "result": {
                        "s3url": "https://s3.example.com/file.txt",
                        "mimetype": "text/plain",
                        "name": "file.txt",
                    }
                },
            )

    def test_substitute_download_with_file_downloadable_in_allof(
        self, file_helper, mock_tool
    ):
        """Test that file_downloadable in allOf triggers file download detection."""
        mock_tool.output_parameters = {
            "type": "object",
            "properties": {
                "image": {
                    "allOf": [
                        {
                            "type": "object",
                            "file_downloadable": True,
                            "properties": {
                                "s3url": {"type": "string"},
                                "mimetype": {"type": "string"},
                                "name": {"type": "string"},
                            },
                        },
                        {"required": ["s3url"]},
                    ]
                }
            },
        }

        with pytest.raises(Exception):
            file_helper._substitute_file_downloads_recursively(
                tool=mock_tool,
                schema=mock_tool.output_parameters,
                request={
                    "image": {
                        "s3url": "https://s3.example.com/image.png",
                        "mimetype": "image/png",
                        "name": "image.png",
                    }
                },
            )

    @patch("composio.core.models._files.FileDownloadable.download")
    def test_substitute_download_anyof_single_or_array_prefers_array_for_list_value(
        self, mock_download, file_helper, mock_tool
    ):
        """List runtime values should use the array branch and download each item."""
        mock_tool.output_parameters = {
            "type": "object",
            "properties": {
                "attachment": {
                    "anyOf": [
                        {"type": "object", "file_downloadable": True},
                        {
                            "type": "array",
                            "items": {"type": "object", "file_downloadable": True},
                        },
                    ]
                }
            },
        }
        mock_download.side_effect = ["/tmp/a.txt", "/tmp/b.txt"]

        request = {
            "attachment": [
                {
                    "s3url": "https://s3.example.com/a.txt",
                    "mimetype": "text/plain",
                    "name": "a.txt",
                },
                {
                    "s3url": "https://s3.example.com/b.txt",
                    "mimetype": "text/plain",
                    "name": "b.txt",
                },
            ]
        }
        result = file_helper._substitute_file_downloads_recursively(
            tool=mock_tool,
            schema=mock_tool.output_parameters,
            request=request,
        )

        assert result is request
        assert result["attachment"] == ["/tmp/a.txt", "/tmp/b.txt"]
        assert mock_download.call_count == 2

    @patch("composio.core.models._files.FileDownloadable.download")
    def test_substitute_download_array_items_with_anyof_file_branch(
        self, mock_download, file_helper, mock_tool
    ):
        """Array item schemas can contain file-downloadable union branches."""
        mock_tool.output_parameters = {
            "type": "object",
            "properties": {
                "attachments": {
                    "type": "array",
                    "items": {
                        "anyOf": [
                            {"type": "object", "file_downloadable": True},
                            {"type": "null"},
                        ]
                    },
                }
            },
        }
        mock_download.side_effect = ["/tmp/a.txt", "/tmp/b.txt"]

        result = file_helper._substitute_file_downloads_recursively(
            tool=mock_tool,
            schema=mock_tool.output_parameters,
            request={
                "attachments": [
                    {
                        "s3url": "https://s3.example.com/a.txt",
                        "mimetype": "text/plain",
                        "name": "a.txt",
                    },
                    None,
                    {
                        "s3url": "https://s3.example.com/b.txt",
                        "mimetype": "text/plain",
                        "name": "b.txt",
                    },
                ]
            },
        )

        assert result["attachments"] == ["/tmp/a.txt", None, "/tmp/b.txt"]
        assert mock_download.call_count == 2


class TestFileHelperFindVariantMethods:
    """Test cases for _find_uploadable_schema_variant and _find_downloadable_schema_variant."""

    def test_find_uploadable_variant_in_anyof(self, file_helper):
        """Test finding uploadable variant in anyOf."""
        schema = {
            "anyOf": [
                {"type": "string", "file_uploadable": True},
                {"type": "null"},
            ]
        }
        result = file_helper._find_uploadable_schema_variant(schema)
        assert result is not None
        assert result.get("file_uploadable") is True

    def test_find_uploadable_variant_in_oneof(self, file_helper):
        """Test finding uploadable variant in oneOf."""
        schema = {
            "oneOf": [
                {"type": "string"},
                {"type": "string", "file_uploadable": True},
            ]
        }
        result = file_helper._find_uploadable_schema_variant(schema)
        assert result is not None
        assert result.get("file_uploadable") is True

    def test_find_uploadable_variant_in_allof(self, file_helper):
        """Test finding uploadable variant in allOf."""
        schema = {
            "allOf": [
                {"type": "string", "file_uploadable": True},
                {"minLength": 1},
            ]
        }
        result = file_helper._find_uploadable_schema_variant(schema)
        assert result is not None
        assert result.get("file_uploadable") is True

    def test_find_uploadable_variant_not_found(self, file_helper):
        """Test that None is returned when no uploadable variant exists."""
        schema = {
            "anyOf": [
                {"type": "string"},
                {"type": "null"},
            ]
        }
        result = file_helper._find_uploadable_schema_variant(schema)
        assert result is None

    def test_find_uploadable_variant_prefers_runtime_value_shape(self, file_helper):
        """Runtime values should select the matching file-bearing schema shape."""
        schema = {
            "anyOf": [
                {"type": "string", "file_uploadable": True},
                {
                    "type": "array",
                    "items": {"type": "string", "file_uploadable": True},
                },
            ]
        }

        result = file_helper._find_uploadable_schema_variant(
            schema,
            value=["/tmp/a.txt", "/tmp/b.txt"],
        )

        assert result is not None
        assert result["type"] == "array"

    def test_find_uploadable_variant_falls_back_to_first_file_bearing_variant(
        self, file_helper
    ):
        """No runtime shape match should preserve historical first-match behavior."""
        schema = {
            "anyOf": [
                {"type": "string", "file_uploadable": True},
                {
                    "type": "array",
                    "items": {"type": "string", "file_uploadable": True},
                },
            ]
        }

        result = file_helper._find_uploadable_schema_variant(schema, value=123)

        assert result is not None
        assert result["type"] == "string"

    def test_find_downloadable_variant_in_anyof(self, file_helper):
        """Test finding downloadable variant in anyOf."""
        schema = {
            "anyOf": [
                {"type": "object", "file_downloadable": True},
                {"type": "null"},
            ]
        }
        result = file_helper._find_downloadable_schema_variant(schema)
        assert result is not None
        assert result.get("file_downloadable") is True

    def test_find_downloadable_variant_in_oneof(self, file_helper):
        """Test finding downloadable variant in oneOf."""
        schema = {
            "oneOf": [
                {"type": "string"},
                {"type": "object", "file_downloadable": True},
            ]
        }
        result = file_helper._find_downloadable_schema_variant(schema)
        assert result is not None
        assert result.get("file_downloadable") is True

    def test_find_downloadable_variant_nested(self, file_helper):
        """Test finding downloadable variant with nested file_downloadable."""
        schema = {
            "anyOf": [
                {
                    "type": "object",
                    "properties": {
                        "file": {"type": "object", "file_downloadable": True}
                    },
                },
                {"type": "null"},
            ]
        }
        result = file_helper._find_downloadable_schema_variant(schema)
        assert result is not None
        assert result.get("type") == "object"


class TestFileUploadWithMixedSchemas:
    """Test cases for schemas with anyOf/oneOf/allOf alongside properties with file_uploadable."""

    def test_upload_from_base_properties_when_anyof_has_no_file_uploadable(
        self, file_helper, mock_tool
    ):
        """Test that file_uploadable in base properties works when sibling has anyOf without file_uploadable."""
        mock_tool.input_parameters = {
            "type": "object",
            "properties": {
                "metadata": {
                    "anyOf": [
                        {"type": "string"},
                        {"type": "null"},
                    ]
                },
                "file": {
                    "type": "string",
                    "file_uploadable": True,
                },
            },
        }

        # Should try to upload the file (will fail because file doesn't exist)
        with pytest.raises(Exception):
            file_helper._substitute_file_uploads_recursively(
                tool=mock_tool,
                schema=mock_tool.input_parameters,
                request={"metadata": "some data", "file": "/path/to/file.txt"},
            )

    def test_upload_when_root_has_anyof_without_file_uploadable_and_properties_with_file_uploadable(
        self, file_helper, mock_tool
    ):
        """Test upload when root schema has anyOf (no file_uploadable) and properties (with file_uploadable)."""
        mock_tool.input_parameters = {
            "type": "object",
            "anyOf": [
                {"required": ["text"]},
                {"required": ["file"]},
            ],
            "properties": {
                "text": {"type": "string"},
                "file": {"type": "string", "file_uploadable": True},
            },
        }

        # Should try to upload the file
        with pytest.raises(Exception):
            file_helper._substitute_file_uploads_recursively(
                tool=mock_tool,
                schema=mock_tool.input_parameters,
                request={"file": "/path/to/document.pdf"},
            )

    def test_metadata_preserved_when_file_upload_with_mixed_schema(
        self, file_helper, mock_tool
    ):
        """Test that non-file properties are preserved when processing mixed schemas."""
        mock_tool.input_parameters = {
            "type": "object",
            "properties": {
                "metadata": {
                    "anyOf": [
                        {"type": "string"},
                        {"type": "null"},
                    ]
                },
                "count": {"type": "integer"},
            },
        }

        result = file_helper._substitute_file_uploads_recursively(
            tool=mock_tool,
            schema=mock_tool.input_parameters,
            request={"metadata": "test data", "count": 42},
        )

        # Both values should be preserved unchanged
        assert result["metadata"] == "test data"
        assert result["count"] == 42


class TestUrlHelperFunctions:
    """Test cases for URL-related helper functions."""

    def test_is_url_with_http(self):
        """Test _is_url correctly identifies HTTP URLs."""
        assert _is_url("http://example.com/file.jpg") is True
        assert _is_url("http://localhost:8080/api/file") is True

    def test_is_url_with_https(self):
        """Test _is_url correctly identifies HTTPS URLs."""
        assert _is_url("https://example.com/file.jpg") is True
        assert _is_url("https://images.pexels.com/photos/123.jpg") is True

    def test_is_url_with_long_real_world_urls(self):
        """Test _is_url handles long real-world URLs with query parameters."""
        assert (
            _is_url(
                "https://encrypted-tbn0.gstatic.com/images?q=tbn:ANd9GcQq0powzLhfi1bakZ0eNSIA_aJ_5UlPsCte1g&s"
            )
            is True
        )
        assert (
            _is_url(
                "https://images.unsplash.com/photo-1503023345310-bd7c1de61c7d?ixid=MnwxMjA3fDB8MHxzZWFyY2h8MXx8bWFjYm9vfGVufDB8fDB8fA%3D%3D&auto=format&fit=crop&w=800&q=60"
            )
            is True
        )

    def test_is_url_with_local_paths(self):
        """Test _is_url correctly rejects local file paths."""
        assert _is_url("/path/to/file.jpg") is False
        assert _is_url("./relative/path.txt") is False
        assert _is_url("file.txt") is False
        assert _is_url("C:\\Windows\\file.txt") is False

    def test_is_url_with_other_schemes(self):
        """Test _is_url rejects non-HTTP schemes."""
        assert _is_url("ftp://ftp.example.com/file.txt") is False
        assert _is_url("file:///local/file.txt") is False
        assert _is_url("mailto:test@example.com") is False

    def test_is_url_with_invalid_inputs(self):
        """Test _is_url handles invalid inputs gracefully."""
        assert _is_url("") is False
        assert _is_url("not a url") is False
        assert _is_url("http://") is False

    def test_get_extension_from_mimetype_common_types(self):
        """Test _get_extension_from_mimetype for common types."""
        assert _get_extension_from_mimetype("image/jpeg") == ".jpg"
        assert _get_extension_from_mimetype("image/png") == ".png"
        assert _get_extension_from_mimetype("application/pdf") == ".pdf"
        assert _get_extension_from_mimetype("text/plain") == ".txt"
        assert _get_extension_from_mimetype("application/json") == ".json"

    def test_get_extension_from_mimetype_unknown_type(self):
        """Test _get_extension_from_mimetype returns empty for unknown types."""
        assert _get_extension_from_mimetype("application/unknown") == ""
        assert _get_extension_from_mimetype("custom/type") == ""

    def test_get_extension_from_mimetype_case_insensitive(self):
        """Test _get_extension_from_mimetype handles case-insensitive mimetypes (RFC 2045)."""
        # Uppercase variations
        assert _get_extension_from_mimetype("IMAGE/JPEG") == ".jpg"
        assert _get_extension_from_mimetype("APPLICATION/PDF") == ".pdf"
        # Mixed case variations
        assert _get_extension_from_mimetype("Image/Jpeg") == ".jpg"
        assert _get_extension_from_mimetype("Application/Pdf") == ".pdf"
        assert _get_extension_from_mimetype("Text/Plain") == ".txt"
        # Edge cases
        assert _get_extension_from_mimetype("IMAGE/png") == ".png"
        assert _get_extension_from_mimetype("video/MP4") == ".mp4"

    def test_generate_timestamped_filename(self):
        """Test _generate_timestamped_filename generates valid filenames."""
        filename = _generate_timestamped_filename(".jpg")
        assert filename.startswith("file_")
        assert filename.endswith(".jpg")
        assert len(filename) > 15  # Should have timestamp and unique ID

    def test_generate_timestamped_filename_no_extension(self):
        """Test _generate_timestamped_filename works without extension."""
        filename = _generate_timestamped_filename("")
        assert filename.startswith("file_")
        assert not filename.endswith(".")


class TestFetchFileFromUrl:
    """Test cases for _fetch_file_from_url function."""

    @patch("composio.core.models._files.safe_get")
    def test_fetch_file_from_url_success(self, mock_get):
        """Test successful file fetch from URL."""
        mock_response = MagicMock()
        mock_response.ok = True
        mock_response.status_code = 200
        mock_response.headers = {"content-type": "image/jpeg"}
        mock_response.iter_content.return_value = [b"test file content"]
        mock_response.close = MagicMock()
        mock_get.return_value = mock_response

        filename, content, mimetype = _fetch_file_from_url(
            "https://example.com/image.jpg"
        )

        assert content == b"test file content"
        assert mimetype == "image/jpeg"
        assert filename == "image.jpg"
        mock_get.assert_called_once_with(
            "https://example.com/image.jpg",
            stream=True,
            timeout=(5, 60),
        )

    @patch("composio.core.models._files.safe_get")
    def test_fetch_file_from_url_with_charset_in_content_type(self, mock_get):
        """Test that charset is stripped from content-type."""
        mock_response = MagicMock()
        mock_response.ok = True
        mock_response.status_code = 200
        mock_response.headers = {"content-type": "text/html; charset=utf-8"}
        mock_response.iter_content.return_value = [b"<html></html>"]
        mock_response.close = MagicMock()
        mock_get.return_value = mock_response

        filename, content, mimetype = _fetch_file_from_url(
            "https://example.com/page.html"
        )

        assert mimetype == "text/html"

    @patch("composio.core.models._files.safe_get")
    def test_fetch_file_from_url_generates_filename_when_missing(self, mock_get):
        """Test filename generation when URL has no filename."""
        mock_response = MagicMock()
        mock_response.ok = True
        mock_response.status_code = 200
        mock_response.headers = {"content-type": "image/png"}
        mock_response.iter_content.return_value = [b"image data"]
        mock_response.close = MagicMock()
        mock_get.return_value = mock_response

        filename, content, mimetype = _fetch_file_from_url("https://example.com/")

        assert filename.startswith("file_")
        assert filename.endswith(".png")

    @patch("composio.core.models._files.safe_get")
    def test_fetch_file_from_url_generates_filename_when_no_extension(self, mock_get):
        """Test filename generation when URL filename has no extension."""
        mock_response = MagicMock()
        mock_response.ok = True
        mock_response.status_code = 200
        mock_response.headers = {"content-type": "application/pdf"}
        mock_response.iter_content.return_value = [b"pdf data"]
        mock_response.close = MagicMock()
        mock_get.return_value = mock_response

        filename, content, mimetype = _fetch_file_from_url(
            "https://example.com/document"
        )

        assert filename.startswith("file_")
        assert filename.endswith(".pdf")

    @patch("composio.core.models._files.safe_get")
    def test_fetch_file_from_url_failure(self, mock_get):
        """Test error handling when URL fetch fails."""
        mock_response = MagicMock()
        mock_response.ok = False
        mock_response.status_code = 404
        mock_response.close = MagicMock()
        mock_get.return_value = mock_response

        with pytest.raises(ErrorUploadingFile) as exc_info:
            _fetch_file_from_url("https://example.com/notfound.jpg")

        assert "Failed to fetch file from URL" in str(exc_info.value)
        assert "404" in str(exc_info.value)

    @patch("composio.core.models._files.safe_get")
    def test_fetch_file_from_url_maps_midstream_failure(self, mock_get):
        def failing_stream(chunk_size=None):
            yield b"partial"
            raise requests.exceptions.ConnectionError("peer reset mid-stream")

        mock_response = MagicMock()
        mock_response.ok = True
        mock_response.status_code = 200
        mock_response.headers = {"content-type": "text/plain"}
        mock_response.iter_content.side_effect = failing_stream
        mock_get.return_value = mock_response

        with pytest.raises(ErrorUploadingFile, match="peer reset mid-stream"):
            _fetch_file_from_url("https://example.com/file.txt")

        mock_response.close.assert_called_once()

    @patch("composio.core.models._files.safe_get")
    def test_fetch_file_from_url_decodes_percent_encoded_filename(self, mock_get):
        """Test that percent-encoded characters in URL filenames are decoded."""
        mock_response = MagicMock()
        mock_response.ok = True
        mock_response.status_code = 200
        mock_response.headers = {"content-type": "application/pdf"}
        mock_response.iter_content.return_value = [b"document content"]
        mock_response.close = MagicMock()
        mock_get.return_value = mock_response

        # URL with percent-encoded spaces and special characters
        filename, content, mimetype = _fetch_file_from_url(
            "https://example.com/My%20Document%20%282024%29.pdf"
        )

        # Filename should be decoded
        assert filename == "My Document (2024).pdf"
        assert content == b"document content"
        assert mimetype == "application/pdf"

    @patch("composio.core.models._files.safe_get")
    def test_fetch_file_from_url_decodes_unicode_filename(self, mock_get):
        """Test that percent-encoded unicode characters in URL filenames are decoded."""
        mock_response = MagicMock()
        mock_response.ok = True
        mock_response.status_code = 200
        mock_response.headers = {"content-type": "image/jpeg"}
        mock_response.iter_content.return_value = [b"image data"]
        mock_response.close = MagicMock()
        mock_get.return_value = mock_response

        # URL with percent-encoded unicode (e.g., Japanese characters)
        filename, content, mimetype = _fetch_file_from_url(
            "https://example.com/%E3%83%95%E3%82%A1%E3%82%A4%E3%83%AB.jpg"
        )

        # Filename should be decoded to unicode
        assert filename == "ファイル.jpg"

    @patch("composio.core.models._files.safe_get")
    def test_fetch_file_from_url_handles_plus_sign_in_filename(self, mock_get):
        """Test that plus signs in URL paths are preserved (not converted to spaces)."""
        mock_response = MagicMock()
        mock_response.ok = True
        mock_response.status_code = 200
        mock_response.headers = {"content-type": "text/plain"}
        mock_response.iter_content.return_value = [b"file content"]
        mock_response.close = MagicMock()
        mock_get.return_value = mock_response

        # Plus signs in path should remain as plus signs (unquote doesn't convert + to space)
        filename, content, mimetype = _fetch_file_from_url(
            "https://example.com/file+name.txt"
        )

        assert filename == "file+name.txt"


class TestUploadBytesToS3:
    """Test cases for _upload_bytes_to_s3 function."""

    @patch("composio.core.models._files.safe_request")
    def test_upload_bytes_to_s3_success(self, mock_safe_request):
        """Test successful upload to S3."""
        mock_client = MagicMock()
        mock_s3_response = MagicMock()
        mock_s3_response.key = "s3-key-123"
        mock_s3_response.new_presigned_url = "https://s3.example.com/upload"
        mock_client.post.return_value = mock_s3_response

        mock_put_response = MagicMock()
        mock_put_response.status_code = 200
        mock_safe_request.return_value = mock_put_response

        result = _upload_bytes_to_s3(
            client=mock_client,
            filename="test.jpg",
            content=b"file content",
            mimetype="image/jpeg",
            tool="TEST_TOOL",
            toolkit="test_toolkit",
        )

        assert result == "s3-key-123"
        mock_client.post.assert_called_once()
        mock_safe_request.assert_called_once_with(
            "PUT",
            "https://s3.example.com/upload",
            data=b"file content",
            headers={"Content-Type": "image/jpeg"},
            timeout=(5, 60),
        )

    @patch("composio.core.models._files.safe_request")
    def test_upload_bytes_to_s3_failure(self, mock_safe_request):
        """Test error handling when S3 upload fails."""
        mock_client = MagicMock()
        mock_s3_response = MagicMock()
        mock_s3_response.key = "s3-key-123"
        mock_s3_response.new_presigned_url = "https://s3.example.com/upload"
        mock_client.post.return_value = mock_s3_response

        mock_put_response = MagicMock()
        mock_put_response.status_code = 500
        mock_safe_request.return_value = mock_put_response

        with pytest.raises(ErrorUploadingFile) as exc_info:
            _upload_bytes_to_s3(
                client=mock_client,
                filename="test.jpg",
                content=b"file content",
                mimetype="image/jpeg",
                tool="TEST_TOOL",
                toolkit="test_toolkit",
            )

        assert "Failed to upload to S3" in str(exc_info.value)

    @patch("composio.core.models._files.safe_request")
    def test_upload_bytes_to_s3_timeout(self, mock_safe_request):
        """Test request timeouts are reported as upload errors."""
        mock_client = MagicMock()
        mock_s3_response = MagicMock()
        mock_s3_response.key = "s3-key-123"
        mock_s3_response.new_presigned_url = "https://s3.example.com/upload?token=abc"
        mock_client.post.return_value = mock_s3_response
        # The exception text itself carries the presigned URL (incl. token), as
        # real urllib3 errors do — the SDK must not surface it in the message.
        mock_safe_request.side_effect = requests.exceptions.Timeout(
            "HTTPSConnectionPool(host='s3.example.com', port=443): "
            "Max retries exceeded with url: /upload?token=abc"
        )

        with pytest.raises(ErrorUploadingFile) as exc_info:
            _upload_bytes_to_s3(
                client=mock_client,
                filename="test.jpg",
                content=b"file content",
                mimetype="image/jpeg",
                tool="TEST_TOOL",
                toolkit="test_toolkit",
            )

        assert "Failed to upload to S3" in str(exc_info.value)
        assert "token=abc" not in str(exc_info.value)


class TestFileUploadableFromUrl:
    """Test cases for FileUploadable.from_url and from_path with URLs."""

    @patch("composio.core.models._files._upload_bytes_to_s3")
    @patch("composio.core.models._files._fetch_file_from_url")
    def test_from_url_success(self, mock_fetch, mock_upload):
        """Test successful FileUploadable creation from URL."""
        mock_fetch.return_value = ("image.jpg", b"image data", "image/jpeg")
        mock_upload.return_value = "s3-key-abc"
        mock_client = MagicMock()

        result = FileUploadable.from_url(
            client=mock_client,
            url="https://example.com/image.jpg",
            tool="TEST_TOOL",
            toolkit="test_toolkit",
        )

        assert result.name == "image.jpg"
        assert result.mimetype == "image/jpeg"
        assert result.s3key == "s3-key-abc"
        mock_fetch.assert_called_once_with("https://example.com/image.jpg")
        mock_upload.assert_called_once()

    @patch("composio.core.models._files._upload_bytes_to_s3")
    @patch("composio.core.models._files._fetch_file_from_url")
    def test_from_path_detects_url(self, mock_fetch, mock_upload):
        """Test that from_path correctly detects and handles URLs."""
        mock_fetch.return_value = ("photo.png", b"photo data", "image/png")
        mock_upload.return_value = "s3-key-xyz"
        mock_client = MagicMock()

        result = FileUploadable.from_path(
            client=mock_client,
            file="https://images.example.com/photo.png",
            tool="SEND_EMAIL",
            toolkit="gmail",
        )

        assert result.name == "photo.png"
        assert result.mimetype == "image/png"
        assert result.s3key == "s3-key-xyz"
        mock_fetch.assert_called_once_with("https://images.example.com/photo.png")

    @patch("composio.core.models._files._fetch_file_from_url")
    def test_from_url_propagates_fetch_error(self, mock_fetch):
        """Test that fetch errors are propagated correctly."""
        mock_fetch.side_effect = ErrorUploadingFile("Fetch failed")
        mock_client = MagicMock()

        with pytest.raises(ErrorUploadingFile) as exc_info:
            FileUploadable.from_url(
                client=mock_client,
                url="https://example.com/missing.jpg",
                tool="TEST_TOOL",
                toolkit="test_toolkit",
            )

        assert "Fetch failed" in str(exc_info.value)

    def test_before_file_upload_hook_receives_source_url(self):
        """from_path emits ``source="url"`` to the hook for http(s) inputs.

        We abort from the hook to avoid the downstream network path; the
        pre-abort capture is what we're asserting on.
        """
        from composio.exceptions import FileUploadAbortedError

        mock_client = MagicMock()
        seen = {}

        def hook(ctx):
            seen.update(ctx)
            return False

        with pytest.raises(FileUploadAbortedError):
            FileUploadable.from_path(
                client=mock_client,
                file="https://example.com/photo.png",
                tool="SEND_EMAIL",
                toolkit="gmail",
                before_file_upload=hook,
            )

        assert seen == {
            "path": "https://example.com/photo.png",
            "source": "url",
            "tool": "SEND_EMAIL",
            "toolkit": "gmail",
        }

    def test_before_file_upload_hook_receives_source_path(self, tmp_path):
        """from_path emits ``source="path"`` to the hook for local inputs."""
        from composio.exceptions import FileUploadAbortedError

        f = tmp_path / "doc.txt"
        f.write_text("hello")
        mock_client = MagicMock()
        seen = {}

        def hook(ctx):
            seen.update(ctx)
            return False

        with pytest.raises(FileUploadAbortedError):
            FileUploadable.from_path(
                client=mock_client,
                file=str(f),
                tool="MY_TOOL",
                toolkit="my_toolkit",
                before_file_upload=hook,
            )

        assert seen == {
            "path": str(f),
            "source": "path",
            "tool": "MY_TOOL",
            "toolkit": "my_toolkit",
        }

    def test_url_hook_returning_local_path_routes_through_path_branch(self, tmp_path):
        """A hook that rewrites a URL into a local path must NOT be fed to
        ``from_url``. It has to route back into the local-file branch so the
        allowlist / denylist / existence checks all run."""
        from composio.exceptions import SDKFileNotFoundError

        # A path that's syntactically a path but does not exist — if routing
        # is correct, we'll get SDKFileNotFoundError from the local branch.
        # If the bug is still there, we'd hit `from_url` and the URL fetch
        # would explode (or worse, succeed) instead.
        rewritten = str(tmp_path / "does-not-exist.txt")

        def hook(ctx):
            assert ctx["source"] == "url"
            return rewritten

        mock_client = MagicMock()

        with pytest.raises(SDKFileNotFoundError):
            FileUploadable.from_path(
                client=mock_client,
                file="https://example.com/photo.png",
                tool="T",
                toolkit="tk",
                before_file_upload=hook,
            )

    @patch("composio.core.models._files._fetch_file_from_url")
    @patch("composio.core.models._files._upload_bytes_to_s3")
    def test_path_hook_returning_url_routes_through_url_branch(
        self, mock_upload, mock_fetch, tmp_path
    ):
        """Inverse of the above: a hook on a local path that returns a URL
        must route through ``from_url``, not stat the URL string as a file."""
        f = tmp_path / "local.txt"
        f.write_text("hi")

        mock_fetch.return_value = ("photo.png", b"x", "image/png")
        mock_upload.return_value = "s3-key"

        def hook(ctx):
            assert ctx["source"] == "path"
            return "https://example.com/photo.png"

        mock_client = MagicMock()
        result = FileUploadable.from_path(
            client=mock_client,
            file=str(f),
            tool="T",
            toolkit="tk",
            before_file_upload=hook,
        )

        # If routing worked, the URL fetch path was taken.
        mock_fetch.assert_called_once_with("https://example.com/photo.png")
        assert result.s3key == "s3-key"


class TestFileHelperWithUrls:
    """Test cases for FileHelper handling URLs in file uploads."""

    @patch("composio.core.models._files.FileUploadable.from_path")
    def test_substitute_file_uploads_with_url(
        self, mock_from_path, file_helper, mock_tool
    ):
        """Test that URLs are correctly processed in substitute_file_uploads."""
        mock_tool.input_parameters = {
            "type": "object",
            "properties": {
                "attachment": {"type": "string", "file_uploadable": True},
            },
        }

        mock_uploadable = MagicMock()
        mock_uploadable.model_dump.return_value = {
            "name": "image.jpg",
            "mimetype": "image/jpeg",
            "s3key": "s3-key-123",
        }
        mock_from_path.return_value = mock_uploadable

        result = file_helper._substitute_file_uploads_recursively(
            tool=mock_tool,
            schema=mock_tool.input_parameters,
            request={"attachment": "https://example.com/image.jpg"},
        )

        assert result["attachment"] == {
            "name": "image.jpg",
            "mimetype": "image/jpeg",
            "s3key": "s3-key-123",
        }
        mock_from_path.assert_called_once()

    @patch("composio.core.models._files.FileUploadable.from_path")
    def test_substitute_file_uploads_with_url_in_anyof(
        self, mock_from_path, file_helper, mock_tool
    ):
        """Test URL handling in anyOf schema variants."""
        mock_tool.input_parameters = {
            "type": "object",
            "properties": {
                "file": {
                    "anyOf": [
                        {"type": "string", "file_uploadable": True},
                        {"type": "null"},
                    ]
                },
            },
        }

        mock_uploadable = MagicMock()
        mock_uploadable.model_dump.return_value = {
            "name": "doc.pdf",
            "mimetype": "application/pdf",
            "s3key": "s3-key-456",
        }
        mock_from_path.return_value = mock_uploadable

        result = file_helper._substitute_file_uploads_recursively(
            tool=mock_tool,
            schema=mock_tool.input_parameters,
            request={"file": "https://docs.example.com/doc.pdf"},
        )

        assert result["file"] == {
            "name": "doc.pdf",
            "mimetype": "application/pdf",
            "s3key": "s3-key-456",
        }

    @patch("composio.core.models._files.FileUploadable.from_path")
    def test_substitute_file_uploads_array_with_urls(
        self, mock_from_path, file_helper, mock_tool
    ):
        """Test URL handling in arrays with file_uploadable items."""
        mock_tool.input_parameters = {
            "type": "object",
            "properties": {
                "attachments": {
                    "type": "array",
                    "items": {"type": "string", "file_uploadable": True},
                },
            },
        }

        mock_uploadable1 = MagicMock()
        mock_uploadable1.model_dump.return_value = {
            "name": "file1.jpg",
            "mimetype": "image/jpeg",
            "s3key": "key1",
        }
        mock_uploadable2 = MagicMock()
        mock_uploadable2.model_dump.return_value = {
            "name": "file2.png",
            "mimetype": "image/png",
            "s3key": "key2",
        }
        mock_from_path.side_effect = [mock_uploadable1, mock_uploadable2]

        result = file_helper._substitute_file_uploads_recursively(
            tool=mock_tool,
            schema=mock_tool.input_parameters,
            request={
                "attachments": [
                    "https://example.com/file1.jpg",
                    "https://example.com/file2.png",
                ]
            },
        )

        assert len(result["attachments"]) == 2
        assert result["attachments"][0]["s3key"] == "key1"
        assert result["attachments"][1]["s3key"] == "key2"


class TestTruncateFilename:
    """Test cases for _truncate_filename function.

    Long filenames are common with public bucket URLs containing hashes or UUIDs.
    These can cause issues, so they are replaced with timestamped filenames.
    """

    def test_truncate_filename_short_unchanged(self):
        """Short filenames should not be modified."""
        assert _truncate_filename("document.pdf") == "document.pdf"
        assert _truncate_filename("image.jpg") == "image.jpg"
        assert _truncate_filename("file.txt") == "file.txt"

    def test_truncate_filename_at_limit_unchanged(self):
        """Filenames exactly at the limit should not be modified."""
        # Create a filename exactly at the limit (100 chars by default)
        name = "a" * 95 + ".pdf"  # 95 + 4 = 99 chars
        assert _truncate_filename(name) == name

        name_at_limit = "a" * 96 + ".pdf"  # 96 + 4 = 100 chars
        assert _truncate_filename(name_at_limit) == name_at_limit

    def test_truncate_filename_long_generates_timestamped(self):
        """Long filenames should be replaced with timestamped filename."""
        long_name = "a" * 150 + ".pdf"
        result = _truncate_filename(long_name)

        assert len(result) < _MAX_FILENAME_LENGTH
        assert result.startswith("file_")
        assert result.endswith(".pdf")

    def test_truncate_filename_preserves_extension(self):
        """Extension should be preserved when generating timestamped filename."""
        test_cases = [
            ("very_long_" * 20 + ".docx", ".docx"),
            ("hash_" * 30 + ".jpg", ".jpg"),
            ("uuid_" * 25 + ".png", ".png"),
            ("data_" * 40 + ".json", ".json"),
        ]

        for long_name, expected_ext in test_cases:
            result = _truncate_filename(long_name)
            assert result.endswith(expected_ext), (
                f"Expected {result} to end with {expected_ext}"
            )

    def test_truncate_filename_long_hash_url(self):
        """Hash-based filenames from URLs should be truncated."""
        # Typical long hash filename from public bucket URLs (needs to be >100 chars)
        hash_filename = "8f14e45fceea167a5a36dedd4bea2543_5d41402abc4b2a76b9719d911017c592_extra_hash_data_to_exceed_limit_download.jpg"
        assert len(hash_filename) > _MAX_FILENAME_LENGTH  # Verify test setup

        result = _truncate_filename(hash_filename)

        assert len(result) <= _MAX_FILENAME_LENGTH
        assert result.startswith("file_")
        assert result.endswith(".jpg")

    def test_truncate_filename_multiple_hashes(self):
        """Multiple concatenated hashes should be truncated."""
        # 32 char hash * 5 = 160 chars + extension
        multi_hash = "8f14e45fceea167a5a36dedd4bea2543" * 5 + ".pdf"
        result = _truncate_filename(multi_hash)

        assert len(result) <= _MAX_FILENAME_LENGTH
        assert result.endswith(".pdf")

    def test_truncate_filename_no_extension(self):
        """Long filenames without extension should still be truncated."""
        long_name = "a" * 150
        result = _truncate_filename(long_name)

        assert len(result) <= _MAX_FILENAME_LENGTH
        assert result.startswith("file_")
        # Should not end with a dot
        assert not result.endswith(".")

    def test_truncate_filename_custom_max_length(self):
        """Custom max_length parameter should be respected."""
        name = "a" * 60 + ".txt"  # 64 chars total

        # With default limit (100), should be unchanged
        assert _truncate_filename(name) == name

        # With custom limit (50), should be truncated
        result = _truncate_filename(name, max_length=50)
        assert len(result) <= 50
        assert result.startswith("file_")
        assert result.endswith(".txt")

    def test_truncate_filename_edge_case_one_over_limit(self):
        """Filename one character over limit should be truncated."""
        # Create filename exactly one char over the limit
        name = "a" * 97 + ".pdf"  # 97 + 4 = 101 chars (one over default 100)
        result = _truncate_filename(name)

        assert len(result) < _MAX_FILENAME_LENGTH
        assert result.startswith("file_")
        assert result.endswith(".pdf")

    def test_truncate_filename_long_extension(self):
        """Long extensions should be preserved when truncating."""
        long_name = "file_" * 30 + ".dockerfile"
        result = _truncate_filename(long_name)

        assert result.endswith(".dockerfile")
        assert result.startswith("file_")

    def test_truncate_filename_multiple_dots_preserves_last_extension(self):
        """Filename with multiple dots should preserve only the last extension."""
        long_name = "archive" * 20 + ".backup.tar.gz"
        result = _truncate_filename(long_name)

        # rsplit(".", 1) takes the part after the last dot
        assert result.endswith(".gz")


class TestFetchFileFromUrlWithTruncation:
    """Test cases for _fetch_file_from_url with filename truncation."""

    @patch("composio.core.models._files.safe_get")
    def test_fetch_truncates_long_filename(self, mock_get):
        """Long filenames from URLs should be truncated."""
        mock_response = MagicMock()
        mock_response.ok = True
        mock_response.status_code = 200
        mock_response.headers = {"content-type": "application/pdf"}
        mock_response.iter_content.return_value = [b"test content"]
        mock_response.close = MagicMock()
        mock_get.return_value = mock_response

        # Create a very long filename (hash-based, common in public buckets)
        long_filename = "8f14e45fceea167a5a36dedd4bea2543" * 5 + ".pdf"

        filename, content, mimetype = _fetch_file_from_url(
            f"https://bucket.example.com/files/{long_filename}"
        )

        assert len(filename) <= _MAX_FILENAME_LENGTH
        assert filename.startswith("file_")
        assert filename.endswith(".pdf")
        assert content == b"test content"

    @patch("composio.core.models._files.safe_get")
    def test_fetch_preserves_short_filename(self, mock_get):
        """Short filenames should be preserved unchanged."""
        mock_response = MagicMock()
        mock_response.ok = True
        mock_response.status_code = 200
        mock_response.headers = {"content-type": "image/jpeg"}
        mock_response.iter_content.return_value = [b"image data"]
        mock_response.close = MagicMock()
        mock_get.return_value = mock_response

        filename, content, mimetype = _fetch_file_from_url(
            "https://example.com/photo.jpg"
        )

        assert filename == "photo.jpg"

    @patch("composio.core.models._files.safe_get")
    def test_fetch_truncates_after_adding_extension(self, mock_get):
        """Truncation should happen after extension is appended."""
        mock_response = MagicMock()
        mock_response.ok = True
        mock_response.status_code = 200
        mock_response.headers = {"content-type": "application/pdf"}
        mock_response.iter_content.return_value = [b"pdf content"]
        mock_response.close = MagicMock()
        mock_get.return_value = mock_response

        # URL without extension - extension will be added from mimetype
        long_name_no_ext = "8f14e45fceea167a5a36dedd4bea2543" * 4
        filename, content, mimetype = _fetch_file_from_url(
            f"https://bucket.example.com/{long_name_no_ext}"
        )

        # Should be truncated and have .pdf extension
        assert len(filename) <= _MAX_FILENAME_LENGTH
        assert filename.endswith(".pdf")

    @patch("composio.core.models._files.safe_get")
    def test_fetch_generated_filename_not_truncated(self, mock_get):
        """Generated timestamped filenames (when URL has no filename) should be short enough."""
        mock_response = MagicMock()
        mock_response.ok = True
        mock_response.status_code = 200
        mock_response.headers = {"content-type": "image/png"}
        mock_response.iter_content.return_value = [b"data"]
        mock_response.close = MagicMock()
        mock_get.return_value = mock_response

        # URL with no filename - will generate a timestamped one
        filename, content, mimetype = _fetch_file_from_url("https://example.com/")

        # Generated filename should be naturally short
        assert filename.startswith("file_")
        assert filename.endswith(".png")
        assert len(filename) < 50  # Timestamped names are short

    @patch("composio.core.models._files.safe_get")
    def test_fetch_long_real_world_url(self, mock_get):
        """Long real-world URLs should be handled correctly."""
        mock_response = MagicMock()
        mock_response.ok = True
        mock_response.status_code = 200
        mock_response.headers = {"content-type": "application/pdf"}
        mock_response.iter_content.return_value = [b"test content"]
        mock_response.close = MagicMock()
        mock_get.return_value = mock_response

        # Real-world long URL example (no extension, relies on mimetype)
        long_url = (
            "https://encrypted-tbn0.gstatic.com/images?q="
            "tbn:ANd9GcQq0powzLhfi1bakZ0eNSIA_aJ_5UlPsCte1g&s"
        )
        filename, content, mimetype = _fetch_file_from_url(long_url)

        assert len(filename) <= _MAX_FILENAME_LENGTH
        assert filename.startswith("file_")
        assert filename.endswith(".pdf")
        assert content == b"test content"


class TestResponseSizeLimit:
    """Test response size limiting."""

    @patch("composio.core.models._files.safe_get")
    def test_rejects_oversized_content_length(self, mock_get):
        """Files with Content-Length > max_size should be rejected early."""
        mock_response = MagicMock()
        mock_response.ok = True
        mock_response.status_code = 200
        mock_response.headers = {"Content-Length": "200000000"}  # 200MB
        mock_response.close = MagicMock()
        mock_get.return_value = mock_response

        with pytest.raises(ResponseTooLargeError):
            _fetch_file_from_url(
                "https://example.com/large.zip", max_size=100 * 1024 * 1024
            )

    @patch("composio.core.models._files.safe_get")
    def test_rejects_oversized_during_streaming(self, mock_get):
        """Files that exceed max_size during download should be rejected."""
        mock_response = MagicMock()
        mock_response.ok = True
        mock_response.status_code = 200
        mock_response.headers = {}  # No Content-Length
        # Return 20MB of data in chunks
        mock_response.iter_content.return_value = [
            b"x" * 1024 * 1024 for _ in range(20)
        ]
        mock_response.close = MagicMock()
        mock_get.return_value = mock_response

        with pytest.raises(ResponseTooLargeError):
            _fetch_file_from_url(
                "https://example.com/large.zip", max_size=10 * 1024 * 1024
            )

    @patch("composio.core.models._files.safe_get")
    def test_accepts_file_within_limit(self, mock_get):
        """Files within size limit should be accepted."""
        mock_response = MagicMock()
        mock_response.ok = True
        mock_response.status_code = 200
        mock_response.headers = {"content-type": "image/jpeg", "Content-Length": "1000"}
        mock_response.iter_content.return_value = [b"x" * 1000]
        mock_response.close = MagicMock()
        mock_get.return_value = mock_response

        filename, content, mimetype = _fetch_file_from_url(
            "https://example.com/image.jpg", max_size=10 * 1024 * 1024
        )

        assert len(content) == 1000
        assert mimetype == "image/jpeg"


class TestDownloadSizeLimit:
    """``FileDownloadable.download`` streams an untrusted body to disk."""

    @staticmethod
    def _downloadable() -> FileDownloadable:
        return FileDownloadable(
            name="report.bin",
            mimetype="application/octet-stream",
            s3url="https://example.com/report.bin",
        )

    @patch("composio.core.models._files.safe_get")
    def test_download_rejects_oversized_content_length(self, mock_get, tmp_path):
        """A self-declared oversized body is rejected before any bytes are read."""
        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.headers = {"Content-Length": "200000000"}
        mock_response.close = MagicMock()
        mock_get.return_value = mock_response

        with pytest.raises(ResponseTooLargeError):
            self._downloadable().download(outdir=tmp_path, root=tmp_path, max_size=1024)

        mock_response.iter_content.assert_not_called()

    @patch("composio.core.models._files.safe_get")
    def test_download_rejects_oversized_during_streaming(self, mock_get, tmp_path):
        """A dishonest (here, absent) Content-Length cannot bypass the cap."""
        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.headers = {}
        mock_response.iter_content.return_value = [b"x" * 512 for _ in range(4)]
        mock_response.close = MagicMock()
        mock_get.return_value = mock_response

        with pytest.raises(ResponseTooLargeError):
            self._downloadable().download(outdir=tmp_path, root=tmp_path, max_size=1024)

    @patch("composio.core.models._files.safe_get")
    def test_download_removes_partial_file_on_failure(self, mock_get, tmp_path):
        """A truncated download must not be left behind as if it succeeded."""
        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.headers = {}
        mock_response.iter_content.return_value = [b"x" * 512 for _ in range(4)]
        mock_response.close = MagicMock()
        mock_get.return_value = mock_response

        with pytest.raises(ResponseTooLargeError):
            self._downloadable().download(outdir=tmp_path, root=tmp_path, max_size=1024)

        assert list(tmp_path.iterdir()) == []

    @patch("composio.core.models._files.safe_get")
    def test_download_accepts_file_within_limit(self, mock_get, tmp_path):
        """A body under the cap is written through unchanged."""
        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.headers = {}
        mock_response.iter_content.return_value = [b"x" * 256, b"y" * 256]
        mock_response.close = MagicMock()
        mock_get.return_value = mock_response

        outfile = self._downloadable().download(
            outdir=tmp_path, root=tmp_path, max_size=1024
        )

        assert outfile.exists()
        assert outfile.read_bytes() == b"x" * 256 + b"y" * 256

    @patch("composio.core.models._files.safe_get")
    def test_download_wraps_stream_failure_and_removes_partial_file(
        self, mock_get, tmp_path
    ):
        """A transport failure mid-stream keeps the documented error contract."""

        def failing_stream(chunk_size=None):
            yield b"x" * 256
            raise requests.exceptions.ConnectionError("connection reset")

        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.headers = {}
        mock_response.iter_content.side_effect = failing_stream
        mock_response.close = MagicMock()
        mock_get.return_value = mock_response

        with pytest.raises(ErrorDownloadingFile):
            self._downloadable().download(outdir=tmp_path, root=tmp_path, max_size=1024)

        assert list(tmp_path.iterdir()) == []

    @patch("composio.core.models._files.safe_get")
    def test_download_wraps_write_failure_and_removes_partial_file(
        self, mock_get, tmp_path
    ):
        """A disk failure while writing is an `ErrorDownloadingFile`, not a raw OSError."""

        def failing_stream(chunk_size=None):
            yield b"x" * 256
            raise OSError(28, "No space left on device")

        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.headers = {}
        mock_response.iter_content.side_effect = failing_stream
        mock_response.close = MagicMock()
        mock_get.return_value = mock_response

        with pytest.raises(ErrorDownloadingFile):
            self._downloadable().download(outdir=tmp_path, root=tmp_path, max_size=1024)

        assert list(tmp_path.iterdir()) == []


class TestRedirectHandling:
    """Test redirect handling (redirects should be rejected)."""

    @patch("composio.core.models._files.safe_get")
    def test_rejects_redirect_302(self, mock_get):
        """302 redirects should be rejected with clear error message."""
        mock_response = MagicMock()
        mock_response.status_code = 302
        mock_response.headers = {"Location": "https://example.com/final.jpg"}
        mock_response.close = MagicMock()
        mock_get.return_value = mock_response

        with pytest.raises(ErrorUploadingFile, match="redirect"):
            _fetch_file_from_url("https://example.com/redirect")

    @patch("composio.core.models._files.safe_get")
    def test_rejects_redirect_301(self, mock_get):
        """301 redirects should be rejected."""
        mock_response = MagicMock()
        mock_response.status_code = 301
        mock_response.headers = {"Location": "https://example.com/"}
        mock_response.close = MagicMock()
        mock_get.return_value = mock_response

        with pytest.raises(ErrorUploadingFile, match="redirect"):
            _fetch_file_from_url("https://example.com/test")

    @patch("composio.core.models._files.safe_get")
    def test_rejects_redirect_307(self, mock_get):
        """307 redirects should be rejected."""
        mock_response = MagicMock()
        mock_response.status_code = 307
        mock_response.headers = {"Location": "https://example.com/"}
        mock_response.close = MagicMock()
        mock_get.return_value = mock_response

        with pytest.raises(ErrorUploadingFile, match="redirect"):
            _fetch_file_from_url("https://example.com/test")

    @patch("composio.core.models._files.safe_get")
    def test_rejects_redirect_308(self, mock_get):
        """308 redirects should be rejected."""
        mock_response = MagicMock()
        mock_response.status_code = 308
        mock_response.headers = {"Location": "https://example.com/"}
        mock_response.close = MagicMock()
        mock_get.return_value = mock_response

        with pytest.raises(ErrorUploadingFile, match="redirect"):
            _fetch_file_from_url("https://example.com/test")


class TestS3UploadErrorHandling:
    """Test S3 upload error handling."""

    @patch("composio.core.models._files.safe_request")
    def test_403_is_treated_as_error(self, mock_safe_request):
        """HTTP 403 should be treated as upload failure."""
        mock_client = MagicMock()
        mock_s3_response = MagicMock()
        mock_s3_response.key = "s3-key"
        mock_s3_response.new_presigned_url = "https://s3.example.com/upload"
        mock_client.post.return_value = mock_s3_response

        mock_put_response = MagicMock()
        mock_put_response.status_code = 403
        mock_safe_request.return_value = mock_put_response

        with pytest.raises(ErrorUploadingFile, match="403"):
            _upload_bytes_to_s3(
                client=mock_client,
                filename="test.jpg",
                content=b"data",
                mimetype="image/jpeg",
                tool="TEST",
                toolkit="test",
            )

    @patch("composio.core.models._files.safe_request")
    def test_200_is_success(self, mock_safe_request):
        """HTTP 200 should be treated as success."""
        mock_client = MagicMock()
        mock_s3_response = MagicMock()
        mock_s3_response.key = "s3-key"
        mock_s3_response.new_presigned_url = "https://s3.example.com/upload"
        mock_client.post.return_value = mock_s3_response

        mock_put_response = MagicMock()
        mock_put_response.status_code = 200
        mock_safe_request.return_value = mock_put_response

        result = _upload_bytes_to_s3(
            client=mock_client,
            filename="test.jpg",
            content=b"data",
            mimetype="image/jpeg",
            tool="TEST",
            toolkit="test",
        )
        assert result == "s3-key"

    @patch("composio.core.models._files.safe_request")
    def test_500_is_treated_as_error(self, mock_safe_request):
        """HTTP 500 should be treated as upload failure."""
        mock_client = MagicMock()
        mock_s3_response = MagicMock()
        mock_s3_response.key = "s3-key"
        mock_s3_response.new_presigned_url = "https://s3.example.com/upload"
        mock_client.post.return_value = mock_s3_response

        mock_put_response = MagicMock()
        mock_put_response.status_code = 500
        mock_safe_request.return_value = mock_put_response

        with pytest.raises(ErrorUploadingFile, match="500"):
            _upload_bytes_to_s3(
                client=mock_client,
                filename="test.jpg",
                content=b"data",
                mimetype="image/jpeg",
                tool="TEST",
                toolkit="test",
            )


class TestUrlSanitization:
    """Test URL sanitization for logging."""

    def test_sanitizes_query_params(self):
        """Query parameters should be redacted in logs."""
        url = "https://example.com/file?token=secret123&key=abc"
        sanitized = _sanitize_url_for_logging(url)
        assert "secret123" not in sanitized
        assert "abc" not in sanitized
        assert "[REDACTED]" in sanitized

    def test_preserves_url_without_query(self):
        """URLs without query params should be unchanged."""
        url = "https://example.com/path/to/file.jpg"
        sanitized = _sanitize_url_for_logging(url)
        assert sanitized == url

    def test_preserves_path(self):
        """Path should be preserved when redacting query params."""
        url = "https://example.com/path/to/file.jpg?token=secret"
        sanitized = _sanitize_url_for_logging(url)
        assert "/path/to/file.jpg" in sanitized
        assert "example.com" in sanitized

    def test_handles_empty_query(self):
        """URLs with empty query string should not have [REDACTED]."""
        url = "https://example.com/file.jpg"
        sanitized = _sanitize_url_for_logging(url)
        assert "[REDACTED]" not in sanitized


class TestFileDownloadablePathTraversal:
    """SEC-316: server-controlled `name` must not escape the output dir."""

    @pytest.fixture(autouse=True)
    def _allow_fetch_target(self):
        yield

    def _mock_response(self, content: bytes = b"data") -> MagicMock:
        response = MagicMock()
        response.status_code = 200
        response.iter_content = lambda chunk_size: [content]
        response.close = MagicMock()
        return response

    def test_relative_traversal_is_neutralized(self, tmp_path):
        outdir = tmp_path / "safe"
        # outdir does not need to exist yet — download() creates it.
        f = FileDownloadable(
            name="../../../PWNED.sh",
            mimetype="application/octet-stream",
            s3url="https://example.com/file",
        )
        with patch(
            "composio.core.models._files.safe_get",
            return_value=self._mock_response(b"#!/bin/sh\n"),
        ):
            written = f.download(outdir, root=outdir)

        # Traversal sequence collapsed to basename and stayed inside outdir.
        assert written == outdir / "PWNED.sh"
        assert written.resolve().is_relative_to(outdir.resolve())
        assert not (tmp_path / "PWNED.sh").exists()
        assert written.read_bytes() == b"#!/bin/sh\n"

    def test_absolute_path_is_neutralized(self, tmp_path):
        outdir = tmp_path / "safe"
        f = FileDownloadable(
            name="/etc/passwd",
            mimetype="application/octet-stream",
            s3url="https://example.com/file",
        )
        with patch(
            "composio.core.models._files.safe_get",
            return_value=self._mock_response(b"x"),
        ):
            written = f.download(outdir, root=outdir)

        # `Path('/etc/passwd').name == 'passwd'` — absolute path is stripped.
        assert written == outdir / "passwd"
        assert written.resolve().is_relative_to(outdir.resolve())

    def test_dotdot_only_name_is_rejected(self, tmp_path):
        """`Path('..').name == '..'`. Basename strip alone wouldn't catch it,
        but the second-line `is_relative_to()` check does."""
        from composio.exceptions import ErrorDownloadingFile

        outdir = tmp_path / "safe"
        f = FileDownloadable(
            name="..",
            mimetype="application/octet-stream",
            s3url="https://example.com/file",
        )
        with patch(
            "composio.core.models._files.safe_get",
            return_value=self._mock_response(),
        ):
            with pytest.raises(ErrorDownloadingFile, match="Path traversal detected"):
                f.download(outdir, root=outdir)
        # No file was written under the parent.
        assert not (tmp_path / "x").exists()

    def test_safe_filename_passes_through(self, tmp_path):
        outdir = tmp_path / "safe"
        f = FileDownloadable(
            name="report.pdf",
            mimetype="application/pdf",
            s3url="https://example.com/file",
        )
        with patch(
            "composio.core.models._files.safe_get",
            return_value=self._mock_response(b"%PDF-1.4"),
        ):
            written = f.download(outdir, root=outdir)

        assert written == outdir / "report.pdf"
        assert written.read_bytes() == b"%PDF-1.4"

    def test_download_uses_timeout(self, tmp_path):
        outdir = tmp_path / "safe"
        f = FileDownloadable(
            name="report.pdf",
            mimetype="application/pdf",
            s3url="https://example.com/file",
        )
        with patch(
            "composio.core.models._files.safe_get",
            return_value=self._mock_response(b"%PDF-1.4"),
        ) as mock_get:
            f.download(outdir, root=outdir)

        mock_get.assert_called_once_with(
            "https://example.com/file",
            stream=True,
            timeout=(5, 60),
        )

    def test_download_timeout_raises_error(self, tmp_path):
        outdir = tmp_path / "safe"
        f = FileDownloadable(
            name="report.pdf",
            mimetype="application/pdf",
            s3url="https://example.com/file?token=abc",
        )
        with patch(
            "composio.core.models._files.safe_get",
            side_effect=requests.exceptions.Timeout(
                "Max retries exceeded with url: /file?token=abc"
            ),
        ):
            with pytest.raises(ErrorDownloadingFile) as exc_info:
                f.download(outdir, root=outdir)

        assert "Error downloading file" in str(exc_info.value)
        assert "token=abc" not in str(exc_info.value)

    def test_download_stream_timeout_raises_error(self, tmp_path):
        outdir = tmp_path / "safe"
        response = self._mock_response()
        response.iter_content = MagicMock(
            side_effect=requests.exceptions.ReadTimeout(
                "Max retries exceeded with url: /file?token=abc"
            )
        )
        f = FileDownloadable(
            name="report.pdf",
            mimetype="application/pdf",
            s3url="https://example.com/file?token=abc",
        )
        with patch(
            "composio.core.models._files.safe_get",
            return_value=response,
        ):
            with pytest.raises(ErrorDownloadingFile) as exc_info:
                f.download(outdir, root=outdir)

        assert "Error downloading file" in str(exc_info.value)
        assert "token=abc" not in str(exc_info.value)
        response.close.assert_called_once()

    def test_download_non_200_redacts_url_and_closes(self, tmp_path):
        outdir = tmp_path / "safe"
        response = self._mock_response()
        response.status_code = 403
        f = FileDownloadable(
            name="report.pdf",
            mimetype="application/pdf",
            s3url="https://example.com/file?token=abc",
        )
        with patch(
            "composio.core.models._files.safe_get",
            return_value=response,
        ):
            with pytest.raises(ErrorDownloadingFile) as exc_info:
                f.download(outdir, root=outdir)

        assert "token=abc" not in str(exc_info.value)
        response.close.assert_called_once()

    def test_basename_collapse_through_subdir_is_rejected(self, tmp_path):
        """`Path('foo/..').name == '..'` — the basename strip of a name that
        traverses *through* a subdir collapses to `..`, then the
        `is_relative_to()` check rejects it. Explicit coverage of the
        residual-`..` path through basename normalization (the plain `..`
        test covers the no-subdir case)."""
        from composio.exceptions import ErrorDownloadingFile

        outdir = tmp_path / "safe"
        f = FileDownloadable(
            name="foo/..",
            mimetype="application/octet-stream",
            s3url="https://example.com/file",
        )
        with patch(
            "composio.core.models._files.safe_get",
            return_value=self._mock_response(),
        ):
            with pytest.raises(ErrorDownloadingFile, match="Path traversal detected"):
                f.download(outdir, root=outdir)
        # SEC-316 P3.1: check runs before mkdir, so outdir is not created
        # as a side effect of a rejected payload.
        assert not outdir.exists()

    @pytest.mark.parametrize("name", ["", "."])
    def test_name_without_usable_basename_is_rejected(self, name, tmp_path):
        """`Path('').name` and `Path('.').name` are both `''`, so `outdir / ''`
        used to resolve to `outdir` itself, pass the containment check (a path
        is relative to itself), and only fail at write time with a raw
        `IsADirectoryError` after the directory had been created.

        These are now refused up front. Documents the fail-closed behavior so a
        future change cannot silently weaken it without breaking this test."""
        from composio.exceptions import ErrorDownloadingFile

        outdir = tmp_path / "safe"
        f = FileDownloadable(
            name=name,
            mimetype="application/octet-stream",
            s3url="https://example.com/file",
        )
        with patch(
            "composio.core.models._files.safe_get",
            return_value=self._mock_response(b"x"),
        ):
            with pytest.raises(ErrorDownloadingFile, match="no usable basename"):
                f.download(outdir, root=outdir)

        # Rejected before `mkdir`, so nothing was created on disk at all.
        assert not outdir.exists()

    @pytest.mark.parametrize(
        "name,reason",
        [
            ("NUL", "reserved device name"),
            ("nul.txt", "reserved device name"),
            ("NUL.tar.gz", "reserved device name"),
            ("COM1", "reserved device name"),
            ("report.txt:payload", "reserved by Windows"),
            ("a\x00b", "NUL byte"),
            ("😀" * 128, "longer than"),
            ("x" * 300, "longer than"),
        ],
    )
    def test_unsafe_filenames_are_rejected(self, name, reason, tmp_path):
        """`self.name` is untrusted by the same rule as the slugs. Without these
        checks a NUL byte escapes as a raw ValueError from `resolve()`, an
        over-long name as a raw OSError mid-write, and `NUL` opens the Windows
        null device — silently discarding the payload while returning a path."""
        from composio.exceptions import ErrorDownloadingFile

        outdir = tmp_path / "safe"
        f = FileDownloadable(
            name=name,
            mimetype="application/octet-stream",
            s3url="https://example.com/file",
        )
        with patch(
            "composio.core.models._files.safe_get",
            return_value=self._mock_response(b"x"),
        ):
            with pytest.raises(ErrorDownloadingFile, match=reason):
                f.download(outdir, root=outdir)
        assert not outdir.exists()


class TestDownloadDirSlugTraversal:
    """Server-controlled tool/toolkit slugs must not relocate the download
    directory.

    The directory is built from API response fields, so containment has to be
    checked against the locally configured root rather than against the built
    directory itself — the latter is a reference those fields can move. These
    tests pin the anchor to the configured root.
    """

    TRAVERSALS = [
        "../../../../../etc/escaped",
        "..",
        ".",
        "",
        "/etc",
        "..\\..\\evil",
        "a/b",
        "CON",
        "x" * 200,
        "a\x00b",
    ]

    def _tool(self, tool_slug="GMAIL_GET_ATTACHMENT", toolkit_slug="GMAIL"):
        tool = MagicMock()
        tool.slug = tool_slug
        tool.toolkit.slug = toolkit_slug
        tool.output_parameters = {
            "type": "object",
            "properties": {"attachment": {"file_downloadable": True}},
        }
        return tool

    def _response(self):
        return {
            "attachment": {
                "name": "composio",
                "mimetype": "text/plain",
                "s3url": "https://example.com/file",
            }
        }

    def _mock_get(self, content: bytes = b"payload"):
        response = MagicMock()
        response.status_code = 200
        response.iter_content = lambda chunk_size: [content]
        response.close = MagicMock()
        return patch("composio.core.models._files.safe_get", return_value=response)

    @pytest.mark.parametrize("slug", TRAVERSALS)
    def test_hostile_tool_slug_is_rejected(self, slug, tmp_path):
        outdir = tmp_path / "files"
        outdir.mkdir()
        helper = FileHelper(client=None, outdir=str(outdir))
        with self._mock_get():
            with pytest.raises(UnsafePathComponentError):
                helper.substitute_file_downloads(
                    tool=self._tool(tool_slug=slug), response=self._response()
                )
        # The write is refused before any directory is created.
        assert list(outdir.iterdir()) == []

    @pytest.mark.parametrize("slug", TRAVERSALS)
    def test_hostile_toolkit_slug_is_rejected(self, slug, tmp_path):
        outdir = tmp_path / "files"
        outdir.mkdir()
        helper = FileHelper(client=None, outdir=str(outdir))
        with self._mock_get():
            with pytest.raises(UnsafePathComponentError):
                helper.substitute_file_downloads(
                    tool=self._tool(toolkit_slug=slug), response=self._response()
                )
        assert list(outdir.iterdir()) == []

    def test_traversal_does_not_escape_the_configured_root(self, tmp_path):
        """A deep `../` chain in `tool.slug` must not land the payload in a
        sibling of the configured directory."""
        outdir = tmp_path / "home" / "app" / ".composio" / "files"
        outdir.mkdir(parents=True)
        helper = FileHelper(client=None, outdir=str(outdir))
        depth = len(outdir.resolve().parts) - len(tmp_path.resolve().parts) + 1
        traversal = "/".join([".."] * depth) + "/etc/escaped"

        with self._mock_get(b"payload-that-must-not-be-written"):
            with pytest.raises(UnsafePathComponentError):
                helper.substitute_file_downloads(
                    tool=self._tool(tool_slug=traversal), response=self._response()
                )

        assert not (tmp_path / "etc").exists()
        assert list(outdir.iterdir()) == []

    def test_legitimate_slugs_still_nest_under_the_root(self, tmp_path):
        outdir = tmp_path / "files"
        outdir.mkdir()
        helper = FileHelper(client=None, outdir=str(outdir))
        with self._mock_get(b"payload"):
            result = helper.substitute_file_downloads(
                tool=self._tool(), response=self._response()
            )

        written = Path(result["attachment"])
        assert (
            written.resolve()
            == (outdir / "GMAIL" / "GMAIL_GET_ATTACHMENT" / "composio").resolve()
        )
        assert written.read_bytes() == b"payload"

    def test_tilde_download_dir_still_works(self, tmp_path, monkeypatch):
        """`secure_join` expands `~` in the root; the containment check in
        `download()` must expand it too. When only one side did, every download
        under `file_download_dir='~/...'` failed as a path traversal."""
        monkeypatch.setenv("HOME", str(tmp_path))
        helper = FileHelper(client=None, outdir="~/downloads")
        with self._mock_get(b"payload"):
            result = helper.substitute_file_downloads(
                tool=self._tool(), response=self._response()
            )

        written = Path(result["attachment"])
        assert (
            written.resolve()
            == (
                tmp_path / "downloads" / "GMAIL" / "GMAIL_GET_ATTACHMENT" / "composio"
            ).resolve()
        )
        assert written.read_bytes() == b"payload"

    def test_symlink_inside_root_cannot_be_used_to_escape(self, tmp_path):
        """Per-component validation cannot see a symlink; the post-resolve
        containment check in `secure_join` is what catches this."""
        outdir = tmp_path / "files"
        outdir.mkdir()
        outside = tmp_path / "outside"
        outside.mkdir()
        (outdir / "GMAIL").symlink_to(outside, target_is_directory=True)

        helper = FileHelper(client=None, outdir=str(outdir))
        with self._mock_get():
            with pytest.raises(UnsafePathComponentError, match="outside"):
                helper.substitute_file_downloads(
                    tool=self._tool(), response=self._response()
                )
        assert list(outside.iterdir()) == []


class TestEnhanceSchemaDescriptionsEmptySchema:
    """Regression tests for the empty-input-parameters crash.

    MCP-backed toolkits can return `input_parameters: {}` for tools with no
    required arguments (the same convention the API uses for missing output
    schemas). Before the fix, `FileHelper.enhance_schema_descriptions` —
    called unconditionally on every fetched tool's input_parameters from
    `Tools._get()` — raised `KeyError: 'properties'` on that shape,
    crashing the public `Composio.tools.get(...)` call.

    The sibling `FileHelper.process_file_uploadable_schema` already guarded
    this with `if "properties" not in schema: return schema`. This adds the
    same guard to `enhance_schema_descriptions`.

    See https://github.com/ComposioHQ/composio/issues/3354 for the related
    TS-side issue.
    """

    def test_empty_schema_does_not_raise(self, file_helper):
        """schema={} must not raise — it means 'no declared schema'."""
        # Pre-fix: KeyError: 'properties'
        result = file_helper.enhance_schema_descriptions(schema={})
        assert result == {}

    def test_schema_without_properties_returns_unchanged(self, file_helper):
        """A schema with metadata but no `properties` key is returned as-is.

        Mirrors how `process_file_uploadable_schema` treats the same input.
        """
        schema = {"type": "object", "additionalProperties": False}
        result = file_helper.enhance_schema_descriptions(schema=schema)
        assert result == {"type": "object", "additionalProperties": False}

    def test_schema_with_empty_properties_dict_is_handled(self, file_helper):
        """`properties: {}` is valid JSON Schema (declares no fields).

        The loop iterates zero times and the schema is returned unchanged.
        """
        schema = {"type": "object", "properties": {}}
        result = file_helper.enhance_schema_descriptions(schema=schema)
        assert result == {"type": "object", "properties": {}}

    def test_populated_schema_still_enhanced(self, file_helper):
        """Non-empty schemas continue to receive type-hint enhancements.

        Guard against regressing the enhancement behavior while fixing
        the empty-schema crash.
        """
        schema = {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Search term"},
            },
            "required": ["query"],
        }
        result = file_helper.enhance_schema_descriptions(schema=schema)
        description = result["properties"]["query"]["description"]
        # The description must have been mutated from the input and must
        # mention the type and the required-marker. Assertions are token-
        # based to avoid coupling to the exact phrasing in
        # enhance_schema_descriptions.
        assert description != "Search term"
        assert "string" in description
        assert "required" in description

    def test_boolean_property_schemas_are_left_unchanged(self, file_helper):
        schema = {
            "type": "object",
            "properties": {
                "anything": True,
                "never": False,
                "query": {"type": "string"},
            },
            "required": ["anything", "query"],
        }

        result = file_helper.enhance_schema_descriptions(schema)

        assert result["properties"]["anything"] is True
        assert result["properties"]["never"] is False
        assert "string" in result["properties"]["query"]["description"]
        assert "required" in result["properties"]["query"]["description"]


class TestFromPathSensitiveGuard:
    """`FileUploadable.from_path` is the single upload primitive in the Python
    SDK; the sensitive-path denylist must fire there before any file read or
    network round-trip (parity with the TS core SDK and the CLI fix for
    issue #3746 / GHSA-hp3h-89pf-5q58)."""

    def test_from_path_blocks_ssh_private_key(self, mock_client):
        p = Path.home() / ".ssh" / "id_rsa"
        with pytest.raises(SensitiveFilePathBlockedError):
            FileUploadable.from_path(
                client=mock_client,
                file=str(p),
                tool="GMAIL_SEND_EMAIL",
                toolkit="gmail",
            )
        # The guard runs before the SDK contacts the API for a presigned URL.
        mock_client.post.assert_not_called()

    def test_from_path_blocks_dotenv_basename(self, mock_client):
        p = Path(tempfile.gettempdir()) / ".env"
        with pytest.raises(SensitiveFilePathBlockedError):
            FileUploadable.from_path(
                client=mock_client,
                file=str(p),
                tool="GMAIL_SEND_EMAIL",
                toolkit="gmail",
            )
        mock_client.post.assert_not_called()

    def test_from_path_opt_out_disables_guard(self, mock_client):
        p = Path.home() / ".ssh" / "composio-does-not-exist-guard-test"
        with pytest.raises(Exception) as exc_info:
            FileUploadable.from_path(
                client=mock_client,
                file=str(p),
                tool="GMAIL_SEND_EMAIL",
                toolkit="gmail",
                sensitive_file_upload_protection=False,
            )
        assert not isinstance(exc_info.value, SensitiveFilePathBlockedError)
        mock_client.post.assert_not_called()


class TestResponseDerivedUrlsAreGuarded:
    def _download_response(self) -> MagicMock:
        response = MagicMock()
        response.status_code = 200
        response.iter_content = lambda chunk_size: [b"%PDF-1.4"]
        response.close = MagicMock()
        return response

    def _s3_client(self, presigned_url: str) -> MagicMock:
        client = MagicMock()
        s3meta = MagicMock()
        s3meta.key = "s3-key"
        s3meta.new_presigned_url = presigned_url
        client.post.return_value = s3meta
        return client

    def test_download_validates_s3url(self, tmp_path):
        f = FileDownloadable(
            name="report.pdf",
            mimetype="application/pdf",
            s3url="https://s3.example.com/file",
        )
        with patch(
            "composio.core.models._files.safe_get",
            return_value=self._download_response(),
        ) as mock_get:
            f.download(tmp_path / "out", root=tmp_path / "out")

        # `safe_get` is the guard: it validates the target and connects to the
        # address it validated, rather than re-resolving the hostname.
        assert mock_get.call_args.args == ("https://s3.example.com/file",)

    def test_download_blocked_url_never_reaches_the_network(self, tmp_path):
        outdir = tmp_path / "out"
        f = FileDownloadable(
            name="report.pdf",
            mimetype="application/pdf",
            s3url="http://169.254.169.254/latest/meta-data",
        )
        with patch(
            "composio.utils.url_safety.assert_safe_fetch_target",
            side_effect=BlockedInternalUrlError("blocked"),
        ):
            with patch(
                "composio.utils.url_safety.requests.Session.request"
            ) as mock_send:
                with pytest.raises(BlockedInternalUrlError):
                    f.download(outdir, root=outdir)

        mock_send.assert_not_called()
        assert not outdir.exists()

    def test_download_refuses_to_follow_redirects(self, tmp_path):
        f = FileDownloadable(
            name="report.pdf",
            mimetype="application/pdf",
            s3url="https://s3.example.com/file",
        )
        with patch(
            "composio.core.models._files.safe_get",
            return_value=self._download_response(),
        ) as mock_get:
            f.download(tmp_path / "out", root=tmp_path / "out")

        # `safe_get` never follows redirects, and passing `allow_redirects`
        # through to it would be a way to turn that off.
        assert "allow_redirects" not in mock_get.call_args.kwargs

    def test_upload_bytes_to_s3_goes_through_safe_request(self):
        client = self._s3_client("https://s3.example.com/upload")
        with patch("composio.core.models._files.safe_request") as mock_safe_request:
            mock_safe_request.return_value.status_code = 200
            _upload_bytes_to_s3(
                client=client,
                filename="test.jpg",
                content=b"data",
                mimetype="image/jpeg",
                tool="TEST",
                toolkit="test",
            )

        assert mock_safe_request.call_args.args == (
            "PUT",
            "https://s3.example.com/upload",
        )

    def test_upload_bytes_to_s3_blocked_url_sends_nothing(self):
        client = self._s3_client("http://169.254.169.254/upload")
        with patch(
            "composio.utils.url_safety.assert_safe_fetch_target",
            side_effect=BlockedInternalUrlError("blocked"),
        ):
            with patch(
                "composio.utils.url_safety.requests.Session.request"
            ) as mock_request:
                with pytest.raises(BlockedInternalUrlError):
                    _upload_bytes_to_s3(
                        client=client,
                        filename="test.jpg",
                        content=b"data",
                        mimetype="image/jpeg",
                        tool="TEST",
                        toolkit="test",
                    )

        mock_request.assert_not_called()

    def test_file_upload_goes_through_safe_request(self, tmp_path):
        source = tmp_path / "report.pdf"
        source.write_bytes(b"%PDF-1.4")

        with patch("composio.core.models._files.safe_request") as mock_safe_request:
            mock_safe_request.return_value.status_code = 200
            assert upload(url="https://s3.example.com/upload", file=source) is True

        assert mock_safe_request.call_args.args == (
            "PUT",
            "https://s3.example.com/upload",
        )

    def test_file_upload_blocked_url_sends_nothing(self, tmp_path):
        source = tmp_path / "report.pdf"
        source.write_bytes(b"%PDF-1.4")

        with patch(
            "composio.utils.url_safety.assert_safe_fetch_target",
            side_effect=BlockedInternalUrlError("blocked"),
        ):
            with patch(
                "composio.utils.url_safety.requests.Session.request"
            ) as mock_request:
                with pytest.raises(BlockedInternalUrlError):
                    upload(url="http://127.0.0.1:9000/upload", file=source)

        mock_request.assert_not_called()