115 lines
4.6 KiB
Python
115 lines
4.6 KiB
Python
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from core.artifacts import ArtifactPathError, ToolExecutionResult, resolve_artifact_path
|
|
from core.executor import ExecCtx
|
|
from core.executor_host import HostExecutor
|
|
from tools.publish_artifacts import PublishArtifactsTool
|
|
from web.routers.files import _task_file_target
|
|
|
|
|
|
class ArtifactPathTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.tmp.name)
|
|
self.wd = self.root / "技术讨论"
|
|
self.wd.mkdir()
|
|
(self.wd / "report.pdf").write_bytes(b"pdf")
|
|
|
|
def tearDown(self) -> None:
|
|
self.tmp.cleanup()
|
|
|
|
def test_canonical_and_legacy_paths_resolve_to_same_file(self) -> None:
|
|
expected = self.wd / "report.pdf"
|
|
for raw in (
|
|
"report.pdf",
|
|
"技术讨论/report.pdf",
|
|
str(expected),
|
|
"/workspace/技术讨论/report.pdf",
|
|
):
|
|
with self.subTest(raw=raw):
|
|
actual, rel = resolve_artifact_path(
|
|
raw, working_dir=self.wd, user_root=self.root,
|
|
)
|
|
self.assertEqual(actual, expected.resolve())
|
|
self.assertEqual(rel, "report.pdf")
|
|
|
|
def test_explicit_dot_slash_disambiguates_same_named_subdirectory(self) -> None:
|
|
nested = self.wd / "技术讨论" / "nested.html"
|
|
nested.parent.mkdir()
|
|
nested.write_text("ok", encoding="utf-8")
|
|
actual, rel = resolve_artifact_path(
|
|
"./技术讨论/nested.html", working_dir=self.wd, user_root=self.root,
|
|
)
|
|
self.assertEqual(actual, nested.resolve())
|
|
self.assertEqual(rel, "技术讨论/nested.html")
|
|
|
|
def test_escape_and_directory_are_rejected(self) -> None:
|
|
for raw in ("../outside.txt", "."):
|
|
with self.subTest(raw=raw), self.assertRaises(ArtifactPathError):
|
|
resolve_artifact_path(
|
|
raw, working_dir=self.wd, user_root=self.root,
|
|
)
|
|
|
|
def test_publish_artifacts_is_explicit_bounded_and_deduplicated(self) -> None:
|
|
tool = PublishArtifactsTool(
|
|
working_dir=self.wd,
|
|
base_dir=self.wd,
|
|
user_root=self.root,
|
|
)
|
|
result = tool.execute({"not": "a list"})
|
|
self.assertIsInstance(result, str)
|
|
published = tool.execute([
|
|
{"path": "report.pdf", "label": "最终报告"},
|
|
{"path": "./report.pdf"},
|
|
])
|
|
self.assertIsInstance(published, ToolExecutionResult)
|
|
self.assertEqual(len(published.artifacts), 1)
|
|
self.assertEqual(published.artifacts[0].path, "report.pdf")
|
|
self.assertEqual(published.artifacts[0].label, "最终报告")
|
|
|
|
executed = HostExecutor({tool.name: tool}).call_tool(
|
|
tool.name,
|
|
{"artifacts": [{"path": "report.pdf"}]},
|
|
ExecCtx(user_id="u", task_id="t", working_dir=self.wd),
|
|
)
|
|
self.assertEqual(executed.content, "[OK] published 1 artifact(s): report.pdf")
|
|
self.assertEqual(executed.artifacts[0]["path"], "report.pdf")
|
|
|
|
def test_publish_path_is_strictly_task_relative_when_names_repeat(self) -> None:
|
|
nested = self.wd / "技术讨论" / "nested.html"
|
|
nested.parent.mkdir()
|
|
nested.write_text("ok", encoding="utf-8")
|
|
tool = PublishArtifactsTool(
|
|
working_dir=self.wd,
|
|
base_dir=self.wd,
|
|
user_root=self.root,
|
|
)
|
|
published = tool.execute([{"path": "技术讨论/nested.html"}])
|
|
self.assertIsInstance(published, ToolExecutionResult)
|
|
self.assertEqual(published.artifacts[0].path, "技术讨论/nested.html")
|
|
|
|
def test_legacy_card_resolution_covers_both_known_shapes_and_rename(self) -> None:
|
|
direct = self.wd / "manual.pdf"
|
|
direct.write_bytes(b"direct")
|
|
nested = self.wd / "技术讨论" / "nested.html"
|
|
nested.parent.mkdir()
|
|
nested.write_text("nested", encoding="utf-8")
|
|
|
|
# 92ac20cf shape: old user-root path already points at the correct file.
|
|
self.assertEqual(
|
|
_task_file_target(self.root, self.wd, "技术讨论/manual.pdf", True),
|
|
direct.resolve(),
|
|
)
|
|
# 9b4502aa shape: the same text was actually task-relative into a repeated dir.
|
|
self.assertEqual(
|
|
_task_file_target(self.root, self.wd, "技术讨论/nested.html", True),
|
|
nested.resolve(),
|
|
)
|
|
# After a top-level rename, dropping the obsolete first component finds the file.
|
|
self.assertEqual(
|
|
_task_file_target(self.root, self.wd, "旧目录/manual.pdf", True),
|
|
direct.resolve(),
|
|
)
|