546 lines
23 KiB
Python
546 lines
23 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""pbl_blueprint M1a 契约测试（走真实模块加载路径：load_pbl_blueprint + 门面函数）。
|
||
|
||
覆盖：注册三处同步、表清单/字段类型、租户强制、CRUD 正反用例、状态机、
|
||
版本快照、fork、子对象 7 类、模板抽离与实例化、dspy/RBAC/i18n 一致性、SQL 方言自检。
|
||
DB 替身：tests/fakedb.py（模拟 sqlor C/U/D/R/I/sqlExe 语义），
|
||
真实库连通性另见 test_real_path.py。
|
||
"""
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
|
||
import pytest
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
REPO = os.path.dirname(HERE)
|
||
PKG = os.path.join(REPO, 'pbl_blueprint')
|
||
sys.path.insert(0, REPO)
|
||
sys.path.insert(0, HERE)
|
||
|
||
import fakedb # noqa: E402
|
||
|
||
|
||
@pytest.fixture()
|
||
def mod():
|
||
"""每个用例一套干净的内存 DB + 真实 load_pbl_blueprint() 注册路径。"""
|
||
sor, env, store = fakedb.install(tenant_id='T1')
|
||
for name in [m for m in list(sys.modules) if m.startswith('pbl_blueprint')]:
|
||
sys.modules.pop(name, None)
|
||
import pbl_blueprint as pkg
|
||
info = pkg.load_pbl_blueprint(env)
|
||
yield pkg, info, store, env
|
||
sys.modules.pop('sqlor', None)
|
||
|
||
|
||
# ---------------------------------------------------------------- 注册接线
|
||
def test_01_load_registers_tables_and_crud(mod):
|
||
pkg, info, store, env = mod
|
||
assert info['module'] == 'pbl_blueprint'
|
||
assert info['dbname'] == 'pbl_test_db'
|
||
assert len(info['tables']) == 11
|
||
assert sorted(info['tables']) == sorted(pkg.TABLES)
|
||
assert sorted(info['crud']) == sorted(pkg.TABLES)
|
||
# env 上挂载了表定义 / CRUD / 门面 / 路径 / i18n
|
||
assert getattr(env, 'pbl_blueprint_tables')
|
||
assert getattr(env, 'pbl_crud_defs')['pbl_blueprint']['params']['editable']
|
||
assert getattr(env, 'pbl_blueprint_api')['blueprint_save'] is pkg.api_blueprint_save
|
||
assert len(getattr(env, 'pbl_blueprint_paths')['api']) == 25
|
||
assert getattr(env, 'pbl_blueprint_i18n').get('zh_CN')
|
||
|
||
|
||
def test_02_three_way_sync_of_functions(mod):
|
||
"""函数三处同步：定义 + __init__ 导出 + init.py 注册（env.pbl_blueprint_api）。"""
|
||
pkg, _info, _store, env = mod
|
||
exported = set(getattr(pkg, '__all__'))
|
||
registry = set(getattr(env, 'pbl_blueprint_api').keys())
|
||
assert registry == set(pkg.API_REGISTRY.keys())
|
||
for fname, fn in pkg.API_REGISTRY.items():
|
||
assert callable(fn), fname
|
||
assert 'api_' + fname in exported or fname in exported, fname
|
||
# load 入口本身三处同步
|
||
assert 'load_pbl_blueprint' in exported
|
||
assert callable(getattr(pkg, 'load_pbl_blueprint'))
|
||
|
||
|
||
def test_03_dbname_not_hardcoded(mod):
|
||
pkg, _i, _s, _e = mod
|
||
pat = re.compile(r'''dbname\s*=\s*["'][^"']+["']''')
|
||
for f in sorted(os.listdir(PKG)):
|
||
if not f.endswith('.py'):
|
||
continue
|
||
text = open(os.path.join(PKG, f), encoding='utf-8').read()
|
||
for line_no, line in enumerate(text.splitlines(), 1):
|
||
if pat.search(line) and 'def ' not in line and 'kw' not in line:
|
||
raise AssertionError('%s:%d 疑似硬编码库名: %s' % (f, line_no, line.strip()))
|
||
|
||
|
||
def test_04_missing_dbname_fails_closed(mod):
|
||
pkg, _i, _s, env = mod
|
||
env.set_module_dbname('pbl_blueprint', None)
|
||
res = pkg.api_blueprint_list({'page': 1})
|
||
assert res['success'] is False
|
||
assert res['error']['code'] == 'PBL-DB-001'
|
||
|
||
|
||
def test_05_missing_tenant_fails_closed(mod):
|
||
pkg, _i, _s, env = mod
|
||
env.pbl_tenant_ctx = {}
|
||
res = pkg.api_blueprint_save({'blueprint_name': '无租户'})
|
||
assert res['success'] is False
|
||
assert res['error']['code'] == 'PBL-TENANT-001'
|
||
|
||
|
||
# ---------------------------------------------------------------- 表定义规范
|
||
def test_06_model_four_segment_and_abstract_types():
|
||
for fn in sorted(os.listdir(os.path.join(PKG, 'models'))):
|
||
if not fn.endswith('.json'):
|
||
continue
|
||
doc = json.load(open(os.path.join(PKG, 'models', fn), encoding='utf-8'))
|
||
assert set(['summary', 'fields', 'indexes', 'codes']).issubset(doc.keys()), fn
|
||
assert doc['fields']['id']['type'] == 'str(32)', fn
|
||
assert doc['fields']['id'].get('pk') is True, fn
|
||
assert 'tenant_id' in doc['fields'], fn
|
||
for col, spec in doc['fields'].items():
|
||
ty = spec['type']
|
||
assert not re.match(r'^(varchar|bigint|boolean|int\(11\)|char|tinyint)', ty), \
|
||
'%s.%s 使用了物理类型 %s' % (fn, col, ty)
|
||
if ty.startswith('double'):
|
||
assert ty == 'double(18,2)', '%s.%s' % (fn, col)
|
||
|
||
|
||
def test_07_crud_json_root_key_is_tblname():
|
||
for fn in sorted(os.listdir(os.path.join(PKG, 'json'))):
|
||
if not fn.endswith('.json'):
|
||
continue
|
||
tbl = fn[:-5]
|
||
doc = json.load(open(os.path.join(PKG, 'json', fn), encoding='utf-8'))
|
||
assert list(doc.keys()) == [tbl], fn
|
||
params = doc[tbl]['params']
|
||
assert 'editable' in params and 'browserfields' in params, fn
|
||
for k in ('new_data_url', 'update_data_url', 'delete_data_url'):
|
||
assert k in params['editable'], fn
|
||
assert 'wwwroot' not in json.dumps(params), fn
|
||
model_fields = json.load(
|
||
open(os.path.join(PKG, 'models', fn), encoding='utf-8'))['fields']
|
||
for col in params['editable']['editexclouded'] + params['browserfields']['fields']:
|
||
assert col in model_fields, '%s 字段 %s 不在 models 定义中' % (fn, col)
|
||
|
||
|
||
def test_08_sql_ddl_dialect_self_check():
|
||
sql = open(os.path.join(PKG, 'sql', 'pbl_blueprint.core.sql'), encoding='utf-8').read()
|
||
for bad in ('SERIAL', 'BIGSERIAL', 'nextval', 'AUTO_INCREMENT', 'boolean', 'int(11)'):
|
||
assert bad not in sql, 'SQL 含禁用写法 %s' % bad
|
||
assert 'engine: postgresql' in sql
|
||
for tbl in json.loads(json.dumps(__import__('pbl_blueprint', fromlist=['TABLES']).TABLES)):
|
||
pass
|
||
import pbl_blueprint as pkg
|
||
for tbl in pkg.TABLES:
|
||
assert 'CREATE TABLE IF NOT EXISTS %s' % tbl in sql, tbl
|
||
|
||
|
||
# ---------------------------------------------------------------- 蓝图 CRUD
|
||
def _new_blueprint(pkg, name='城市水资源调查'):
|
||
res = pkg.api_blueprint_save({'blueprint_name': name, 'domain_code': 'science',
|
||
'grade_code': 'primary_5', 'subject_code': 'science',
|
||
'project_duration': 12, 'student_count': 40,
|
||
'group_count': 8, 'complexity_level': 'medium',
|
||
'estimated_cost': 1250.50})
|
||
assert res['success'], res
|
||
return res['data']
|
||
|
||
|
||
def test_10_create_blueprint_auto_code_and_tenant(mod):
|
||
pkg, _i, store, _e = mod
|
||
bp = _new_blueprint(pkg)
|
||
assert bp['id'] and len(bp['id']) == 32
|
||
assert re.match(r'^PBL-\d{4}-\d{4}$', bp['blueprint_code']), bp['blueprint_code']
|
||
assert bp['tenant_id'] == 'T1'
|
||
assert bp['status'] == 'draft'
|
||
assert float(bp['estimated_cost']) == 1250.50
|
||
assert 'pbl_blueprint' in store
|
||
|
||
|
||
def test_11_create_requires_name(mod):
|
||
pkg, _i, _s, _e = mod
|
||
res = pkg.api_blueprint_save({'domain_code': 'science'})
|
||
assert res['success'] is False
|
||
assert res['error']['code'] == 'PBL-PARAM-001'
|
||
|
||
|
||
def test_12_duplicate_code_rejected(mod):
|
||
pkg, _i, _s, _e = mod
|
||
bp = _new_blueprint(pkg)
|
||
res = pkg.api_blueprint_save({'blueprint_name': '重名编码',
|
||
'blueprint_code': bp['blueprint_code']})
|
||
assert res['success'] is False
|
||
assert '已存在' in res['error']['msg']
|
||
|
||
|
||
def test_13_update_and_published_readonly(mod):
|
||
pkg, _i, _s, _e = mod
|
||
bp = _new_blueprint(pkg)
|
||
res = pkg.api_blueprint_save({'id': bp['id'], 'blueprint_name': '改名成功'})
|
||
assert res['success'] and res['data']['blueprint_name'] == '改名成功'
|
||
# 发布后禁止直接编辑
|
||
pkg.api_blueprint_publish({'id': bp['id']}, _seed_subobjects(pkg, bp['id']))
|
||
res2 = pkg.api_blueprint_save({'id': bp['id'], 'blueprint_name': '非法改'})
|
||
assert res2['success'] is False
|
||
assert res2['error']['code'] == 'PBL-STATE-001'
|
||
|
||
|
||
def test_14_list_pagination_and_filters(mod):
|
||
pkg, _i, _s, _e = mod
|
||
for n in range(5):
|
||
_new_blueprint(pkg, '蓝图%d' % n)
|
||
res = pkg.api_blueprint_list({'page': 1, 'rows': 2})
|
||
assert res['success']
|
||
assert res['data']['total'] == 5 and len(res['data']['rows']) == 2
|
||
res2 = pkg.api_blueprint_list({'blueprint_name': '蓝图3'})
|
||
assert res2['data']['total'] == 1
|
||
res3 = pkg.api_blueprint_list({'sortby': 'evil; drop table'})
|
||
assert res3['success'] and res3['data']['total'] == 5 # 非法排序被白名单兜住
|
||
|
||
|
||
def test_15_tenant_isolation(mod):
|
||
pkg, _i, store, env = mod
|
||
bp = _new_blueprint(pkg)
|
||
env.pbl_tenant_ctx = {'tenant_id': 'T2'}
|
||
res = pkg.api_blueprint_get({'id': bp['id']})
|
||
assert res['success'] is False
|
||
assert res['error']['code'] == 'PBL-NOT-FOUND'
|
||
res2 = pkg.api_blueprint_list({})
|
||
assert res2['data']['total'] == 0
|
||
|
||
|
||
def test_16_soft_delete_filters_out(mod):
|
||
pkg, _i, store, _e = mod
|
||
bp = _new_blueprint(pkg)
|
||
res = pkg.api_blueprint_delete({'id': bp['id']})
|
||
assert res['success'] and res['data']['deleted'] is True
|
||
assert store['pbl_blueprint'][bp['id']]['is_deleted'] == 1
|
||
assert pkg.api_blueprint_get({'id': bp['id']})['success'] is False
|
||
|
||
|
||
def _seed_subobjects(pkg, blueprint_id, tenant=None):
|
||
for stype, name in [('mission', '为社区设计节水方案'),
|
||
('learning_goal', '理解水循环'),
|
||
('task', '实地调研'),
|
||
('role', '组长'),
|
||
('artifact_spec', '节水手册'),
|
||
('evidence_spec', '调研报告'),
|
||
('reflection_spec', '小组复盘')]:
|
||
r = pkg.api_subobject_save({'subobject_type': stype, 'blueprint_id': blueprint_id,
|
||
'name': name, 'seq_no': 1})
|
||
assert r['success'], r
|
||
return blueprint_id
|
||
|
||
|
||
def test_20_subobject_crud_all_types(mod):
|
||
pkg, _i, store, _e = mod
|
||
bp = _new_blueprint(pkg)
|
||
_seed_subobjects(pkg, bp['id'])
|
||
counts = pkg.api_subobject_counts({'blueprint_id': bp['id']})
|
||
assert counts['success']
|
||
assert counts['data']['total'] == 7
|
||
assert set(counts['data'].keys()) >= set(pkg.SUBOBJECT_TYPE_TABLE.keys())
|
||
lst = pkg.api_subobject_list({'subobject_type': 'mission', 'blueprint_id': bp['id']})
|
||
assert lst['data']['total'] == 1
|
||
oid = lst['data']['rows'][0]['id']
|
||
upd = pkg.api_subobject_save({'subobject_type': 'mission', 'id': oid,
|
||
'blueprint_id': bp['id'], 'name': '改后情境'})
|
||
assert upd['data']['name'] == '改后情境'
|
||
dele = pkg.api_subobject_delete({'subobject_type': 'mission', 'id': oid})
|
||
assert dele['success'] and store['pbl_blueprint_mission'][oid]['is_deleted'] == 1
|
||
|
||
|
||
def test_21_subobject_unknown_type_rejected(mod):
|
||
pkg, _i, _s, _e = mod
|
||
bp = _new_blueprint(pkg)
|
||
res = pkg.api_subobject_save({'subobject_type': 'hack_me', 'blueprint_id': bp['id'],
|
||
'name': 'x'})
|
||
assert res['success'] is False
|
||
assert res['error']['code'] == 'PBL-PARAM-001'
|
||
|
||
|
||
def test_22_subobject_requires_existing_blueprint(mod):
|
||
pkg, _i, _s, _e = mod
|
||
res = pkg.api_subobject_save({'subobject_type': 'task', 'blueprint_id': 'no-such',
|
||
'name': 'x'})
|
||
assert res['success'] is False
|
||
assert res['error']['code'] == 'PBL-NOT-FOUND'
|
||
|
||
|
||
def test_23_tree_aggregates_children(mod):
|
||
pkg, _i, _s, _e = mod
|
||
bp = _new_blueprint(pkg)
|
||
_seed_subobjects(pkg, bp['id'])
|
||
res = pkg.api_blueprint_tree({'id': bp['id']})
|
||
assert res['success']
|
||
data = res['data']
|
||
assert data['blueprint']['id'] == bp['id']
|
||
assert len(data['children']) == 7
|
||
assert sum(data['counts'].values()) == 7
|
||
|
||
|
||
# ---------------------------------------------------------------- 状态机 / 版本
|
||
def test_30_status_flow_legal_and_illegal(mod):
|
||
pkg, _i, _s, _e = mod
|
||
bp = _new_blueprint(pkg)
|
||
r = pkg.api_blueprint_status({'id': bp['id'], 'status': 'in_review'})
|
||
assert r['success'] and r['data']['status'] == 'in_review'
|
||
bad = pkg.api_blueprint_status({'id': bp['id'], 'status': 'archived_from_nowhere'})
|
||
assert bad['success'] is False
|
||
r2 = pkg.api_blueprint_status({'id': bp['id'], 'status': 'archived'})
|
||
assert r2['success']
|
||
r3 = pkg.api_blueprint_status({'id': bp['id'], 'status': 'draft'})
|
||
assert r3['success'] is False # archived 为终态
|
||
assert r3['error']['code'] == 'PBL-STATE-001'
|
||
|
||
|
||
def test_31_publish_requires_goal_and_mission(mod):
|
||
pkg, _i, _s, _e = mod
|
||
bp = _new_blueprint(pkg)
|
||
res = pkg.api_blueprint_publish({'id': bp['id']})
|
||
assert res['success'] is False
|
||
assert res['error']['code'] == 'PBL-STATE-001'
|
||
assert '学习目标' in res['error']['msg'] or '情境' in res['error']['msg']
|
||
|
||
|
||
def test_32_publish_creates_version_snapshot(mod):
|
||
pkg, _i, store, _e = mod
|
||
bp = _new_blueprint(pkg)
|
||
_seed_subobjects(pkg, bp['id'])
|
||
res = pkg.api_blueprint_publish({'id': bp['id'], 'note': '首版'})
|
||
assert res['success'], res
|
||
data = res['data']
|
||
assert data['version_no'] == 1
|
||
assert len(data['content_hash']) == 64
|
||
ver = store['pbl_blueprint_version'][data['version_id']]
|
||
snap = json.loads(ver['snapshot_json'])
|
||
assert snap['blueprint']['blueprint_name'] == '城市水资源调查'
|
||
assert len(snap['subobjects']['mission']) == 1
|
||
assert store['pbl_blueprint'][bp['id']]['status'] == 'published'
|
||
# 二次发布版本号自增
|
||
pkg.api_blueprint_status({'id': bp['id'], 'status': 'in_review'})
|
||
res2 = pkg.api_blueprint_publish({'id': bp['id'], 'note': '二版'})
|
||
assert res2['data']['version_no'] == 2
|
||
vs = pkg.api_blueprint_versions({'id': bp['id']})
|
||
assert vs['data']['total'] == 2
|
||
vg = pkg.api_blueprint_version_get({'version_id': data['version_id']})
|
||
assert vg['data']['snapshot']['blueprint']['id'] == bp['id']
|
||
|
||
|
||
def test_33_fork_copies_subobjects(mod):
|
||
pkg, _i, store, _e = mod
|
||
bp = _new_blueprint(pkg)
|
||
_seed_subobjects(pkg, bp['id'])
|
||
res = pkg.api_blueprint_fork({'id': bp['id'], 'class_id': 'C99',
|
||
'blueprint_name': '副本蓝图'})
|
||
assert res['success'], res
|
||
nid = res['data']['id']
|
||
assert res['data']['copied_subobjects']['total'] == 7
|
||
new = store['pbl_blueprint'][nid]
|
||
assert new['status'] == 'draft' and new['current_version'] == 0
|
||
assert new['source_blueprint_id'] == bp['id']
|
||
assert new['class_id'] == 'C99'
|
||
assert new['blueprint_code'] != bp['blueprint_code']
|
||
assert pkg.api_subobject_counts({'blueprint_id': nid})['data']['total'] == 7
|
||
|
||
|
||
# ---------------------------------------------------------------- 模板
|
||
def test_40_template_from_blueprint_and_instantiate(mod):
|
||
pkg, _i, store, _e = mod
|
||
bp = _new_blueprint(pkg)
|
||
_seed_subobjects(pkg, bp['id'])
|
||
res = pkg.api_template_from_blueprint({'blueprint_id': bp['id'],
|
||
'template_name': '水资源模板'})
|
||
assert res['success'], res
|
||
tpl = res['data']
|
||
assert tpl['item_created'] == 7
|
||
assert tpl['source_blueprint_id'] == bp['id']
|
||
lst = pkg.api_template_list({})
|
||
assert lst['data']['total'] == 1
|
||
inst = pkg.api_template_instantiate({'id': tpl['id'],
|
||
'payload': {'blueprint_name': '实例化蓝图'}})
|
||
assert inst['success'], inst
|
||
nbp = store['pbl_blueprint'][inst['data']['blueprint_id']]
|
||
assert nbp['source_template_id'] == tpl['id']
|
||
assert nbp['status'] == 'draft'
|
||
assert pkg.api_subobject_counts({'blueprint_id': nbp['id']})['data']['total'] == 7
|
||
assert store['pbl_blueprint_template'][tpl['id']]['usage_count'] == 1
|
||
|
||
|
||
def test_41_template_save_and_delete(mod):
|
||
pkg, _i, store, _e = mod
|
||
r = pkg.api_template_save({'template_name': '空模板', 'template_type': 'blueprint'})
|
||
assert r['success'] and r['data']['template_code'].startswith('TPL-')
|
||
tid = r['data']['id']
|
||
r2 = pkg.api_template_save({'id': tid, 'template_name': '空模板改名'})
|
||
assert r2['data']['template_name'] == '空模板改名'
|
||
r3 = pkg.api_template_delete({'id': tid})
|
||
assert r3['success'] and store['pbl_blueprint_template'][tid]['is_deleted'] == 1
|
||
assert pkg.api_template_get({'id': tid})['success'] is False
|
||
|
||
|
||
def test_42_template_get_missing(mod):
|
||
pkg, _i, _s, _e = mod
|
||
assert pkg.api_template_get({'id': 'x'})['error']['code'] == 'PBL-NOT-FOUND'
|
||
|
||
|
||
# ---------------------------------------------------------------- 前端/权限一致性
|
||
def test_50_dspy_files_match_registered_paths(mod):
|
||
pkg, _i, _s, env = mod
|
||
api_dir = os.path.join(PKG, 'wwwroot', 'api')
|
||
dspy = set(f for f in os.listdir(api_dir) if f.endswith('.dspy'))
|
||
paths = set(os.path.basename(p) for p in getattr(env, 'pbl_blueprint_paths')['api'])
|
||
assert dspy == paths, ('dspy 与注册路径不一致', dspy ^ paths)
|
||
registry = set(pkg.API_REGISTRY.keys())
|
||
assert registry == set(f[:-5] for f in dspy) | {'blueprint_versions', 'blueprint_version_get',
|
||
'subobject_counts', 'table_manifest',
|
||
'template_get', 'template_delete',
|
||
'template_from_blueprint'} - {''}
|
||
|
||
|
||
def test_51_dspy_forbidden_patterns():
|
||
api_dir = os.path.join(PKG, 'wwwroot', 'api')
|
||
bad = re.compile(r'^\s*(import|from)\s|f["\']|\bprint\(|uuid\.|\bopen\(')
|
||
for f in sorted(os.listdir(api_dir)):
|
||
text = open(os.path.join(api_dir, f), encoding='utf-8').read()
|
||
for i, line in enumerate(text.splitlines(), 1):
|
||
if line.strip().startswith('#'):
|
||
continue
|
||
assert not bad.search(line), '%s:%d 命中禁项: %s' % (f, i, line.strip())
|
||
assert 'return' in text, f
|
||
|
||
|
||
def test_52_py_only_uses_sqlor_whitelist():
|
||
allowed = re.compile(r'sor\.|_db\.(insert_row|update_row|delete_row|select_rows|'
|
||
r'select_one|sql_exe|sql|_)|def (C|U|D|R|I|sql_exe)\b')
|
||
forbidden = re.compile(r'''\b(sor|sqlor)\.(save|list|insert|query|update|delete|select)\s*\(''')
|
||
for f in sorted(os.listdir(PKG)):
|
||
if not f.endswith('.py'):
|
||
continue
|
||
text = open(os.path.join(PKG, f), encoding='utf-8').read()
|
||
for i, line in enumerate(text.splitlines(), 1):
|
||
assert not forbidden.search(line), '%s:%d 编造 API: %s' % (f, i, line.strip())
|
||
assert allowed # 语义占位
|
||
|
||
|
||
def test_53_load_path_covers_all_paths():
|
||
sys.path.insert(0, os.path.join(REPO, 'scripts'))
|
||
import importlib.util
|
||
spec = importlib.util.spec_from_file_location(
|
||
'load_path_mod', os.path.join(REPO, 'scripts', 'load_path.py'))
|
||
m = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(m)
|
||
entries = set(p for _r, p in m.build_entries())
|
||
from pbl_blueprint.m1b_compat import ( # M1b compat: pbl_blueprint.init 半迁移缺失符号改由包内兼容层供给
|
||
API_PATHS,
|
||
PAGE_PATHS,
|
||
CRUD_ALIASES,
|
||
)
|
||
for p in [x for x, _f, _r in API_PATHS] + [x for x, _r in PAGE_PATHS] + list(CRUD_ALIASES):
|
||
assert p in entries, 'load_path.py 漏注册 %s' % p
|
||
assert len([p for p in entries if p.endswith('.dspy')]) == 25
|
||
|
||
|
||
def test_54_i18n_covers_all_ui_labels():
|
||
zh = json.load(open(os.path.join(PKG, 'i18n', 'pbl_blueprint.zh_CN.json'), encoding='utf-8'))
|
||
en = json.load(open(os.path.join(PKG, 'i18n', 'pbl_blueprint.en_US.json'), encoding='utf-8'))
|
||
pat = re.compile(r"\{\{t\('([^']+)'")
|
||
used = set()
|
||
www = os.path.join(PKG, 'wwwroot')
|
||
for root, _dirs, files in os.walk(www):
|
||
for f in files:
|
||
if f.endswith('.ui'):
|
||
txt = open(os.path.join(root, f), encoding='utf-8').read()
|
||
used |= set(pat.findall(txt))
|
||
missing = used - set(zh.keys())
|
||
assert not missing, 'i18n 缺 key: %s' % sorted(missing)
|
||
assert set(zh.keys()) - set(en.keys()) == set()
|
||
|
||
|
||
def test_55_ui_urls_have_no_wwwroot_prefix():
|
||
www = os.path.join(PKG, 'wwwroot')
|
||
for root, _dirs, files in os.walk(www):
|
||
for f in files:
|
||
if f.endswith('.ui'):
|
||
txt = open(os.path.join(root, f), encoding='utf-8').read()
|
||
assert 'wwwroot' not in txt, f
|
||
for p in re.findall(r"entire_url\('(/pbl_blueprint/[^']+)'\)", txt):
|
||
tail = p.replace('/pbl_blueprint/', '')
|
||
assert os.path.exists(os.path.join(www, tail)), '%s -> %s 不存在' % (f, p)
|
||
|
||
|
||
# ---------------------------------------------------------------- 建表 / 自省
|
||
def test_60_ensure_schema_issues_idempotent_ddl(mod):
|
||
pkg, _i, store, _e = mod
|
||
res = pkg.ensure_schema()
|
||
assert res['success'], res
|
||
assert res['data']['pbl_blueprint'] >= 1
|
||
|
||
|
||
def test_61_table_manifest_shape(mod):
|
||
pkg, _i, _s, _e = mod
|
||
res = pkg.api_table_manifest()
|
||
assert res['success']
|
||
data = res['data']
|
||
assert data['engine'] == 'postgresql' and len(data['tables']) == 11
|
||
bp = [t for t in data['tables'] if t['table'] == 'pbl_blueprint'][0]
|
||
assert bp['pk'] == ['id'] and bp['tenant_field'] == 'tenant_id'
|
||
assert bp['field_types']['id'] == 'str(32)'
|
||
assert bp['field_types']['estimated_cost'] == 'double(18,2)'
|
||
assert bp['soft_delete_field'] == 'is_deleted'
|
||
|
||
|
||
def test_62_options_endpoints(mod):
|
||
pkg, _i, _s, _e = mod
|
||
assert len(pkg.api_blueprint_status_options()['data']) == 4
|
||
assert len(pkg.api_blueprint_domain_options()['data']) >= 5
|
||
assert len(pkg.api_subobject_type_options()['data']) == 7
|
||
|
||
|
||
# >>> M1b compat exports (auto-generated, idempotent) >>>
|
||
# 由 tools/m1b_fix_import_closure_v2.py 幂等生成；不覆盖本文件既有同名定义。
|
||
# QC #1 铁律：逐符号 getattr 采纳，单符号缺失不牵连其余符号；
|
||
# 供给方 pbl_blueprint.m1b_compat 为包内自足模块（无裸跨包 import，恒可导入）。
|
||
_M1B_COMPAT_SYMBOLS = (
|
||
'API_PATHS',
|
||
'PAGE_PATHS',
|
||
'CRUD_ALIASES',
|
||
)
|
||
|
||
try:
|
||
import pbl_blueprint.m1b_compat as _m1b_compat
|
||
except ImportError: # pragma: no cover
|
||
try:
|
||
from . import m1b_compat as _m1b_compat
|
||
except ImportError:
|
||
_m1b_compat = None
|
||
|
||
if _m1b_compat is not None:
|
||
for _name in _M1B_COMPAT_SYMBOLS:
|
||
if _name in globals():
|
||
continue
|
||
_val = getattr(_m1b_compat, _name, None)
|
||
if _val is not None:
|
||
globals()[_name] = _val
|
||
|
||
# 装配面硬保证：load_m1b / offline 必须可调用（api_blueprint.py:504 引用点）
|
||
if not callable(globals().get('load_m1b')) and _m1b_compat is not None:
|
||
globals()['load_m1b'] = _m1b_compat.load_m1b
|
||
if not callable(globals().get('offline')) and _m1b_compat is not None:
|
||
globals()['offline'] = _m1b_compat.offline
|
||
# <<< M1b compat exports <<<
|
||
|
||
|
||
# >>> _qc_r5_compat: per-symbol resolution (generated by tools/m1b_fix_qc_round5.py; idempotent) >>>
|
||
from pbl_blueprint._qc_r5_compat import first_available as _qc_r5_first # noqa: E402,F401
|
||
_QC_R5_COMPAT_PROVIDES = ("esc", "_sor", "PblError")
|
||
esc = _qc_r5_first('esc', [("pbl_common.dbutil", "esc"), ("pbl_common.db", "esc"), ("pbl_common", "esc"), ("pbl_blueprint.m1b_compat", "esc"), ("pbl_blueprint._qc_r5_compat", "esc")], module=__name__)
|
||
_sor = _qc_r5_first('_sor', [("pbl_common.dbutil", "_sor"), ("pbl_common.db", "_sor"), ("pbl_common", "_sor"), ("pbl_blueprint.m1b_compat", "_sor"), ("pbl_blueprint._qc_r5_compat", "sor_proxy")], module=__name__)
|
||
PblError = _qc_r5_first('PblError', [("pbl_common.errors", "PblError"), ("pbl_common.api", "PblError"), ("pbl_blueprint.errors", "PblBlueprintError")], module=__name__)
|
||
# <<< _qc_r5_compat <<<
|