199 lines
10 KiB
Python
199 lines
10 KiB
Python
"""Configuration-cache hits, isolation, revocation and cross-catalog invalidation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from test_api_read_cache import FakeRedis, _config
|
|
|
|
from app.gateway import api_read_cache
|
|
from app.gateway.catalog_cache import cached_catalog
|
|
from app.gateway.routers import compact_menu_overrides, html_templates, light_apps, menu_overrides, models, tags, task_buttons
|
|
from deerflow.persistence.compact_menu_overrides import MemoryCompactMenuOverrideStore
|
|
from deerflow.persistence.light_apps import MemoryLightAppStore
|
|
from deerflow.persistence.menu_overrides import MemoryMenuOverrideStore
|
|
from deerflow.persistence.tags import MemoryTagStore
|
|
from deerflow.persistence.task_buttons import MemoryTaskButtonStore
|
|
|
|
|
|
@pytest.fixture
|
|
def cached_request(monkeypatch):
|
|
redis = FakeRedis()
|
|
monkeypatch.setattr("deerflow.persistence.engine.get_session_factory", lambda: None)
|
|
monkeypatch.setattr(api_read_cache, "_api_read_cache_settings_cache", None)
|
|
monkeypatch.setattr(api_read_cache, "_client", lambda *_args, **_kwargs: (redis, "test:catalog"))
|
|
request = SimpleNamespace(
|
|
app=SimpleNamespace(state=SimpleNamespace(config=_config())),
|
|
state=SimpleNamespace(user=SimpleNamespace(id="user-1", system_role="user")),
|
|
)
|
|
|
|
async def current_user(req):
|
|
return req.state.user
|
|
|
|
for router in (menu_overrides, compact_menu_overrides, light_apps, task_buttons, html_templates):
|
|
monkeypatch.setattr(router, "get_optional_user_from_request", current_user)
|
|
for router in (light_apps, task_buttons):
|
|
monkeypatch.setattr(router, "get_current_user", AsyncMock(return_value="admin"))
|
|
return request, redis
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_personal_menu_cache_isolates_users_roles_and_checks_auth(cached_request):
|
|
request, _ = cached_request
|
|
store = request.app.state.menu_override_store = MemoryMenuOverrideStore()
|
|
await store.replace_all([{"node_id": "n", "label": "default"}])
|
|
await store.replace_user_overrides("user-1", [{"node_id": "n", "label": "private"}])
|
|
store.list_overrides = AsyncMock(wraps=store.list_overrides)
|
|
assert (await menu_overrides.list_menu_overrides(request)).overrides[0].label == "private"
|
|
await menu_overrides.list_menu_overrides(request)
|
|
assert store.list_overrides.await_count == 1
|
|
|
|
request.state.user = SimpleNamespace(id="user-2", system_role="user")
|
|
assert (await menu_overrides.list_menu_overrides(request)).overrides[0].label == "default"
|
|
request.state.user = SimpleNamespace(id="user-1", system_role="admin")
|
|
assert (await menu_overrides.list_menu_overrides(request)).editable_scope == "default"
|
|
request.state.user = None
|
|
with pytest.raises(HTTPException) as exc:
|
|
await menu_overrides.list_menu_overrides(request)
|
|
assert exc.value.status_code == 401
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_default_menu_edit_invalidates_all_personal_views(cached_request):
|
|
request, _ = cached_request
|
|
store = request.app.state.menu_override_store = MemoryMenuOverrideStore()
|
|
await store.replace_all([{"node_id": "n", "label": "old"}])
|
|
assert (await menu_overrides.list_menu_overrides(request)).overrides[0].label == "old"
|
|
request.state.user = SimpleNamespace(id="admin", system_role="admin")
|
|
await menu_overrides.replace_menu_overrides(request, menu_overrides.MenuOverrideReplaceRequest(
|
|
overrides=[menu_overrides.MenuOverride(node_id="n", label="new")],
|
|
hierarchy=[menu_overrides.MenuHierarchyNode(node_id="n")],
|
|
))
|
|
request.state.user = SimpleNamespace(id="user-1", system_role="user")
|
|
assert (await menu_overrides.list_menu_overrides(request)).overrides[0].label == "new"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_compact_menu_cache_isolates_variant_and_reset(cached_request):
|
|
request, _ = cached_request
|
|
store = request.app.state.compact_menu_override_store = MemoryCompactMenuOverrideStore()
|
|
await store.replace_all([{"node_id": "n", "label": "A"}], menu_type="A")
|
|
await store.replace_all([{"node_id": "n", "label": "B"}], menu_type="B")
|
|
assert (await compact_menu_overrides.list_compact_menu_overrides(request, "A")).overrides[0].label == "A"
|
|
assert (await compact_menu_overrides.list_compact_menu_overrides(request, "B")).overrides[0].label == "B"
|
|
await compact_menu_overrides.replace_compact_menu_overrides(request, compact_menu_overrides.CompactMenuOverrideReplaceRequest(
|
|
menu_type="B", overrides=[compact_menu_overrides.CompactMenuOverride(node_id="n", label="personal")],
|
|
))
|
|
assert (await compact_menu_overrides.list_compact_menu_overrides(request, "B")).overrides[0].label == "personal"
|
|
await compact_menu_overrides.reset_compact_menu_overrides(request, "B")
|
|
assert (await compact_menu_overrides.list_compact_menu_overrides(request, "B")).overrides[0].label == "B"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_light_app_filters_detail_and_tags_are_invalidated(cached_request):
|
|
request, _ = cached_request
|
|
request.state.user.system_role = "admin"
|
|
request.app.state.light_app_store = MemoryLightAppStore()
|
|
tag_store = request.app.state.tag_store = MemoryTagStore()
|
|
app = await light_apps.create_light_app(request, light_apps.LightAppCreateRequest(
|
|
name="One", app_type="iframe", url="https://example.test", mount_location="root",
|
|
))
|
|
query = dict(app_type=None, mount_location=None, enabled_only=False, search=None, tag_ids=None)
|
|
assert len((await light_apps.list_light_apps(request, **query)).apps) == 1
|
|
assert not (await light_apps.list_light_apps(request, **{**query, "search": "missing"})).apps
|
|
assert (await light_apps.get_light_app(request, app.app_id)).name == "One"
|
|
tag = await tags.create_tag(tags.TagCreateRequest(name="Topic", tag_type="light_app"), request, tag_store)
|
|
await tags.set_resource_tags("light_app", app.app_id, tags.TagAssignmentRequest(tag_ids=[tag.id]), request, tag_store)
|
|
assert (await light_apps.get_light_app(request, app.app_id)).tags[0]["name"] == "Topic"
|
|
await tags.update_tag(tag.id, tags.TagUpdateRequest(name="Renamed"), request, tag_store)
|
|
assert (await light_apps.list_light_apps(request, **query)).apps[0].tags[0]["name"] == "Renamed"
|
|
await light_apps.delete_light_app(request, app.app_id)
|
|
assert not (await light_apps.list_light_apps(request, **query)).apps
|
|
with pytest.raises(HTTPException) as exc:
|
|
await light_apps.get_light_app(request, app.app_id)
|
|
assert exc.value.status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_writes_while_disabled_do_not_resurrect_old_buttons(cached_request):
|
|
request, _ = cached_request
|
|
request.state.user.system_role = "admin"
|
|
store = request.app.state.task_button_store = MemoryTaskButtonStore()
|
|
await store.replace_all([{"id": "b", "business": "rwfx", "label": "old"}])
|
|
assert (await task_buttons.list_task_buttons(request)).buttons[0].label == "old"
|
|
api_read_cache.set_api_read_cache_settings_local(False, 300)
|
|
await task_buttons.replace_task_buttons(request, task_buttons.TaskButtonReplaceRequest(
|
|
buttons=[task_buttons.TaskButton(id="b", business="rwfx", label="new")],
|
|
))
|
|
api_read_cache.set_api_read_cache_settings_local(True, 300)
|
|
assert (await task_buttons.list_task_buttons(request)).buttons[0].label == "new"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_html_edit_invalidates_metadata_content_and_list(cached_request, monkeypatch):
|
|
request, _ = cached_request
|
|
request.state.user.system_role = "admin"
|
|
state = {"html": "old"}
|
|
monkeypatch.setattr(html_templates.template_store, "read_template_html", lambda _id: state["html"])
|
|
# content 端点默认走引擎的「完整文档」读取(结构化模板会组合默认数据)。
|
|
monkeypatch.setattr(html_templates.template_fill, "read_template_text", lambda _id: state["html"])
|
|
monkeypatch.setattr(html_templates.template_store, "list_templates", lambda: [{"name": state["html"]}])
|
|
monkeypatch.setattr(
|
|
html_templates.template_store, "get_template_meta", lambda _id: {"id": _id, "name": state["html"]}
|
|
)
|
|
|
|
def update(_id, **kwargs):
|
|
state["html"] = kwargs["html"]
|
|
return {"name": state["html"]}
|
|
|
|
monkeypatch.setattr(html_templates.template_store, "update_template", update)
|
|
assert (await html_templates.get_html_template_content("one", request))["html"] == "old"
|
|
await html_templates.list_html_templates(request)
|
|
await html_templates.get_html_template("one", request)
|
|
await html_templates.update_html_template("one", html_templates.TemplateUpdateRequest(html="new"), request)
|
|
assert (await html_templates.get_html_template_content("one", request))["html"] == "new"
|
|
assert (await html_templates.get_html_template("one", request))["name"] == "new"
|
|
assert (await html_templates.list_html_templates(request))["templates"][0]["name"] == "new"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_models_cache_capability_reads_and_do_not_cache_failures(cached_request):
|
|
request, _ = cached_request
|
|
store = request.app.state.model_capability_store = SimpleNamespace(list_capabilities=AsyncMock(
|
|
side_effect=[RuntimeError("unavailable"), [{"model_name": "one", "supports_reasoning_effort": True}]],
|
|
))
|
|
assert await models._reasoning_capabilities(request) == {}
|
|
assert (await models._reasoning_capabilities(request))["one"]["supports_reasoning_effort"] is True
|
|
await models._reasoning_capabilities(request)
|
|
assert store.list_capabilities.await_count == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_redis_outage_falls_back_to_config_store(cached_request, monkeypatch):
|
|
request, redis = cached_request
|
|
request.app.state.task_button_store = MemoryTaskButtonStore()
|
|
monkeypatch.setattr(redis, "get", AsyncMock(side_effect=ConnectionError("offline")))
|
|
assert (await task_buttons.list_task_buttons(request)).buttons == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_catalogue_burst_shares_one_database_fill(cached_request):
|
|
request, _ = cached_request
|
|
calls = 0
|
|
|
|
async def load():
|
|
nonlocal calls
|
|
calls += 1
|
|
await asyncio.sleep(0.03)
|
|
return {"buttons": ["shared"]}
|
|
|
|
results = await asyncio.gather(*(
|
|
cached_catalog(request, "task_buttons", "all", load) for _ in range(30)
|
|
))
|
|
assert calls == 1
|
|
assert all(result == {"buttons": ["shared"]} for result in results)
|