Archived
snowy-yang
fc7fc21284
tests/test_golden.py 的 fixture 直接指向 data/site.db 并注释为「只读」，但快照里 有 2 条 auth=user 的 POST /download（均返回 400），这条路径会 reserve_and_open 开账、失败后 settle_refund 退款——于是**每次 pytest 都往生产库插 2 行 billing_ledger**（status='rejected', reserved=1, refunded=1），并把该用户的 last_active 顶到测试时刻。 发现时生产库已累积 83 行这样的行（时间戳与历次 pytest 运行一一对应），库体积 也因此长了 2 个页（151552 → 159744）。净余额为 0（预扣 1、退款 1），不丢钱， 但污染台账与后台台账视图，而且让「跑测试不动生产数据」这个前提不再成立—— 排查时无法再用「库指纹没变」来证明冒烟/测试是安全的。 改法：复制一份 data/site.db（含 -wal/-shm 边车文件）到 tmp_path，再把 auth_system.DB_PATH/DATA_DIR 与 core.db.DB_PATH/DATA_DIR 一并指向副本。 契约保真度不变（同一份真实数据、同一 schema），写入全部落在临时目录。 _session_for 也改为读 auth_system.DB_PATH，保证「造 Cookie」与「发请求」 命中的是同一个库（原先写死路径，换库后会静默读生产库拿到错误 login_seq）。 验证：改前改后各跑一次完整套件，data/site.db 的 sha1 / 体积 / mtime 三项完全 不变，billing_ledger 行数停在 83；144 passed；ruff All checks passed； pyright 0 errors / 0 warnings。 注：生产库里既有的 83 行测试残留**未删除**（那是生产数据，需确认后再动）； ledger 自带保留期清扫（auth_system/ledger.py 的 DELETE ... settled_at <= ?） 到期会自行回收。
139 lines
5.4 KiB
Python
139 lines
5.4 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）。有意改模板后必须同步刷新这两项：
|
||
服务端渲染 = templates/index.html 去掉末尾换行，故
|
||
html_len = 文件字节数 - 1，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_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, 'DATA_DIR', tmp_path)
|
||
monkeypatch.setattr(auth_system, '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']
|