"""Tests for compact/sidebar-summary menu overrides (简介模式菜单). Overrides are stored per ``menu_type`` (A / B) so two website deployments can keep independent left-nav layouts. Saving type B must not wipe type A. """ from __future__ import annotations from types import SimpleNamespace import pytest import pytest_asyncio from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from app.gateway.routers import compact_menu_overrides as compact_router from app.gateway.routers.compact_menu_overrides import _merge_overrides from deerflow.persistence.base import Base from deerflow.persistence.compact_menu_overrides import ( CompactMenuOverrideRepository, MemoryCompactMenuOverrideStore, normalize_compact_menu_type, ) from deerflow.persistence.compact_menu_overrides.keys import public_node_id, storage_node_id @pytest_asyncio.fixture async def store(tmp_path): engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'compact-menu.db'}") async with engine.begin() as conn: import deerflow.persistence.compact_menu_overrides.model # noqa: F401 await conn.run_sync(Base.metadata.create_all) sf = async_sessionmaker(engine, expire_on_commit=False) return CompactMenuOverrideRepository(sf) def test_normalize_compact_menu_type(): assert normalize_compact_menu_type(None) == "A" assert normalize_compact_menu_type("") == "A" assert normalize_compact_menu_type("a") == "A" assert normalize_compact_menu_type("default") == "A" assert normalize_compact_menu_type("B") == "B" assert normalize_compact_menu_type("alt") == "B" assert normalize_compact_menu_type("2") == "B" def test_storage_keys_keep_type_a_unprefixed(): assert storage_node_id("A", "workspace-chats") == "workspace-chats" assert storage_node_id("B", "workspace-chats") == "B::workspace-chats" assert public_node_id("B::workspace-chats", "B") == "workspace-chats" assert public_node_id("workspace-chats", "A") == "workspace-chats" @pytest.mark.asyncio async def test_replace_all_roundtrips(store: CompactMenuOverrideRepository): out = await store.replace_all( [ { "node_id": "workspace-chats", "placement": "settings", "sort_order": 1, "label": "会话", "disabled": True, }, {"node_id": "workspace-agents", "placement": "sidebar", "sort_order": 0, "label": None}, {"node_id": "workspace-skills", "placement": "toolbox", "sort_order": 0, "label": "技能工具"}, ], updated_by="admin-1", ) assert len(out) == 3 rows = {r["node_id"]: r for r in await store.list_overrides()} assert rows["workspace-chats"]["label"] == "会话" assert rows["workspace-chats"]["placement"] == "settings" assert rows["workspace-chats"]["menu_type"] == "A" assert rows["workspace-chats"]["disabled"] is True assert rows["workspace-agents"]["updated_by"] == "admin-1" assert rows["workspace-skills"]["placement"] == "toolbox" @pytest.mark.asyncio async def test_replace_all_is_full_replacement_within_type(store: CompactMenuOverrideRepository): await store.replace_all([{"node_id": "a"}, {"node_id": "b"}], menu_type="A") await store.replace_all([{"node_id": "c"}], menu_type="A") ids = {r["node_id"] for r in await store.list_overrides("A")} assert ids == {"c"} @pytest.mark.asyncio async def test_types_are_isolated(store: CompactMenuOverrideRepository): await store.replace_all( [{"node_id": "workspace-chats", "placement": "sidebar", "sort_order": 0, "label": "对话A"}], menu_type="A", ) await store.replace_all( [{"node_id": "workspace-chats", "placement": "settings", "sort_order": 2, "label": "对话B"}], menu_type="B", ) a_rows = await store.list_overrides("A") b_rows = await store.list_overrides("B") assert len(a_rows) == 1 assert a_rows[0]["node_id"] == "workspace-chats" assert a_rows[0]["label"] == "对话A" assert a_rows[0]["placement"] == "sidebar" assert len(b_rows) == 1 assert b_rows[0]["node_id"] == "workspace-chats" assert b_rows[0]["label"] == "对话B" assert b_rows[0]["placement"] == "settings" await store.replace_all([], menu_type="B") assert await store.list_overrides("B") == [] leftover = await store.list_overrides("A") assert len(leftover) == 1 assert leftover[0]["label"] == "对话A" @pytest.mark.asyncio async def test_omitted_menu_type_reads_legacy_rows(): """Old pages that do not send menu_type still see the original compact menu.""" mem = MemoryCompactMenuOverrideStore() mem._rows["workspace-chats"] = { "node_id": "workspace-chats", "placement": "settings", "sort_order": 1, "label": "旧对话", } rows = await mem.list_overrides(None) assert len(rows) == 1 assert rows[0]["node_id"] == "workspace-chats" assert rows[0]["label"] == "旧对话" assert rows[0]["menu_type"] == "A" assert await mem.list_overrides("B") == [] @pytest.mark.asyncio async def test_sql_blank_menu_type_is_treated_as_original(store: CompactMenuOverrideRepository): from sqlalchemy import update from deerflow.persistence.compact_menu_overrides.model import CompactMenuOverrideRow await store.replace_all( [{"node_id": "legacy", "placement": "sidebar", "sort_order": 0, "label": "旧菜单"}], ) async with store._sf() as session: await session.execute(update(CompactMenuOverrideRow).values(menu_type="")) await session.commit() rows = await store.list_overrides(None) assert {r["node_id"] for r in rows} == {"legacy"} assert rows[0]["label"] == "旧菜单" assert await store.list_overrides("B") == [] @pytest.mark.asyncio async def test_memory_store_parity(): mem = MemoryCompactMenuOverrideStore() await mem.replace_all( [{"node_id": "x", "placement": "settings", "sort_order": 3}], menu_type="B", updated_by="u", ) await mem.replace_all([{"node_id": "y", "sort_order": 0}], menu_type="A") b_rows = await mem.list_overrides("B") assert b_rows[0]["node_id"] == "x" assert b_rows[0]["menu_type"] == "B" assert b_rows[0]["placement"] == "settings" a_ids = {r["node_id"] for r in await mem.list_overrides("A")} assert a_ids == {"y"} @pytest.mark.asyncio async def test_personal_compact_snapshots_are_isolated_by_user_and_type(store: CompactMenuOverrideRepository): await store.replace_user_overrides( "user-1", [ { "node_id": "chat", "placement": "toolbox", "sort_order": 0, "label": "我的对话", "disabled": True, } ], menu_type="A", updated_by="user-1", ) await store.replace_user_overrides( "user-1", [{"node_id": "chat", "placement": "settings", "sort_order": 0, "label": "B 对话"}], menu_type="B", updated_by="user-1", ) assert (await store.list_user_overrides("user-1", "A"))[0]["placement"] == "toolbox" assert (await store.list_user_overrides("user-1", "A"))[0]["disabled"] is True assert (await store.list_user_overrides("user-1", "B"))[0]["placement"] == "settings" assert await store.list_user_overrides("user-2", "A") is None assert await store.replace_user_overrides("user-1", [], menu_type="A") is None assert await store.list_user_overrides("user-1", "A") is None assert (await store.list_user_overrides("user-1", "B"))[0]["label"] == "B 对话" def test_personal_compact_nodes_win_while_new_default_nodes_flow_through(): default_rows = [ {"node_id": "existing", "placement": "settings", "sort_order": 0, "label": "管理员名称"}, {"node_id": "new", "placement": "toolbox", "sort_order": 1, "label": "新增菜单"}, ] personal_rows = [ { "node_id": "existing", "placement": "sidebar", "sort_order": 3, "label": "我的名称", "disabled": True, }, ] merged = {row["node_id"]: row for row in _merge_overrides(default_rows, personal_rows)} assert merged["existing"]["label"] == "我的名称" assert merged["existing"]["placement"] == "sidebar" assert merged["existing"]["disabled"] is True assert merged["new"]["label"] == "新增菜单" @pytest.mark.asyncio async def test_user_compact_put_does_not_replace_admin_default(monkeypatch): store = MemoryCompactMenuOverrideStore() await store.replace_all( [{"node_id": "chat", "placement": "settings", "sort_order": 0, "label": "默认对话"}], menu_type="A", updated_by="admin-1", ) request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(compact_menu_override_store=store))) async def current_user(_request): return SimpleNamespace(id="user-1", system_role="user") monkeypatch.setattr(compact_router, "get_optional_user_from_request", current_user) body = compact_router.CompactMenuOverrideReplaceRequest( menu_type="A", overrides=[ compact_router.CompactMenuOverride( node_id="chat", placement="toolbox", sort_order=0, label="我的对话", ) ], ) saved = await compact_router.replace_compact_menu_overrides(request, body) assert saved.editable_scope == "user" assert saved.overrides[0].placement == "toolbox" assert (await store.list_overrides("A"))[0]["placement"] == "settings"