Hermes CI Fix
fb4609be14
ci/woodpecker/push/main Pipeline failed
- ① kpi_templates instantiate 构造点显式传会话账套（原缺 entity_id ⇒ 静默落 e1），kpi_code 判重按账套收紧
- ② kpi_subject_map 3 个读取点加护栏注释（不加列、不改语义）
- ③ subjects 分层：读口 union（NULL=全局标准准则科目 / N=账套私有），UPSERT 唯一键改 (entity_id, subject_code)
- ④ analysis_results 账套化：读/写/删三端点按会话账套判定归属（列未落地时 kpi_code JOIN 止血），
无归属行进 unattributed 登记、删除一律 404
- 模型层按现网 DDL 补齐：KPIDefinition(template_id/is_system + uk_entity_kpi)、AnalysisResult.entity_id、Subject.entity_id
- 新增 migrate_{subjects,analysis_results}_entity_id_20260923.py（--check/--apply/--down）、
22 条对账用例 test_entity_dimension_convergence_20260923.py
DDL 已由 ops 受控通道落地（subjects 21:31:56 / analysis_results 21:31:59 CST，审批 1e9b4d9a/5ffb1640）
本提交为切片提交，未 push（发布须另取放行）
613 lines
31 KiB
Python
613 lines
31 KiB
Python
"""账套维度收口批 · 2026-09-23 — 四段验收用例
|
||
|
||
任务：`0-P1-project-20260923-entity-dimension-convergence-fullstack`
|
||
（裁定件 `project-entity-dimension-ruling-20260923`）
|
||
|
||
A 段（零 DDL）
|
||
① `kpi_templates` instantiate 构造点显式挂会话账套 + `kpi_code` 判重按账套收紧
|
||
② `kpi_subject_map` 3 个读取点护栏注释（静态判据：`grep -n "账套维度经"` ≥3 命中）
|
||
③ `analysis_results` 读侧止血（`entity_id` 列未落地时按 `kpi_code` JOIN 判归属）
|
||
B 段（需 DDL，代码侧先落）
|
||
③ `subjects` 分层读取口径（全局 `entity_id IS NULL` ∪ 本账套私有）+ UPSERT 按账套判重
|
||
④ `analysis_results` 加列后的读/写/删三端点账套化
|
||
D 段：两个迁移脚本的旧形态/新形态/回填闸门/回滚/生产库守卫（SQLite 沙箱演练）
|
||
|
||
全部用例只跑内存 SQLite / 临时文件 SQLite（`tests/conftest.py` 机制），
|
||
**零生产库 / 零生产 Redis 连接**；迁移脚本用 `--url sqlite:///…` 沙箱，不碰 `cma` 库。
|
||
"""
|
||
import importlib
|
||
import json
|
||
import os
|
||
import sys
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy import create_engine, text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app import models as models_pkg
|
||
from app.api import analysis_results as ar_mod
|
||
from app.models import (
|
||
AnalysisResult, Entity, KPIDefinition, KPISubjectMap, KPITemplate, OperationLog, Subject,
|
||
)
|
||
from tests.conftest import TEST_USER_PASSWORD, auth_header, create_test_user
|
||
|
||
BACKEND = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
TEMPLATES = "/api/cma/templates"
|
||
ANALYSIS = "/api/cma/analysis"
|
||
SUBJECTS = "/api/cma/subjects"
|
||
BOT_SUBJECTS = "/api/cma/bot/subjects"
|
||
DATA_CLS = "/api/cma/data-classification"
|
||
BOT_KEY = {"X-BOT-KEY": "cma-bot-finance-2026"}
|
||
MIG_SUBJECTS = "scripts.migrate_subjects_entity_id_20260923"
|
||
MIG_ANALYSIS = "scripts.migrate_analysis_results_entity_id_20260923"
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 夹具
|
||
# ──────────────────────────────────────────────
|
||
|
||
def _ensure_entity(db: Session, eid: int, name: str, status: str = "active") -> Entity:
|
||
ent = db.query(Entity).filter(Entity.id == eid).first()
|
||
if not ent:
|
||
ent = Entity(id=eid, name=name, short_name=name, status=status)
|
||
db.add(ent)
|
||
db.commit()
|
||
db.refresh(ent)
|
||
return ent
|
||
|
||
|
||
def _seed_two_entities(db: Session):
|
||
_ensure_entity(db, 1, "陕西酣客文化传媒")
|
||
_ensure_entity(db, 2, "陕西博海网络科技")
|
||
|
||
|
||
def _login_as(client: TestClient, username: str, entity_id: int) -> str:
|
||
resp = client.post("/api/cma/auth/login", json={
|
||
"username": username, "password": TEST_USER_PASSWORD, "entity_id": entity_id,
|
||
})
|
||
assert resp.status_code == 200, f"登录 {username}@entity{entity_id} 失败: {resp.text[:300]}"
|
||
return resp.json()["token"]
|
||
|
||
|
||
def _two_tokens(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
create_test_user(db)
|
||
return _login_as(client, "testadmin", 1), _login_as(client, "testadmin", 2)
|
||
|
||
|
||
def _seed_kpi(db: Session, code: str, entity_id: int, name: str = "测试KPI") -> KPIDefinition:
|
||
kpi = KPIDefinition(kpi_code=code, kpi_name=name, dimension="finance",
|
||
unit="元", target_value=100.0, status="active", entity_id=entity_id)
|
||
db.add(kpi)
|
||
db.commit()
|
||
db.refresh(kpi)
|
||
return kpi
|
||
|
||
|
||
def _seed_analysis(db: Session, kpi_code, entity_id, conclusion: str, period: str = "2026-06"):
|
||
row = AnalysisResult(period=period, conclusion=conclusion, confidence=70,
|
||
kpi_code=kpi_code, entity_id=entity_id)
|
||
db.add(row)
|
||
db.commit()
|
||
db.refresh(row)
|
||
return row
|
||
|
||
|
||
def _seed_subject(db: Session, code: str, name: str, entity_id=None, remark=None) -> Subject:
|
||
s = Subject(subject_code=code, subject_name=name, entity_id=entity_id, remark=remark,
|
||
is_active=1, level=1)
|
||
db.add(s)
|
||
db.commit()
|
||
db.refresh(s)
|
||
return s
|
||
|
||
|
||
# ══════════════════════════════════════════════
|
||
# ① kpi_templates instantiate 构造点
|
||
# ══════════════════════════════════════════════
|
||
|
||
class TestTemplateInstantiateEntityScope:
|
||
|
||
def _seed_template(self, db: Session, code: str = "F_TPL_REV") -> KPITemplate:
|
||
t = KPITemplate(kpi_code=code, kpi_name="模板_营业收入", dimension="finance",
|
||
unit="万元", formula="SUM(收入)", formula_desc="收入合计",
|
||
target_value=100.0, is_system=1, usage_count=0)
|
||
db.add(t)
|
||
db.commit()
|
||
db.refresh(t)
|
||
return t
|
||
|
||
def _instantiate(self, client: TestClient, token: str, tid: int, code: str):
|
||
return client.post(f"{TEMPLATES}/{tid}/instantiate", headers=auth_header(token),
|
||
json={"kpi_code": code, "kpi_name": "实例KPI", "dimension": "finance",
|
||
"unit": "万元", "target_value": 100.0})
|
||
|
||
def test_each_entity_gets_its_own_row(self, client: TestClient, db: Session):
|
||
"""e1 / e2 各实例化一次（同码）→ 回库各 1 行且归属正确"""
|
||
t1, t2 = _two_tokens(client, db)
|
||
tpl = self._seed_template(db, "F_TPL_REV")
|
||
|
||
r1 = self._instantiate(client, t1, tpl.id, "F_TPL_REV")
|
||
assert r1.status_code == 200, r1.text
|
||
assert r1.json()["entity_id"] == 1
|
||
# e2 允许建同码（uk_entity_kpi 复合唯一；全表判重会让博海建不了同码 KPI）
|
||
r2 = self._instantiate(client, t2, tpl.id, "F_TPL_REV")
|
||
assert r2.status_code == 200, r2.text
|
||
assert r2.json()["entity_id"] == 2
|
||
|
||
db.expire_all()
|
||
rows = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == "F_TPL_REV").all()
|
||
assert len(rows) == 2, [r.id for r in rows]
|
||
assert sorted(r.entity_id for r in rows) == [1, 2]
|
||
|
||
def test_same_entity_duplicate_code_rejected(self, client: TestClient, db: Session):
|
||
"""同账套内同码 → 400（判重按账套收紧，不放过本账套重复）"""
|
||
t1, _ = _two_tokens(client, db)
|
||
tpl = self._seed_template(db, "F_TPL_DUP")
|
||
assert self._instantiate(client, t1, tpl.id, "F_TPL_DUP").status_code == 200
|
||
dup = self._instantiate(client, t1, tpl.id, "F_TPL_DUP")
|
||
assert dup.status_code == 400, dup.text
|
||
assert "在本账套已存在" in dup.text
|
||
db.expire_all()
|
||
assert db.query(KPIDefinition).filter(
|
||
KPIDefinition.kpi_code == "F_TPL_DUP", KPIDefinition.entity_id == 1).count() == 1
|
||
|
||
def test_constructor_carries_entity_id_in_source(self):
|
||
"""构造点显式传账套（防再次退回模型 default=1 的静默兜底）"""
|
||
src = open(os.path.join(BACKEND, "app/api/templates.py"), encoding="utf-8").read()
|
||
assert "entity_id=entity_id" in src
|
||
assert "KPIDefinition.kpi_code == kpi_code,\n KPIDefinition.entity_id == entity_id," in src
|
||
|
||
|
||
# ══════════════════════════════════════════════
|
||
# ② kpi_subject_map 护栏（静态判据）
|
||
# ══════════════════════════════════════════════
|
||
|
||
class TestKpiSubjectMapGuardrail:
|
||
|
||
GUARD_FILES = ("app/api/ontology.py", "app/utils/alert_attribution.py",
|
||
"scripts/kpi_value_source_derive.py")
|
||
|
||
def test_guard_comment_present_in_three_read_points(self):
|
||
hits = {}
|
||
for rel in self.GUARD_FILES:
|
||
src = open(os.path.join(BACKEND, rel), encoding="utf-8").read()
|
||
hits[rel] = src.count("账套维度经")
|
||
assert all(v >= 1 for v in hits.values()), hits
|
||
assert len(hits) == 3
|
||
|
||
def test_no_subject_code_reverse_lookup_introduced(self):
|
||
"""护栏语义：三个读取点都仍按 kpi_id 取数（不得出现按 subject_code 过滤的 kpi_subject_map 查询）"""
|
||
for rel in self.GUARD_FILES:
|
||
src = open(os.path.join(BACKEND, rel), encoding="utf-8").read()
|
||
assert "KPISubjectMap.subject_code ==" not in src, rel
|
||
|
||
|
||
# ══════════════════════════════════════════════
|
||
# ③④ analysis_results 账套化（列口径 + join 止血口径）
|
||
# ══════════════════════════════════════════════
|
||
|
||
class TestAnalysisResultEntityScope:
|
||
|
||
def _seed(self, db: Session):
|
||
_seed_kpi(db, "F_AN_1", 1, "酣客指标")
|
||
_seed_kpi(db, "F_AN_2", 2, "博海指标")
|
||
_seed_kpi(db, "F_AN_SHARED", 1, "同码A")
|
||
_seed_kpi(db, "F_AN_SHARED", 2, "同码B")
|
||
r1 = _seed_analysis(db, "F_AN_1", 1, "酣客结论")
|
||
r2 = _seed_analysis(db, "F_AN_2", 2, "博海结论")
|
||
r3 = _seed_analysis(db, None, None, "无归属（kpi_code 为空）", period="2026-07")
|
||
r4 = _seed_analysis(db, "F_AN_SHARED", None, "归属二义（同码跨账套）", period="2026-08")
|
||
r5 = _seed_analysis(db, "F_AN_NOPE", None, "库内无该码", period="2026-09")
|
||
return r1, r2, r3, r4, r5
|
||
|
||
def _ids(self, payload, key="results"):
|
||
return sorted(x["id"] for x in payload[key])
|
||
|
||
# ── 列口径（DDL 已落地） ──────────────────────────────
|
||
|
||
def test_get_two_tokens_intersection_empty(self, client: TestClient, db: Session):
|
||
r1, r2, r3, r4, r5 = self._seed(db)
|
||
t1, t2 = _two_tokens(client, db)
|
||
v1 = client.get(f"{ANALYSIS}/result", headers=auth_header(t1)).json()
|
||
v2 = client.get(f"{ANALYSIS}/result", headers=auth_header(t2)).json()
|
||
|
||
assert self._ids(v1) == [r1.id], v1
|
||
assert self._ids(v2) == [r2.id], v2
|
||
assert set(self._ids(v1)) & set(self._ids(v2)) == set()
|
||
assert v1["entity_id"] == 1 and v2["entity_id"] == 2
|
||
assert v1["entity_scope"] == "column"
|
||
# 无归属行（kpi_code 空 / 无定义 / 二义）对两账套都不进 results，只在登记池
|
||
assert sorted(x["id"] for x in v1["unattributed"]) == sorted([r3.id, r4.id, r5.id])
|
||
assert v1["unattributed"] == v2["unattributed"]
|
||
assert v1["unattributed_total"] == 3
|
||
assert all("结论" not in x for x in v1["unattributed"])
|
||
|
||
def test_get_join_stopgap_before_ddl(self, client: TestClient, db: Session, monkeypatch):
|
||
"""entity_id 列未落地时（DDL 前置）按 kpi_code JOIN 判归属，止血口径与列口径一致"""
|
||
r1, r2, r3, r4, r5 = self._seed(db)
|
||
t1, t2 = _two_tokens(client, db)
|
||
monkeypatch.setattr(ar_mod, "_has_entity_column", lambda db_: False)
|
||
|
||
v1 = client.get(f"{ANALYSIS}/result", headers=auth_header(t1)).json()
|
||
v2 = client.get(f"{ANALYSIS}/result", headers=auth_header(t2)).json()
|
||
assert v1["entity_scope"] == "kpi_join"
|
||
assert self._ids(v1) == [r1.id]
|
||
assert self._ids(v2) == [r2.id]
|
||
assert set(self._ids(v1)) & set(self._ids(v2)) == set()
|
||
# 二义码（F_AN_SHARED 属两账套）在 join 口径下不猜账套 → 登记池
|
||
assert sorted(x["id"] for x in v1["unattributed"]) == sorted([r3.id, r4.id, r5.id])
|
||
|
||
def test_delete_cross_entity_is_404_and_row_survives(self, client: TestClient, db: Session):
|
||
r1, r2, _, _, _ = self._seed(db)
|
||
t1, t2 = _two_tokens(client, db)
|
||
|
||
bad = client.delete(f"{ANALYSIS}/result/{r1.id}", headers=auth_header(t2))
|
||
assert bad.status_code == 404, bad.text
|
||
db.expire_all()
|
||
assert db.query(AnalysisResult).filter(AnalysisResult.id == r1.id).count() == 1, "对端行被删"
|
||
# 反向：e1 删 e2 的行同样 404
|
||
bad2 = client.delete(f"{ANALYSIS}/result/{r2.id}", headers=auth_header(t1))
|
||
assert bad2.status_code == 404
|
||
# 正对照：本账套删自己的行 → 200
|
||
ok = client.delete(f"{ANALYSIS}/result/{r1.id}", headers=auth_header(t1))
|
||
assert ok.status_code == 200, ok.text
|
||
db.expire_all()
|
||
assert db.query(AnalysisResult).filter(AnalysisResult.id == r1.id).count() == 0
|
||
|
||
def test_delete_unattributed_is_404(self, client: TestClient, db: Session):
|
||
_, _, r3, _, _ = self._seed(db)
|
||
t1, _ = _two_tokens(client, db)
|
||
resp = client.delete(f"{ANALYSIS}/result/{r3.id}", headers=auth_header(t1))
|
||
assert resp.status_code == 404
|
||
db.expire_all()
|
||
assert db.query(AnalysisResult).filter(AnalysisResult.id == r3.id).count() == 1
|
||
|
||
def test_post_writes_session_entity(self, client: TestClient, db: Session):
|
||
_two_tokens(client, db)
|
||
_seed_kpi(db, "F_AN_1", 1)
|
||
_seed_kpi(db, "F_AN_2", 2)
|
||
t1, t2 = _login_as(client, "testadmin", 1), _login_as(client, "testadmin", 2)
|
||
|
||
p2 = client.post(f"{ANALYSIS}/result", headers=auth_header(t2),
|
||
json={"period": "2026-09", "conclusion": "博海写入", "kpi_code": "F_AN_2"})
|
||
assert p2.status_code == 200, p2.text
|
||
assert p2.json()["entity_id"] == 2
|
||
assert p2.json()["entity_persisted"] == "column"
|
||
|
||
p1 = client.post(f"{ANALYSIS}/result", headers=auth_header(t1),
|
||
json={"period": "2026-09", "conclusion": "酣客写入", "kpi_code": "F_AN_1"})
|
||
assert p1.status_code == 200
|
||
db.expire_all()
|
||
new2 = db.query(AnalysisResult).filter(AnalysisResult.id == p2.json()["id"]).first()
|
||
assert new2.entity_id == 2, "create 未写会话账套"
|
||
# 交叉可见性：各自的 GET 只看到自己那一行
|
||
v1 = client.get(f"{ANALYSIS}/result", headers=auth_header(t1)).json()
|
||
v2 = client.get(f"{ANALYSIS}/result", headers=auth_header(t2)).json()
|
||
assert self._ids(v1) == [p1.json()["id"]]
|
||
assert self._ids(v2) == [p2.json()["id"]]
|
||
|
||
def test_post_pre_ddl_logs_entity_and_does_not_guess_column(self, client: TestClient, db: Session,
|
||
monkeypatch):
|
||
"""列未落地时：账套只落 OperationLog.detail，行内 entity_id 保持 NULL（不猜列）"""
|
||
_two_tokens(client, db)
|
||
_seed_kpi(db, "F_AN_1", 1)
|
||
t1 = _login_as(client, "testadmin", 1)
|
||
monkeypatch.setattr(ar_mod, "_has_entity_column", lambda db_: False)
|
||
|
||
resp = client.post(f"{ANALYSIS}/result", headers=auth_header(t1),
|
||
json={"period": "2026-09", "conclusion": "过渡期写入", "kpi_code": "F_AN_1"})
|
||
assert resp.status_code == 200, resp.text
|
||
assert resp.json()["entity_persisted"] == "operation_log"
|
||
db.expire_all()
|
||
row = db.query(AnalysisResult).filter(AnalysisResult.id == resp.json()["id"]).first()
|
||
assert row.entity_id is None
|
||
logs = db.query(OperationLog).filter(OperationLog.target_type == "analysis_result").all()
|
||
assert len(logs) == 1, [l.id for l in logs]
|
||
detail = logs[0].detail
|
||
detail = json.loads(detail) if isinstance(detail, str) else detail
|
||
assert detail["entity_id"] == 1
|
||
|
||
|
||
# ══════════════════════════════════════════════
|
||
# ③ subjects 分层（读取口径 + UPSERT 判重）
|
||
# ══════════════════════════════════════════════
|
||
|
||
class TestSubjectsLayering:
|
||
|
||
def _seed(self, db: Session):
|
||
_seed_subject(db, "1001", "库存现金", None) # 全局标准
|
||
_seed_subject(db, "1002", "银行存款", None) # 全局标准
|
||
_seed_subject(db, "100106", "酒窖备用金", 1, remark="zxt-sync") # 酣客私有
|
||
_seed_subject(db, "100107", "现金备用卡（唐）", 1, remark="zxt-sync")
|
||
_seed_subject(db, "200201", "博海专户", 2) # 博海私有
|
||
_seed_subject(db, "100199", "酣客同码科目", 1) # 同码不同账套
|
||
_seed_subject(db, "100199", "博海同码科目", 2)
|
||
|
||
def test_get_subjects_union_scope(self, client: TestClient, db: Session):
|
||
self._seed(db)
|
||
t1, t2 = _two_tokens(client, db)
|
||
v1 = client.get(SUBJECTS, headers=auth_header(t1)).json()
|
||
v2 = client.get(SUBJECTS, headers=auth_header(t2)).json()
|
||
c1 = {d["subject_code"] for d in v1["data"]}
|
||
c2 = {d["subject_code"] for d in v2["data"]}
|
||
|
||
assert {"1001", "1002", "100106", "100107", "100199"} <= c1
|
||
assert "200201" not in c1, "博海私有明细泄漏给酣客"
|
||
assert {"1001", "1002", "200201", "100199"} <= c2
|
||
assert "100106" not in c2 and "100107" not in c2, "酣客明细泄漏给博海"
|
||
# 同码不同账套 → 各看自己的名字
|
||
n1 = {d["subject_code"]: d["subject_name"] for d in v1["data"]}
|
||
n2 = {d["subject_code"]: d["subject_name"] for d in v2["data"]}
|
||
assert n1["100199"] == "酣客同码科目"
|
||
assert n2["100199"] == "博海同码科目"
|
||
# 全局标准科目两账套都在
|
||
assert {"1001", "1002"} <= c1 and {"1001", "1002"} <= c2
|
||
|
||
def test_data_classification_inventory_and_stats_scoped(self, client: TestClient, db: Session):
|
||
self._seed(db)
|
||
t1, t2 = _two_tokens(client, db)
|
||
inv1 = client.get(f"{DATA_CLS}/inventory", params={"data_type": "subject"},
|
||
headers=auth_header(t1)).json()
|
||
inv2 = client.get(f"{DATA_CLS}/inventory", params={"data_type": "subject"},
|
||
headers=auth_header(t2)).json()
|
||
codes = lambda v: {(i.get("subject_code") or i.get("code")) for i in (v.get("items") or v.get("data") or [])}
|
||
assert "100106" in codes(inv1) and "100106" not in codes(inv2)
|
||
assert "200201" in codes(inv2) and "200201" not in codes(inv1)
|
||
|
||
s1 = client.get(f"{DATA_CLS}/stats", headers=auth_header(t1)).json()
|
||
s2 = client.get(f"{DATA_CLS}/stats", headers=auth_header(t2)).json()
|
||
assert s1["subjects"]["total"] == 5, s1 # 2 全局 + 3 酣客私有
|
||
assert s2["subjects"]["total"] == 4, s2 # 2 全局 + 2 博海私有
|
||
assert s1["total"] == 5 and s2["total"] == 4 # 无 KPI 数据时总数为科目数
|
||
|
||
def test_bot_upsert_per_entity_judgement(self, client: TestClient, db: Session):
|
||
"""UPSERT 判重按 (entity_id, subject_code)：不同账套可各有同码，同账套内才 skip"""
|
||
self._seed(db)
|
||
_seed_two_entities(db)
|
||
body = {"subjects": [{"subject_code": "300301", "subject_name": "博海新专户",
|
||
"parent_code": None, "level": 1, "category": "asset", "is_active": 1}],
|
||
"update_existing": False}
|
||
h2 = dict(BOT_KEY); h2["X-Entity-Id"] = "2"
|
||
h1 = dict(BOT_KEY); h1["X-Entity-Id"] = "1"
|
||
|
||
r2 = client.post(BOT_SUBJECTS, headers=h2, json=body)
|
||
assert r2.status_code == 200, r2.text
|
||
assert r2.json()["inserted"] == 1 and r2.json()["entity_id"] == 2
|
||
r1 = client.post(BOT_SUBJECTS, headers=h1, json=body)
|
||
assert r1.status_code == 200, r1.text
|
||
assert r1.json()["inserted"] == 1, "同码但不同账套应可插入（按账套判重）"
|
||
# 同账套重复 → skip
|
||
again = client.post(BOT_SUBJECTS, headers=h1, json=body)
|
||
assert again.json()["inserted"] == 0 and again.json()["skipped"] == 1
|
||
|
||
db.expire_all()
|
||
rows = db.query(Subject).filter(Subject.subject_code == "300301").all()
|
||
assert sorted(r.entity_id for r in rows) == [1, 2]
|
||
# 全局科目（entity_id IS NULL）不被他账套 update 改写
|
||
s = client.post(BOT_SUBJECTS, headers=h1,
|
||
json={"subjects": [{"subject_code": "1001", "subject_name": "库存现金改名",
|
||
"parent_code": None, "level": 1,
|
||
"category": "asset", "is_active": 1}],
|
||
"update_existing": True})
|
||
assert s.json()["updated"] == 0, "全局标准科目不该被计入 updated（分层后只更新本账套私有行）"
|
||
assert s.json()["skipped_global"] == 1
|
||
db.expire_all()
|
||
glob = db.query(Subject).filter(Subject.subject_code == "1001",
|
||
Subject.entity_id.is_(None)).first()
|
||
assert glob.subject_name == "库存现金", "全局标准科目被账套 UPSERT 改写"
|
||
|
||
def test_bot_upsert_global_row_is_never_rewritten_nor_duplicated(self, client: TestClient, db: Session):
|
||
"""全局标准科目（entity_id IS NULL）：同码既不被账套 UPSERT 改写，也不插本账套副本
|
||
|
||
回归保护（本单 A③ 修过的一处缺陷）：原先 `existing` 把全局行也算「已存在」，
|
||
但 UPDATE 只匹配 `entity_id == 会话账套` ⇒ 对全局行「报 updated=1 却零改动」（响应失真）。
|
||
分层口径：全局行 → 计入 `skipped_global`，永不改写、永不重复插入。
|
||
"""
|
||
self._seed(db)
|
||
_seed_two_entities(db)
|
||
h1 = dict(BOT_KEY); h1["X-Entity-Id"] = "1"
|
||
need = {"subject_code": "1001", "subject_name": "库存现金（改名）", "parent_code": None,
|
||
"level": 1, "category": "asset", "is_active": 1}
|
||
|
||
for update_existing in (False, True):
|
||
r = client.post(BOT_SUBJECTS, headers=h1,
|
||
json={"subjects": [need], "update_existing": update_existing})
|
||
assert r.status_code == 200, r.text
|
||
d = r.json()
|
||
assert (d["inserted"], d["updated"], d["skipped_global"]) == (0, 0, 1), d
|
||
|
||
db.expire_all()
|
||
rows = db.query(Subject).filter(Subject.subject_code == "1001").all()
|
||
assert len(rows) == 1, "全局标准科目被插入了账套副本（同码重复）"
|
||
assert rows[0].entity_id is None and rows[0].subject_name == "库存现金", "全局标准科目被改写"
|
||
|
||
# ── dry_run 口径与真实执行一致 ──
|
||
def test_bot_upsert_dry_run_reports_global_skip(self, client: TestClient, db: Session):
|
||
self._seed(db)
|
||
_seed_two_entities(db)
|
||
h1 = dict(BOT_KEY); h1["X-Entity-Id"] = "1"
|
||
r = client.post(BOT_SUBJECTS, headers=h1,
|
||
json={"subjects": [{"subject_code": "1001", "subject_name": "库存现金",
|
||
"parent_code": None, "level": 1,
|
||
"category": "asset", "is_active": 1}],
|
||
"update_existing": True, "dry_run": True})
|
||
assert r.status_code == 200, r.text
|
||
d = r.json()
|
||
assert d["dry_run"] is True and d["would_update"] == 0 and d["would_skip_global"] == 1, d
|
||
|
||
def test_read_sites_carry_layer_filter(self):
|
||
"""四个读取口都必须带分层口径（防回退成整表读）"""
|
||
for rel in ("app/api/subjects.py", "app/api/data_classification.py",
|
||
"app/services/report_reclass.py", "app/api/ontology.py"):
|
||
src = open(os.path.join(BACKEND, rel), encoding="utf-8").read()
|
||
assert "Subject.entity_id.is_(None)" in src, rel
|
||
|
||
|
||
# ══════════════════════════════════════════════
|
||
# 迁移脚本：沙箱演练（SQLite 旧形态 → apply → backfill → down）
|
||
# ══════════════════════════════════════════════
|
||
|
||
OLD_SUBJECTS_DDL = """
|
||
CREATE TABLE subjects (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
subject_code VARCHAR(20) NOT NULL UNIQUE,
|
||
subject_name VARCHAR(200) NOT NULL,
|
||
parent_code VARCHAR(20),
|
||
level INTEGER DEFAULT 1,
|
||
category VARCHAR(50),
|
||
new_standard_category VARCHAR(20),
|
||
important_flag INTEGER DEFAULT 0,
|
||
data_level VARCHAR(20),
|
||
data_category VARCHAR(50),
|
||
is_active INTEGER DEFAULT 1,
|
||
remark VARCHAR(500),
|
||
created_at DATETIME,
|
||
updated_at DATETIME
|
||
)
|
||
"""
|
||
|
||
OLD_ANALYSIS_DDL = """
|
||
CREATE TABLE analysis_results (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
period VARCHAR(20) NOT NULL,
|
||
conclusion VARCHAR(1000) NOT NULL,
|
||
confidence INTEGER NOT NULL,
|
||
data_source VARCHAR(500),
|
||
calculation_logic VARCHAR(1000),
|
||
comparable_benchmark VARCHAR(500),
|
||
limitations VARCHAR(1000),
|
||
has_actual INTEGER DEFAULT 0,
|
||
has_target INTEGER DEFAULT 0,
|
||
has_trend INTEGER DEFAULT 0,
|
||
has_review INTEGER DEFAULT 0,
|
||
kpi_code VARCHAR(50),
|
||
kpi_name VARCHAR(200),
|
||
created_at DATETIME
|
||
)
|
||
"""
|
||
|
||
|
||
class TestMigrationScriptsSandbox:
|
||
|
||
def _run(self, mod, argv):
|
||
old = sys.argv
|
||
sys.argv = ["migrate"] + argv
|
||
try:
|
||
return mod.main()
|
||
finally:
|
||
sys.argv = old
|
||
|
||
@pytest.fixture
|
||
def subj_env(self, tmp_path):
|
||
sys.path.insert(0, BACKEND)
|
||
mod = importlib.import_module(MIG_SUBJECTS)
|
||
eng = create_engine(f"sqlite:///{tmp_path/'subj.db'}")
|
||
with eng.begin() as conn:
|
||
conn.execute(text(OLD_SUBJECTS_DDL))
|
||
for i in range(164):
|
||
conn.execute(text("INSERT INTO subjects (subject_code, subject_name, remark) "
|
||
f"VALUES ('9{i:05d}', '酣客明细{i}', 'zxt-sync')"))
|
||
for i in range(43):
|
||
conn.execute(text("INSERT INTO subjects (subject_code, subject_name) "
|
||
f"VALUES ('1{i:03d}', '标准科目{i}')"))
|
||
return mod, eng, tmp_path
|
||
|
||
@pytest.fixture
|
||
def analysis_env(self, tmp_path):
|
||
sys.path.insert(0, BACKEND)
|
||
mod = importlib.import_module(MIG_ANALYSIS)
|
||
eng = create_engine(f"sqlite:///{tmp_path/'ar.db'}")
|
||
with eng.begin() as conn:
|
||
conn.execute(text(OLD_ANALYSIS_DDL))
|
||
conn.execute(text("INSERT INTO analysis_results (period, conclusion, confidence, kpi_code) "
|
||
"VALUES ('2026-06','测试',70,'TEST001')"))
|
||
conn.execute(text("INSERT INTO analysis_results (period, conclusion, confidence, kpi_code) "
|
||
"VALUES ('2026-07','测试2',70,NULL)"))
|
||
return mod, eng, tmp_path
|
||
|
||
def _cols(self, eng, table):
|
||
return {r[1] for r in eng.connect().execute(text(f"PRAGMA table_info({table})")).fetchall()}
|
||
|
||
# ── subjects ──────────────────────────────────
|
||
|
||
def test_subjects_check_is_readonly(self, subj_env):
|
||
mod, eng, _ = subj_env
|
||
assert self._run(mod, ["--check", "--url", str(eng.url)]) == 0
|
||
assert "entity_id" not in self._cols(eng, "subjects")
|
||
with eng.connect() as conn:
|
||
assert conn.execute(text("SELECT COUNT(*) FROM subjects")).scalar() == 207
|
||
|
||
def test_subjects_apply_backfill_down_roundtrip(self, subj_env):
|
||
mod, eng, tmp = subj_env
|
||
url = str(eng.url)
|
||
assert self._run(mod, ["--apply", "--url", url]) == 0
|
||
assert "entity_id" in self._cols(eng, "subjects")
|
||
# 二跑幂等
|
||
assert self._run(mod, ["--apply", "--url", url]) == 0
|
||
with eng.connect() as conn:
|
||
names = {r[0] for r in conn.execute(
|
||
text("SELECT name FROM sqlite_master WHERE type='index'")).fetchall()}
|
||
assert "uk_entity_subject" in names and "ix_subjects_entity_id" in names
|
||
assert conn.execute(text("SELECT COUNT(*) FROM subjects WHERE entity_id IS NOT NULL")).scalar() == 0
|
||
|
||
# 回填闸门：无取证件 / 取证件行数不符 → 拒绝（fail-closed）
|
||
assert self._run(mod, ["--backfill", "--url", url]) == 3
|
||
bad = tmp / "ev_bad.json"
|
||
bad.write_text(json.dumps({"dama_evidence": "x", "finance_confirmation": "y",
|
||
"counts": {"e1": 1, "global_null": 43}}), encoding="utf-8")
|
||
assert self._run(mod, ["--backfill", "--url", url, "--evidence", str(bad)]) == 3
|
||
with eng.connect() as conn:
|
||
assert conn.execute(text("SELECT COUNT(*) FROM subjects WHERE entity_id=1")).scalar() == 0
|
||
|
||
# 正确取证件 → 164 行归 e1、43 行留 NULL
|
||
good = tmp / "ev_ok.json"
|
||
good.write_text(json.dumps({"dama_evidence": "dama-...out", "finance_confirmation": "任总...",
|
||
"counts": {"e1": 164, "global_null": 43}}), encoding="utf-8")
|
||
assert self._run(mod, ["--backfill", "--url", url, "--evidence", str(good),
|
||
"--target-entity", "1"]) == 0
|
||
with eng.connect() as conn:
|
||
dist = dict(conn.execute(text(
|
||
"SELECT IFNULL(entity_id,'NULL'), COUNT(*) FROM subjects GROUP BY 1")).fetchall())
|
||
assert dist == {1: 164, "NULL": 43}, dist
|
||
|
||
# 回滚：有分账套数据须 --yes；回滚后列消失、code 唯一性恢复
|
||
assert self._run(mod, ["--down", "--url", url]) == 3
|
||
assert self._run(mod, ["--down", "--url", url, "--yes"]) == 0
|
||
assert "entity_id" not in self._cols(eng, "subjects")
|
||
with eng.connect() as conn:
|
||
assert conn.execute(text("SELECT COUNT(*) FROM subjects")).scalar() == 207
|
||
|
||
def test_subjects_prod_name_guard(self, subj_env, tmp_path, monkeypatch):
|
||
mod, _, _ = subj_env
|
||
monkeypatch.chdir(tmp_path)
|
||
assert self._run(mod, ["--apply", "--url", "sqlite:///cma"]) == 3 # 未放行
|
||
monkeypatch.setenv("CMA_DDL_APPROVED", "1")
|
||
assert self._run(mod, ["--apply", "--url", "sqlite:///cma", "--allow-prod"]) == 3 # 表不存在 → 3
|
||
|
||
# ── analysis_results ──────────────────────────
|
||
|
||
def test_analysis_apply_and_down(self, analysis_env):
|
||
mod, eng, _ = analysis_env
|
||
url = str(eng.url)
|
||
assert self._run(mod, ["--check", "--url", url]) == 0
|
||
assert self._run(mod, ["--apply", "--url", url]) == 0
|
||
assert "entity_id" in self._cols(eng, "analysis_results")
|
||
assert self._run(mod, ["--apply", "--url", url]) == 0 # 幂等
|
||
with eng.connect() as conn:
|
||
names = {r[0] for r in conn.execute(
|
||
text("SELECT name FROM sqlite_master WHERE type='index'")).fetchall()}
|
||
assert "ix_analysis_results_entity_id" in names
|
||
# 存量 2 行不被代猜账套
|
||
assert conn.execute(text(
|
||
"SELECT COUNT(*) FROM analysis_results WHERE entity_id IS NOT NULL")).scalar() == 0
|
||
assert conn.execute(text("SELECT COUNT(*) FROM analysis_results")).scalar() == 2
|
||
assert self._run(mod, ["--down", "--url", url, "--yes"]) == 0
|
||
assert "entity_id" not in self._cols(eng, "analysis_results")
|
||
|
||
def test_analysis_prod_name_guard(self, analysis_env, tmp_path, monkeypatch):
|
||
mod, _, _ = analysis_env
|
||
monkeypatch.chdir(tmp_path)
|
||
assert self._run(mod, ["--apply", "--url", "sqlite:///cma"]) == 3
|