94 lines
3 KiB
Python
94 lines
3 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
from adapter.matrix.files import (
|
|
build_agent_workspace_path,
|
|
download_matrix_attachment,
|
|
)
|
|
from core.protocol import Attachment
|
|
|
|
|
|
async def test_download_matrix_attachment_persists_file_and_returns_workspace_path(tmp_path: Path):
|
|
async def download(url: str):
|
|
assert url == "mxc://server/id"
|
|
return SimpleNamespace(body=b"%PDF-1.7")
|
|
|
|
client = SimpleNamespace(download=download)
|
|
attachment = Attachment(
|
|
type="document",
|
|
url="mxc://server/id",
|
|
filename="report.pdf",
|
|
mime_type="application/pdf",
|
|
)
|
|
|
|
saved = await download_matrix_attachment(
|
|
client=client,
|
|
workspace_root=tmp_path,
|
|
matrix_user_id="@alice:example.org",
|
|
room_id="!room:example.org",
|
|
attachment=attachment,
|
|
timestamp="20260420-153000",
|
|
)
|
|
|
|
assert saved.workspace_path == "report.pdf"
|
|
assert (tmp_path / "report.pdf").read_bytes() == b"%PDF-1.7"
|
|
|
|
|
|
def test_build_agent_workspace_path_uses_agent_workspace_volume(tmp_path: Path):
|
|
rel_path, abs_path = build_agent_workspace_path(
|
|
workspace_root=tmp_path / "agents" / "17",
|
|
filename="quarterly status.pdf",
|
|
)
|
|
|
|
assert rel_path == "quarterly status.pdf"
|
|
assert abs_path == tmp_path / "agents" / "17" / rel_path
|
|
|
|
|
|
def test_build_agent_workspace_path_uses_windows_style_copy_index(tmp_path: Path):
|
|
workspace_root = tmp_path / "agents" / "17"
|
|
workspace_root.mkdir(parents=True)
|
|
(workspace_root / "report.pdf").write_bytes(b"old")
|
|
(workspace_root / "report (1).pdf").write_bytes(b"older")
|
|
|
|
rel_path, abs_path = build_agent_workspace_path(
|
|
workspace_root=workspace_root,
|
|
filename="report.pdf",
|
|
)
|
|
|
|
assert rel_path == "report (2).pdf"
|
|
assert abs_path == workspace_root / "report (2).pdf"
|
|
|
|
|
|
def test_build_agent_workspace_path_sanitizes_to_basename(tmp_path: Path):
|
|
rel_path, abs_path = build_agent_workspace_path(
|
|
workspace_root=tmp_path / "agents" / "17",
|
|
filename="../../quarterly: status?.pdf",
|
|
)
|
|
|
|
assert rel_path == "quarterly_ status_.pdf"
|
|
assert abs_path == tmp_path / "agents" / "17" / "quarterly_ status_.pdf"
|
|
|
|
|
|
async def test_download_matrix_attachment_uses_agent_workspace_root(tmp_path: Path):
|
|
async def download(url: str):
|
|
assert url == "mxc://server/id"
|
|
return SimpleNamespace(body=b"%PDF-1.7")
|
|
|
|
saved = await download_matrix_attachment(
|
|
client=SimpleNamespace(download=download),
|
|
workspace_root=tmp_path / "agents" / "17",
|
|
matrix_user_id="@alice:example.org",
|
|
room_id="!room:example.org",
|
|
attachment=Attachment(
|
|
type="document",
|
|
url="mxc://server/id",
|
|
filename="report.pdf",
|
|
mime_type="application/pdf",
|
|
),
|
|
timestamp="20260428-110000",
|
|
)
|
|
|
|
assert saved.workspace_path == "report.pdf"
|
|
assert (tmp_path / "agents" / "17" / saved.workspace_path).read_bytes() == b"%PDF-1.7"
|