GaoYiYuan_0626
f056fcd007
AIcoding 第 5 步 todo 开发（开发计划 v2.0）前半段： - T-1 3 张新表 DDL（core_convert_request 6 态 ENUM / core_trade_calendar / core_share_rule）+ 种子 + sqlite 单一事实源同步 + 列清单断言 - T-2/T-2b calc 扩展（product_round/redeem_amount/partial_qty）+ 真实净值实算回填 - T-3 convert_request_repository（6 态 + 条件 UPDATE 守卫）+ core_ro 三读方法 + share_lot_repository.available_qty_with_inflight（R-3 在途占用推导） - T-4 convert_repository.sync_mirror 成为 risk_convert_detail 唯一进度镜像写入口 （旧三方法标 Deprecated，T-7 后删） - T-5 locks.py 锁键构造器 convert_req_lock_key / convert_confirm_lock_key - T-6 convert_service.accept_convert 受理事务（八步：锁→幂等→校验→受理日顺延 →在途占用校验→落单→镜像+审计→202；不扣份额/不折算/不写流水） + tests/test_convert_accept.py（16 用例） + scripts/dev/verify_convert_accept.py（真库 36/36 一致） + trading_calendar.py 纯函数包（R-5）+ 21 用例 T-6 真库实测暴露并修复：confirm_eta 在日历数据边界抛 ValueError，会让已落库 的受理单在调用方眼里变 500；改为展示性字段容错 + 单测守护。 基线：798 passed / 10 skipped，零回归。
304 lines
12 KiB
Python
304 lines
12 KiB
Python
"""引擎工厂（utils/db.py）单测（B7 评审 P1-1）：单例复用 + dispose 真实释放。
|
||
|
||
create_engine 与缓存字典均 monkeypatch 替换，不触网；验证 dispose_engines
|
||
对每个缓存 Engine 显式调用 dispose() 并清空缓存（不再是仅 cache_clear）。
|
||
|
||
另含 T-0 门禁用例（sqlite 结构对齐 + 门禁自身有效性反向验证）、T-0b 引擎角色
|
||
隔离用例（同库不同角色 / 账号未配置回退 / 非法角色报错），以及 3 条真 MySQL
|
||
权限断言（`xh_core_ro` 只读 · `audit_log` 只追加 · `xh_core_rw` 限 4 表）——
|
||
后者在账号未配置或真库不可达时自动 skip，不阻塞单测基线。
|
||
"""
|
||
|
||
import pytest
|
||
from sqlalchemy import text
|
||
|
||
from _ddl import EXPECTED_CORE_HOLDING_COLUMNS, create_sqlite_engine
|
||
from app.utils import db
|
||
|
||
|
||
def _unique_column_sets(conn, table: str) -> set[tuple[str, ...]]:
|
||
"""PRAGMA 取某表全部唯一约束的列组合（sqlite 自动索引 unique 字段为 1/'u'）。"""
|
||
result: set[tuple[str, ...]] = set()
|
||
for row in conn.execute(text(f"PRAGMA index_list({table})")):
|
||
# 行结构：seq, name, unique, origin, partial
|
||
if row[2]:
|
||
cols = tuple(r[2] for r in conn.execute(text(f"PRAGMA index_info({row[1]})")))
|
||
result.add(cols)
|
||
return result
|
||
|
||
|
||
class SpyEngine:
|
||
def __init__(self):
|
||
self.dispose_calls = 0
|
||
|
||
def dispose(self, close: bool = True) -> None:
|
||
self.dispose_calls += 1
|
||
|
||
|
||
def _patch(monkeypatch):
|
||
created_urls = []
|
||
|
||
def fake_create_engine(url, **kwargs):
|
||
created_urls.append(url)
|
||
return SpyEngine()
|
||
|
||
monkeypatch.setattr(db, "create_engine", fake_create_engine)
|
||
monkeypatch.setattr(db, "_engines", {})
|
||
return created_urls
|
||
|
||
|
||
def test_get_engine_caches_one_engine_per_database(monkeypatch):
|
||
created = _patch(monkeypatch)
|
||
e1 = db.get_engine("db_a")
|
||
e2 = db.get_engine("db_a")
|
||
e3 = db.get_engine("db_b")
|
||
assert e1 is e2
|
||
assert e1 is not e3
|
||
assert len(created) == 2
|
||
|
||
|
||
def test_dispose_engines_calls_dispose_and_clears_cache(monkeypatch):
|
||
_patch(monkeypatch)
|
||
e = db.get_engine("db_a")
|
||
db.dispose_engines()
|
||
assert e.dispose_calls == 1
|
||
assert db._engines == {}
|
||
# 缓存已清：重建新实例而非复用已 dispose 的旧实例
|
||
fresh = db.get_engine("db_a")
|
||
assert fresh is not e
|
||
assert fresh.dispose_calls == 0
|
||
|
||
|
||
def test_dispose_engines_on_empty_cache_is_noop(monkeypatch):
|
||
_patch(monkeypatch)
|
||
db.dispose_engines() # 不抛异常即可
|
||
|
||
|
||
# ---------- T-0 门禁：sqlite 结构对齐（R1）----------
|
||
|
||
|
||
def test_core_holding_columns(sqlite_engine):
|
||
"""sqlite core_holding 与 MySQL scripts/core/01-ddl.sql:132-145 逐列对齐。
|
||
|
||
convert 的首个阻断项正是列名失配（sqlite 曾为 quantity、MySQL 为 qty），
|
||
本用例把这类失配钉死在单测层，不让它留到集成测试期。
|
||
"""
|
||
with sqlite_engine.connect() as conn:
|
||
info = conn.execute(text("PRAGMA table_info(core_holding)")).all()
|
||
columns = {row[1] for row in info}
|
||
assert EXPECTED_CORE_HOLDING_COLUMNS <= columns, (
|
||
f"core_holding 缺列：{sorted(EXPECTED_CORE_HOLDING_COLUMNS - columns)}"
|
||
)
|
||
primary_key = {row[1] for row in info if row[5] > 0}
|
||
assert primary_key == {"id"}
|
||
assert ("customer_id", "product_id") in _unique_column_sets(conn, "core_holding")
|
||
|
||
|
||
def test_core_product_nav_columns(sqlite_engine):
|
||
"""core_product_nav 存在且含 (product_id, nav_date) 唯一约束（R-f）。"""
|
||
with sqlite_engine.connect() as conn:
|
||
tables = {
|
||
row[0]
|
||
for row in conn.execute(text("SELECT name FROM sqlite_master WHERE type = 'table'"))
|
||
}
|
||
assert "core_product_nav" in tables
|
||
columns = {row[1] for row in conn.execute(text("PRAGMA table_info(core_product_nav)"))}
|
||
assert {"id", "product_id", "nav", "daily_chg_pct", "nav_date"} <= columns
|
||
assert ("product_id", "nav_date") in _unique_column_sets(conn, "core_product_nav")
|
||
|
||
|
||
def test_t1_new_tables_and_columns(sqlite_engine):
|
||
"""T-1 新增 3 表 + core_product 4 列在 sqlite 建表源中存在（R-回归 14）。
|
||
|
||
MySQL 侧落点 scripts/core/01-ddl.sql；两侧双份同步的一致性由
|
||
REQUIRED_CONVERT_TABLES + 本用例列清单守护（R-g）。核心断言：
|
||
· core_convert_request 存在且有 status（6 态受理单权威状态机）
|
||
· core_trade_calendar 存在且有 cal_date/is_open（Q11 交易日历）
|
||
· core_share_rule 存在且有 product_id/business_type/share_digits（D27）
|
||
· core_product 含 T-1 补的 4 列（share_class/allow_ac_convert/
|
||
rounding_mode/share_digits）
|
||
"""
|
||
with sqlite_engine.connect() as conn:
|
||
tables = {
|
||
row[0]
|
||
for row in conn.execute(text("SELECT name FROM sqlite_master WHERE type = 'table'"))
|
||
}
|
||
for t in ("core_convert_request", "core_trade_calendar", "core_share_rule"):
|
||
assert t in tables, f"T-1 新表缺失：{t}"
|
||
req_cols = {row[1] for row in conn.execute(text("PRAGMA table_info(core_convert_request)"))}
|
||
assert {"convert_group_id", "client_request_id", "customer_id", "from_product_id",
|
||
"to_product_id", "qty", "actual_qty", "status", "cancel_before",
|
||
"requested_at", "confirmed_at", "coupon", "remark", "created_at", "updated_at"
|
||
} <= req_cols, f"core_convert_request 缺列：{sorted(req_cols)}"
|
||
cal_cols = {row[1] for row in conn.execute(text("PRAGMA table_info(core_trade_calendar)"))}
|
||
assert {"cal_date", "is_open", "remark"} <= cal_cols
|
||
rule_cols = {row[1] for row in conn.execute(text("PRAGMA table_info(core_share_rule)"))}
|
||
assert {"product_id", "business_type", "share_digits", "rounding_mode"} <= rule_cols
|
||
prod_cols = {row[1] for row in conn.execute(text("PRAGMA table_info(core_product)"))}
|
||
assert {"share_class", "allow_ac_convert", "rounding_mode", "share_digits"} <= prod_cols
|
||
|
||
|
||
def test_ddl_alignment_guard_fails_when_column_missing(monkeypatch):
|
||
"""反向验证门禁自身有效：抽掉 qty 列 → 建库即 AssertionError。"""
|
||
import _ddl
|
||
|
||
broken = dict(_ddl.SQLITE_TABLES)
|
||
broken["core_holding"] = """
|
||
CREATE TABLE core_holding (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
customer_id VARCHAR(64) NOT NULL, product_id VARCHAR(64) NOT NULL,
|
||
market_value DECIMAL NOT NULL, UNIQUE (customer_id, product_id))
|
||
"""
|
||
monkeypatch.setattr(_ddl, "SQLITE_TABLES", broken)
|
||
with pytest.raises(AssertionError, match="core_holding 缺列"):
|
||
create_sqlite_engine()
|
||
|
||
|
||
# ---------- T-0b：引擎角色隔离（D20 · 缓存键 (库名, 角色)）----------
|
||
|
||
|
||
def test_get_engine_role_isolation(monkeypatch):
|
||
"""同库不同角色 → 不同 engine（缓存键由「库名」改为「(库名, 角色)」）。"""
|
||
created = _patch(monkeypatch)
|
||
ro = db.get_engine("db_a", "ro")
|
||
rw = db.get_engine("db_a", "rw")
|
||
admin = db.get_engine("db_a", "admin")
|
||
assert ro is not rw
|
||
assert rw is not admin
|
||
assert ro is not admin
|
||
assert db.get_engine("db_a", "ro") is ro # 同角色仍复用
|
||
assert len(created) == 3
|
||
|
||
|
||
def test_get_engine_role_accounts_and_fallback(monkeypatch):
|
||
"""ro/rw 各用自己账号；账号留空 → 回退 mysql_user（渐进启用，不阻塞开发）。"""
|
||
from app.config.settings import settings
|
||
|
||
for name, value in (
|
||
("mysql_host", "127.0.0.1"),
|
||
("mysql_port", 3306),
|
||
("mysql_user", "root"),
|
||
("mysql_password", "rootpw"),
|
||
("mysql_database", "jinrong_agent"),
|
||
("mysql_core_database", "jinrong_core"),
|
||
("mysql_core_ro_user", ""),
|
||
("mysql_core_ro_password", ""),
|
||
("mysql_core_rw_user", ""),
|
||
("mysql_core_rw_password", ""),
|
||
("mysql_agent_user", ""),
|
||
("mysql_agent_password", ""),
|
||
):
|
||
monkeypatch.setattr(settings, name, value)
|
||
created = _patch(monkeypatch)
|
||
|
||
db.get_engine("jinrong_core", "ro")
|
||
db.get_engine("jinrong_core", "rw")
|
||
db.get_engine("jinrong_agent", "rw")
|
||
assert len(created) == 3
|
||
# 三组账号均未配置 → 全部回退 mysql_user（与单账号时代行为一致）
|
||
assert all("root:rootpw@" in url for url in created)
|
||
|
||
# 配置账号后：各角色取各自账号
|
||
db.dispose_engines()
|
||
created.clear()
|
||
for name, value in (
|
||
("mysql_core_ro_user", "xh_core_ro"),
|
||
("mysql_core_ro_password", "p1"),
|
||
("mysql_core_rw_user", "xh_core_rw"),
|
||
("mysql_core_rw_password", "p2"),
|
||
("mysql_agent_user", "xh_agent_rw"),
|
||
("mysql_agent_password", "p3"),
|
||
):
|
||
monkeypatch.setattr(settings, name, value)
|
||
|
||
db.get_engine("jinrong_core", "ro")
|
||
db.get_engine("jinrong_core", "rw")
|
||
db.get_engine("jinrong_agent", "rw")
|
||
assert "xh_core_ro:p1@" in created[0]
|
||
assert "xh_core_rw:p2@" in created[1]
|
||
assert "xh_agent_rw:p3@" in created[2]
|
||
assert "/jinrong_core?" in created[0]
|
||
assert "/jinrong_core?" in created[1]
|
||
assert "/jinrong_agent?" in created[2]
|
||
# agent 库无独立只读账号 → ro 亦走 mysql_agent_user（不静默落到 root），
|
||
# 但缓存键 (库, 角色) 不同，因此仍是独立 engine
|
||
agent_ro = db.get_engine("jinrong_agent", "ro")
|
||
assert "xh_agent_rw:p3@" in created[3]
|
||
assert agent_ro is not db.get_engine("jinrong_agent", "rw")
|
||
|
||
|
||
def test_get_engine_rejects_unknown_role(monkeypatch):
|
||
"""非法角色立即报错：防拼写错误静默取到默认账号。"""
|
||
_patch(monkeypatch)
|
||
with pytest.raises(ValueError, match="unknown db role"):
|
||
db.get_engine("db_a", "readonly")
|
||
|
||
|
||
# ---------- T-0b：真 MySQL 权限断言（账号未配置则 skip）----------
|
||
# 前置：管理员执行 scripts/core/00-grant.sql，把 3 组账号写入 .env。
|
||
# 未配置时不阻塞单测基线（本文件默认全绿）。
|
||
|
||
|
||
def _role_engine_or_skip(role: str, database_setting: str, user_setting: str):
|
||
"""取角色 engine；账号未配置或真库不可达 → skip 并给出 bootstrap 提示。"""
|
||
from app.config.settings import settings
|
||
from app.utils.db import get_engine
|
||
|
||
if not getattr(settings, user_setting):
|
||
pytest.skip(
|
||
f"未配置 {user_setting}：先执行 scripts/core/00-grant.sql 并写入 .env"
|
||
)
|
||
engine = get_engine(getattr(settings, database_setting), role)
|
||
try:
|
||
with engine.connect() as conn:
|
||
conn.execute(text("SELECT 1"))
|
||
except Exception as exc:
|
||
pytest.skip(f"真 MySQL 不可用（{exc}）")
|
||
return engine
|
||
|
||
|
||
def test_core_ro_account_is_readonly():
|
||
"""xh_core_ro 只有 SELECT：INSERT 被数据库拒绝（Core 只读 DB 级强制）。"""
|
||
from sqlalchemy.exc import SQLAlchemyError
|
||
|
||
engine = _role_engine_or_skip("ro", "mysql_core_database", "mysql_core_ro_user")
|
||
with pytest.raises(SQLAlchemyError) as exc:
|
||
with engine.begin() as conn:
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_trade (trade_id, customer_id, product_id, trade_type,"
|
||
" amount, trade_status, traded_at)"
|
||
" VALUES ('TRD-TEST-RO-DENY', '__x__', '__x__', 'subscribe', 1,"
|
||
" 'confirmed', NOW())"
|
||
)
|
||
)
|
||
assert "denied" in str(exc.value).lower()
|
||
|
||
|
||
def test_audit_log_append_only():
|
||
"""audit_log 只允许追加：UPDATE / DELETE 均被拒（审计红线 DB 级强制）。"""
|
||
from sqlalchemy.exc import SQLAlchemyError
|
||
|
||
engine = _role_engine_or_skip("rw", "mysql_database", "mysql_agent_user")
|
||
for stmt in (
|
||
"UPDATE audit_log SET decision = 'tampered' WHERE id = -1",
|
||
"DELETE FROM audit_log WHERE id = -1",
|
||
):
|
||
with pytest.raises(SQLAlchemyError) as exc:
|
||
with engine.begin() as conn:
|
||
conn.execute(text(stmt))
|
||
assert "denied" in str(exc.value).lower(), stmt
|
||
|
||
|
||
def test_core_rw_scope():
|
||
"""xh_core_rw 限 4 表：对第 5 张表 core_product 的写被拒（最小权限）。"""
|
||
from sqlalchemy.exc import SQLAlchemyError
|
||
|
||
engine = _role_engine_or_skip("rw", "mysql_core_database", "mysql_core_rw_user")
|
||
with pytest.raises(SQLAlchemyError) as exc:
|
||
with engine.begin() as conn:
|
||
conn.execute(
|
||
text("UPDATE core_product SET product_name = '__x__' WHERE product_id = '__x__'")
|
||
)
|
||
assert "denied" in str(exc.value).lower()
|