fix: 补齐知识库元数据管理与生命周期校验

ai_knowledge_base_v2/apps/backend/tests/test_knowledge_admin.py

@@ -0,0 +1,142 @@
+from __future__ import annotations
+
+import pytest
+from fastapi import HTTPException
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import Session
+from sqlalchemy.pool import StaticPool
+
+from app.api.admin_knowledge_lifecycle import batch_update_metadata, update_metadata
+from app.models import Base
+from app.models.admin import Admin, Role
+from app.models.knowledge import Knowledge, KnowledgeManifest, KnowledgeSourceSnapshot, KnowledgeVersion
+from app.schemas.knowledge import KnowledgeBatchMetadataUpdateRequest, KnowledgeMetadataUpdateRequest
+from app.services.knowledge_agent_service import KnowledgeAgentService
+from app.services.knowledge_pipeline_service import KnowledgePipelineService
+
+
+def _database() -> Session:
+    engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+    Base.metadata.create_all(engine)
+    return Session(engine)
+
+
+def _admin(role_code: str | None = None) -> Admin:
+    role = Role(id=10, code=role_code, name=role_code) if role_code else None
+    return Admin(id=1, username="admin", password="hash", name="管理员", status=1, role=role)
+
+
+def _knowledge(db: Session, knowledge_id: int = 1) -> Knowledge:
+    item = Knowledge(
+        id=knowledge_id,
+        name="原名称",
+        feishu_space_id="space-1",
+        feishu_node_id="node-1",
+        status=1,
+        knowledge_type="general",
+        review_mode="manual",
+        source_status="normal",
+        manifest_confirmed=1,
+    )
+    db.add(item)
+    db.flush()
+    snapshot = KnowledgeSourceSnapshot(
+        knowledge_id=item.id,
+        source_title="源标题",
+        source_content="源内容",
+        content_hash=f"hash-{knowledge_id}",
+        source_identifier="space-1/node-1",
+    )
+    db.add(snapshot)
+    db.flush()
+    version = KnowledgeVersion(
+        knowledge_id=item.id,
+        snapshot_id=snapshot.id,
+        version_no=1,
+        status="published",
+        quality_status="passed",
+        processing_rule_version="test-v1",
+    )
+    db.add(version)
+    db.flush()
+    item.current_version_id = version.id
+    db.add(
+        KnowledgeManifest(
+            knowledge_id=item.id,
+            version_id=version.id,
+            purpose="原用途",
+            applicable_questions="课程问题",
+            inapplicable_questions="天气",
+            core_topics="课程",
+            content_hash=f"manifest-{knowledge_id}",
+            confirmed=1,
+        )
+    )
+    db.commit()
+    return item
+
+
+def test_metadata_update_preserves_source_version_and_open_state(monkeypatch: pytest.MonkeyPatch):
+    monkeypatch.setattr("app.api.admin_knowledge_lifecycle.KnowledgeCatalogCacheService.invalidate", lambda: None)
+    with _database() as db:
+        item = _knowledge(db)
+        original = (item.feishu_space_id, item.feishu_node_id, item.current_version_id, item.status)
+        update_metadata(
+            item.id,
+            KnowledgeMetadataUpdateRequest(name="新后台名称", knowledgeType="course", reviewMode="automatic"),
+            db,
+            _admin(),
+        )
+        db.refresh(item)
+        assert (item.name, item.knowledge_type, item.review_mode) == ("新后台名称", "course", "automatic")
+        assert (item.feishu_space_id, item.feishu_node_id, item.current_version_id, item.status) == original
+        assert KnowledgeAgentService.get_knowledge_catalog(db)[0]["name"] == "新后台名称"
+
+
+def test_metadata_update_requires_super_admin():
+    with _database() as db:
+        item = _knowledge(db)
+        with pytest.raises(HTTPException) as exc:
+            update_metadata(
+                item.id,
+                KnowledgeMetadataUpdateRequest(name="新名称", knowledgeType="qa", reviewMode="manual"),
+                db,
+                _admin("operator"),
+            )
+        assert exc.value.status_code == 403
+
+
+def test_batch_metadata_update_reports_missing_items(monkeypatch: pytest.MonkeyPatch):
+    monkeypatch.setattr("app.api.admin_knowledge_lifecycle.KnowledgeCatalogCacheService.invalidate", lambda: None)
+    with _database() as db:
+        item = _knowledge(db)
+        response = batch_update_metadata(
+            KnowledgeBatchMetadataUpdateRequest(knowledgeIds=[item.id, 999], knowledgeType="qa"),
+            db,
+            _admin(),
+        )
+        assert response["data"]["success"] == 1
+        assert response["data"]["failed"] == 1
+        assert db.scalar(select(Knowledge).where(Knowledge.id == item.id)).knowledge_type == "qa"
+
+
+def test_rollback_switches_published_version_statuses():
+    with _database() as db:
+        item = _knowledge(db)
+        old_version = db.get(KnowledgeVersion, item.current_version_id)
+        snapshot = db.scalar(select(KnowledgeSourceSnapshot).where(KnowledgeSourceSnapshot.knowledge_id == item.id))
+        target = KnowledgeVersion(
+            knowledge_id=item.id,
+            snapshot_id=snapshot.id,
+            version_no=0,
+            status="superseded",
+            quality_status="passed",
+            processing_rule_version="test-v1",
+        )
+        db.add(target)
+        db.flush()
+        KnowledgePipelineService.rollback(db, item, target, admin_id=1, keep_open=True)
+        db.commit()
+        assert item.current_version_id == target.id
+        assert target.status == "published"
+        assert old_version.status == "superseded"