deerflow-code/offline-backend-20260512/backend/tests/test_compact_menu_overrides.py

"""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

import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

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": "会话"},
            {"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-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"}