feat(knowledge): 企业记忆分区——知识三表entity_id多租户隔离 (entity-memory-partition P1)

- knowledge_events/knowledge_summaries: entity_id NOT NULL DEFAULT 1 + 索引（企业记忆抽屉） - knowledge_articles: entity_id 可空——NULL=全局共享(CMA教学知识), 有值=企业专属 - knowledge_service 全链路 entity_id: extract_events(KPIAlert直过滤+OperationLog经user_entities归属)/_kpi_snapshot(修跨企业KPI泄漏)/generate_summary/get_last_summary/run_*/sync - api/knowledge.py 8端点 Depends(get_entity_id) 强制过滤；详情越权404 - api/knowledge_articles.py 全局∪本企业查询；企业专属越权404 - 生产DB三表ALTER+索引已完成; 存量38篇教学文章保持NULL全局共享 - 测试 test_security_knowledge_partition.py 8条(双实体互查越权/生成自动带entity/全局文章双可见) - 方案文档 docs/knowledge-entity-partition-plan-20260903.md（附评审决策点）
2026-09-03 23:00:15 +08:00
parent a3111a25f6
commit e22e1702f8
7 changed files with 461 additions and 60 deletions
@@ -0,0 +1,220 @@
+"""企业记忆分区（knowledge 侧 entity_id 多租户隔离）测试（2026-09-03 P1）
+
+覆盖任务 DoD：
+- knowledge_events / knowledge_summaries / knowledge_articles 三表 entity_id 隔离
+- 摘要列表/详情强制 entity 过滤（entity1 查不到 entity2 → 404/空集）
+- 摘要生成自动带 entity_id（写入当前账套抽屉）
+- 知识文章：全局共享(NULL) ∪ 本企业专属；他企专属 404
+- extract_events 按企业隔离（KPIAlert 直过滤 + OperationLog 经 user_entities 归属）
+"""
+import pytest
+from datetime import datetime, timedelta
+from fastapi.testclient import TestClient
+from sqlalchemy.orm import Session
+
+from app.models import KPIDefinition, KPIAlert, Entity
+from app.models.knowledge import KnowledgeSummary, KnowledgeEvent
+from app.models.knowledge_article import KnowledgeArticle
+from app.services import knowledge_service
+from tests.conftest import create_test_kpi, create_test_user, get_token_for_user, auth_header
+
+SUMMARY_FIELDS = dict(
+    level="daily", period_key="2026-09-01", content="测试摘要内容",
+    generated_by="test", is_stale=0,
+)
+
+
+def _ensure_entity(db: Session, eid: int, name: str = None) -> Entity:
+    ent = db.query(Entity).filter(Entity.id == eid).first()
+    if not ent:
+        ent = Entity(id=eid, name=name or f"企业{eid}", short_name=f"E{eid}", status="active")
+        db.add(ent)
+        db.commit()
+        db.refresh(ent)
+    return ent
+
+
+def _login_token(client: TestClient, username: str = "testadmin",
+                 password: str = "admin123", entity_id: int = 1) -> str:
+    """登录指定账套返回 token（conftest.get_token_for_user 固定 entity1，这里参数化）"""
+    resp = client.post("/api/cma/auth/login", json={
+        "username": username, "password": password, "entity_id": entity_id,
+    })
+    assert resp.status_code == 200, resp.text
+    return resp.json().get("token")
+
+
+# ============================================================
+# summaries 抽屉隔离
+# ============================================================
+class TestKnowledgeSummaryIsolation:
+    def test_list_summaries_entity_scoped(self, client: TestClient, db: Session):
+        """摘要列表按 entity 隔离：entity1 看不到 entity2 的摘要"""
+        _ensure_entity(db, 2)
+        s1 = KnowledgeSummary(**SUMMARY_FIELDS, title="E1摘要", entity_id=1)
+        s2 = KnowledgeSummary(**SUMMARY_FIELDS, title="E2专属摘要", entity_id=2)
+        db.add_all([s1, s2])
+        db.commit()
+        db.refresh(s1)
+        db.refresh(s2)
+
+        create_test_user(db)
+        token1 = _login_token(client, entity_id=1)
+        r1 = client.get("/api/cma/knowledge/summaries", headers=auth_header(token1))
+        assert r1.status_code == 200, r1.text
+        items = r1.json()["items"]
+        titles = [i["title"] for i in items]
+        assert any("摘要" in t for t in titles)
+        assert "E2专属摘要" not in titles
+
+    def test_get_summary_cross_entity_404(self, client: TestClient, db: Session):
+        """越权：entity1 直取 entity2 摘要详情 → 404（防跨企业探测）"""
+        _ensure_entity(db, 2)
+        s2 = KnowledgeSummary(**SUMMARY_FIELDS, title="E2秘密摘要", entity_id=2)
+        db.add(s2)
+        db.commit()
+        db.refresh(s2)
+
+        create_test_user(db)
+        token1 = _login_token(client, entity_id=1)
+        r = client.get(f"/api/cma/knowledge/summaries/{s2.id}", headers=auth_header(token1))
+        assert r.status_code == 404
+
+        # 反向：entity2 取 entity1 摘要同样 404
+        s1 = KnowledgeSummary(**SUMMARY_FIELDS, title="E1摘要", entity_id=1)
+        db.add(s1)
+        db.commit()
+        db.refresh(s1)
+        token2 = _login_token(client, entity_id=2)
+        r2 = client.get(f"/api/cma/knowledge/summaries/{s1.id}", headers=auth_header(token2))
+        assert r2.status_code == 404
+
+    def test_get_summary_own_entity_ok(self, client: TestClient, db: Session):
+        """同企业取自己摘要详情 → 200"""
+        s1 = KnowledgeSummary(**SUMMARY_FIELDS, title="E1自己的摘要", entity_id=1)
+        db.add(s1)
+        db.commit()
+        db.refresh(s1)
+        create_test_user(db)
+        token1 = _login_token(client, entity_id=1)
+        r = client.get(f"/api/cma/knowledge/summaries/{s1.id}", headers=auth_header(token1))
+        assert r.status_code == 200
+        assert r.json()["title"] == "E1自己的摘要"
+
+
+# ============================================================
+# 摘要生成自动带 entity_id
+# ============================================================
+class TestKnowledgeGenerateWritesEntity:
+    def test_generate_daily_writes_current_entity(self, client: TestClient, db: Session, monkeypatch):
+        """generate/daily 写入当前账套抽屉：entity1 触发 → 摘要 entity_id=1；entity2 触发 → =2"""
+        _ensure_entity(db, 2)
+
+        async def _fake_deepseek(prompt: str, timeout: int = 30) -> str:
+            return "测试摘要：本期无重大变化。"
+        monkeypatch.setattr(knowledge_service, "_call_deepseek", _fake_deepseek)
+
+        create_test_user(db)
+        token1 = _login_token(client, entity_id=1)
+        r1 = client.post("/api/cma/knowledge/generate/daily", headers=auth_header(token1))
+        assert r1.status_code == 200, r1.text
+        assert r1.json()["summary"]["entity_id"] == 1
+
+        token2 = _login_token(client, entity_id=2)
+        r2 = client.post("/api/cma/knowledge/generate/daily", headers=auth_header(token2))
+        assert r2.status_code == 200, r2.text
+        assert r2.json()["summary"]["entity_id"] == 2
+
+        # DB 回查：两账套抽屉各一条自己的
+        e1 = db.query(KnowledgeSummary).filter(KnowledgeSummary.entity_id == 1).count()
+        e2 = db.query(KnowledgeSummary).filter(KnowledgeSummary.entity_id == 2).count()
+        assert e1 >= 1
+        assert e2 >= 1
+
+
+# ============================================================
+# 知识文章：全局共享 ∪ 本企业专属
+# ============================================================
+class TestKnowledgeArticlesGlobalUnion:
+    def test_articles_global_union_scope(self, client: TestClient, db: Session):
+        """文章查询 = 本企业专属 ∪ 全局共享；他企专属不可见"""
+        _ensure_entity(db, 2)
+        g = KnowledgeArticle(title="全局教学文章", content="c", category="term", entity_id=None)
+        e1 = KnowledgeArticle(title="E1专属文章", content="c", category="term", entity_id=1)
+        e2 = KnowledgeArticle(title="E2专属文章", content="c", category="term", entity_id=2)
+        db.add_all([g, e1, e2])
+        db.commit()
+        for a in (g, e1, e2):
+            db.refresh(a)
+
+        create_test_user(db)
+        token1 = _login_token(client, entity_id=1)
+        r1 = client.get("/api/cma/knowledge-articles", headers=auth_header(token1))
+        assert r1.status_code == 200, r1.text
+        titles1 = {i["title"] for i in r1.json()["data"]}
+        assert "全局教学文章" in titles1          # 全局共享可见
+        assert "E1专属文章" in titles1            # 本企业专属可见
+        assert "E2专属文章" not in titles1        # 他企专属不可见
+
+        token2 = _login_token(client, entity_id=2)
+        r2 = client.get("/api/cma/knowledge-articles", headers=auth_header(token2))
+        titles2 = {i["title"] for i in r2.json()["data"]}
+        assert "全局教学文章" in titles2
+        assert "E2专属文章" in titles2
+        assert "E1专属文章" not in titles2
+
+    def test_get_article_cross_entity_404(self, client: TestClient, db: Session):
+        """越权：entity1 直取 entity2 专属文章 → 404；全局文章两账套可取"""
+        _ensure_entity(db, 2)
+        g = KnowledgeArticle(title="全局文章", content="c", entity_id=None)
+        e2 = KnowledgeArticle(title="E2秘密文章", content="c", entity_id=2)
+        db.add_all([g, e2])
+        db.commit()
+        db.refresh(g)
+        db.refresh(e2)
+
+        create_test_user(db)
+        token1 = _login_token(client, entity_id=1)
+        assert client.get(f"/api/cma/knowledge-articles/{g.id}", headers=auth_header(token1)).status_code == 200
+        assert client.get(f"/api/cma/knowledge-articles/{e2.id}", headers=auth_header(token1)).status_code == 404
+
+
+# ============================================================
+# extract_events 企业隔离（service 层精确断言）
+# ============================================================
+class TestExtractEventsIsolation:
+    def _seed_alerts(self, db: Session):
+        k1 = create_test_kpi(db, kpi_code="KE_E1", entity_id=1, status="active")
+        k2 = create_test_kpi(db, kpi_code="KE_E2", entity_id=2, status="active")
+        db.add(KPIAlert(kpi_id=k1.id, alert_level="red", alert_message="E1企业预警信息",
+                        status="pending", entity_id=1))
+        db.add(KPIAlert(kpi_id=k2.id, alert_level="red", alert_message="E2企业预警信息",
+                        status="pending", entity_id=2))
+        db.commit()
+        return k1, k2
+
+    def test_extract_events_only_own_entity(self, client: TestClient, db: Session):
+        """extract_events(entity_id=1) 只含 E1 的预警，不含 E2"""
+        _ensure_entity(db, 2)
+        self._seed_alerts(db)
+        since = datetime.utcnow() - timedelta(days=1)
+        ev1 = knowledge_service.extract_events(db, since, None, entity_id=1)
+        ev2 = knowledge_service.extract_events(db, since, None, entity_id=2)
+        txt1 = "\n".join(str(e) for e in ev1)
+        txt2 = "\n".join(str(e) for e in ev2)
+        assert "E1企业预警信息" in txt1
+        assert "E2企业预警信息" not in txt1   # 核心越权断言：E1 抽不到 E2
+        assert "E2企业预警信息" in txt2
+        assert "E1企业预警信息" not in txt2
+
+    def test_api_events_entity_scoped(self, client: TestClient, db: Session):
+        """API /knowledge/events 按 token 账套过滤事件"""
+        _ensure_entity(db, 2)
+        self._seed_alerts(db)
+        create_test_user(db)
+        token1 = _login_token(client, entity_id=1)
+        r = client.get("/api/cma/knowledge/events", headers=auth_header(token1))
+        assert r.status_code == 200, r.text
+        body = "\n".join(str(e) for e in r.json()["events"])
+        assert "E1企业预警信息" in body
+        assert "E2企业预警信息" not in body