@@ -0,0 +1,174 @@
""" ASK-20260924-15 / -17 新判据回归（2026-09-25，任务单
`yanxue-20260924-ask15-17-fullstack-engine-and-bot-default`）。
- **#17**： ）
显式 `?entity_id=` / `X-Entity-Id:` / body `entity_id` 行为不变；无 key → 401；写端点缺省 → 400 且零写入。
- **#15**：告警引擎差异/趋势入口 `run_deviation_check(db, period=None)` 的期间口径
**回落「该 KPI 自己的最近可用期间」**，不再硬编码当月（当月无 kpi_values ⇒ 永久 0 新增）。
"""
from datetime import datetime , timedelta
import pytest
from fastapi . testclient import TestClient
from starlette . requests import Request
from app . deps import get_entity_id , require_bot_entity_id
from app . models import Entity , KPIDefinition , KPIValue
from app . utils . alert_entity_scope import resolve_latest_available_period
from tests . conftest import create_test_kpi
BOT_KEY = " cma-bot-finance-2026 "
HEAD = { " X-BOT-KEY " : BOT_KEY }
def _prev_month ( ) - > str :
return ( datetime . now ( ) . replace ( day = 1 ) - timedelta ( days = 1 ) ) . strftime ( " % Y- % m " )
def _cur_month ( ) - > str :
return datetime . now ( ) . strftime ( " % Y- % m " )
def _raw_request ( headers = None , method = " GET " , query_string = b " " ) - > Request :
return Request ( {
" type " : " http " , " method " : method , " path " : " /api/cma/bot/overview " ,
" headers " : ( headers or [ ] ) + [ ( b " host " , b " test " ) ] ,
" query_string " : query_string , " client " : ( " 127.0.0.1 " , 1 ) ,
" server " : ( " test " , 80 ) , " scheme " : " http " ,
} )
# ─────────────────────────── #17：Bot 通道缺省账套 → 400 ───────────────────────────
class TestBotChannelEntityRequired :
def test_overview_without_entity_400 ( self , client , db ) :
""" ① 带 key、不带 entity_id → 400 且 message 说明须显式传账套 """
r = client . get ( " /api/cma/bot/overview " , headers = HEAD )
assert r . status_code == 400 , r . text
assert " 缺少账套参数 entity_id " in r . text
def test_overview_entity1_ok ( self , client , db ) :
""" ② ?entity_id=1 → 200 且 entity_id=1（酣客口径） """
r = client . get ( " /api/cma/bot/overview?entity_id=1 " , headers = HEAD )
assert r . status_code == 200 , r . text
assert r . json ( ) [ " entity_id " ] == 1
def test_overview_entity2_ok ( self , client , db ) :
""" ③ ?entity_id=2 → 200 且 entity_id=2（博海口径，另建账套） """
if not db . query ( Entity ) . filter ( Entity . id == 2 ) . first ( ) :
db . add ( Entity ( id = 2 , name = " 另一企业 " , short_name = " B " , status = " active " ) )
db . commit ( )
r = client . get ( " /api/cma/bot/overview?entity_id=2 " , headers = HEAD )
assert r . status_code == 200 , r . text
assert r . json ( ) [ " entity_id " ] == 2
def test_x_entity_id_header_ok ( self , client , db ) :
""" 显式 header 形态同样放行（三种传参形态之一） """
r = client . get ( " /api/cma/bot/overview " ,
headers = { * * HEAD , " X-Entity-Id " : " 1 " } )
assert r . status_code == 200 , r . text
assert r . json ( ) [ " entity_id " ] == 1
def test_no_bot_key_401 ( self , client , db ) :
""" ④ 无 key → 401（鉴权在前；本单未改鉴权） """
r = client . get ( " /api/cma/bot/overview?entity_id=1 " )
assert r . status_code == 401
def test_write_endpoint_without_entity_400_and_no_write ( self , client , db ) :
""" ⑤ 写端点缺省账套 → 400 且**库内零写入**（旧写法会静默写 entity 1） """
kpi = create_test_kpi ( db , kpi_code = " ENTITY_REQ_1 " )
before = db . query ( KPIValue ) . filter ( KPIValue . kpi_id == kpi . id ) . count ( )
r = client . post ( " /api/cma/bot/kpi-value-with-check " ,
json = { " kpi_id " : kpi . id , " actual_value " : 1.0 , " period " : _prev_month ( ) } ,
headers = HEAD )
assert r . status_code == 400 , r . text
assert " 缺少账套参数 entity_id " in r . text
db . expire_all ( )
after = db . query ( KPIValue ) . filter ( KPIValue . kpi_id == kpi . id ) . count ( )
assert after == before == 0
def test_require_bot_entity_id_helper ( self ) :
""" helper 语义：显式 int/数字串放行；缺失/空串 400；非数字 422 """
from fastapi import HTTPException
assert require_bot_entity_id ( { " entity_id " : 2 } ) == 2
assert require_bot_entity_id ( { " entity_id " : " 2 " } ) == 2
for bad in ( None , { } , { " entity_id " : " " } , { " entity_id " : " " } ) :
with pytest . raises ( HTTPException ) as e :
require_bot_entity_id ( bad , source = " unit " )
assert e . value . status_code == 400
with pytest . raises ( HTTPException ) as e :
require_bot_entity_id ( { " entity_id " : " abc " } )
assert e . value . status_code == 422
def test_legacy_no_key_fallback_kept ( self ) :
""" 边界：**无 X-BOT-KEY** 的存量公开接口仍兜底 entity 1（本单未动） """
assert get_entity_id ( request = _raw_request ( ) , entity_id = None ,
x_entity_id = None , db = None ) == 1
with pytest . raises ( Exception ) as e :
get_entity_id ( request = _raw_request ( headers = [ ( b " x-bot-key " , b " k " ) ] ) ,
entity_id = None , x_entity_id = None , db = None )
assert getattr ( e . value , " status_code " , None ) == 400
# ─────────────────────── #15：差异/趋势入口期间口径（裁定2b） ───────────────────────
class TestDeviationEnginePeriodCaliber :
def test_fallback_is_kpi_latest_available_period ( self , db ) :
""" 单一事实源：回落期间 = 该 KPI 自己的最近可用期间（最大 YYYY-MM） """
kpi = create_test_kpi ( db , kpi_code = " PERIOD_FB_1 " )
assert resolve_latest_available_period ( db , kpi . id ) is None
db . add ( KPIValue ( kpi_id = kpi . id , entity_id = 1 , period = " 2026-03 " , actual_value = 1.0 ) )
db . add ( KPIValue ( kpi_id = kpi . id , entity_id = 1 , period = " 2026-05 " , actual_value = 2.0 ) )
db . commit ( )
assert resolve_latest_available_period ( db , kpi . id ) == " 2026-05 "
def test_deviation_check_uses_fallback_not_current_month ( self , db , monkeypatch ) :
""" 改前用当月（2026-09 无值 ⇒ 永久 0 新增）；改后逐 KPI 回落上一可用月 """
prev , cur = _prev_month ( ) , _cur_month ( )
assert prev != cur
kpi = create_test_kpi ( db , kpi_code = " PERIOD_FB_2 " )
db . add ( KPIValue ( kpi_id = kpi . id , entity_id = 1 , period = prev , actual_value = 100.0 ) )
db . commit ( )
import app . utils . deviation_engine as de
seen = [ ]
def _fake_build ( db_ , kpi_ , period ) :
seen . append ( ( kpi_ . id , period ) )
return { " triggered " : False }
monkeypatch . setattr ( de , " build_deviation_alert " , _fake_build )
monkeypatch . setattr ( de , " check_trend_anomaly " ,
lambda * a , * * k : { " anomaly " : False } )
de . run_deviation_check ( db ) # period=None ⇒ 逐 KPI 回落
assert ( kpi . id , prev ) in seen , " 缺省期间应回落该 KPI 的最近可用期间 %s " % prev
assert ( kpi . id , cur ) not in seen , " 不应再拿当月（无数据月）当口径 "
def test_deviation_check_explicit_period_still_fixed ( self , db , monkeypatch ) :
""" 显式传 period → 固定该期间（原语义不变，抽查/测试用） """
prev = _prev_month ( )
kpi = create_test_kpi ( db , kpi_code = " PERIOD_FB_3 " )
db . add ( KPIValue ( kpi_id = kpi . id , entity_id = 1 , period = prev , actual_value = 100.0 ) )
db . commit ( )
import app . utils . deviation_engine as de
seen = [ ]
monkeypatch . setattr ( de , " build_deviation_alert " ,
lambda db_ , kpi_ , period : ( seen . append ( ( kpi_ . id , period ) ) ,
{ " triggered " : False } ) [ 1 ] )
monkeypatch . setattr ( de , " check_trend_anomaly " , lambda * a , * * k : { " anomaly " : False } )
de . run_deviation_check ( db , period = " 2026-06 " )
assert ( kpi . id , " 2026-06 " ) in seen
def test_active_kpi_without_values_is_skipped_not_crashed ( self , db , monkeypatch ) :
""" 无任何 kpi_values 的 active KPI：跳过（不抛、不计入期间分布） """
kpi = create_test_kpi ( db , kpi_code = " PERIOD_FB_4 " )
import app . utils . deviation_engine as de
seen = [ ]
monkeypatch . setattr ( de , " build_deviation_alert " ,
lambda db_ , kpi_ , period : ( seen . append ( ( kpi_ . id , period ) ) ,
{ " triggered " : False } ) [ 1 ] )
monkeypatch . setattr ( de , " check_trend_anomaly " , lambda * a , * * k : { " anomaly " : False } )
de . run_deviation_check ( db )
assert ( kpi . id , None ) not in seen
wecom-fullstack