DemonGatanjieu-Anomalous_Model_Browser/tests/test_mcp_directory.py

"""The directory of tools by what an AI wants to do (api/mcp_directory.py): built from the catalog,
in begin_task's answer and in get_situation's while there is no task; and the daily set, where
an app holding an older tool list is told so."""
import os
import sys
import unittest
from pathlib import Path
from unittest import mock

PLUGIN = Path(__file__).resolve().parents[1]
sys.path[:0] = [str(PLUGIN), str(PLUGIN.parents[1]), str(PLUGIN / "tests")]

from api import action_catalog, ai_tasks, mcp_apps, mcp_directory, mcp_server  # noqa: E402
from test_mcp import rpc  # noqa: E402  (its module keeps tasks and gates in temporary folders)


def call(name, arguments):
    return rpc("tools/call", {"name": name, "arguments": arguments}, name)["result"]


class DirectoryTests(unittest.TestCase):
    def test_every_stage_has_a_heading_and_every_tool_a_line(self):
        stages = {stage for stage, _heading in mcp_directory.SECTIONS}
        self.assertTrue({entry.stage for entry in action_catalog.TOOLS.values()} <= stages, "a new stage needs a heading")
        specs = {name: tool["spec"] for name, tool in mcp_server.TOOLS.items() if name not in mcp_apps.APP_TOOLS}
        text = " ".join(mcp_directory.directory(specs).values())
        self.assertTrue(all(f"{name} (" in text for name in specs), "every tool is in it")
        self.assertIn("set_model (Swap a node's model)", text)

    def test_only_the_tools_offered_in_the_catalogs_order(self):
        specs = {name: mcp_server.TOOLS[name]["spec"] for name in ("run_status", "run_workflow", "zoom_image")}
        found = mcp_directory.directory(specs)
        self.assertEqual(found["Run a workflow"], "run_workflow (Run the workflow), run_status (How the run is going)")
        self.assertEqual(set(found), {"note", "Run a workflow", "Pictures"})

    def test_names_alone_when_the_answer_would_be_too_long(self):
        specs = {name: tool["spec"] for name, tool in mcp_server.TOOLS.items() if name not in mcp_apps.APP_TOOLS}
        result = {"content": [{"type": "text", "text": ""}], "structuredContent": {"task": "t", "asked": "x" * 2500}, "isError": False}
        found = mcp_directory.attach(result, "begin_task", specs)["structuredContent"]["directory"]
        self.assertIn("set_model", found["Models"])
        self.assertNotIn("(Swap a node's model)", str(found), "the titles go first")
        self.assertLessEqual(mcp_directory.mcp_fit.size(mcp_directory.directory(specs, titles=False)),
                             mcp_directory.mcp_fit.size(mcp_directory.directory(specs)) * 2 // 3)

    def test_begin_task_carries_the_directory(self):
        result = call("begin_task", {"title": "目录", "new_task": True})
        found = result["structuredContent"]["directory"]
        self.assertIn("run_workflow", found["Run a workflow"])
        self.assertIn('"directory"', result["content"][0]["text"], "the text says it too")
        self.assertLess(len(result["content"][0]["text"]), 4000, "small enough to stay in the chat")

    def test_get_situation_carries_it_only_without_a_task(self):
        with mock.patch.object(mcp_server.ai_tasks, "active_task_id", return_value=None), \
                mock.patch("api.mcp_situation.ask_page", side_effect=mcp_server.mcp_bridge.PageUnavailable("no page")):
            self.assertIn("directory", call("get_situation", {})["structuredContent"])
        call("begin_task", {"title": "有任务", "new_task": True})
        with mock.patch("api.mcp_situation.ask_page", side_effect=mcp_server.mcp_bridge.PageUnavailable("no page")):
            self.assertNotIn("directory", call("get_situation", {})["structuredContent"])


class StaleListTests(unittest.TestCase):
    """An app going on with the tool list it fetched before Anomalous was updated (mcp_directory.stale)."""

    def setUp(self):
        mcp_directory._lists = None
        if os.path.exists(mcp_directory._path()):
            os.remove(mcp_directory._path())

    def test_an_app_never_seen_fetching_is_not_told(self):
        self.assertNotIn("tool_list", call("begin_task", {"title": "没见过", "new_task": True})["structuredContent"])

    def test_the_ai_and_the_workbench_are_told_until_the_app_fetches_the_list(self):
        rpc("tools/list")
        task = call("begin_task", {"title": "旧列表", "new_task": True})["structuredContent"]
        self.assertNotIn("tool_list", call("report", {"text": "好", "task": task["task"]})["structuredContent"], "today's list")
        with mock.patch.dict(mcp_server.TOOLS["report"]["spec"], {"description": "Anomalous was updated"}):
            answer = call("report", {"text": "还在", "task": task["task"]})
            self.assertIn("older list", answer["structuredContent"]["tool_list"])
            self.assertIn("tool_list", answer["content"][0]["text"], "the text says it too")
            self.assertTrue(ai_tasks.get_task(task["task"])["stale_tools"])
            rpc("tools/list")
            self.assertNotIn("stale_tools", ai_tasks.get_task(task["task"]), "fetched again: the note goes")
            self.assertNotIn("tool_list", call("report", {"text": "好了", "task": task["task"]})["structuredContent"])
        mcp_directory._lists = None
        self.assertTrue(mcp_directory.stale("test", mcp_server._tools()), "kept on disk: the app now holds the other list")


if __name__ == "__main__":
    unittest.main()