180 lines
6.9 KiB
Python
180 lines
6.9 KiB
Python
"""Regression coverage for default and per-user sidebar menu layouts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from fastapi import HTTPException
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app.gateway.routers import menu_overrides as menu_router
|
|
from deerflow.persistence.base import Base
|
|
from deerflow.persistence.menu_overrides import MemoryMenuOverrideStore, MenuOverrideRepository
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def store(tmp_path):
|
|
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'menu-overrides.db'}")
|
|
async with engine.begin() as conn:
|
|
import deerflow.persistence.menu_overrides.model # noqa: F401
|
|
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
sf = async_sessionmaker(engine, expire_on_commit=False)
|
|
try:
|
|
yield MenuOverrideRepository(sf)
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sql_personal_snapshot_roundtrip_and_reset(store: MenuOverrideRepository):
|
|
personal = [
|
|
{"node_id": "old", "parent_id": None, "sort_order": 0, "label": "我的菜单", "disabled": False},
|
|
{"node_id": "child", "parent_id": "old", "sort_order": 0, "label": "子菜单", "disabled": True},
|
|
]
|
|
saved = await store.replace_user_overrides("user-1", personal, updated_by="user-1")
|
|
assert saved is not None
|
|
assert {row["node_id"] for row in saved} == {"old", "child"}
|
|
assert (await store.list_user_overrides("user-1"))[1]["disabled"] is True
|
|
assert await store.list_user_overrides("user-2") is None
|
|
|
|
assert await store.replace_user_overrides("user-1", [], updated_by="user-1") is None
|
|
assert await store.list_user_overrides("user-1") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_effective_menu_keeps_personal_nodes_and_inherits_new_default(monkeypatch):
|
|
store = MemoryMenuOverrideStore()
|
|
await store.replace_all(
|
|
[
|
|
{"node_id": "existing", "parent_id": None, "sort_order": 0, "label": "管理员新名称", "disabled": True},
|
|
{"node_id": "new", "parent_id": None, "sort_order": 1, "label": "管理员新增", "disabled": False},
|
|
],
|
|
updated_by="admin",
|
|
)
|
|
await store.replace_user_overrides(
|
|
"user-1",
|
|
[
|
|
{"node_id": "existing", "parent_id": None, "sort_order": 3, "label": "我的旧名称", "disabled": False},
|
|
],
|
|
updated_by="user-1",
|
|
)
|
|
|
|
async def current_user(_request):
|
|
return SimpleNamespace(id="user-1", system_role="user")
|
|
|
|
monkeypatch.setattr(menu_router, "get_optional_user_from_request", current_user)
|
|
response = await menu_router._read_response(SimpleNamespace(), store)
|
|
effective = {row.node_id: row for row in response.overrides}
|
|
|
|
assert response.editable_scope == "user"
|
|
assert response.has_user_override is True
|
|
assert effective["existing"].label == "我的旧名称"
|
|
assert effective["existing"].disabled is False
|
|
assert effective["new"].label == "管理员新增"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_reads_default_and_does_not_consume_personal_snapshot(monkeypatch):
|
|
store = MemoryMenuOverrideStore()
|
|
await store.replace_all(
|
|
[{"node_id": "menu", "parent_id": None, "sort_order": 0, "label": "默认", "disabled": False}]
|
|
)
|
|
await store.replace_user_overrides(
|
|
"admin-1",
|
|
[{"node_id": "menu", "parent_id": None, "sort_order": 0, "label": "个人", "disabled": False}],
|
|
)
|
|
|
|
async def current_admin(_request):
|
|
return SimpleNamespace(id="admin-1", system_role="admin")
|
|
|
|
monkeypatch.setattr(menu_router, "get_optional_user_from_request", current_admin)
|
|
response = await menu_router._read_response(SimpleNamespace(), store)
|
|
assert response.editable_scope == "default"
|
|
assert response.has_user_override is False
|
|
assert response.overrides[0].label == "默认"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_user_put_and_reset_only_touch_personal_snapshot(monkeypatch):
|
|
store = MemoryMenuOverrideStore()
|
|
await store.replace_all(
|
|
[{"node_id": "menu", "parent_id": None, "sort_order": 0, "label": "管理员默认", "disabled": False}],
|
|
updated_by="admin-1",
|
|
)
|
|
request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(menu_override_store=store)))
|
|
|
|
async def current_user(_request):
|
|
return SimpleNamespace(id="user-1", system_role="user")
|
|
|
|
monkeypatch.setattr(menu_router, "get_optional_user_from_request", current_user)
|
|
body = menu_router.MenuOverrideReplaceRequest(
|
|
overrides=[
|
|
menu_router.MenuOverride(
|
|
node_id="menu",
|
|
parent_id=None,
|
|
sort_order=0,
|
|
label="我的菜单",
|
|
disabled=False,
|
|
)
|
|
]
|
|
)
|
|
saved = await menu_router.replace_menu_overrides(request, body)
|
|
assert saved.editable_scope == "user"
|
|
assert saved.overrides[0].label == "我的菜单"
|
|
assert (await store.list_overrides())[0]["label"] == "管理员默认"
|
|
|
|
reset = await menu_router.reset_menu_overrides(request)
|
|
assert reset.has_user_override is False
|
|
assert reset.overrides[0].label == "管理员默认"
|
|
assert await store.list_user_overrides("user-1") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_auth_never_falls_through_to_default_admin(monkeypatch):
|
|
async def no_user(_request):
|
|
return None
|
|
|
|
monkeypatch.setattr(menu_router, "get_optional_user_from_request", no_user)
|
|
monkeypatch.setattr(menu_router, "is_auth_disabled", lambda: False)
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await menu_router._actor(SimpleNamespace())
|
|
assert exc_info.value.status_code == 401
|
|
|
|
|
|
def test_personal_snapshot_rejects_parent_missing_from_snapshot():
|
|
rows = [
|
|
menu_router.MenuOverride(
|
|
node_id="child",
|
|
parent_id="unknown-third-level-parent",
|
|
sort_order=0,
|
|
label="子菜单",
|
|
disabled=False,
|
|
)
|
|
]
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
menu_router._validate_depth(rows, require_known_parents=True)
|
|
assert exc_info.value.status_code == 400
|
|
|
|
|
|
def test_admin_sparse_override_uses_full_hierarchy_for_depth_validation():
|
|
override = menu_router.MenuOverride(
|
|
node_id="moved",
|
|
parent_id="level-3",
|
|
sort_order=0,
|
|
label=None,
|
|
disabled=False,
|
|
)
|
|
hierarchy = [
|
|
menu_router.MenuHierarchyNode(node_id="level-1", parent_id=None),
|
|
menu_router.MenuHierarchyNode(node_id="level-2", parent_id="level-1"),
|
|
menu_router.MenuHierarchyNode(node_id="level-3", parent_id="level-2"),
|
|
menu_router.MenuHierarchyNode(node_id="moved", parent_id="level-3"),
|
|
]
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
menu_router._validate_depth([override], hierarchy=hierarchy)
|
|
assert exc_info.value.status_code == 400
|
|
assert "超过 3 级" in str(exc_info.value.detail)
|