from __future__ import annotations import asyncio import json from types import SimpleNamespace from uuid import uuid4 import pytest import pytest_asyncio import yaml from sqlalchemy import select from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from app.gateway.routers.agents import ( AgentsListResponse, _agent_list_cache, _agent_list_inflight, _agent_owner_name_cache, _backfill_listing_cache, _cached_paginated_agent_list, _load_config_extras, _resolve_usernames, _write_agent_config, ) from app.gateway.services import build_run_config from deerflow.config import paths as paths_module from deerflow.config.agents_config import AgentConfig, normalize_agent_chat_link from deerflow.persistence.agents import AgentRepository, MemoryAgentStore from deerflow.persistence.agents.model import AgentFavoriteRow, AgentRow from deerflow.persistence.base import Base from deerflow.persistence.positions.memory import MemoryPositionStore from deerflow.persistence.roundtable_chains.model import RoundtableChainRow from deerflow.persistence.tags.memory import MemoryTagStore from deerflow.services.position_sync import PositionSyncService @pytest.fixture def isolated_paths(tmp_path, monkeypatch): """Redirect get_paths() at a fresh temp directory for the duration of one test.""" monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path)) monkeypatch.setattr(paths_module, "_paths", None) yield tmp_path monkeypatch.setattr(paths_module, "_paths", None) @pytest_asyncio.fixture async def sql_session_factory(tmp_path): engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'agent-tests.db'}") async with engine.begin() as conn: # Importing PositionSyncService above registers the agent / skill / # scheduled-task rows it materializes into. await conn.run_sync(Base.metadata.create_all) sf = async_sessionmaker(engine, expire_on_commit=False) try: yield sf finally: await engine.dispose() async def _add_sql_agent(sf, *, agent_id: str, user_id: str | None) -> None: async with sf() as session: session.add( AgentRow( id=agent_id, name=agent_id, description="", user_id=user_id, published=True, is_private=False, square_id="", ) ) await session.commit() @pytest.mark.asyncio async def test_sidebar_agent_query_returns_only_lightweight_visible_fields(sql_session_factory): await _add_sql_agent(sql_session_factory, agent_id="owned-agent", user_id="user-1") await _add_sql_agent(sql_session_factory, agent_id="public-agent", user_id=None) agents = await AgentRepository(sql_session_factory).list_sidebar("user-1", limit=3) assert agents == [ {"id": "owned-agent", "name": "owned-agent"}, {"id": "public-agent", "name": "public-agent"}, ] assert all(set(item) == {"id", "name"} for item in agents) @pytest.mark.asyncio async def test_memory_sidebar_query_returns_only_lightweight_fields(): store = MemoryAgentStore() await store.create({"id": "owned-agent", "name": "我的智能体", "user_id": "user-1"}) agents = await store.list_sidebar("user-1", limit=3) assert agents == [{"id": "owned-agent", "name": "我的智能体"}] async def _add_sql_favorite(sf, *, user_id: str, agent_id: str, origin: str) -> None: async with sf() as session: session.add( AgentFavoriteRow( id=uuid4().hex, user_id=user_id, agent_id=agent_id, origin=origin, position_id="default" if origin == "position" else None, ) ) await session.commit() async def _add_sql_chain(sf, *, chain_id: str, agent_ids: list[str]) -> None: async with sf() as session: session.add( RoundtableChainRow( id=chain_id, user_id="chain-owner", title=chain_id, description="", seats=json.dumps( [{"agent_id": agent_id, "name": agent_id, "description": ""} for agent_id in agent_ids], ensure_ascii=False, ), ) ) await session.commit() async def _sql_favorite_rows(sf, user_id: str) -> set[tuple[str, str]]: async with sf() as session: rows = ( await session.execute( select(AgentFavoriteRow.agent_id, AgentFavoriteRow.origin).where( AgentFavoriteRow.user_id == user_id ) ) ).all() return {(agent_id, origin) for agent_id, origin in rows} async def _default_positions() -> MemoryPositionStore: positions = MemoryPositionStore() await positions.create_position( { "id": "default", "name": "Default", "description": "", "is_default": True, } ) return positions @pytest.mark.asyncio async def test_agent_store_allows_duplicate_chinese_names_and_filters_visibility() -> None: store = MemoryAgentStore() chinese_name = "\u4e2d\u6587\u52a9\u624b" await store.ensure_builtin({"id": "builtin-agent", "name": chinese_name, "description": "built in"}) await store.create({"id": "private-a", "name": chinese_name, "description": "private", "user_id": "user-a"}) await store.create({"id": "published-b", "name": chinese_name, "description": "published", "user_id": "user-b", "published": True}) user_a_ids = {agent["id"] for agent in await store.list_visible("user-a")} assert user_a_ids == {"builtin-agent", "private-a", "published-b"} user_c_ids = {agent["id"] for agent in await store.list_visible("user-c")} assert user_c_ids == {"builtin-agent", "published-b"} @pytest.mark.asyncio async def test_position_sync_default_does_not_collect_builtin_agents(sql_session_factory) -> None: await _add_sql_agent(sql_session_factory, agent_id="builtin-auto", user_id=None) await _add_sql_agent(sql_session_factory, agent_id="builtin-manual", user_id=None) await _add_sql_favorite( sql_session_factory, user_id="user-a", agent_id="builtin-auto", origin="position", ) await _add_sql_favorite( sql_session_factory, user_id="user-a", agent_id="builtin-manual", origin="user", ) sync = PositionSyncService( sql_session_factory, await _default_positions(), MemoryTagStore(), ) await sync.sync_user("user-a") assert await _sql_favorite_rows(sql_session_factory, "user-a") == { ("builtin-manual", "user") } @pytest.mark.asyncio async def test_position_sync_tagged_agents_are_not_auto_favorited(sql_session_factory) -> None: await _add_sql_agent(sql_session_factory, agent_id="builtin-tagged", user_id=None) await _add_sql_agent(sql_session_factory, agent_id="custom-tagged", user_id="owner") await _add_sql_agent(sql_session_factory, agent_id="chain-tagged", user_id="owner") await _add_sql_chain(sql_session_factory, chain_id="chain-1", agent_ids=["chain-tagged"]) positions = await _default_positions() tags = MemoryTagStore() await tags.create_tag({"id": "agent-tag", "name": "Agent tag", "tag_type": "agent"}) await tags.set_target_tags("agent", "builtin-tagged", ["agent-tag"]) await tags.set_target_tags("agent", "custom-tagged", ["agent-tag"]) await tags.set_target_tags("agent", "chain-tagged", ["agent-tag"]) await positions.set_tag_bindings("default", {"agent": ["agent-tag"]}) sync = PositionSyncService(sql_session_factory, positions, tags) await sync.sync_user("user-a") assert await _sql_favorite_rows(sql_session_factory, "user-a") == { ("custom-tagged", "position") } @pytest.mark.asyncio async def test_mine_scope_ignores_auto_agent_favorites_but_keeps_manual_builtin(sql_session_factory) -> None: await _add_sql_agent(sql_session_factory, agent_id="builtin-auto", user_id=None) await _add_sql_agent(sql_session_factory, agent_id="builtin-manual", user_id=None) await _add_sql_agent(sql_session_factory, agent_id="custom-auto", user_id="owner") await _add_sql_agent(sql_session_factory, agent_id="chain-auto", user_id="owner") await _add_sql_chain(sql_session_factory, chain_id="chain-1", agent_ids=["chain-auto"]) await _add_sql_favorite( sql_session_factory, user_id="user-a", agent_id="builtin-auto", origin="position", ) await _add_sql_favorite( sql_session_factory, user_id="user-a", agent_id="builtin-manual", origin="user", ) await _add_sql_favorite( sql_session_factory, user_id="user-a", agent_id="custom-auto", origin="position", ) await _add_sql_favorite( sql_session_factory, user_id="user-a", agent_id="chain-auto", origin="position", ) store = AgentRepository(sql_session_factory) rows, total = await store.list_paginated( user_id="user-a", scope="mine", search=None, tag_ids=None, page=1, page_size=20, ) assert total == 2 assert {row["id"] for row in rows} == {"builtin-manual", "custom-auto"} assert await store.list_favorite_ids("user-a") == {"builtin-manual", "custom-auto"} @pytest.mark.asyncio async def test_manual_favorite_upgrades_existing_auto_builtin_favorite(sql_session_factory) -> None: await _add_sql_agent(sql_session_factory, agent_id="builtin-agent", user_id=None) await _add_sql_favorite( sql_session_factory, user_id="user-a", agent_id="builtin-agent", origin="position", ) store = AgentRepository(sql_session_factory) assert await store.list_favorite_ids("user-a") == set() await store.add_favorite("user-a", "builtin-agent") assert await _sql_favorite_rows(sql_session_factory, "user-a") == { ("builtin-agent", "user") } assert await store.list_favorite_ids("user-a") == {"builtin-agent"} rows, total = await store.list_paginated( user_id="user-a", scope="mine", search=None, tag_ids=None, page=1, page_size=20, ) assert total == 1 assert [row["id"] for row in rows] == ["builtin-agent"] async def _add_sql_hidden_agent(sf, *, agent_id: str, user_id: str) -> None: """他人创建、普通列表不可见的条目（未发布 / 私密广场）。""" async with sf() as session: session.add( AgentRow( id=agent_id, name=agent_id, description="", user_id=user_id, published=not agent_id.startswith("hidden-unpublished"), is_private=agent_id.startswith("hidden-private"), square_id="", ) ) await session.commit() @pytest.mark.asyncio async def test_sql_list_paginated_admin_pages_every_agent(sql_session_factory) -> None: """管理端分页（GET /api/agents?admin=true&page=N 的 store 支撑）： admin=True 跳过可见性过滤 — 内置 / 他人未发布 / 私密广场条目全部参与分页， 搜索在服务端过滤，越界页返回空页而非报错。""" await _add_sql_agent(sql_session_factory, agent_id="builtin-1", user_id=None) await _add_sql_agent(sql_session_factory, agent_id="owned-a", user_id="user-a") await _add_sql_hidden_agent(sql_session_factory, agent_id="hidden-unpublished", user_id="user-b") await _add_sql_hidden_agent(sql_session_factory, agent_id="hidden-private", user_id="user-b") store = AgentRepository(sql_session_factory) # 非 admin 对照：user-a 只看得到自己 + 内置（user-b 的两条不可见）。 _, visible_total = await store.list_paginated( user_id="user-a", scope=None, search=None, tag_ids=None, page=1, page_size=20, ) assert visible_total == 2 page1, total = await store.list_paginated( user_id="user-a", scope=None, search=None, tag_ids=None, page=1, page_size=2, admin=True, sort="time_asc", ) page2, total2 = await store.list_paginated( user_id="user-a", scope=None, search=None, tag_ids=None, page=2, page_size=2, admin=True, sort="time_asc", ) assert total == 4 and total2 == 4 assert len(page1) == 2 and len(page2) == 2 assert {r["id"] for r in page1} | {r["id"] for r in page2} == { "builtin-1", "owned-a", "hidden-unpublished", "hidden-private", } # 越界页：空页但 total 不变（前端据此收敛回末页）。 page3, total3 = await store.list_paginated( user_id="user-a", scope=None, search=None, tag_ids=None, page=3, page_size=2, admin=True, ) assert total3 == 4 and page3 == [] # 服务端搜索：管理端目录同样按名称过滤。 hits, hits_total = await store.list_paginated( user_id="user-a", scope=None, search="hidden", tag_ids=None, page=1, page_size=20, admin=True, ) assert hits_total == 2 assert {r["id"] for r in hits} == {"hidden-unpublished", "hidden-private"} @pytest.mark.asyncio async def test_memory_store_admin_pagination() -> None: """内存版 store 的 admin 分页行为与 SQL 版一致（单元测试/回退路径）。""" store = MemoryAgentStore() await store.ensure_builtin({"id": "builtin-1", "name": "Built In"}) await store.create({"id": "owned-a", "name": "Owned A", "user_id": "user-a"}) await store.create( {"id": "hidden-b", "name": "Hidden B", "user_id": "user-b", "published": False} ) await store.create( { "id": "private-b", "name": "Private B", "user_id": "user-b", # 内存版 _is_visible 不看 is_private，这里保持未发布， # 使「非 admin 对照 = 内置 + 自己」在两种 store 语义下都成立。 "published": False, "is_private": True, } ) _, visible_total = await store.list_paginated( user_id="user-a", scope=None, search=None, tag_ids=None, page=1, page_size=20 ) assert visible_total == 2 page1, total = await store.list_paginated( user_id="user-a", scope=None, search=None, tag_ids=None, page=1, page_size=3, admin=True, sort="time_asc", ) page2, total2 = await store.list_paginated( user_id="user-a", scope=None, search=None, tag_ids=None, page=2, page_size=3, admin=True, sort="time_asc", ) assert total == 4 and total2 == 4 assert len(page1) == 3 and len(page2) == 1 assert {r["id"] for r in page1} | {r["id"] for r in page2} == { "builtin-1", "owned-a", "hidden-b", "private-b", } async def _add_sql_hidden_flagged_agent(sf, *, agent_id: str, user_id: str | None) -> None: """创建时即 hidden=True 的条目（绕过 API，直接落库）。""" async with sf() as session: session.add( AgentRow( id=agent_id, name=agent_id, description="", user_id=user_id, published=True, is_private=False, square_id="", hidden=True, ) ) await session.commit() @pytest.mark.asyncio async def test_sql_hidden_agent_vanishes_from_non_admin_listings_but_stays_usable( sql_session_factory, ) -> None: """管理端隐藏开关的 store 语义：hidden=True 后从所有非 admin 列表消失 （list_visible / 各分页 scope / selector / 私密广场），但单条访问 （get_visible）不受影响，admin 列表仍可见且可按 hidden 筛选。""" await _add_sql_agent(sql_session_factory, agent_id="builtin-1", user_id=None) await _add_sql_agent(sql_session_factory, agent_id="owned-a", user_id="user-a") await _add_sql_agent(sql_session_factory, agent_id="published-b", user_id="user-b") await _add_sql_hidden_flagged_agent(sql_session_factory, agent_id="flagged-1", user_id="user-b") store = AgentRepository(sql_session_factory) # list_visible（默认列表）：flagged-1 消失 visible_ids = {r["id"] for r in await store.list_visible("user-a")} assert visible_ids == {"builtin-1", "owned-a", "published-b"} # 非_admin 分页（默认 scope 与 square/mine scope 一致排除） for scope in (None, "square", "mine"): rows, total = await store.list_paginated( user_id="user-a", scope=scope, search=None, tag_ids=None, page=1, page_size=20, ) assert "flagged-1" not in {r["id"] for r in rows}, scope # 属主自己也看不到自己被隐藏的智能体（非 admin 一律过滤） rows, _ = await store.list_paginated( user_id="user-b", scope="mine", search=None, tag_ids=None, page=1, page_size=20, ) assert "flagged-1" not in {r["id"] for r in rows} # selector（featured）与私密广场同样排除 await store.set_featured("flagged-1", 1) featured_ids = {r["id"] for r in await store.list_featured("user-a")} assert "flagged-1" not in featured_ids assert await store.update_any("flagged-1", {"is_private": True}) is not None private_ids = {r["id"] for r in await store.list_private()} assert "flagged-1" not in private_ids # 单条访问不受影响（隐藏不是禁用：既有会话/详情继续可用） assert await store.get_visible("flagged-1", "user-a") is not None assert await store.get_visible("flagged-1", "user-b") is not None # admin：全量可见，且可按 hidden 筛选 rows, total = await store.list_paginated( user_id="user-a", scope=None, search=None, tag_ids=None, page=1, page_size=20, admin=True, ) assert total == 4 and "flagged-1" in {r["id"] for r in rows} rows, total = await store.list_paginated( user_id="user-a", scope=None, search=None, tag_ids=None, page=1, page_size=20, admin=True, hidden=True, ) assert total == 1 and {r["id"] for r in rows} == {"flagged-1"} rows, total = await store.list_paginated( user_id="user-a", scope=None, search=None, tag_ids=None, page=1, page_size=20, admin=True, hidden=False, ) assert total == 3 and "flagged-1" not in {r["id"] for r in rows} # 开关往返：取消隐藏后回到列表。注：flagged-1 中途被挪进了私密广场 #（is_private=True），普通列表本就不含私密条目，先移回公共广场再断言。 assert (await store.update_any("flagged-1", {"hidden": False}))["hidden"] is False await store.update_any("flagged-1", {"is_private": False}) visible_ids = {r["id"] for r in await store.list_visible("user-a")} assert "flagged-1" in visible_ids @pytest.mark.asyncio async def test_memory_hidden_agent_filtered_from_non_admin_listings() -> None: """内存版 store 的 hidden 行为与 SQL 版一致。""" store = MemoryAgentStore() await store.ensure_builtin({"id": "builtin-1", "name": "Built In"}) await store.create({"id": "owned-a", "name": "Owned A", "user_id": "user-a"}) await store.create( {"id": "flagged-1", "name": "Flagged", "user_id": "user-b", "published": True, "hidden": True} ) assert {r["id"] for r in await store.list_visible("user-a")} == {"builtin-1", "owned-a"} rows, total = await store.list_paginated( user_id="user-a", scope=None, search=None, tag_ids=None, page=1, page_size=20, ) assert total == 2 and "flagged-1" not in {r["id"] for r in rows} # 单条访问 + admin 可见 + hidden 筛选 assert await store.get_visible("flagged-1", "user-a") is not None rows, total = await store.list_paginated( user_id="user-a", scope=None, search=None, tag_ids=None, page=1, page_size=20, admin=True, ) assert total == 3 and "flagged-1" in {r["id"] for r in rows} rows, total = await store.list_paginated( user_id="user-a", scope=None, search=None, tag_ids=None, page=1, page_size=20, admin=True, hidden=True, ) assert total == 1 and {r["id"] for r in rows} == {"flagged-1"} # 开关往返 assert (await store.update_any("flagged-1", {"hidden": False}))["hidden"] is False assert "flagged-1" in {r["id"] for r in await store.list_visible("user-a")} @pytest.mark.asyncio async def test_agent_store_only_owner_can_update_or_delete() -> None: store = MemoryAgentStore() await store.ensure_builtin({"id": "builtin-agent", "name": "Built in"}) await store.create({"id": "private-a", "name": "Owned", "user_id": "user-a"}) assert await store.update("builtin-agent", "user-a", {"name": "Nope"}) is None assert await store.update("private-a", "user-b", {"name": "Nope"}) is None updated = await store.update("private-a", "user-a", {"name": "Renamed"}) assert updated is not None assert updated["name"] == "Renamed" assert await store.delete("private-a", "user-b") is False assert await store.delete("private-a", "user-a") is True @pytest.mark.asyncio async def test_delete_builtin_only_targets_builtins() -> None: store = MemoryAgentStore() await store.ensure_builtin({"id": "builtin-agent", "name": "Built in"}) await store.create({"id": "owned-agent", "name": "Owned", "user_id": "user-a"}) # delete_builtin refuses to touch a user-owned agent assert await store.delete_builtin("owned-agent") is False assert await store.get_any("owned-agent") is not None # delete_builtin succeeds on a built-in assert await store.delete_builtin("builtin-agent") is True assert await store.get_any("builtin-agent") is None # And no-ops on a missing id assert await store.delete_builtin("does-not-exist") is False @pytest.mark.asyncio async def test_update_builtin_only_targets_builtins() -> None: store = MemoryAgentStore() await store.ensure_builtin({"id": "builtin-agent", "name": "Built in"}) await store.create({"id": "owned-agent", "name": "Owned", "user_id": "user-a"}) # update_builtin succeeds on a built-in agent updated = await store.update_builtin("builtin-agent", {"name": "Renamed by admin"}) assert updated is not None assert updated["name"] == "Renamed by admin" # update_builtin refuses to touch a user-owned agent (returns None) assert await store.update_builtin("owned-agent", {"name": "Hijacked"}) is None # And on a missing agent assert await store.update_builtin("does-not-exist", {"name": "Ghost"}) is None @pytest.mark.asyncio async def test_get_editable_or_raise_admin_can_edit_builtin() -> None: """Route helper: admin gets edit access to any agent; non-admin still 403.""" from fastapi import HTTPException from app.gateway.routers.agents import _get_editable_or_raise store = MemoryAgentStore() await store.ensure_builtin({"id": "builtin-agent", "name": "Built in"}) await store.create({"id": "owned-agent", "name": "Owned", "user_id": "user-a"}) # Owner edits their own agent → returns ('owner', record) record, mode = await _get_editable_or_raise( store, "owned-agent", "user-a", is_admin=False ) assert mode == "owner" and record["id"] == "owned-agent" # Admin edits a built-in agent → returns ('admin', record) record, mode = await _get_editable_or_raise( store, "builtin-agent", "user-a", is_admin=True ) assert mode == "admin" and record["id"] == "builtin-agent" # Non-admin tries to edit a built-in → 403 with pytest.raises(HTTPException) as exc: await _get_editable_or_raise(store, "builtin-agent", "user-a", is_admin=False) assert exc.value.status_code == 403 # Admin edits someone else's custom agent → returns ('admin', record). # Behavior changed: admins manage the full agent catalog (built-in + any # user's agents), not just built-ins. Non-admins are still owner-only. await store.create( {"id": "published-other", "name": "Pub", "user_id": "user-x", "published": True} ) record, mode = await _get_editable_or_raise( store, "published-other", "user-a", is_admin=True ) assert mode == "admin" and record["id"] == "published-other" def test_build_run_config_injects_custom_assistant_as_agent_id() -> None: config = build_run_config("thread-1", None, None, assistant_id="agent-123") assert config["configurable"]["thread_id"] == "thread-1" assert config["configurable"]["agent_id"] == "agent-123" assert "agent_name" not in config["configurable"] def test_write_agent_config_persists_skills_into_config_yaml(isolated_paths) -> None: agent_id = "agent-001" _write_agent_config( agent_id, name="中文助手", description="desc", skills=["s1", "s2"], model="gpt-x", tool_groups=["tg-a"], ) config_file = paths_module.get_paths().agent_dir(agent_id) / "config.yaml" raw = yaml.safe_load(config_file.read_text(encoding="utf-8")) assert raw["id"] == agent_id assert raw["name"] == "中文助手" assert raw["description"] == "desc" assert raw["skills"] == ["s1", "s2"] assert raw["model"] == "gpt-x" assert raw["model_prompt_suffix_enabled"] is True assert raw["tool_groups"] == ["tg-a"] def test_write_agent_config_persists_prompt_suffix_switch(isolated_paths) -> None: _write_agent_config( "agent-suffix-off", name="助手", description="", model_prompt_suffix_enabled=False, ) assert _load_config_extras("agent-suffix-off")["model_prompt_suffix_enabled"] is False def test_load_config_extras_returns_values_written_to_yaml(isolated_paths) -> None: agent_id = "agent-002" _write_agent_config( agent_id, name="agent-002", description="", skills=["s1"], model=None, tool_groups=None, ) extras = _load_config_extras(agent_id) assert extras == { "skills": ["s1"], "model": None, "model_prompt_suffix_enabled": True, "tool_groups": None, "common_questions": None, "template_name": None, "seat_skill_directive": None, "llmwiki_knowledge_base_ids": None, "chat_link": None, } def test_load_config_extras_missing_directory_returns_none_triple(isolated_paths) -> None: extras = _load_config_extras("does-not-exist") assert extras == { "skills": None, "model": None, "model_prompt_suffix_enabled": True, "tool_groups": None, "common_questions": None, "template_name": None, "seat_skill_directive": None, "llmwiki_knowledge_base_ids": None, "chat_link": None, } def test_write_agent_config_persists_common_questions(isolated_paths) -> None: agent_id = "agent-cq-1" # Structured {title, prompt} questions round-trip as clean dict mappings. questions = [ {"title": "能做什么", "prompt": "这个智能体能做什么？"}, {"title": "上传文件", "prompt": "如何上传文件？"}, ] _write_agent_config(agent_id, name="助手", description="", common_questions=questions) config_file = paths_module.get_paths().agent_dir(agent_id) / "config.yaml" raw = yaml.safe_load(config_file.read_text(encoding="utf-8")) assert raw["common_questions"] == questions extras = _load_config_extras(agent_id) assert [q.model_dump() for q in extras["common_questions"]] == questions def test_write_agent_config_coerces_legacy_string_questions(isolated_paths) -> None: """Legacy ``list[str]`` configs coerce to {title, prompt} with title=content.""" agent_id = "agent-cq-legacy" _write_agent_config( agent_id, name="助手", description="", common_questions=["如何上传文件？"] ) config_file = paths_module.get_paths().agent_dir(agent_id) / "config.yaml" raw = yaml.safe_load(config_file.read_text(encoding="utf-8")) assert raw["common_questions"] == [ {"title": "如何上传文件？", "prompt": "如何上传文件？"} ] def test_write_agent_config_omits_common_questions_when_none(isolated_paths) -> None: agent_id = "agent-cq-2" _write_agent_config(agent_id, name="x", description="") config_file = paths_module.get_paths().agent_dir(agent_id) / "config.yaml" raw = yaml.safe_load(config_file.read_text(encoding="utf-8")) assert "common_questions" not in raw assert _load_config_extras(agent_id)["common_questions"] is None def test_write_agent_config_explicit_empty_common_questions_is_persisted(isolated_paths) -> None: agent_id = "agent-cq-3" _write_agent_config(agent_id, name="x", description="", common_questions=[]) assert _load_config_extras(agent_id)["common_questions"] == [] def test_write_agent_config_explicit_empty_skills_is_persisted(isolated_paths) -> None: agent_id = "agent-003" _write_agent_config(agent_id, name="x", description="", skills=[]) extras = _load_config_extras(agent_id) assert extras["skills"] == [] def test_write_agent_config_omits_skills_when_none(isolated_paths) -> None: agent_id = "agent-004" _write_agent_config(agent_id, name="x", description="") config_file = paths_module.get_paths().agent_dir(agent_id) / "config.yaml" raw = yaml.safe_load(config_file.read_text(encoding="utf-8")) assert "skills" not in raw assert "model" not in raw assert "tool_groups" not in raw assert "chat_link" not in raw def test_write_agent_config_persists_chat_link_and_drops_blank(isolated_paths) -> None: agent_id = "agent-chat-link" link = { "url": "https://example.test/desk", "open_mode": "self", "route_params": [ {"key": "userName", "paramType": "login", "source": "userName"}, {"key": "theme", "paramType": "theme", "themeValue": "red=red;dark-blue=dark"}, ], } _write_agent_config(agent_id, name="助手", description="", chat_link=link) extras = _load_config_extras(agent_id) assert extras["chat_link"]["url"] == "https://example.test/desk" assert extras["chat_link"]["open_mode"] == "self" assert extras["chat_link"]["route_params"][0]["key"] == "userName" assert extras["chat_link"]["route_params"][1]["paramType"] == "theme" _write_agent_config(agent_id, name="助手", description="", chat_link=None) raw = yaml.safe_load((paths_module.get_paths().agent_dir(agent_id) / "config.yaml").read_text(encoding="utf-8")) assert "chat_link" not in raw assert _load_config_extras(agent_id)["chat_link"] is None def test_normalize_agent_chat_link_rejects_unsafe_urls() -> None: assert normalize_agent_chat_link({"url": ""}) is None assert normalize_agent_chat_link({"url": "javascript:alert(1)"}, strict=False) is None with pytest.raises(ValueError): normalize_agent_chat_link({"url": "javascript:alert(1)"}, strict=True) kept = normalize_agent_chat_link({"url": "/page/workspace/agents", "open_mode": "blank"}) assert kept is not None assert kept["url"] == "/page/workspace/agents" loaded = AgentConfig(name="x", chat_link={"url": "javascript:alert(1)"}) assert loaded.chat_link is None @pytest.mark.asyncio async def test_backfill_listing_cache_mirrors_chat_link() -> None: store = MemoryAgentStore() link = {"url": "https://example.test/a", "open_mode": "blank", "route_params": []} cfg = AgentConfig(id="a1", name="Agent", chat_link=link) await _backfill_listing_cache(store, [("a1", cfg)]) extras = await store.get_extras_for(["a1"]) assert extras["a1"]["chat_link"]["url"] == "https://example.test/a" await _backfill_listing_cache(store, [("a1", cfg)]) assert (await store.get_extras_for(["a1"]))["a1"]["chat_link"]["url"] == "https://example.test/a" @pytest.mark.asyncio async def test_backfill_listing_cache_mirrors_config_values() -> None: store = MemoryAgentStore() cfg = AgentConfig(id="a1", name="Agent", skills=["s1", "s2"], tool_groups=["g1"], model="m1") await _backfill_listing_cache(store, [("a1", cfg)]) extras = await store.get_extras_for(["a1"]) assert extras["a1"]["skills"] == ["s1", "s2"] assert extras["a1"]["tool_groups"] == ["g1"] assert extras["a1"]["model"] == "m1" @pytest.mark.asyncio async def test_backfill_listing_cache_is_idempotent_and_skips_unchanged(monkeypatch) -> None: store = MemoryAgentStore() cfg = AgentConfig(id="a1", name="Agent", skills=["s1"], tool_groups=None, model=None) await _backfill_listing_cache(store, [("a1", cfg)]) # On a second run with identical config, no re-write should occur. calls = {"replace": 0, "set": 0} orig_replace = store.replace_skills orig_set = store.set_extras async def _replace(agent_id, skills): calls["replace"] += 1 await orig_replace(agent_id, skills) async def _set(agent_id, *, tool_groups, model): calls["set"] += 1 await orig_set(agent_id, tool_groups=tool_groups, model=model) monkeypatch.setattr(store, "replace_skills", _replace) monkeypatch.setattr(store, "set_extras", _set) await _backfill_listing_cache(store, [("a1", cfg)]) assert calls == {"replace": 0, "set": 0} # Changing the config triggers a re-write. cfg2 = AgentConfig(id="a1", name="Agent", skills=["s1", "s2"], tool_groups=None, model=None) await _backfill_listing_cache(store, [("a1", cfg2)]) assert calls["replace"] == 1 @pytest.mark.asyncio async def test_paginated_agent_list_coalesces_concurrent_reads_and_returns_copies() -> None: _agent_list_cache.clear() _agent_list_inflight.clear() calls = 0 async def loader() -> AgentsListResponse: nonlocal calls calls += 1 await asyncio.sleep(0.01) return AgentsListResponse(agents=[], total=3, page=1, page_size=20) first, second = await asyncio.gather( _cached_paginated_agent_list(("user", 1), loader), _cached_paginated_agent_list(("user", 1), loader), ) assert calls == 1 assert {first[1], second[1]} == {"未命中", "合并等待"} first[0].total = 99 cached, cache_status = await _cached_paginated_agent_list(("user", 1), loader) assert cache_status == "命中" assert cached.total == 3 assert calls == 1 @pytest.mark.asyncio async def test_agent_owner_usernames_are_batched_and_cached(monkeypatch) -> None: from app.gateway import deps as deps_module _agent_owner_name_cache.clear() class Provider: def __init__(self): self.calls = [] async def get_users_by_ids(self, user_ids): self.calls.append(list(user_ids)) return [SimpleNamespace(id="owner-1", email="alice@example.com")] provider = Provider() monkeypatch.setattr(deps_module, "get_local_provider", lambda: provider) assert await _resolve_usernames({"owner-1", "deleted-user"}) == {"owner-1": "alice"} assert await _resolve_usernames({"owner-1", "deleted-user"}) == {"owner-1": "alice"} assert provider.calls == [["deleted-user", "owner-1"]]