mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-10 03:28:53 +00:00
Merge pull request #27794 from BerriAI/litellm_/fervent-archimedes-7e86be
Some checks are pending
Unit Tests: Caching (Redis) / caching-redis (push) Waiting to run
Unit Tests: Proxy DB Operations / assert-shard-coverage (push) Waiting to run
Unit Tests: Proxy DB Operations / auth-checks (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / budgets (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / custom-logging (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / db-and-spend (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / endpoints-and-responses (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / guardrails-hooks (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / jwt-and-keys (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / key-generation (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / logging-misc (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / proxy-runtime (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / proxy-server-core (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / schema-migration (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / proxy-utils (push) Blocked by required conditions
Unit Tests: Security / security (push) Waiting to run
Some checks are pending
Unit Tests: Caching (Redis) / caching-redis (push) Waiting to run
Unit Tests: Proxy DB Operations / assert-shard-coverage (push) Waiting to run
Unit Tests: Proxy DB Operations / auth-checks (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / budgets (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / custom-logging (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / db-and-spend (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / endpoints-and-responses (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / guardrails-hooks (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / jwt-and-keys (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / key-generation (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / logging-misc (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / proxy-runtime (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / proxy-server-core (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / schema-migration (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / proxy-utils (push) Blocked by required conditions
Unit Tests: Security / security (push) Waiting to run
cherry-pick: reject bare str at file-input sinks (#27762) onto litellm_1.84.0rc2
This commit is contained in:
commit
037d4c505b
7 changed files with 185 additions and 65 deletions
|
|
@ -53,8 +53,19 @@ def process_audio_file(audio_file: FileTypes) -> ProcessedAudioFile:
|
|||
# Raw bytes
|
||||
filename = "audio.wav"
|
||||
file_content = bytes(audio_file)
|
||||
elif isinstance(audio_file, (str, os.PathLike)):
|
||||
# File path or PathLike
|
||||
elif isinstance(audio_file, str):
|
||||
# Bare strings are rejected — see extract_file_data for the same
|
||||
# rationale: in a proxy request handler the string is
|
||||
# attacker-controlled, and opening it as a path is an arbitrary
|
||||
# file read.
|
||||
raise ValueError(
|
||||
"process_audio_file does not accept bare str inputs. Pass bytes, "
|
||||
"an open file handle, a (filename, content) tuple, or a "
|
||||
"pathlib.Path."
|
||||
)
|
||||
elif isinstance(audio_file, os.PathLike):
|
||||
# File path or PathLike — PathLike is a Python-level type that
|
||||
# HTTP form values can't fabricate.
|
||||
file_path = str(audio_file)
|
||||
with open(file_path, "rb") as f:
|
||||
file_content = f.read()
|
||||
|
|
@ -66,8 +77,14 @@ def process_audio_file(audio_file: FileTypes) -> ProcessedAudioFile:
|
|||
content = audio_file[1]
|
||||
if isinstance(content, (bytes, bytearray)):
|
||||
file_content = bytes(content)
|
||||
elif isinstance(content, (str, os.PathLike)):
|
||||
# File path or PathLike
|
||||
elif isinstance(content, str):
|
||||
raise ValueError(
|
||||
"process_audio_file does not accept bare str tuple "
|
||||
"contents. Pass bytes, an open file handle, or a "
|
||||
"pathlib.Path."
|
||||
)
|
||||
elif isinstance(content, os.PathLike):
|
||||
# PathLike: SDK convenience for local-file uploads.
|
||||
with open(str(content), "rb") as f:
|
||||
file_content = f.read()
|
||||
elif hasattr(content, "read"):
|
||||
|
|
@ -149,7 +166,14 @@ def get_audio_file_content_hash(file_obj: FileTypes) -> str:
|
|||
try:
|
||||
if isinstance(file_content_obj, (bytes, bytearray)):
|
||||
file_content = bytes(file_content_obj)
|
||||
elif isinstance(file_content_obj, (str, os.PathLike)):
|
||||
elif isinstance(file_content_obj, str):
|
||||
# Bare strings are not treated as file paths in this helper —
|
||||
# the cache-key path is reached from request handlers where the
|
||||
# value is attacker-controlled. Fall back to hashing the string
|
||||
# itself rather than opening it.
|
||||
fallback_filename = file_content_obj
|
||||
file_content = None
|
||||
elif isinstance(file_content_obj, os.PathLike):
|
||||
try:
|
||||
with open(str(file_content_obj), "rb") as f:
|
||||
file_content = f.read()
|
||||
|
|
@ -229,8 +253,15 @@ def calculate_request_duration(file: FileTypes) -> Optional[float]:
|
|||
if isinstance(file, (bytes, bytearray)):
|
||||
# Raw bytes
|
||||
file_content = bytes(file)
|
||||
elif isinstance(file, (str, os.PathLike)):
|
||||
# File path
|
||||
elif isinstance(file, str):
|
||||
# Bare strings are rejected — see extract_file_data.
|
||||
raise ValueError(
|
||||
"calculate_request_duration does not accept bare str inputs. "
|
||||
"Pass bytes, an open file handle, a (filename, content) "
|
||||
"tuple, or a pathlib.Path."
|
||||
)
|
||||
elif isinstance(file, os.PathLike):
|
||||
# File path (PathLike): SDK convenience.
|
||||
with open(str(file), "rb") as f:
|
||||
file_content = f.read()
|
||||
elif isinstance(file, tuple):
|
||||
|
|
|
|||
|
|
@ -726,14 +726,25 @@ def extract_file_data(file_data: FileTypes) -> ExtractedFileData:
|
|||
else:
|
||||
file_content = file_data
|
||||
# Convert content to bytes
|
||||
if isinstance(file_content, (str, PathLike)):
|
||||
# If it's a path, open and read the file
|
||||
# Extract filename from path if not already set
|
||||
if isinstance(file_content, str):
|
||||
# Bare string inputs are rejected: when this helper runs in a proxy
|
||||
# request handler the string came from an attacker-controlled form
|
||||
# field, and opening it as a path is an arbitrary file read on the
|
||||
# proxy host. SDK callers who want to upload from a path should
|
||||
# either pass a pathlib.Path (a PathLike instance — see the branch
|
||||
# below) or open the file themselves and pass the handle / bytes.
|
||||
raise ValueError(
|
||||
"extract_file_data does not accept bare str inputs. Pass bytes, "
|
||||
"an open file handle, a (filename, content) tuple, or a "
|
||||
"pathlib.Path. To upload a local file from a path, call "
|
||||
"open(path, 'rb') yourself."
|
||||
)
|
||||
if isinstance(file_content, PathLike):
|
||||
# PathLike (pathlib.Path) is a Python-level type that HTTP form
|
||||
# values can't fabricate. Treat as a local file path for SDK
|
||||
# convenience.
|
||||
if filename is None:
|
||||
if isinstance(file_content, PathLike):
|
||||
filename = Path(file_content).name
|
||||
else:
|
||||
filename = Path(str(file_content)).name
|
||||
filename = Path(file_content).name
|
||||
with open(file_content, "rb") as f:
|
||||
content = f.read()
|
||||
elif isinstance(file_content, io.IOBase):
|
||||
|
|
|
|||
|
|
@ -10,7 +10,6 @@ import os
|
|||
import re
|
||||
from functools import partial
|
||||
from io import IOBase
|
||||
from pathlib import Path
|
||||
from typing import Any, Coroutine, Dict, Optional, Union
|
||||
|
||||
import httpx
|
||||
|
|
@ -376,11 +375,13 @@ def convert_file_document_to_url_document(document: Dict[str, Any]) -> Dict[str,
|
|||
with an inline base64 data URI.
|
||||
|
||||
Accepts document dicts like:
|
||||
{"type": "file", "file": "/path/to/document.pdf"} # file path string
|
||||
{"type": "file", "file": Path("/path/to/doc.pdf")} # pathlib.Path
|
||||
{"type": "file", "file": <binary file-like object>} # file-like object (BinaryIO)
|
||||
{"type": "file", "file": b"raw bytes"} # raw bytes
|
||||
|
||||
Bare ``str`` paths are not accepted — pass a ``pathlib.Path`` or
|
||||
``open(path, "rb")`` instead. See the str check below for the rationale.
|
||||
|
||||
Returns:
|
||||
{"type": "document_url", "document_url": "data:<mime>;base64,<data>"}
|
||||
or {"type": "image_url", "image_url": "data:<mime>;base64,<data>"}
|
||||
|
|
@ -389,14 +390,28 @@ def convert_file_document_to_url_document(document: Dict[str, Any]) -> Dict[str,
|
|||
if file_input is None:
|
||||
raise ValueError(
|
||||
"document with type='file' must include a 'file' field containing "
|
||||
"a file path (str), pathlib.Path, file-like object, or bytes"
|
||||
"a pathlib.Path, file-like object, or bytes"
|
||||
)
|
||||
|
||||
file_bytes: bytes
|
||||
mime_type: str = "application/octet-stream"
|
||||
file_name: Optional[str] = None
|
||||
|
||||
if isinstance(file_input, (str, Path)):
|
||||
if isinstance(file_input, str):
|
||||
# Bare strings are rejected here. The OCR ``document`` accepts a
|
||||
# ``{"type": "file", "file": <value>}`` shape, and when this helper
|
||||
# runs in a proxy request handler ``<value>`` is attacker-controlled.
|
||||
# Opening it as a path is an arbitrary local file read on the proxy
|
||||
# host, which is then base64-encoded and forwarded to the OCR
|
||||
# provider — an exfiltration primitive.
|
||||
raise ValueError(
|
||||
"OCR file input does not accept bare str values. Pass bytes, "
|
||||
"a pathlib.Path, or a file-like object. To OCR a local file "
|
||||
"from a path, call open(path, 'rb') yourself."
|
||||
)
|
||||
if isinstance(file_input, os.PathLike):
|
||||
# os.PathLike (pathlib.Path and custom __fspath__ classes) is a
|
||||
# Python-level type that HTTP form values can't fabricate.
|
||||
file_path = str(file_input)
|
||||
if not os.path.isfile(file_path):
|
||||
raise FileNotFoundError(f"File not found: {file_path}")
|
||||
|
|
@ -417,7 +432,7 @@ def convert_file_document_to_url_document(document: Dict[str, Any]) -> Dict[str,
|
|||
else:
|
||||
raise ValueError(
|
||||
f"Unsupported file input type: {type(file_input)}. "
|
||||
"Expected str (file path), pathlib.Path, bytes, or a file-like object."
|
||||
"Expected pathlib.Path, bytes, or a file-like object."
|
||||
)
|
||||
|
||||
if not file_bytes:
|
||||
|
|
|
|||
|
|
@ -367,3 +367,57 @@ def test_update_messages_with_model_file_ids_skips_non_openai_file_blocks():
|
|||
|
||||
# Messages pass through unchanged when there is no `file` sub-dict to remap.
|
||||
assert updated == messages
|
||||
|
||||
|
||||
class TestExtractFileDataBareStr:
|
||||
"""``extract_file_data`` used to accept bare ``str`` values and ``open()``
|
||||
them server-side. When the helper runs inside a proxy request handler the
|
||||
value is attacker-controlled, so the open() call was a textbook arbitrary
|
||||
local file read. Lock the new contract: bare ``str`` is rejected with a
|
||||
clear migration message; ``pathlib.Path`` is still accepted for SDK
|
||||
ergonomics because it's a Python-level type that HTTP form values can't
|
||||
fabricate."""
|
||||
|
||||
def test_rejects_bare_str(self):
|
||||
from litellm.litellm_core_utils.prompt_templates.common_utils import (
|
||||
extract_file_data,
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match="does not accept bare str inputs"):
|
||||
extract_file_data("/etc/passwd")
|
||||
|
||||
def test_accepts_pathlib_path(self):
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
from litellm.litellm_core_utils.prompt_templates.common_utils import (
|
||||
extract_file_data,
|
||||
)
|
||||
|
||||
content = b"hello"
|
||||
with tempfile.NamedTemporaryFile(suffix=".txt", delete=False) as f:
|
||||
f.write(content)
|
||||
tmp_path = Path(f.name)
|
||||
|
||||
try:
|
||||
extracted = extract_file_data(tmp_path)
|
||||
assert extracted.get("content") == content
|
||||
finally:
|
||||
os.unlink(str(tmp_path))
|
||||
|
||||
def test_accepts_bytes(self):
|
||||
from litellm.litellm_core_utils.prompt_templates.common_utils import (
|
||||
extract_file_data,
|
||||
)
|
||||
|
||||
extracted = extract_file_data(b"raw bytes content")
|
||||
assert extracted.get("content") == b"raw bytes content"
|
||||
|
||||
def test_accepts_tuple(self):
|
||||
from litellm.litellm_core_utils.prompt_templates.common_utils import (
|
||||
extract_file_data,
|
||||
)
|
||||
|
||||
extracted = extract_file_data(("foo.txt", b"raw bytes content"))
|
||||
assert extracted.get("filename") == "foo.txt"
|
||||
assert extracted.get("content") == b"raw bytes content"
|
||||
|
|
|
|||
|
|
@ -42,8 +42,10 @@ class TestProcessAudioFile:
|
|||
assert result.filename == "audio.wav"
|
||||
assert result.content_type == "audio/wav"
|
||||
|
||||
def test_process_file_path_input(self):
|
||||
"""Test processing file path input"""
|
||||
def test_process_pathlib_input(self):
|
||||
"""pathlib.Path is a Python-level type HTTP form values can't fabricate."""
|
||||
from pathlib import Path
|
||||
|
||||
test_content = b"test audio content"
|
||||
|
||||
with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as temp_file:
|
||||
|
|
@ -51,15 +53,22 @@ class TestProcessAudioFile:
|
|||
temp_file_path = temp_file.name
|
||||
|
||||
try:
|
||||
result = process_audio_file(temp_file_path)
|
||||
result = process_audio_file(Path(temp_file_path))
|
||||
|
||||
assert isinstance(result, ProcessedAudioFile)
|
||||
assert result.file_content == test_content
|
||||
assert result.filename == os.path.basename(temp_file_path)
|
||||
assert result.content_type == "audio/mpeg" # .mp3 should map to audio/mpeg
|
||||
assert result.content_type == "audio/mpeg"
|
||||
finally:
|
||||
os.unlink(temp_file_path)
|
||||
|
||||
def test_process_bare_str_path_rejected(self):
|
||||
"""Bare str paths are rejected — when this runs in a proxy request
|
||||
handler the value is attacker-controlled, and opening it as a path
|
||||
is an arbitrary local file read."""
|
||||
with pytest.raises(ValueError, match="does not accept bare str inputs"):
|
||||
process_audio_file("/etc/passwd")
|
||||
|
||||
def test_process_tuple_input_with_bytes(self):
|
||||
"""Test processing tuple input with bytes content"""
|
||||
filename = "test.wav"
|
||||
|
|
@ -73,8 +82,10 @@ class TestProcessAudioFile:
|
|||
assert result.filename == filename
|
||||
assert result.content_type == "audio/wav"
|
||||
|
||||
def test_process_tuple_input_with_file_path(self):
|
||||
"""Test processing tuple input with file path content"""
|
||||
def test_process_tuple_input_with_pathlib_content(self):
|
||||
"""Tuple input with pathlib.Path content is allowed; bare str content is not."""
|
||||
from pathlib import Path
|
||||
|
||||
test_content = b"test audio content"
|
||||
|
||||
with tempfile.NamedTemporaryFile(suffix=".flac", delete=False) as temp_file:
|
||||
|
|
@ -83,7 +94,7 @@ class TestProcessAudioFile:
|
|||
|
||||
try:
|
||||
filename = "custom_name.flac"
|
||||
audio_tuple = (filename, temp_file_path)
|
||||
audio_tuple = (filename, Path(temp_file_path))
|
||||
|
||||
result = process_audio_file(audio_tuple)
|
||||
|
||||
|
|
|
|||
|
|
@ -1454,40 +1454,34 @@ def test_extract_file_data_with_path_object():
|
|||
os.unlink(tmp_path)
|
||||
|
||||
|
||||
def test_extract_file_data_with_string_path():
|
||||
"""Test that filename is correctly extracted from string paths."""
|
||||
def test_extract_file_data_with_pathlib_path():
|
||||
"""Test that filename is correctly extracted from pathlib.Path inputs.
|
||||
Bare str paths are rejected — when this runs in a proxy request handler
|
||||
the value is attacker-controlled and opening it as a path is an LFI."""
|
||||
import os
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
from litellm.litellm_core_utils.prompt_templates.common_utils import (
|
||||
extract_file_data,
|
||||
)
|
||||
|
||||
# Create a temporary WAV file
|
||||
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
|
||||
tmp.write(b"fake wav content")
|
||||
tmp_path = tmp.name
|
||||
tmp_path = Path(tmp.name)
|
||||
|
||||
try:
|
||||
# Test with string path
|
||||
extracted = extract_file_data(tmp_path)
|
||||
|
||||
# Verify filename was extracted
|
||||
assert extracted["filename"] is not None
|
||||
assert extracted["filename"].endswith(".wav")
|
||||
|
||||
# Verify MIME type was correctly detected (can be audio/wav or audio/x-wav depending on system)
|
||||
assert extracted["content_type"] in [
|
||||
"audio/wav",
|
||||
"audio/x-wav",
|
||||
], f"Expected 'audio/wav' or 'audio/x-wav' but got '{extracted['content_type']}'"
|
||||
|
||||
# Verify content was read
|
||||
assert extracted["content"] == b"fake wav content"
|
||||
|
||||
finally:
|
||||
# Clean up temporary file
|
||||
os.unlink(tmp_path)
|
||||
os.unlink(str(tmp_path))
|
||||
|
||||
|
||||
def test_extract_file_data_with_tuple_format():
|
||||
|
|
@ -1510,35 +1504,29 @@ def test_extract_file_data_with_tuple_format():
|
|||
|
||||
|
||||
def test_extract_file_data_fallback_to_octet_stream():
|
||||
"""Test that unknown file types fall back to application/octet-stream."""
|
||||
"""Unknown file types fall back to application/octet-stream."""
|
||||
import os
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
from litellm.litellm_core_utils.prompt_templates.common_utils import (
|
||||
extract_file_data,
|
||||
)
|
||||
|
||||
# Create a temporary file with unknown extension
|
||||
with tempfile.NamedTemporaryFile(suffix=".xyz123", delete=False) as tmp:
|
||||
tmp.write(b"unknown content")
|
||||
tmp_path = tmp.name
|
||||
tmp_path = Path(tmp.name)
|
||||
|
||||
try:
|
||||
# Test with unknown file type
|
||||
extracted = extract_file_data(tmp_path)
|
||||
|
||||
# Verify filename was extracted
|
||||
assert extracted["filename"] is not None
|
||||
assert extracted["filename"].endswith(".xyz123")
|
||||
|
||||
# Verify MIME type falls back to octet-stream
|
||||
assert (
|
||||
extracted["content_type"] == "application/octet-stream"
|
||||
), f"Expected 'application/octet-stream' for unknown type, got '{extracted['content_type']}'"
|
||||
|
||||
finally:
|
||||
# Clean up temporary file
|
||||
os.unlink(tmp_path)
|
||||
os.unlink(str(tmp_path))
|
||||
|
||||
|
||||
def test_convert_tool_response_with_pdf_file():
|
||||
|
|
|
|||
|
|
@ -60,14 +60,15 @@ class TestGetMimeType:
|
|||
|
||||
|
||||
class TestConvertFileDocumentToUrlDocument:
|
||||
def test_should_convert_pdf_file_path_to_document_url(self):
|
||||
"""File path to a PDF should produce type=document_url with base64 data URI."""
|
||||
def test_should_convert_pdf_pathlib_path_to_document_url(self):
|
||||
"""pathlib.Path to a PDF should produce type=document_url with base64 data URI.
|
||||
Bare str paths are rejected — see test_should_reject_bare_str_path below."""
|
||||
pdf_content = b"%PDF-1.4 test content"
|
||||
|
||||
with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as f:
|
||||
f.write(pdf_content)
|
||||
f.flush()
|
||||
tmp_path = f.name
|
||||
tmp_path = Path(f.name)
|
||||
|
||||
try:
|
||||
result = convert_file_document_to_url_document(
|
||||
|
|
@ -80,16 +81,16 @@ class TestConvertFileDocumentToUrlDocument:
|
|||
b64_data = result["document_url"].split(";base64,")[1]
|
||||
assert base64.b64decode(b64_data) == pdf_content
|
||||
finally:
|
||||
os.unlink(tmp_path)
|
||||
os.unlink(str(tmp_path))
|
||||
|
||||
def test_should_convert_image_file_path_to_image_url(self):
|
||||
"""File path to a PNG image should produce type=image_url with base64 data URI."""
|
||||
def test_should_convert_image_pathlib_path_to_image_url(self):
|
||||
"""pathlib.Path to a PNG image should produce type=image_url with base64 data URI."""
|
||||
png_content = b"\x89PNG\r\n\x1a\n fake png content"
|
||||
|
||||
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f:
|
||||
f.write(png_content)
|
||||
f.flush()
|
||||
tmp_path = f.name
|
||||
tmp_path = Path(f.name)
|
||||
|
||||
try:
|
||||
result = convert_file_document_to_url_document(
|
||||
|
|
@ -102,7 +103,16 @@ class TestConvertFileDocumentToUrlDocument:
|
|||
b64_data = result["image_url"].split(";base64,")[1]
|
||||
assert base64.b64decode(b64_data) == png_content
|
||||
finally:
|
||||
os.unlink(tmp_path)
|
||||
os.unlink(str(tmp_path))
|
||||
|
||||
def test_should_reject_bare_str_path(self):
|
||||
"""Bare str ``file`` values are rejected — when this runs in a proxy
|
||||
request handler the value is attacker-controlled, and opening it as
|
||||
a path is an arbitrary local file read on the proxy host."""
|
||||
with pytest.raises(ValueError, match="does not accept bare str values"):
|
||||
convert_file_document_to_url_document(
|
||||
{"type": "file", "file": "/etc/passwd"}
|
||||
)
|
||||
|
||||
def test_should_convert_pathlib_path(self):
|
||||
"""pathlib.Path objects should work the same as string paths."""
|
||||
|
|
@ -189,17 +199,17 @@ class TestConvertFileDocumentToUrlDocument:
|
|||
with pytest.raises(ValueError, match="must include a 'file' field"):
|
||||
convert_file_document_to_url_document({"type": "file"})
|
||||
|
||||
def test_should_raise_error_for_nonexistent_file_path(self):
|
||||
"""Non-existent file path should raise FileNotFoundError."""
|
||||
def test_should_raise_error_for_nonexistent_pathlib_path(self):
|
||||
"""Non-existent pathlib.Path should raise FileNotFoundError."""
|
||||
with pytest.raises(FileNotFoundError, match="File not found"):
|
||||
convert_file_document_to_url_document(
|
||||
{"type": "file", "file": "/nonexistent/path/to/file.pdf"}
|
||||
{"type": "file", "file": Path("/nonexistent/path/to/file.pdf")}
|
||||
)
|
||||
|
||||
def test_should_raise_error_for_empty_file(self):
|
||||
"""Empty file should raise ValueError."""
|
||||
with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as f:
|
||||
tmp_path = f.name
|
||||
tmp_path = Path(f.name)
|
||||
|
||||
try:
|
||||
with pytest.raises(ValueError, match="File is empty"):
|
||||
|
|
@ -207,7 +217,7 @@ class TestConvertFileDocumentToUrlDocument:
|
|||
{"type": "file", "file": tmp_path}
|
||||
)
|
||||
finally:
|
||||
os.unlink(tmp_path)
|
||||
os.unlink(str(tmp_path))
|
||||
|
||||
def test_should_raise_error_for_unsupported_type(self):
|
||||
"""Unsupported file input types should raise ValueError."""
|
||||
|
|
@ -226,14 +236,14 @@ class TestConvertFileDocumentToUrlDocument:
|
|||
}
|
||||
)
|
||||
|
||||
def test_should_override_mime_type_for_file_path(self):
|
||||
def test_should_override_mime_type_for_pathlib_path(self):
|
||||
"""Explicit mime_type should override auto-detection from extension."""
|
||||
content = b"some content"
|
||||
|
||||
with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as f:
|
||||
f.write(content)
|
||||
f.flush()
|
||||
tmp_path = f.name
|
||||
tmp_path = Path(f.name)
|
||||
|
||||
try:
|
||||
result = convert_file_document_to_url_document(
|
||||
|
|
@ -243,7 +253,7 @@ class TestConvertFileDocumentToUrlDocument:
|
|||
assert result["type"] == "image_url"
|
||||
assert result["image_url"].startswith("data:image/png;base64,")
|
||||
finally:
|
||||
os.unlink(tmp_path)
|
||||
os.unlink(str(tmp_path))
|
||||
|
||||
|
||||
class TestBuildDocumentFromUpload:
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue