Archived
snowy-yang
17b4a99680
为什么：首页响应源从 Jinja2 渲染模板（去尾换行）变为 frontend/dist/index.html 原样字节，html_len/位移到 967 字节。 快照以固定产物名重建，docstring 同步更新口径说明。
139 lines
5.5 KiB
Python
139 lines
5.5 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""黄金契约回归：FastAPI 严格复现 Flask 时代的响应。
|
||
|
||
tests/golden/responses.json 是**删除 Flask 之前**从 Flask 应用捕获的快照
|
||
（状态码 + 响应体）。本测试让 FastAPI（根路径模式）跑同一批请求，
|
||
逐字段断言一致——这样删除 Flask 巨石后，行为回归仍有硬证据。
|
||
|
||
说明：
|
||
- 会话 Cookie 在运行时用当前库的 login_seq 动态构造（快照不存 Cookie）。
|
||
- 请求为只读/确定性调用，不含实时上游数据（那些由 parity/单测覆盖）。
|
||
- `index` 条目的 html_len/html_sha256 记的是**当前前端**的首页产物，不是 Flask 期数值
|
||
（该条目 body 为 null）。有意改前端后必须同步刷新这两项：
|
||
首页现服务 React 构建产物 frontend/dist/index.html（vite 固定产物文件名，
|
||
字节级稳定），html_len = 文件字节数，html_sha256 = sha256(文件字节)。
|
||
"""
|
||
import hashlib
|
||
import json
|
||
import shutil
|
||
import sqlite3
|
||
import time
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
import auth_system
|
||
import core.db as core_db
|
||
from core import session as cs
|
||
|
||
_GOLDEN = Path(__file__).parent / 'golden' / 'responses.json'
|
||
SECRET = 'golden-secret'
|
||
|
||
|
||
def _load_cases():
|
||
return json.loads(_GOLDEN.read_text(encoding='utf-8'))
|
||
|
||
|
||
def _session_for(role):
|
||
"""按当前库中该角色用户的 login_seq 造一份合法会话 Cookie。
|
||
|
||
读的是 `auth_system.DB_PATH`（而非写死 'data/site.db'）——fixture 已把它
|
||
指向临时副本，这样造 Cookie 与请求命中的是同一个库。
|
||
"""
|
||
conn = sqlite3.connect(str(auth_system.db.DB_PATH))
|
||
conn.row_factory = sqlite3.Row
|
||
try:
|
||
where = "role = 'admin'" if role == 'admin' else "role != 'admin' AND status = 1"
|
||
row = conn.execute(
|
||
f'SELECT id, login_seq FROM users WHERE {where} AND login_seq IS NOT NULL LIMIT 1'
|
||
).fetchone()
|
||
finally:
|
||
conn.close()
|
||
if row is None:
|
||
return None
|
||
return cs.dumps(SECRET, {
|
||
'site_uid': int(row['id']), 'site_uid_seq': int(row['login_seq']),
|
||
'site_uid_ts': int(time.time()),
|
||
})
|
||
|
||
|
||
def _copy_db(src: Path, dst: Path) -> None:
|
||
"""复制站点库到临时目录（含 WAL 边车文件，保证副本与生产一致）。"""
|
||
shutil.copyfile(src, dst)
|
||
for suffix in ('-wal', '-shm'):
|
||
side = Path(str(src) + suffix)
|
||
if side.exists():
|
||
shutil.copyfile(side, Path(str(dst) + suffix))
|
||
|
||
|
||
@pytest.fixture()
|
||
def client(tmp_path, monkeypatch):
|
||
# 站点库取生产数据的**副本**，不是生产库本身。
|
||
# 曾经直接指向 data/site.db 并注释为「只读」——但快照里 2 条 POST /download
|
||
# 会走 reserve_and_open 开账、失败后 settle_refund 退款，即**每次跑测试都往
|
||
# 生产库插 2 行 rejected 台账**（净余额为 0，但污染台账与后台视图，还会
|
||
# 把 users.last_active 顶到测试时刻）。改成副本后契约保真度不变（同一份数据），
|
||
# 写入全部落在 tmp_path。
|
||
_copy_db(Path('data/site.db'), tmp_path / 'site.db')
|
||
monkeypatch.setattr(auth_system.db, 'DATA_DIR', tmp_path)
|
||
monkeypatch.setattr(auth_system.db, 'DB_PATH', tmp_path / 'site.db')
|
||
monkeypatch.setattr(core_db, 'DATA_DIR', tmp_path)
|
||
monkeypatch.setattr(core_db, 'DB_PATH', tmp_path / 'site.db')
|
||
monkeypatch.setattr('web.deps.secret_key', lambda: SECRET)
|
||
monkeypatch.setattr(auth_system, 'get_secret_key', lambda: SECRET)
|
||
from web.app import create_app
|
||
return TestClient(create_app(api_prefix=''))
|
||
|
||
|
||
# 响应对比前剔除的易变字段（时钟）
|
||
_VOLATILE = {'timestamp'}
|
||
|
||
|
||
def _strip(obj):
|
||
if isinstance(obj, dict):
|
||
return {k: _strip(v) for k, v in obj.items() if k not in _VOLATILE}
|
||
if isinstance(obj, list):
|
||
return [_strip(v) for v in obj]
|
||
return obj
|
||
|
||
|
||
def _request(client, method, path, payload, cookie):
|
||
headers = {'Cookie': f'session={cookie}'} if cookie else {}
|
||
if method == 'GET':
|
||
return client.get(path, headers=headers)
|
||
return client.post(path, json=payload or {}, headers=headers)
|
||
|
||
|
||
def test_golden_contract(client):
|
||
"""逐例复现 Flask 快照。"""
|
||
cases = _load_cases()
|
||
failures = []
|
||
for c in cases:
|
||
if c.get('html_len'):
|
||
r = client.get(c['path'])
|
||
if r.status_code != c['status'] or len(r.content) != c['html_len']:
|
||
failures.append(f"{c['name']}: html {r.status_code}/{len(r.content)} "
|
||
f"!= {c['status']}/{c['html_len']}")
|
||
continue
|
||
cookie = None
|
||
if c['auth'] == 'admin':
|
||
cookie = _session_for('admin')
|
||
elif c['auth'] == 'user':
|
||
cookie = _session_for('user')
|
||
r = _request(client, c['method'], c['path'], c['payload'], cookie)
|
||
body = r.json() if r.headers.get('content-type', '').startswith('application/json') else None
|
||
if r.status_code != c['status']:
|
||
failures.append(f"{c['name']}: status {r.status_code} != {c['status']}")
|
||
elif _strip(body) != _strip(c['body']):
|
||
failures.append(f"{c['name']}: body\n fastapi={body}\n golden ={c['body']}")
|
||
assert not failures, '黄金契约回归失败:\n' + '\n'.join(failures)
|
||
|
||
|
||
def test_index_html_stable(client):
|
||
"""首页渲染逐字节稳定（模板未改时）。"""
|
||
cases = {c['name']: c for c in _load_cases()}
|
||
c = cases['index']
|
||
r = client.get('/')
|
||
assert hashlib.sha256(r.content).hexdigest() == c['html_sha256']
|