fix(cron): refresh official category for official-gateway MCP servers

Author: cszmzh

src/dashboard/apigateway/apigateway/apps/mcp_server/tasks.py

@@ -29,7 +29,7 @@
     ADD_STAGE_MCP_SERVER_PERMISSIONS_BEFORE_RELEASE_UPDATE_TASK_NAME,
     RECONCILE_STAGE_MCP_SERVER_PERMISSIONS_AFTER_RELEASE_TASK_NAME,
 )
-from apigateway.apps.mcp_server.models import MCPServer
+from apigateway.apps.mcp_server.models import MCPServer, MCPServerCategory
 from apigateway.biz.mcp_server import (
     MCPServerHandler,
     MCPServerPromptHandler,
@@ -444,3 +444,25 @@ def sync_mcp_server_after_release(
             stage_id,
             release_history_id,
         )
+
+
+@shared_task(name="apigateway.apps.mcp_server.tasks.refresh_official_mcp_server_category", ignore_result=True)
+def refresh_official_mcp_server_category() -> None:
+    """定时给官方网关下的 MCP Server 补齐「官方」分类，只补不删。"""
+    official_category = MCPServerCategory.objects.filter(name="Official", is_active=True).first()
+    if official_category is None:
+        return
+
+    missing_ids = list(
+        MCPServer.objects.filter(gateway__is_official=True)
+        .exclude(categories=official_category)
+        .values_list("id", flat=True)
+    )
+    if not missing_ids:
+        return
+
+    batch_size = 500
+    for start in range(0, len(missing_ids), batch_size):
+        official_category.mcp_servers.add(*missing_ids[start : start + batch_size])
+
+    logger.info("refresh_official_mcp_server_category completed, added=%d", len(missing_ids))

src/dashboard/apigateway/apigateway/conf/celery_conf.py

@@ -83,6 +83,10 @@
         "task": "apigateway.apps.mcp_server.tasks.sync_mcp_server_prompts",
         "schedule": crontab(minute="*/10"),
     },
+    "apigateway.apps.mcp_server.tasks.refresh_official_mcp_server_category": {
+        "task": "apigateway.apps.mcp_server.tasks.refresh_official_mcp_server_category",
+        "schedule": crontab(minute=0),
+    },
 }
 
 CELERY_CHORD_UNLOCK_MAX_RETRIES = 60

src/dashboard/apigateway/apigateway/tests/apps/mcp_server/test_tasks.py

@@ -16,23 +16,32 @@
 # to the current version of the project delivered to anyone in the future.
 #
 from unittest.mock import Mock, call, patch
+from uuid import uuid4
 
 import pytest
+from celery.schedules import crontab
 from ddf import G
 
-from apigateway.apps.mcp_server.models import MCPServer
+from apigateway.apps.mcp_server.constants import (
+    FEATURED_MCP_CATEGORY_NAME,
+    OFFICIAL_MCP_CATEGORY_NAME,
+    MCPServerStatusEnum,
+)
+from apigateway.apps.mcp_server.models import MCPServer, MCPServerCategory
 from apigateway.apps.mcp_server.tasks import (
     _fetch_updated_prompts,
     add_stage_mcp_server_permissions_before_release_update,
     reconcile_stage_mcp_server_permissions_after_release,
+    refresh_official_mcp_server_category,
     sync_mcp_server_after_release,
 )
+from apigateway.conf.celery_conf import CELERY_BEAT_SCHEDULE
 from apigateway.core.constants import (
     PublishEventNameTypeEnum,
     PublishEventStatusEnum,
     ReleaseHistoryStatusEnum,
 )
-from apigateway.core.models import PublishEvent, Release, ReleaseHistory, ResourceVersion
+from apigateway.core.models import Gateway, PublishEvent, Release, ReleaseHistory, ResourceVersion, Stage
 
 
 class TestFetchUpdatedPrompts:
@@ -367,3 +376,146 @@ def test_save_exception_logged(self, fake_gateway, fake_stage, release_history):
                 release_history_id=release_history.id,
                 mcp_servers_data=mcp_data,
             )
