fd7174a70e
## 后端 admin/stats.py
- compute_summary 在单条聚合 SQL 中加两个新字段:
- pending_overdue_24h: status='pending' AND created_at < (now - 24h)
(超时未付,需主动催付)
- pending_missing_intake: status='pending' AND
(order_intakes 无记录 OR order_intakes.status='draft')
(资料待补,需主动跟进)
- LEFT JOIN order_intakes AS i ON i.order_id = orders.id
- 关键 bug 修复: SQL ? 顺序与参数顺序必须匹配,
(overdue_cutoff 在 IN 子句后) 之前误用 (overdue_cutoff, *_REVENUE_STATUSES)
导致 overdue_cutoff datetime 字符串误入 IN 子句,
'completed' 误入 created_at < ? 子句, 收入少算 30%。
已修正为 (*_REVENUE_STATUSES, overdue_cutoff)
- 列名全部限定为 orders.status / orders.created_at,避免与 i.status 歧义
## 前端 admin/static/dashboard.html
- KPI 4 待处理订单卡加 .pending-breakdown 区域
- 两个 pill 标签: pending-tag-overdue(红色,超时) + pending-tag-missing(橙色,资料待补)
- 仅在对应字段 > 0 时才显示
- CSS .pending-tag[hidden] { display: none; } 确保 hidden 属性生效
## 前端 admin/static/dashboard.js
- renderSummary 新增 pending_overdue_24h / pending_missing_intake 渲染
- 至少一个非零才展开 breakdown 区域
## 测试
- conftest.py orders_db fixture 同时建 order_intakes 表,避免 LEFT JOIN 缺表
- test_routes_stats_dashboard.py:
- test_dashboard_empty_db_shape 锁住 summary 必须含 12 字段
- test_dashboard_summary_pending_orders 更新 revenue 期望 (paid + serving 计入)
- 新增 test_dashboard_summary_pending_overdue_24h (2天前 vs 12h前)
- 新增 test_dashboard_summary_pending_intake (submitted/draft/无 intake 4 种组合)
- test_app.py 加 pending-breakdown 结构断言
- pytest 52 passed
## 真实环境验证
- admin/test-pass-123 登录后:
pending=7 overdue=0 missing_intake=7
- 视觉验证: KPI 4 显示 '7' 主值 + '资料待补 7' 橙色 pill
'超时' pill 因 overdue=0 自动隐藏
- 数据一致性: by_status.pending=7 与 summary.pending_orders=7 一致
133 lines
4.3 KiB
Python
133 lines
4.3 KiB
Python
"""pytest 配置 (T6.1).
|
||
|
||
提供 fixture:
|
||
- settings: 内存 SQLite + 安全 JWT 密钥 + 短过期
|
||
- app: FastAPI 实例(lifespan 已运行 → admin 表已建 + bootstrap 用户)
|
||
- client: httpx TestClient
|
||
- auth_token: 登录后的 Bearer JWT
|
||
- auth_headers: {"Authorization": f"Bearer ..."}
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
# 确保项目根在 sys.path
|
||
_ROOT = Path(__file__).resolve().parent.parent.parent
|
||
if str(_ROOT) not in sys.path:
|
||
sys.path.insert(0, str(_ROOT))
|
||
|
||
|
||
@pytest.fixture
|
||
def secure_secret() -> str:
|
||
"""64-char 安全 JWT 密钥。"""
|
||
return "x" * 64 # 确定性,便于测试断言
|
||
|
||
|
||
@pytest.fixture
|
||
def settings(tmp_path, secure_secret, monkeypatch):
|
||
"""隔离的 Settings 实例:tmp_path SQLite + 安全密钥 + 短过期。
|
||
|
||
用真实文件而不是 :memory: 是因为 admin/db.py 中每次 get_connection
|
||
都新建连接,:memory: 在 SQLite 下不共享状态。
|
||
"""
|
||
db_path = str(tmp_path / "admin.db")
|
||
orders_db_path = str(tmp_path / "orders.db")
|
||
share_db_path = str(tmp_path / "short_links.db")
|
||
share_report_dir = str(tmp_path / "share_reports")
|
||
portal_upload_dir = str(tmp_path / "portal_uploads")
|
||
ops_alert_log = str(tmp_path / "ops-alerts.jsonl")
|
||
deletion_request_log = str(tmp_path / "deletion-requests.jsonl")
|
||
monkeypatch.setenv("GAOKAO_ENV", "dev")
|
||
monkeypatch.setenv("GAOKAO_DB_PATH", db_path)
|
||
monkeypatch.setenv("GAOKAO_ORDERS_DB_PATH", orders_db_path)
|
||
monkeypatch.setenv("GAOKAO_SHARE_DB_PATH", share_db_path)
|
||
monkeypatch.setenv("GAOKAO_SHARE_REPORT_DIR", share_report_dir)
|
||
monkeypatch.setenv("GAOKAO_PORTAL_UPLOAD_DIR", portal_upload_dir)
|
||
monkeypatch.setenv("GAOKAO_PORTAL_UPLOAD_MAX_BYTES", "5242880")
|
||
monkeypatch.setenv("GAOKAO_PORTAL_UPLOAD_MAX_FILES", "5")
|
||
monkeypatch.setenv("GAOKAO_OPS_ALERT_LOG", ops_alert_log)
|
||
monkeypatch.setenv("GAOKAO_ALERT_RECIPIENTS", "")
|
||
monkeypatch.setenv("GAOKAO_ALERT_WEBHOOK_URLS", "")
|
||
monkeypatch.setenv("GAOKAO_ORDERS_FERNET_KEY", "test-secret-for-web-self-service")
|
||
monkeypatch.setenv("GAOKAO_JWT_SECRET", secure_secret)
|
||
monkeypatch.setenv("GAOKAO_JWT_EXP_MIN", "5")
|
||
monkeypatch.setenv("GAOKAO_ADMIN_PASS", "test-pass-123")
|
||
monkeypatch.setenv("GAOKAO_OPS_ALERT_LOG", ops_alert_log)
|
||
monkeypatch.setenv("GAOKAO_DELETION_REQUEST_LOG", deletion_request_log)
|
||
|
||
from admin.config import load_settings
|
||
|
||
return load_settings()
|
||
|
||
|
||
@pytest.fixture
|
||
def orders_db(settings):
|
||
"""T6.2 起:管理后台统计端点会读 orders 表,因此 conftest 顺带建一个
|
||
空 orders DB (T4.1 schema),后续 dashboard 测试可在里面塞 fixture 数据。
|
||
|
||
T-pending-intake 起: 同时建 order_intakes 表,确保 summary.pending_missing_intake
|
||
的 LEFT JOIN 不会因为缺表而失败。
|
||
"""
|
||
from data.orders.intake_store import SCHEMA_SQL as INTAKE_SCHEMA_SQL
|
||
from data.orders.schema import apply_schema
|
||
|
||
conn = apply_schema(settings.orders_db_path)
|
||
conn.executescript(INTAKE_SCHEMA_SQL)
|
||
conn.commit()
|
||
conn.close()
|
||
return settings.orders_db_path
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _auto_orders_db(settings, orders_db):
|
||
"""所有测试都默认有空 orders DB 可用,避免 T6.2 端点在不相关测试里
|
||
报 'no such table: orders'。``orders_db`` 显式依赖确保已建。
|
||
"""
|
||
return orders_db
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _reset_login_rate_limit():
|
||
from admin.routes.auth import reset_login_rate_limit_for_tests
|
||
|
||
reset_login_rate_limit_for_tests()
|
||
yield
|
||
reset_login_rate_limit_for_tests()
|
||
|
||
|
||
@pytest.fixture
|
||
def app(settings):
|
||
"""FastAPI 实例（lifespan 已运行）。"""
|
||
from admin.app import create_app
|
||
|
||
return create_app(settings)
|
||
|
||
|
||
@pytest.fixture
|
||
def client(app):
|
||
"""httpx TestClient。"""
|
||
from fastapi.testclient import TestClient
|
||
|
||
with TestClient(app) as c:
|
||
yield c
|
||
|
||
|
||
@pytest.fixture
|
||
def auth_token(client) -> str:
|
||
"""登录后返回 Bearer JWT。"""
|
||
resp = client.post(
|
||
"/api/auth/login",
|
||
json={"username": "admin", "password": "test-pass-123"},
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
return resp.json()["access_token"]
|
||
|
||
|
||
@pytest.fixture
|
||
def auth_headers(auth_token) -> dict:
|
||
return {"Authorization": f"Bearer {auth_token}"}
|