123 lines
4.1 KiB
Python
123 lines
4.1 KiB
Python
"""tests/fake_db.py — 离线测试夹具：sqlite3 承载 pbl_domain_ref + 三张复用基表只读投影。
|
||
|
||
用途：不依赖生产 MySQL / 宿主挂载即可跑真实 SQL 逻辑（唯一约束、软删、过滤、分页、
|
||
悬挂引用、跨租户隔离）。通过 db.set_adapter(SqliteAdapter(conn)) 注入。
|
||
|
||
注意：sqlite 不支持 information_schema，base.table_columns 会自动退化到 PRAGMA table_info。
|
||
"""
|
||
|
||
import os
|
||
import sqlite3
|
||
import sys
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from pbl_domain_ext import base as base_mod # noqa: E402
|
||
from pbl_domain_ext import db as db_mod # noqa: E402
|
||
from pbl_domain_ext.db import SqliteAdapter # noqa: E402
|
||
|
||
DDL_PBL_DOMAIN_REF = """
|
||
CREATE TABLE pbl_domain_ref (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
tenant_id TEXT NOT NULL,
|
||
ref_type TEXT NOT NULL,
|
||
ref_id INTEGER NOT NULL,
|
||
blueprint_id TEXT,
|
||
class_id TEXT,
|
||
team_id TEXT,
|
||
ext TEXT,
|
||
is_deleted INTEGER NOT NULL DEFAULT 0,
|
||
created_at TEXT NOT NULL DEFAULT (datetime('now')),
|
||
updated_at TEXT NOT NULL DEFAULT (datetime('now')),
|
||
UNIQUE (tenant_id, ref_type, ref_id)
|
||
)
|
||
"""
|
||
|
||
# 复用基表（测试替身）：结构对齐 world/scene/entity 关键列，本模块只读
|
||
DDL_WORLD = "CREATE TABLE world (id INTEGER PRIMARY KEY, name TEXT, owner TEXT)"
|
||
DDL_SCENE = "CREATE TABLE scene (id INTEGER PRIMARY KEY, name TEXT, world_id INTEGER)"
|
||
DDL_ENTITY = ("CREATE TABLE entity (id INTEGER PRIMARY KEY, name TEXT, "
|
||
"scene_id INTEGER, kind TEXT)")
|
||
DDL_TEAM_MEMBER = ("CREATE TABLE pbl_team_member (id INTEGER PRIMARY KEY, "
|
||
"tenant_id TEXT, team_id TEXT, user_id TEXT, role TEXT)")
|
||
|
||
|
||
def build_conn(with_team_member=True):
|
||
conn = sqlite3.connect(':memory:')
|
||
conn.execute(DDL_PBL_DOMAIN_REF)
|
||
conn.execute(DDL_WORLD)
|
||
conn.execute(DDL_SCENE)
|
||
conn.execute(DDL_ENTITY)
|
||
if with_team_member:
|
||
conn.execute(DDL_TEAM_MEMBER)
|
||
conn.commit()
|
||
return conn
|
||
|
||
|
||
def seed_base_data(conn):
|
||
"""灌入基表测试数据：2 world / 3 scene / 4 entity。"""
|
||
conn.executemany("INSERT INTO world (id, name, owner) VALUES (?,?,?)", [
|
||
(1, '世界A-火星基地', 'teacher01'),
|
||
(2, '世界B-海洋生态', 'teacher02'),
|
||
])
|
||
conn.executemany("INSERT INTO scene (id, name, world_id) VALUES (?,?,?)", [
|
||
(11, '场景A1-着陆区', 1),
|
||
(12, '场景A2-实验舱', 1),
|
||
(21, '场景B1-珊瑚礁', 2),
|
||
])
|
||
conn.executemany(
|
||
"INSERT INTO entity (id, name, scene_id, kind) VALUES (?,?,?,?)", [
|
||
(101, '实体A1a-探测车', 11, 'vehicle'),
|
||
(102, '实体A1b-宇航员', 11, 'avatar'),
|
||
(121, '实体A2a-培养皿', 12, 'prop'),
|
||
(211, '实体B1a-海龟', 21, 'creature'),
|
||
])
|
||
conn.commit()
|
||
|
||
|
||
def seed_team_members(conn, tenant_id='T1'):
|
||
conn.executemany(
|
||
"INSERT INTO pbl_team_member (tenant_id, team_id, user_id, role) VALUES (?,?,?,?)",
|
||
[(tenant_id, 'TEAM_A', 'stu01', 'leader'),
|
||
(tenant_id, 'TEAM_A', 'stu02', 'member'),
|
||
(tenant_id, 'TEAM_B', 'stu03', 'member')])
|
||
conn.commit()
|
||
|
||
|
||
def setup(with_team_member=True, seed=True):
|
||
"""构建夹具并注入适配器；返回 (conn, adapter)。"""
|
||
conn = build_conn(with_team_member=with_team_member)
|
||
if seed:
|
||
seed_base_data(conn)
|
||
if with_team_member:
|
||
seed_team_members(conn)
|
||
adapter = SqliteAdapter(conn)
|
||
db_mod.set_adapter(adapter)
|
||
db_mod.set_dbname('pbls_test')
|
||
base_mod.clear_cache()
|
||
return conn, adapter
|
||
|
||
|
||
def teardown():
|
||
db_mod.set_adapter(None)
|
||
db_mod.set_dbname(None)
|
||
base_mod.clear_cache()
|
||
|
||
|
||
def raw_sql(conn, sql, params=()):
|
||
"""测试辅助：临时摘掉 dict row_factory，按位置取标量/元组。"""
|
||
saved = conn.row_factory
|
||
conn.row_factory = None
|
||
try:
|
||
cur = conn.execute(sql, params)
|
||
rows = cur.fetchall()
|
||
cur.close()
|
||
return rows
|
||
finally:
|
||
conn.row_factory = saved
|
||
|
||
|
||
def scalar(conn, sql, params=()):
|
||
rows = raw_sql(conn, sql, params)
|
||
return rows[0][0] if rows else None
|