# -*- coding: utf-8 -*- """yanxue-20260924-cma-fetch-caliber-fixes-fullstack 取数/接口层两处口径缺陷的回归用例（2026-09-26，wecom-fullstack）： 1. **伪期间排序陷阱**：`kpi_values` 里的半年/预测伪期间（`2026-H2` 等）字符串序**大于**同年任何 月份（'H' > '0'）⇒ 任何 `ORDER BY period DESC` 取「最新值」的查询会命中伪期间，结论变成预测值。 本单把 R3 期间口径（仅 `YYYY-MM`）延伸到取数层：`app.utils.alert_entity_scope.month_period_clause()`。 改前本文件的前两个用例必红（返回 2026-H2 / -55.0）。 2. **`GET /api/cma/subjects` 响应缺 `entity_id`**：共享层（`entity_id IS NULL`）与账套私有行 在界面上不可区分。本单只加字段，既有 10 个字段语义与名称不变。 """ from datetime import datetime import pytest from fastapi.testclient import TestClient from sqlalchemy.orm import Session from tests.conftest import ( TEST_USER_PASSWORD, auth_header, create_test_kpi, create_test_user, get_token_for_user, ) from app.models import Entity, KPIValue, Subject from app.utils.alert_entity_scope import is_month_period, month_period_clause GLOSSARY = "/api/cma/kpis/glossary" SUBJECTS = "/api/cma/subjects" # R3 判据样本：规整月 / 半年期 / 年期间 / 纯年份 / 非法月 / 近似格式 # （`kpi_values.period` 为 NOT NULL，空值/None 由 is_month_period 直接断言，不入库） PERIOD_SAMPLES = [ "2026-08", "2026-09", "2026-01", "2026-12", "2025-12", "2026-H1", "2026-H2", "2025-FY", "2026-Q2", "2025", "2021", "2026-13", "2026-00", "2026-8", "2026-08-31", ] def _mk_value(db: Session, kpi_id: int, period, value, entity_id: int = 1, source_type: str = "manual", data_status: str = "verified") -> KPIValue: v = KPIValue(entity_id=entity_id, kpi_id=kpi_id, period=period, actual_value=value, source_type=source_type, data_status=data_status) db.add(v) db.commit() db.refresh(v) return v def _token(client: TestClient, db: Session, entity_id: int = 1) -> str: """按账套登录（1=酣客 / 2=博海）：确保企业存在 + 授权 + 取 token""" from app.models import UserEntity ent = db.query(Entity).filter(Entity.id == entity_id).first() if not ent: db.add(Entity(id=entity_id, name=f"企业{entity_id}", short_name=f"E{entity_id}", status="active")) db.commit() user = create_test_user(db) if not db.query(UserEntity).filter(UserEntity.user_id == user.id, UserEntity.entity_id == entity_id).first(): db.add(UserEntity(user_id=user.id, entity_id=entity_id)) db.commit() return get_token_for_user(client, "testadmin", TEST_USER_PASSWORD) class TestMonthPeriodClause: """SQL 等价式（取数层过滤）必须与 Python 侧 R3 判据逐值一致""" def test_clause_equivalent_to_python_regex(self, db: Session): kpi = create_test_kpi(db, kpi_code="T_PC_001") for i, p in enumerate(PERIOD_SAMPLES): _mk_value(db, kpi.id, p, float(i + 1)) rows = db.query(KPIValue.period).filter( KPIValue.kpi_id == kpi.id, month_period_clause(KPIValue.period) ).all() sql_side = sorted(r[0] for r in rows) py_side = sorted(p for p in PERIOD_SAMPLES if is_month_period(p)) assert sql_side == py_side, f"SQL 侧={sql_side} Python 侧={py_side}" assert "2026-H2" not in sql_side and "2026-H1" not in sql_side assert "2025-FY" not in sql_side and "2025" not in sql_side and "2026-13" not in sql_side assert len(sql_side) == 5 def test_null_and_empty_are_not_month_periods(self): """空值/None：Python 侧判据（列 NOT NULL，不入库）""" assert is_month_period(None) is False assert is_month_period("") is False assert is_month_period(" 2026-08") is False class TestLatestActualPeriodOnFetchLayer: """取数层「最新实际值」不得被伪期间顶掉（改前必红）""" def test_glossary_current_value_skips_forecast_pseudo_period(self, client: TestClient, db: Session): token = _token(client, db, 1) kpi = create_test_kpi(db, kpi_code="F_NET_PROFIT", kpi_name="净利润", entity_id=1) # 复刻生产证据（_evidence/hanke-data-gap-20260924/p11.out:56-57） _mk_value(db, kpi.id, "2026-08", -10.46) _mk_value(db, kpi.id, "2026-H1", -90.86) _mk_value(db, kpi.id, "2026-H2", -55.0, source_type="forecast", data_status="estimated") resp = client.get(GLOSSARY, headers=auth_header(token)) assert resp.status_code == 200, resp.text[:300] item = [g for g in resp.json()["glossary"] if g["kpi_code"] == "F_NET_PROFIT"][0] assert item["latest_period"] == "2026-08", f"latest_period={item['latest_period']}" assert item["current_value"] == -10.46, f"current_value={item['current_value']}" def test_glossary_kpi_with_only_pseudo_periods_has_no_current_value(self, client: TestClient, db: Session): """只有伪期间数据 → 当前值必须是「无」，不得拿预测值/半年值充数""" token = _token(client, db, 1) kpi = create_test_kpi(db, kpi_code="T_PC_ONLY_PSEUDO", entity_id=1) _mk_value(db, kpi.id, "2026-H2", 455.0, source_type="forecast", data_status="estimated") _mk_value(db, kpi.id, "2026-H1", 713.27) resp = client.get(GLOSSARY, headers=auth_header(token)) assert resp.status_code == 200, resp.text[:300] item = [g for g in resp.json()["glossary"] if g["kpi_code"] == "T_PC_ONLY_PSEUDO"][0] assert item["latest_period"] is None, f"latest_period={item['latest_period']}" assert item["current_value"] is None, f"current_value={item['current_value']}" def test_dashboard_latest_kpi_value_helper(self, db: Session): """dashboard.latest_kpi_value（原 MySQL-only REGEXP，已收口到单一事实源）""" from app.api.dashboard import latest_kpi_value kpi = create_test_kpi(db, kpi_code="T_PC_HELPER") _mk_value(db, kpi.id, "2026-06", 6.0) _mk_value(db, kpi.id, "2026-H2", 99.0, source_type="forecast", data_status="estimated") got = latest_kpi_value(db, kpi.id) assert got is not None and got.period == "2026-06", f"period={getattr(got, 'period', None)}" assert got.actual_value == 6.0 class TestSubjectsEntityIdField: """subjects 响应补 entity_id，既有 10 字段语义与名称不变""" ORIGINAL_KEYS = {"id", "subject_code", "subject_name", "parent_code", "level", "category", "new_standard_category", "is_active", "remark", "updated_at"} def _seed(self, db: Session): db.add(Subject(entity_id=None, subject_code="1001", subject_name="库存现金", level=1, category="asset", is_active=1)) db.add(Subject(entity_id=1, subject_code="100106", subject_name="酒窖备用金", level=1, category="asset", is_active=1, remark="zxt-sync")) db.add(Subject(entity_id=2, subject_code="100108", subject_name="友赞商户", level=1, category="asset", is_active=1, remark="zxt-sync")) db.commit() def test_response_contains_entity_id_and_keeps_original_fields(self, client: TestClient, db: Session): token = _token(client, db, 1) self._seed(db) resp = client.get(SUBJECTS, headers=auth_header(token)) assert resp.status_code == 200, resp.text[:300] data = resp.json()["data"] assert data, "responses 为空" for item in data: assert self.ORIGINAL_KEYS.issubset(set(item.keys())), sorted(item.keys()) assert "entity_id" in item, sorted(item.keys()) by_code = {i["subject_code"]: i["entity_id"] for i in data} assert by_code["1001"] is None # 全局标准准则科目 assert by_code["100106"] == 1 # 会话账套（酣客）私有 assert "100108" not in by_code # 他账套私有不可见（分层读取口径不变） def test_bohai_session_sees_only_global(self, client: TestClient, db: Session): self._seed(db) ent = db.query(Entity).filter(Entity.id == 2).first() if not ent: db.add(Entity(id=2, name="企业2", short_name="E2", status="active")) db.commit() from app.models import UserEntity user = create_test_user(db) db.add(UserEntity(user_id=user.id, entity_id=2)) db.commit() login = client.post("/api/cma/auth/login", json={ "username": "testadmin", "password": TEST_USER_PASSWORD, "entity_id": 2, }) assert login.status_code == 200, login.text[:300] token = login.json()["token"] resp = client.get(SUBJECTS, headers=auth_header(token)) assert resp.status_code == 200, resp.text[:300] data = resp.json()["data"] by_code = {i["subject_code"]: i["entity_id"] for i in data} assert by_code["1001"] is None assert by_code["100108"] == 2 assert all(i["entity_id"] in (None, 2) for i in data)