Hermes CI Fix
e22e1702f8
- 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（附评审决策点）
221 lines
10 KiB
Python
221 lines
10 KiB
Python
"""企业记忆分区（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
|