mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:11:59 +00:00
ThearchyHelios
8059cd5147
连接层 - 所有 SQLite 数据源关闭双引号字符串兼容（DQS），写错的双引号列名直接报错；设置失败就拒绝连接，启动时自检 - 连接缓存按数据源加版本快照分键、按源整体失效、最多 32 个；版本快照以 immutable 只读打开，建连接前核对文件哈希，并在借出连接时核对文件指纹，被改过就拒绝查询 - 结构探查遇到取不了列的视图时逐个对象重取，只跳过坏的那个并注明原因 解析（新 xlsx_scan + tabular） - 先流式扫描真实边界和版式（不信 dimension），再按边界读值，不整表进内存；按 Excel 自身上限设硬上限，并发解析最多两个 - 隐藏和深度隐藏的工作表不导入；稀疏区域、无缓存值的公式、同名工作表、DTD 拒收 - 去掉空行和两侧空列；列名表名统一规范化（新 names 模块）并保留原表头，撞名不再互相覆盖 - 前导零、超长数字按文本；千分位按数字；日期不带 T；带单位的数字文本写进回执 - 数字列混入非数字、交叉表和多块结构必须由用户决定（422），可选存空值或按原样导入（未规整）；STRICT 建表，整份在一个事务里写完 版本底座（新 table_versions、raw_store） - 每次上传写新库文件、落盘、0444，事务里写构建、导入记录、快照后才切指针；失败时旧版完整可查 - 原件按哈希存档，不提供下载，可以清除；上传源不许改连接字段、不再重新探查；手工 SQLite 源不许指向上传目录和应用自己的数据文件（按文件身份判断） - 启动时清理半成品与孤儿文件、给老上传补初始版本；被运行引用的版本不回收，发起运行与回收互斥 - 上传接口的解析移到线程池 运行固定数据版本 - 发起时把图里（含子工作流）用到的上传源的当前快照记进运行记录和 run.started，续跑、审批恢复沿用；数据源工具按固定的快照生成描述、表结构并查询 - 固定的版本没了、源被删除后同名重建，都明确报错，不改用当前版本或同名的新源；查询快照记 data_version 证据层 - 新报告记 entity_syntax: 2，[[t:明细]]、[[c:明细.金额]] 和 SQL 里不加引号的中文表名都认得，名字按 SQLite 的口径查找；老报告按原规则复核，结果不变 - 写作规则改为中文表名、字段名一律用 [[t:]] / [[c:]]；未规整的表在工具描述、写作目录和裁判摘录里标出 前端 - 上传弹窗接上导入前的决定（混合列、交叉表）和导入回执（跳过的隐藏表、类型转换、警告、原表头对照、未规整标记），上传前告知原件会保存 - 数据源卡片按 origin 认上传的表格，显示当前版本；上传的表格不再给「探查结构」
230 lines
9.9 KiB
Python
230 lines
9.9 KiB
Python
"""SQLite 连接一律关掉 DQS：写错的双引号列名必须报错，而不是被当成字符串。
|
||
|
||
DQS（double-quoted string）是 SQLite 的历史兼容：双引号里的名字找不到列时当成字符串。
|
||
于是 `SUM("金颔")` 得 0、`GROUP BY "金颔"` 把整张表归成一组，查询不报错，结论却是错的。
|
||
中文列名常被加上双引号，正好踩中。
|
||
|
||
这组测试覆盖项目实际用到的三种连接形式（只读源的 mode=ro URI、可写源的普通路径、
|
||
表单里的一次性测试连接），以及关不掉时连接被拒、启动自检。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import sqlite3
|
||
import uuid
|
||
from types import SimpleNamespace
|
||
|
||
import aiosqlite
|
||
import pytest
|
||
from sqlalchemy import text
|
||
from sqlalchemy.exc import OperationalError
|
||
from sqlalchemy.ext.asyncio import create_async_engine
|
||
from sqlalchemy.pool import NullPool
|
||
|
||
from app.data import engine as engine_mod
|
||
from app.data.engine import engine_args, engines, run_query, sqlite_dqs_self_check
|
||
|
||
|
||
def _db(path) -> str:
|
||
conn = sqlite3.connect(path)
|
||
conn.execute("CREATE TABLE 时段客流 (分区 TEXT, 客流 INTEGER)")
|
||
conn.executemany("INSERT INTO 时段客流 VALUES (?, ?)",
|
||
[("分区甲", 120), ("分区乙", 80), ("分区甲", 30)])
|
||
conn.commit()
|
||
conn.close()
|
||
return str(path)
|
||
|
||
|
||
def _source(path: str, *, readonly: bool = True):
|
||
return SimpleNamespace(
|
||
id=f"dqs-{uuid.uuid4().hex[:8]}", kind="sqlite", database=path, readonly=readonly,
|
||
name="dqs_probe", options={}, password=None, username=None, host=None, port=None,
|
||
)
|
||
|
||
|
||
@pytest.fixture
|
||
async def ro_source(tmp_path):
|
||
src = _source(_db(tmp_path / "客流 样例.db"))
|
||
yield src
|
||
await engines.invalidate(src.id)
|
||
|
||
|
||
@pytest.fixture
|
||
async def rw_source(tmp_path):
|
||
src = _source(_db(tmp_path / "rw.db"), readonly=False)
|
||
yield src
|
||
await engines.invalidate(src.id)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 三种连接形式下，写错的双引号列名都报错
|
||
# ---------------------------------------------------------------------------
|
||
|
||
async def test_misspelled_quoted_column_errors_on_readonly_uri_form(ro_source):
|
||
"""只读源走 sqlite+aiosqlite:///file:…?mode=ro&uri=true——critic C4 实测踩坑的正是这个形式。"""
|
||
url, _ = engine_args(ro_source)
|
||
assert url.startswith("sqlite+aiosqlite:///file:") and url.endswith("?mode=ro&uri=true")
|
||
|
||
for sql in ('SELECT SUM("客六") FROM 时段客流',
|
||
'SELECT "客六", COUNT(*) FROM 时段客流 GROUP BY "客六"',
|
||
'SELECT 分区 FROM 时段客流 WHERE "分去" = 1'):
|
||
with pytest.raises(OperationalError, match="no such column"):
|
||
await run_query(ro_source, sql)
|
||
|
||
# 写对的双引号列名照常可用
|
||
ok = await run_query(ro_source, 'SELECT SUM("客流") AS n FROM "时段客流"')
|
||
assert ok.rows == [[230]]
|
||
# 单引号字符串照常是字符串
|
||
lit = await run_query(ro_source, "SELECT COUNT(*) FROM 时段客流 WHERE 分区 = '分区甲'")
|
||
assert lit.rows == [[2]]
|
||
|
||
|
||
async def test_misspelled_quoted_column_errors_on_writable_form(rw_source):
|
||
"""可写源走普通路径；DDL 里的双引号字符串也一并关掉。"""
|
||
url, _ = engine_args(rw_source)
|
||
assert "mode=ro" not in url
|
||
|
||
with pytest.raises(OperationalError, match="no such column"):
|
||
await run_query(rw_source, 'SELECT SUM("客六") FROM 时段客流')
|
||
|
||
engine = await engines.get(rw_source)
|
||
async with engine.connect() as conn:
|
||
with pytest.raises(OperationalError, match="no such column"):
|
||
await conn.execute(text('CREATE INDEX 错索引 ON 时段客流 ("不存在")'))
|
||
|
||
|
||
async def test_one_off_probe_engine_is_covered_too(ro_source):
|
||
"""表单里的「测试连接」用 engine_args 现建一次性 engine（datasources._probe），不走缓存，
|
||
也得关 DQS：关闭措施挂在 engine_args 给出的连接参数上，而不是只挂在缓存建的 engine 上。"""
|
||
url, extra = engine_args(ro_source)
|
||
engine = create_async_engine(url, poolclass=NullPool, **extra)
|
||
try:
|
||
async with engine.connect() as conn:
|
||
with pytest.raises(OperationalError, match="no such column"):
|
||
await conn.execute(text('SELECT SUM("客六") FROM 时段客流'))
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
async def test_aiosqlite_passes_factory_through():
|
||
"""钉住中间那一层：aiosqlite.connect 把关键字参数原样交给 sqlite3.connect。
|
||
|
||
关 DQS 靠 sqlite3.connect 的 factory 参数，一路经 SQLAlchemy connect_args →
|
||
aiosqlite.connect(**kwargs) → sqlite3.connect(factory=…) 传下去。哪天 aiosqlite
|
||
不再转交，这里先失败。
|
||
"""
|
||
async with aiosqlite.connect(":memory:", factory=engine_mod._DqsOffConnection) as conn:
|
||
with pytest.raises(sqlite3.OperationalError, match="no such column"):
|
||
await conn.execute('SELECT "不存在"')
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 关不掉就拒绝这条连接
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _raise_on_setconfig(self, op, enable=True):
|
||
raise sqlite3.OperationalError("模拟：setconfig 失败")
|
||
|
||
|
||
def _ignore_setconfig(self, op, enable=True):
|
||
return None # 调用成功，但什么也没设上
|
||
|
||
|
||
@pytest.mark.parametrize("fake", [_raise_on_setconfig, _ignore_setconfig],
|
||
ids=["setconfig 抛错", "setconfig 没生效"])
|
||
async def test_connection_is_refused_when_dqs_cannot_be_disabled(ro_source, monkeypatch, fake):
|
||
monkeypatch.setattr(engine_mod._DqsOffConnection, "setconfig", fake)
|
||
engine = await engines.get(ro_source)
|
||
with pytest.raises(OperationalError, match="未能关闭 SQLite 的双引号字符串兼容，已拒绝连接"):
|
||
async with engine.connect() as conn:
|
||
await conn.execute(text("SELECT 1"))
|
||
# 走查询入口同样连不上，不会退回一条没关 DQS 的连接
|
||
with pytest.raises(OperationalError, match="已拒绝连接"):
|
||
await run_query(ro_source, "SELECT COUNT(*) FROM 时段客流")
|
||
|
||
|
||
async def test_connection_is_refused_when_sqlite_lacks_the_option(ro_source, monkeypatch):
|
||
"""Python 3.12 之前的 sqlite3 没有这两个常量：照样拒绝，而不是当作没这回事。"""
|
||
monkeypatch.setattr(engine_mod, "_DQS_OPTIONS", (("DML", None), ("DDL", None)))
|
||
engine = await engines.get(ro_source)
|
||
with pytest.raises(OperationalError, match="已拒绝连接"):
|
||
async with engine.connect() as conn:
|
||
await conn.execute(text("SELECT 1"))
|
||
|
||
|
||
def test_refusal_is_a_driver_error():
|
||
"""拒绝用的异常是 sqlite3.OperationalError 的子类：SQLAlchemy 按驱动异常包装，
|
||
上层照常当作「连接失败」处理。"""
|
||
assert issubclass(engine_mod.SqliteHardeningError, sqlite3.OperationalError)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 启动自检
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_self_check_passes_on_this_platform():
|
||
sqlite_dqs_self_check()
|
||
|
||
|
||
@pytest.mark.parametrize("fake", [_raise_on_setconfig, _ignore_setconfig],
|
||
ids=["setconfig 抛错", "setconfig 没生效"])
|
||
def test_self_check_fails_when_dqs_cannot_be_disabled(monkeypatch, fake):
|
||
monkeypatch.setattr(engine_mod._DqsOffConnection, "setconfig", fake)
|
||
with pytest.raises(RuntimeError, match="SQLite 自检失败"):
|
||
sqlite_dqs_self_check()
|
||
|
||
|
||
def test_self_check_fails_when_quoted_names_still_pass(monkeypatch):
|
||
"""连接建得起来、却没关 DQS（比如关闭措施被人改没了）：自检得看结果，不只看有没有抛错。"""
|
||
monkeypatch.setattr(engine_mod, "_disable_dqs", lambda conn: None)
|
||
with pytest.raises(RuntimeError, match="没有报错"):
|
||
sqlite_dqs_self_check()
|
||
|
||
|
||
def test_self_check_catches_ddl_left_on(monkeypatch):
|
||
"""只关了 DML 那项、DDL 里的双引号字符串还开着：查询探针照常报错，建索引探针不报——
|
||
自检得靠建索引那条把它认出来。"""
|
||
dml_only = tuple(item for item in engine_mod._DQS_OPTIONS if item[0] == "DML")
|
||
assert len(dml_only) == 1
|
||
monkeypatch.setattr(engine_mod, "_DQS_OPTIONS", dml_only)
|
||
with pytest.raises(RuntimeError, match="CREATE INDEX") as info:
|
||
sqlite_dqs_self_check()
|
||
assert "没有报错" in str(info.value)
|
||
|
||
|
||
@pytest.mark.parametrize("probe", ["SELECT a FROM no_such_table",
|
||
'SELECT "no_such_column" FROM'],
|
||
ids=["查不存在的表", "语法错误"])
|
||
def test_self_check_rejects_the_wrong_kind_of_error(monkeypatch, probe):
|
||
"""探针因为别的原因报错，说明它没走到双引号名字那一步，什么也没验到：自检照样失败，
|
||
不能见到报错就当通过。"""
|
||
monkeypatch.setattr(engine_mod, "_SELF_CHECK_PROBES", (probe,))
|
||
with pytest.raises(RuntimeError, match="不是预期的「no such column」"):
|
||
sqlite_dqs_self_check()
|
||
|
||
|
||
def test_self_check_fails_when_the_probe_table_cannot_be_made(monkeypatch):
|
||
monkeypatch.setattr(engine_mod, "_SELF_CHECK_SETUP", "CREATE TABLE")
|
||
with pytest.raises(RuntimeError, match="SQLite 自检失败"):
|
||
sqlite_dqs_self_check()
|
||
|
||
|
||
async def test_startup_runs_the_self_check_first(monkeypatch):
|
||
"""main 的启动流程第一步就自检，失败时服务起不来（异常从 lifespan 抛出）。"""
|
||
from app.main import app, lifespan, settings
|
||
|
||
calls: list[str] = []
|
||
|
||
def failing() -> None:
|
||
calls.append("self_check")
|
||
raise RuntimeError("SQLite 自检失败：模拟")
|
||
|
||
def must_not_run(self) -> None:
|
||
calls.append("ensure_dirs")
|
||
|
||
monkeypatch.setattr(engine_mod, "sqlite_dqs_self_check", failing)
|
||
monkeypatch.setattr(type(settings), "ensure_dirs", must_not_run)
|
||
with pytest.raises(RuntimeError, match="SQLite 自检失败"):
|
||
async with lifespan(app):
|
||
pass
|
||
assert calls == ["self_check"]
|