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