94 lines
3.1 KiB
Python
94 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from app.gateway.routers import artifacts
|
|
from deerflow.uploads.policy import default_upload_policy, set_runtime_upload_policy
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _restore_upload_policy():
|
|
set_runtime_upload_policy(default_upload_policy())
|
|
yield
|
|
set_runtime_upload_policy(default_upload_policy())
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_artifact_preview_reads_small_text_without_unbounded_artifact_response(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
source = tmp_path / "notes.txt"
|
|
source.write_text("第一行\n第二行", encoding="utf-8")
|
|
|
|
async def resolve(_thread_id: str, _path: str) -> Path:
|
|
return source
|
|
|
|
monkeypatch.setattr(artifacts, "aresolve_thread_virtual_path", resolve)
|
|
response = await artifacts.preview_artifact.__wrapped__(
|
|
thread_id="thread-1",
|
|
request=SimpleNamespace(),
|
|
path="mnt/user-data/uploads/notes.txt",
|
|
max_chars=1024,
|
|
)
|
|
|
|
assert response.content.splitlines() == ["第一行", "第二行"]
|
|
assert response.truncated is False
|
|
assert response.strategy == "complete-text"
|
|
assert response.total_bytes == source.stat().st_size
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_artifact_preview_samples_large_text_with_full_file_statistics(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
source = tmp_path / "words.txt"
|
|
source.write_text("".join(f"word-{index:04d}\n" for index in range(1, 1001)), encoding="utf-8")
|
|
policy = default_upload_policy()
|
|
policy["large_file"] = {
|
|
"threshold_bytes": 100,
|
|
"max_result_chars": 4096,
|
|
"hard_max_result_chars": 4096,
|
|
}
|
|
set_runtime_upload_policy(policy)
|
|
|
|
async def resolve(_thread_id: str, _path: str) -> Path:
|
|
return source
|
|
|
|
monkeypatch.setattr(artifacts, "aresolve_thread_virtual_path", resolve)
|
|
response = await artifacts.preview_artifact.__wrapped__(
|
|
thread_id="thread-1",
|
|
request=SimpleNamespace(),
|
|
path="mnt/user-data/uploads/words.txt",
|
|
max_chars=4096,
|
|
)
|
|
|
|
assert response.truncated is True
|
|
assert response.stats is not None
|
|
assert response.stats["lines"] == 1000
|
|
assert "流式扫描" in response.content
|
|
assert "word-0001" in response.content
|
|
assert len(response.content) <= 4096
|
|
assert "sampl" in response.strategy
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_artifact_preview_rejects_binary_content(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
source = tmp_path / "payload.bin"
|
|
source.write_bytes(b"abc\x00def")
|
|
|
|
async def resolve(_thread_id: str, _path: str) -> Path:
|
|
return source
|
|
|
|
monkeypatch.setattr(artifacts, "aresolve_thread_virtual_path", resolve)
|
|
with pytest.raises(artifacts.HTTPException) as raised:
|
|
await artifacts.preview_artifact.__wrapped__(
|
|
thread_id="thread-1",
|
|
request=SimpleNamespace(),
|
|
path="mnt/user-data/uploads/payload.bin",
|
|
max_chars=1024,
|
|
)
|
|
assert raised.value.status_code == 415
|