Archived
snowy-yang
70ca21974b
为什么：auth_system 的 11 个子模块此前构成一个强连通分量——db.py 与 backup.py
为了「动态读取路径常量」而在模块级 `import auth_system`（包自身），而包 __init__
又再导出这两个子模块的全部符号，互相指成环。
这个环当前并不崩（实测 8/8 种导入姿势全通过），它的代价是**双绑定**：
`auth_system.DB_PATH is auth_system.db.DB_PATH` 初始为 True，但只 patch 其一
另一处不跟随。于是测试不得不把 4 个路径常量**两处都 patch**
（setattr(auth,...) + setattr(db,...)）——而漏 patch 的后果是静默写生产库，
commit fc7fc21 修的那次「黄金测试写进生产库」正是这么来的。
改法：让路径常量回归**模块属性单一来源**。
- db.py：删掉 `import auth_system`，内部改用本模块常量（DATA_DIR / DB_PATH /
SECRET_KEY_PATH）。这些常量本就定义在本模块，绕道包属性纯属多余。
- backup.py：改用 `import auth_system.db as db` 做模块限定访问，路径走
db.BACKUP_DIR / db.DB_PATH——保留动态读取能力，来源从「包属性」换成
「db 模块」，与唯一来源一致。
注意**不能**写成 `from . import db`：该形式在静态图里同时指向包本身与子模块，
会把这个环重新连上（实测确实如此，SCC 从 11 降到 2 就卡住了）。
- admin_site.py：5 处 `auth.DB_PATH` / `auth.BACKUP_DIR`（auth 是包别名）同样
改为经 db 模块访问。**这处不修就会出事**——它读的是包属性快照，而测试已改为
patch db 模块，于是备份上传会静默落到生产 data/backups/（本次实测真的发生了，
2 个 upload_*.db 已清理，副本留在临时目录）。
- 测试：7 个文件的 patch 目标从包属性迁到 auth_system.db.*。
- __init__.py：补注释说明这些常量在此只是 import 快照，禁止经包属性重定向。
验证：
- Tarjan 重算：非平凡强连通分量 11 → **0**（import 图已是 DAG）
- AST 全仓扫描：经包属性访问路径常量的点 **0 处残留**
- 行为：子进程内 patch auth_system.db.* 后，get_conn 连的库 / create_backup
落盘目录 / get_secret_key 目标路径**三者全部跟随**
- 导入姿势回归 8/8 通过
- 门禁：ruff 0 · pyright 0 errors 0 warnings · pytest 175 passed
82 lines
3.0 KiB
Python
82 lines
3.0 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""备份上传端点回归：钉住「用哪个 UploadFile 做 isinstance」这个坑。
|
||
|
||
坑的由来：`fastapi.UploadFile` 是 `starlette.datastructures.UploadFile` 的**子类**
|
||
（MRO: fastapi.datastructures.UploadFile → starlette.datastructures.UploadFile），
|
||
而 `request.form()` 返回的是 **starlette 那个基类**的实例。所以
|
||
`isinstance(f, fastapi.UploadFile)` 对真实上传恒为 False ——端点会把每一次上传
|
||
都判成「请选择数据库文件」，功能静默失效且不报错。
|
||
|
||
本测试用真实 multipart 请求穿过端点，确保三种分支（接受 / 后缀拒 / 内容拒）行为正确。
|
||
"""
|
||
|
||
import io
|
||
|
||
import pytest
|
||
from fastapi import FastAPI
|
||
from fastapi.testclient import TestClient
|
||
|
||
from web.deps import SiteUser
|
||
from web.routers import admin_site
|
||
|
||
UPLOAD_URL = '/admin/site/backup/upload'
|
||
SQLITE_MAGIC = b'SQLite format 3\x00'
|
||
|
||
|
||
@pytest.fixture()
|
||
def client(tmp_path, monkeypatch):
|
||
"""沙箱库 + 绕过管理员闸（本测试只关心 Form/UploadFile 处理，不测鉴权）。"""
|
||
import auth_system as auth
|
||
import core.db as db
|
||
|
||
monkeypatch.setattr(auth.db, 'DATA_DIR', tmp_path)
|
||
monkeypatch.setattr(auth.db, 'DB_PATH', tmp_path / 'site.db')
|
||
monkeypatch.setattr(auth.db, 'BACKUP_DIR', tmp_path / 'backups')
|
||
monkeypatch.setattr(auth.db, 'SECRET_KEY_PATH', tmp_path / 'secret.key')
|
||
monkeypatch.setattr(db, 'DATA_DIR', tmp_path)
|
||
monkeypatch.setattr(db, 'DB_PATH', tmp_path / 'site.db')
|
||
auth.init_db()
|
||
|
||
app = FastAPI()
|
||
app.include_router(admin_site.router)
|
||
app.dependency_overrides[admin_site._require_admin] = lambda: SiteUser(
|
||
id=1, username='boss', role='admin', credits=-1)
|
||
return TestClient(app)
|
||
|
||
|
||
def _upload(client, filename, payload):
|
||
return client.post(
|
||
UPLOAD_URL,
|
||
files={'file': (filename, io.BytesIO(payload), 'application/octet-stream')},
|
||
)
|
||
|
||
|
||
def test_real_upload_is_accepted(client, tmp_path):
|
||
"""真实 SQLite 文件必须被接收并落盘（若 isinstance 用错类，这里会拿到「请选择数据库文件」）。"""
|
||
r = _upload(client, 'up.db', SQLITE_MAGIC + b'\x00' * 512)
|
||
assert r.status_code == 200
|
||
body = r.json()
|
||
assert body['success'] is True, body
|
||
assert body['data']['name'].endswith('.db')
|
||
assert (tmp_path / 'backups' / body['data']['name']).exists()
|
||
|
||
|
||
def test_missing_file_field_is_rejected(client):
|
||
r = client.post(UPLOAD_URL, data={'note': 'no file here'})
|
||
assert r.json()['success'] is False
|
||
|
||
|
||
def test_non_db_extension_is_rejected(client):
|
||
r = _upload(client, 'evil.txt', SQLITE_MAGIC + b'\x00' * 64)
|
||
body = r.json()
|
||
assert body['success'] is False
|
||
assert '.db' in body['message']
|
||
|
||
|
||
def test_valid_extension_but_bad_content_is_rejected(client, tmp_path):
|
||
"""后缀对但内容不是 SQLite：必须拒绝，且不得把半成品留在备份目录。"""
|
||
r = _upload(client, 'fake.db', b'this is definitely not a sqlite database')
|
||
body = r.json()
|
||
assert body['success'] is False
|
||
assert not list((tmp_path / 'backups').glob('*.db'))
|