mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
c9d3e48a4a
* fix(uploads): never truncate an existing upload on same-name collision Same-name uploads silently destroyed the earlier file on disk: the shared write path (open_upload_file_no_symlink) opened the destination with O_WRONLY|O_CREAT (no O_EXCL) and ftruncate(0), so re-uploading report.pdf opened and emptied the report.pdf already stored for the thread, and two concurrent same-name uploads could race on one inode. The Gateway ingestion pipeline already refuses overwrites (staged no-overwrite link commit), but the embedded client upload path (copy_upload_file_no_symlink) and the channel ingress paths (write_upload_file_no_symlink) still went through the truncating open. Open every upload destination with O_EXCL. On FileExistsError, claim the next free _N suffix against the directory's current contents via the existing claim_unique_filename helper and retry — the same deterministic behavior the Gateway pipeline documents (name_1.ext). The existing file is never opened, so its bytes survive intact under any future naming policy. Embedded-client and channel inbound-file responses now report the name the bytes actually landed under instead of the requested name. * test(uploads): cover concurrent same-name writers * fix(uploads): skip known copy collisions before staging --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
106 lines
4.2 KiB
Python
106 lines
4.2 KiB
Python
"""Converted uploads must preserve Markdown from earlier upload calls."""
|
|
|
|
import asyncio
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from openpyxl import Workbook
|
|
from pptx import Presentation
|
|
|
|
from deerflow.client import DeerFlowClient
|
|
from deerflow.uploads.companions import resolve_companion
|
|
|
|
|
|
def _document(path: Path, content: str) -> None:
|
|
if path.suffix == ".md":
|
|
path.write_text(content, encoding="utf-8")
|
|
elif path.suffix == ".xlsx":
|
|
book = Workbook()
|
|
book.active["A1"] = content
|
|
book.save(path)
|
|
book.close()
|
|
else:
|
|
slides = Presentation()
|
|
slide = slides.slides.add_slide(slides.slide_layouts[1])
|
|
slide.shapes.title.text = content
|
|
slides.save(path)
|
|
|
|
|
|
@pytest.fixture
|
|
def upload_case(tmp_path, monkeypatch):
|
|
import deerflow.client as client_module
|
|
|
|
uploads = tmp_path / "thread/user-data/uploads"
|
|
uploads.mkdir(parents=True)
|
|
monkeypatch.setattr(client_module, "ensure_uploads_dir", lambda *args, **kwargs: uploads)
|
|
return DeerFlowClient.__new__(DeerFlowClient), uploads
|
|
|
|
|
|
@pytest.mark.parametrize("in_loop", [False, True])
|
|
@pytest.mark.parametrize("first_ext,second_ext", [(".md", ".xlsx"), (".md", ".pptx"), (".xlsx", ".pptx"), (".pptx", ".xlsx")])
|
|
def test_separate_uploads_preserve_existing_markdown(upload_case, tmp_path, in_loop, first_ext, second_ext):
|
|
client, uploads = upload_case
|
|
first = tmp_path / f"report{first_ext}"
|
|
second = tmp_path / f"report{second_ext}"
|
|
_document(first, "KEEP FIRST DOCUMENT")
|
|
_document(second, "SECOND DOCUMENT")
|
|
|
|
def upload(source):
|
|
if not in_loop:
|
|
return client.upload_files("companion-thread", [source])
|
|
|
|
# Run the synchronous upload inside an active event loop to exercise conversion_pool.
|
|
async def run():
|
|
return client.upload_files("companion-thread", [source])
|
|
|
|
return asyncio.run(run())
|
|
|
|
first_response = upload(first)
|
|
original_markdown = uploads / "report.md"
|
|
original_bytes = original_markdown.read_bytes()
|
|
second_response = upload(second)
|
|
|
|
assert first_response["success"] and second_response["success"]
|
|
assert original_markdown.read_bytes() == original_bytes
|
|
assert "KEEP FIRST DOCUMENT" in original_markdown.read_text(encoding="utf-8")
|
|
info = second_response["files"][0]
|
|
assert info["filename"] == second.name
|
|
assert info["markdown_file"] == "report_1.md"
|
|
companion = uploads / info["markdown_file"]
|
|
assert "SECOND DOCUMENT" in companion.read_text(encoding="utf-8")
|
|
assert resolve_companion(uploads / second.name) == companion
|
|
assert Path(info["markdown_path"]) == companion
|
|
assert info["markdown_virtual_path"].endswith("/report_1.md")
|
|
assert info["markdown_artifact_url"].endswith("/report_1.md")
|
|
assert (uploads / first.name).read_bytes() == first.read_bytes()
|
|
assert (uploads / second.name).read_bytes() == second.read_bytes()
|
|
if first_ext != ".md":
|
|
assert resolve_companion(uploads / first.name) == original_markdown
|
|
|
|
|
|
def test_reupload_preserves_original_and_occupied_companion_names(upload_case, tmp_path):
|
|
client, uploads = upload_case
|
|
source = tmp_path / "report.xlsx"
|
|
_document(source, "FIRST DOCUMENT")
|
|
client.upload_files("companion-thread", [source])
|
|
first_original = (uploads / source.name).read_bytes()
|
|
first_markdown = (uploads / "report.md").read_bytes()
|
|
occupied = uploads / "report_1.md"
|
|
occupied.write_text("INDEPENDENT NOTES", encoding="utf-8")
|
|
|
|
_document(source, "REPLACEMENT DOCUMENT")
|
|
response = client.upload_files("companion-thread", [source])
|
|
|
|
assert response["success"]
|
|
info = response["files"][0]
|
|
assert info["filename"] == "report_1.xlsx"
|
|
assert (uploads / source.name).read_bytes() == first_original
|
|
assert (uploads / info["filename"]).read_bytes() == source.read_bytes()
|
|
assert info["markdown_file"] == "report_1_1.md"
|
|
assert (uploads / "report.md").read_bytes() == first_markdown
|
|
assert occupied.read_text(encoding="utf-8") == "INDEPENDENT NOTES"
|
|
companion = uploads / info["markdown_file"]
|
|
assert "REPLACEMENT DOCUMENT" in companion.read_text(encoding="utf-8")
|
|
assert resolve_companion(uploads / info["filename"]) == companion
|
|
assert resolve_companion(uploads / source.name) == uploads / "report.md"
|