NeteaseUrl/tests/test_golden.py

# -*- 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'


async def _fake_secret_key() -> str:
    """web.deps.secret_key / auth_system.get_secret_key 均已异步化：
    替身须为协程函数，否则 await / asyncio.run 处会报 TypeError。"""
    return 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', _fake_secret_key)
    monkeypatch.setattr(auth_system, 'get_secret_key', _fake_secret_key)
    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']