+
+
+class TestRefreshOfficialMcpServerCategory:
+    def _category(self, name: str, display_name: str, *, is_active: bool = True) -> MCPServerCategory:
+        category, _ = MCPServerCategory.objects.get_or_create(
+            name=name,
+            defaults={"display_name": display_name, "is_active": is_active},
+        )
+        if category.is_active != is_active or category.display_name != display_name:
+            category.is_active = is_active
+            category.display_name = display_name
+            category.save(update_fields=["is_active", "display_name"])
+        return category
+
+    def _gateway(self, *, is_official: bool) -> tuple[Gateway, Stage]:
+        gateway = G(Gateway, is_official=is_official)
+        return gateway, G(Stage, gateway=gateway)
+
+    def _server(self, gateway: Gateway, stage: Stage, **kwargs) -> MCPServer:
+        return G(
+            MCPServer,
+            gateway=gateway,
+            stage=stage,
+            name=f"mcp-{uuid4().hex[:12]}",
+            **kwargs,
+        )
+
+    def _category_names(self, mcp_server: MCPServer) -> set[str]:
+        return set(mcp_server.categories.values_list("name", flat=True))
+
+    def test_adds_official_category_to_all_servers_on_official_gateway(self):
+        official = self._category(OFFICIAL_MCP_CATEGORY_NAME, "官方资源")
+        gateway, stage = self._gateway(is_official=True)
+        public_active = self._server(
+            gateway,
+            stage,
+            is_public=True,
+            status=MCPServerStatusEnum.ACTIVE.value,
+        )
+        private_inactive = self._server(
+            gateway,
+            stage,
+            is_public=False,
+            status=MCPServerStatusEnum.INACTIVE.value,
+        )
+
+        refresh_official_mcp_server_category()
+
+        assert official in public_active.categories.all()
+        assert official in private_inactive.categories.all()
+
+    def test_does_not_duplicate_existing_official_category(self):
+        official = self._category(OFFICIAL_MCP_CATEGORY_NAME, "官方资源")
+        gateway, stage = self._gateway(is_official=True)
+        mcp_server = self._server(gateway, stage)
+        mcp_server.categories.add(official)
+
+        refresh_official_mcp_server_category()
+
+        assert list(mcp_server.categories.all()) == [official]
+
+    def test_keeps_official_category_on_non_official_gateway(self):
+        official = self._category(OFFICIAL_MCP_CATEGORY_NAME, "官方资源")
+        gateway, stage = self._gateway(is_official=False)
+        mcp_server = self._server(gateway, stage)
+        mcp_server.categories.add(official)
+
+        refresh_official_mcp_server_category()
+
+        assert official in mcp_server.categories.all()
+
+    def test_does_not_add_official_category_to_non_official_gateway(self):
+        self._category(OFFICIAL_MCP_CATEGORY_NAME, "官方资源")
+        gateway, stage = self._gateway(is_official=False)
+        mcp_server = self._server(gateway, stage)
+
+        refresh_official_mcp_server_category()
+
+        assert not mcp_server.categories.filter(name=OFFICIAL_MCP_CATEGORY_NAME).exists()
+
+    def test_preserves_featured_and_business_categories(self):
+        official = self._category(OFFICIAL_MCP_CATEGORY_NAME, "官方资源")
+        featured = self._category(FEATURED_MCP_CATEGORY_NAME, "精选推荐")
+        devops = G(
+            MCPServerCategory,
+            name=f"DevOps-{uuid4().hex[:8]}",
+            display_name="持续交付",
+            is_active=True,
+        )
+        gateway, stage = self._gateway(is_official=True)
+        mcp_server = self._server(gateway, stage)
+        mcp_server.categories.set([featured, devops])
+
+        refresh_official_mcp_server_category()
+
+        assert self._category_names(mcp_server) == {
+            official.name,
+            featured.name,
+            devops.name,
+        }
+
+    def test_skips_when_official_category_missing(self):
+        MCPServerCategory.objects.filter(name=OFFICIAL_MCP_CATEGORY_NAME).delete()
+        gateway, stage = self._gateway(is_official=True)
+        mcp_server = self._server(gateway, stage)
+
+        refresh_official_mcp_server_category()
+
+        assert not mcp_server.categories.exists()
+
+    def test_skips_when_official_category_inactive(self):
+        official = self._category(OFFICIAL_MCP_CATEGORY_NAME, "官方资源", is_active=False)
+        gateway, stage = self._gateway(is_official=True)
+        mcp_server = self._server(gateway, stage)
+
+        refresh_official_mcp_server_category()
+
+        assert official not in mcp_server.categories.all()
+
+    def test_does_not_update_mcp_server_business_fields(self):
+        self._category(OFFICIAL_MCP_CATEGORY_NAME, "官方资源")
+        gateway, stage = self._gateway(is_official=True)
+        mcp_server = self._server(
+            gateway,
+            stage,
+            title="keep-title",
+            description="keep-desc",
+            is_public=False,
+        )
+        original_updated_time = mcp_server.updated_time
+
+        refresh_official_mcp_server_category()
+        mcp_server.refresh_from_db()
+
+        assert mcp_server.title == "keep-title"
+        assert mcp_server.description == "keep-desc"
+        assert mcp_server.is_public is False
+        assert mcp_server.updated_time == original_updated_time
+
+    def test_hourly_schedule_is_registered(self):
+        entry = CELERY_BEAT_SCHEDULE["apigateway.apps.mcp_server.tasks.refresh_official_mcp_server_category"]
+        assert entry["task"] == "apigateway.apps.mcp_server.tasks.refresh_official_mcp_server_category"
+        assert entry["schedule"] == crontab(minute=0)