zcbot/tests/test_artifacts.py

import tempfile
import unittest
from contextlib import contextmanager
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
from uuid import uuid4

from core.artifacts import (
    ArtifactRef,
    ArtifactPathError,
    ToolExecutionResult,
    resolve_artifact_path,
)
from core.executor import ExecCtx
from core.executor_host import HostExecutor
from core.artifact_lifecycle import trash_active_artifacts
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_version_two_ref_carries_stable_artifact_identity(self) -> None:
        artifact_id = uuid4()
        ref = ArtifactRef(
            path="report.pdf",
            label="最终报告",
            artifact_id=artifact_id,
            version=2,
        ).as_dict()
        self.assertEqual(ref["version"], 2)
        self.assertEqual(ref["artifact_id"], str(artifact_id))

    def test_trash_moves_file_and_marks_lifecycle_row_deleted(self) -> None:
        artifact = self.wd / "report.pdf"
        row = SimpleNamespace(
            current_path="技术讨论/report.pdf",
            status="active",
            deleted_at=None,
            trash_path=None,
        )
        session = MagicMock()
        session.execute.return_value.scalars.return_value.all.return_value = [row]

        @contextmanager
        def fake_scope():
            yield session

        with patch("core.artifact_lifecycle.session_scope", fake_scope):
            count = trash_active_artifacts(
                user_id=uuid4(),
                user_root=self.root,
                target=artifact,
            )

        self.assertEqual(count, 1)
        self.assertFalse(artifact.exists())
        self.assertEqual(row.status, "deleted")
        self.assertIsNotNone(row.deleted_at)
        trashed = self.root / row.trash_path
        self.assertTrue(trashed.is_file())
        self.assertEqual(trashed.read_bytes(), b"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(),
        )