"""tests/test_domain_ref.py — M8 薄扩展 13 个契约接口的真实断言测试（离线 sqlite）。 覆盖： §3.1 bind_ref / unbind_ref / get_ref / list_refs / update_ref §3.2 list_worlds_by_tenant / list_scenes_by_world / list_entities_by_scene / get_world_with_pbl_context / check_ref_access §3.3 list_teams_by_class / bind_team_to_world / get_team_worlds + 薄扩展铁律：不改基表（基表行数/结构前后一致）、跨租户隔离（US-21）、 悬挂引用过滤、表总账（OWN_TABLES == models/*.json == 1 张 pbl_domain_ref）。 运行：python3 tests/test_domain_ref.py """ import asyncio import json import os import sys import unittest HERE = os.path.dirname(os.path.abspath(__file__)) ROOT = os.path.dirname(HERE) sys.path.insert(0, ROOT) sys.path.insert(0, HERE) import fake_db # noqa: E402 from pbl_domain_ext import api # noqa: E402 from pbl_domain_ext import base as base_mod # noqa: E402 from pbl_domain_ext import init as init_mod # noqa: E402 from pbl_domain_ext.errors import (E_DUPLICATE, E_FORBIDDEN, E_NOT_FOUND, # noqa: E402 E_VALIDATION, PblError) T1 = 'T1' T2 = 'T2' def run(coro): return asyncio.get_event_loop().run_until_complete(coro) if False else asyncio.run(coro) class BaseCase(unittest.TestCase): def setUp(self): self.conn, self.adapter = fake_db.setup() def tearDown(self): fake_db.teardown() self.conn.close() def assertPblError(self, code, coro): with self.assertRaises(Exception) as ctx: run(coro) exc = ctx.exception self.assertEqual(getattr(exc, 'code', None), code, '期望 %s，实际 %s(%s)' % (code, getattr(exc, 'code', None), exc)) return exc class TestBindRef(BaseCase): """接口 1：bind_ref""" def test_bind_world_ok(self): ref = run(api.bind_ref('world', 1, blueprint_id='BP-1', class_id='CLS-1', team_id='TEAM_A', ext_json={'stage': 2}, tenant_id=T1)) self.assertIsNotNone(ref) self.assertEqual(ref['tenant_id'], T1) self.assertEqual(ref['ref_type'], 'world') self.assertEqual(ref['ref_id'], 1) self.assertEqual(ref['blueprint_id'], 'BP-1') self.assertEqual(ref['class_id'], 'CLS-1') self.assertEqual(ref['team_id'], 'TEAM_A') self.assertEqual(ref['ext_json'], {'stage': 2}) self.assertEqual(ref['is_deleted'], 0) def test_bind_duplicate_raises(self): run(api.bind_ref('world', 1, tenant_id=T1)) self.assertPblError(E_DUPLICATE, api.bind_ref('world', 1, tenant_id=T1)) def test_bind_same_ref_other_tenant_ok(self): """同一基表记录可被不同租户各自绑定（UNIQUE 含 tenant_id）。""" run(api.bind_ref('world', 1, tenant_id=T1)) ref2 = run(api.bind_ref('world', 1, tenant_id=T2)) self.assertEqual(ref2['tenant_id'], T2) def test_bind_base_missing_raises_not_found(self): self.assertPblError(E_NOT_FOUND, api.bind_ref('world', 999, tenant_id=T1)) def test_bind_invalid_ref_type(self): self.assertPblError(E_VALIDATION, api.bind_ref('script', 1, tenant_id=T1)) def test_bind_invalid_ref_id(self): self.assertPblError(E_VALIDATION, api.bind_ref('world', 'abc', tenant_id=T1)) self.assertPblError(E_VALIDATION, api.bind_ref('world', 0, tenant_id=T1)) def test_bind_missing_tenant(self): self.assertPblError(E_VALIDATION, api.bind_ref('world', 1)) def test_rebind_after_unbind_revives(self): run(api.bind_ref('scene', 11, class_id='CLS-1', tenant_id=T1)) run(api.unbind_ref('scene', 11, tenant_id=T1)) ref = run(api.bind_ref('scene', 11, class_id='CLS-2', tenant_id=T1)) self.assertEqual(ref['class_id'], 'CLS-2') self.assertEqual(ref['is_deleted'], 0) def test_bind_ext_invalid_json_string(self): self.assertPblError(E_VALIDATION, api.bind_ref('world', 1, ext_json='{bad json', tenant_id=T1)) class TestUnbindGetUpdate(BaseCase): """接口 2/3/5：unbind_ref / get_ref / update_ref""" def test_unbind_soft_delete(self): run(api.bind_ref('entity', 101, tenant_id=T1)) self.assertTrue(run(api.unbind_ref('entity', 101, tenant_id=T1))) flag = fake_db.scalar(self.conn, "SELECT is_deleted FROM pbl_domain_ref WHERE ref_id=101") self.assertEqual(flag, 1, 'unbind 必须软删而非物理删除') self.assertPblError(E_NOT_FOUND, api.get_ref('entity', 101, tenant_id=T1)) def test_unbind_missing(self): self.assertPblError(E_NOT_FOUND, api.unbind_ref('world', 1, tenant_id=T1)) def test_get_ref_ok(self): run(api.bind_ref('world', 2, blueprint_id='BP-9', tenant_id=T1)) ref = run(api.get_ref('world', 2, tenant_id=T1)) self.assertEqual(ref['blueprint_id'], 'BP-9') def test_get_ref_cross_tenant_not_found(self): run(api.bind_ref('world', 2, tenant_id=T1)) self.assertPblError(E_NOT_FOUND, api.get_ref('world', 2, tenant_id=T2)) def test_update_ref_ok(self): run(api.bind_ref('world', 1, class_id='CLS-1', tenant_id=T1)) ref = run(api.update_ref('world', 1, {'class_id': 'CLS-2', 'team_id': 'TEAM_B', 'ext_json': {'k': 'v'}}, tenant_id=T1)) self.assertEqual(ref['class_id'], 'CLS-2') self.assertEqual(ref['team_id'], 'TEAM_B') self.assertEqual(ref['ext_json'], {'k': 'v'}) def test_update_ref_rejects_key_fields(self): run(api.bind_ref('world', 1, tenant_id=T1)) self.assertPblError(E_VALIDATION, api.update_ref('world', 1, {'ref_id': 2}, tenant_id=T1)) self.assertPblError(E_VALIDATION, api.update_ref('world', 1, {'tenant_id': T2}, tenant_id=T1)) def test_update_ref_missing(self): self.assertPblError(E_NOT_FOUND, api.update_ref('world', 1, {'class_id': 'X'}, tenant_id=T1)) class TestListRefs(BaseCase): """接口 4：list_refs（过滤 + 分页 + 悬挂引用标记）""" def _seed(self): run(api.bind_ref('world', 1, blueprint_id='BP-1', class_id='CLS-1', team_id='TEAM_A', tenant_id=T1)) run(api.bind_ref('world', 2, blueprint_id='BP-1', class_id='CLS-1', team_id='TEAM_B', tenant_id=T1)) run(api.bind_ref('scene', 11, blueprint_id='BP-2', class_id='CLS-1', tenant_id=T1)) run(api.bind_ref('entity', 101, class_id='CLS-2', tenant_id=T1)) run(api.bind_ref('world', 1, class_id='CLS-9', tenant_id=T2)) def test_list_all_tenant_scoped(self): self._seed() res = run(api.list_refs({}, 1, 20, tenant_id=T1)) self.assertEqual(res['total'], 4, 'T1 只应看到自己的 4 条关联') self.assertTrue(all(i['tenant_id'] == T1 for i in res['items'])) def test_list_filter_by_type_and_class(self): self._seed() res = run(api.list_refs({'ref_type': 'world'}, 1, 20, tenant_id=T1)) self.assertEqual(res['total'], 2) res2 = run(api.list_refs({'class_id': 'CLS-2'}, 1, 20, tenant_id=T1)) self.assertEqual(res2['total'], 1) self.assertEqual(res2['items'][0]['ref_type'], 'entity') res3 = run(api.list_refs({'blueprint_id': 'BP-1'}, 1, 20, tenant_id=T1)) self.assertEqual(res3['total'], 2) res4 = run(api.list_refs({'team_id': 'TEAM_B'}, 1, 20, tenant_id=T1)) self.assertEqual(res4['total'], 1) def test_list_pagination(self): self._seed() page1 = run(api.list_refs({}, 1, 2, tenant_id=T1)) page2 = run(api.list_refs({}, 2, 2, tenant_id=T1)) self.assertEqual(len(page1['items']), 2) self.assertEqual(len(page2['items']), 2) self.assertEqual(page1['total'], 4) ids1 = {i['id'] for i in page1['items']} ids2 = {i['id'] for i in page2['items']} self.assertFalse(ids1 & ids2, '分页结果不得重叠') def test_list_page_size_capped(self): self._seed() res = run(api.list_refs({}, 1, 9999, tenant_id=T1)) self.assertEqual(res['size'], api.MAX_PAGE_SIZE) def test_list_invalid_ref_type_filter(self): self.assertPblError(E_VALIDATION, api.list_refs({'ref_type': 'bogus'}, 1, 20, tenant_id=T1)) def test_list_dangling_flag(self): run(api.bind_ref('world', 1, tenant_id=T1)) self.conn.execute("DELETE FROM world WHERE id=1") # 复用模块删除基表记录 self.conn.commit() res = run(api.list_refs({'ref_type': 'world'}, 1, 20, tenant_id=T1)) self.assertEqual(res['total'], 1) self.assertTrue(res['items'][0]['dangling'], '基表已删的 ref 必须标记 dangling') self.assertEqual(res['valid_total'], 0) class TestTenantIsolationQueries(BaseCase): """接口 6~9：租户隔离查询封装（基表 + 扩展联合）""" def test_list_worlds_by_tenant_only_bound(self): run(api.bind_ref('world', 1, class_id='CLS-1', tenant_id=T1)) worlds = run(api.list_worlds_by_tenant(T1)) self.assertEqual(len(worlds), 1) self.assertEqual(worlds[0]['id'], 1) self.assertEqual(worlds[0]['name'], '世界A-火星基地') self.assertEqual(worlds[0]['class_id'], 'CLS-1') self.assertIsNotNone(worlds[0]['pbl_ref']) # 未绑定的 world 2 对 T1 不可见 self.assertNotIn(2, [w['id'] for w in worlds]) def test_list_worlds_by_tenant_empty_when_no_binding(self): self.assertEqual(run(api.list_worlds_by_tenant(T2)), []) def test_list_worlds_by_class_filter(self): run(api.bind_ref('world', 1, class_id='CLS-1', tenant_id=T1)) run(api.bind_ref('world', 2, class_id='CLS-2', tenant_id=T1)) got = run(api.list_worlds_by_tenant(T1, class_id='CLS-2')) self.assertEqual([w['id'] for w in got], [2]) def test_list_worlds_filters_dangling(self): run(api.bind_ref('world', 1, tenant_id=T1)) self.conn.execute("DELETE FROM world WHERE id=1") self.conn.commit() self.assertEqual(run(api.list_worlds_by_tenant(T1)), [], '悬挂引用必须从联合结果中过滤') def test_list_scenes_by_world_ok(self): run(api.bind_ref('world', 1, tenant_id=T1)) run(api.bind_ref('scene', 11, tenant_id=T1)) run(api.bind_ref('scene', 12, tenant_id=T1)) scenes = run(api.list_scenes_by_world(1, T1)) self.assertEqual(sorted(s['id'] for s in scenes), [11, 12]) self.assertEqual(scenes[0]['world_id'], 1) def test_list_scenes_only_bound_visible(self): run(api.bind_ref('world', 1, tenant_id=T1)) run(api.bind_ref('scene', 11, tenant_id=T1)) # 12 未绑定 scenes = run(api.list_scenes_by_world(1, T1)) self.assertEqual([s['id'] for s in scenes], [11]) def test_list_scenes_cross_tenant_forbidden(self): run(api.bind_ref('world', 1, tenant_id=T1)) self.assertPblError(E_FORBIDDEN, api.list_scenes_by_world(1, T2)) def test_list_entities_by_scene_ok(self): run(api.bind_ref('scene', 11, tenant_id=T1)) run(api.bind_ref('entity', 101, tenant_id=T1)) run(api.bind_ref('entity', 102, tenant_id=T1)) ents = run(api.list_entities_by_scene(11, T1)) self.assertEqual(sorted(e['id'] for e in ents), [101, 102]) self.assertEqual(ents[0]['kind'], 'vehicle') def test_list_entities_cross_tenant_forbidden(self): run(api.bind_ref('scene', 11, tenant_id=T1)) self.assertPblError(E_FORBIDDEN, api.list_entities_by_scene(11, T2)) def test_get_world_with_pbl_context_ok(self): run(api.bind_ref('world', 1, blueprint_id='BP-1', class_id='CLS-1', team_id='TEAM_A', ext_json={'mode': 'coop'}, tenant_id=T1)) ctx = run(api.get_world_with_pbl_context(1, T1)) self.assertEqual(ctx['name'], '世界A-火星基地') self.assertEqual(ctx['pbl_context']['blueprint_id'], 'BP-1') self.assertEqual(ctx['pbl_context']['class_id'], 'CLS-1') self.assertEqual(ctx['pbl_context']['team_id'], 'TEAM_A') self.assertEqual(ctx['pbl_context']['ext_json'], {'mode': 'coop'}) self.assertEqual(ctx['pbl_context']['tenant_id'], T1) def test_get_world_context_cross_tenant_forbidden(self): run(api.bind_ref('world', 1, tenant_id=T1)) self.assertPblError(E_FORBIDDEN, api.get_world_with_pbl_context(1, T2)) def test_get_world_context_base_missing_not_found(self): run(api.bind_ref('world', 1, tenant_id=T1)) self.conn.execute("DELETE FROM world WHERE id=1") self.conn.commit() self.assertPblError(E_NOT_FOUND, api.get_world_with_pbl_context(1, T1)) class TestCheckRefAccess(BaseCase): """接口 10：check_ref_access（租户+班级+团队三重匹配）""" def test_access_granted(self): run(api.bind_ref('world', 1, class_id='CLS-1', team_id='TEAM_A', tenant_id=T1)) self.assertTrue(run(api.check_ref_access('world', 1, T1))) self.assertTrue(run(api.check_ref_access('world', 1, T1, class_id='CLS-1', team_id='TEAM_A'))) def test_access_denied_cross_tenant(self): run(api.bind_ref('world', 1, class_id='CLS-1', team_id='TEAM_A', tenant_id=T1)) self.assertFalse(run(api.check_ref_access('world', 1, T2))) def test_access_denied_wrong_class_or_team(self): run(api.bind_ref('world', 1, class_id='CLS-1', team_id='TEAM_A', tenant_id=T1)) self.assertFalse(run(api.check_ref_access('world', 1, T1, class_id='CLS-X'))) self.assertFalse(run(api.check_ref_access('world', 1, T1, team_id='TEAM-X'))) def test_access_denied_unbound(self): self.assertFalse(run(api.check_ref_access('world', 2, T1))) def test_access_denied_after_unbind(self): run(api.bind_ref('scene', 11, tenant_id=T1)) run(api.unbind_ref('scene', 11, tenant_id=T1)) self.assertFalse(run(api.check_ref_access('scene', 11, T1))) def test_access_denied_invalid_input_no_raise(self): self.assertFalse(run(api.check_ref_access('bogus', 1, T1))) self.assertFalse(run(api.check_ref_access('world', 'x', T1))) self.assertFalse(run(api.check_ref_access('world', 1, None))) def test_access_denied_dangling_base(self): run(api.bind_ref('entity', 101, tenant_id=T1)) self.conn.execute("DELETE FROM entity WHERE id=101") self.conn.commit() self.assertFalse(run(api.check_ref_access('entity', 101, T1))) class TestTeamClassDimension(BaseCase): """接口 11~13：团队/班级维度（US-13 共享世界支撑）""" def test_list_teams_by_class_groups(self): run(api.bind_ref('world', 1, class_id='CLS-1', team_id='TEAM_A', tenant_id=T1)) run(api.bind_ref('scene', 11, class_id='CLS-1', team_id='TEAM_A', tenant_id=T1)) run(api.bind_ref('world', 2, class_id='CLS-1', team_id='TEAM_B', tenant_id=T1)) teams = run(api.list_teams_by_class('CLS-1', T1)) self.assertEqual(len(teams), 2) by_id = {t['team_id']: t for t in teams} self.assertEqual(sorted(by_id['TEAM_A']['world_ids']), [1]) self.assertEqual(sorted(by_id['TEAM_A']['scene_ids']), [11]) self.assertEqual(by_id['TEAM_A']['entity_ids'], []) self.assertEqual(len(by_id['TEAM_A']['ref_ids']), 2) self.assertEqual(sorted(by_id['TEAM_B']['world_ids']), [2]) def test_list_teams_by_class_members_from_governance(self): run(api.bind_ref('world', 1, class_id='CLS-1', team_id='TEAM_A', tenant_id=T1)) teams = run(api.list_teams_by_class('CLS-1', T1)) members = teams[0]['members'] self.assertEqual(len(members), 2, 'members 应只读取自 pbl_governance.pbl_team_member') self.assertEqual({m['user_id'] for m in members}, {'stu01', 'stu02'}) def test_list_teams_by_class_tenant_scoped(self): run(api.bind_ref('world', 1, class_id='CLS-1', team_id='TEAM_A', tenant_id=T1)) self.assertEqual(run(api.list_teams_by_class('CLS-1', T2)), []) def test_list_teams_by_class_requires_class_id(self): self.assertPblError(E_VALIDATION, api.list_teams_by_class(None, T1)) def test_list_teams_without_governance_table_degrades(self): """治理模块未部署（无 pbl_team_member 表）时降级为空成员列表，不报错。""" fake_db.teardown() self.conn.close() self.conn, self.adapter = fake_db.setup(with_team_member=False) run(api.bind_ref('world', 1, class_id='CLS-1', team_id='TEAM_A', tenant_id=T1)) teams = run(api.list_teams_by_class('CLS-1', T1)) self.assertEqual(len(teams), 1) self.assertEqual(teams[0]['members'], []) def test_bind_team_to_world_creates_ref(self): ref = run(api.bind_team_to_world(1, 'TEAM_A', T1, class_id='CLS-1', blueprint_id='BP-1')) self.assertEqual(ref['ref_type'], 'world') self.assertEqual(ref['ref_id'], 1) self.assertEqual(ref['team_id'], 'TEAM_A') self.assertEqual(ref['class_id'], 'CLS-1') def test_bind_team_to_world_updates_existing(self): run(api.bind_ref('world', 1, team_id='TEAM_A', tenant_id=T1)) ref = run(api.bind_team_to_world(1, 'TEAM_B', T1)) self.assertEqual(ref['team_id'], 'TEAM_B', '已有关联应幂等更新 team_id') cnt = fake_db.scalar( self.conn, "SELECT COUNT(*) FROM pbl_domain_ref WHERE ref_type='world' AND ref_id=1") self.assertEqual(cnt, 1, '不得产生重复关联行') def test_bind_team_to_world_duplicate_same_team(self): run(api.bind_team_to_world(1, 'TEAM_A', T1)) self.assertPblError(E_DUPLICATE, api.bind_team_to_world(1, 'TEAM_A', T1)) def test_bind_team_to_world_base_missing(self): self.assertPblError(E_NOT_FOUND, api.bind_team_to_world(999, 'TEAM_A', T1)) def test_bind_team_to_world_requires_team_id(self): self.assertPblError(E_VALIDATION, api.bind_team_to_world(1, None, T1)) def test_get_team_worlds(self): run(api.bind_team_to_world(1, 'TEAM_A', T1)) run(api.bind_team_to_world(2, 'TEAM_B', T1)) worlds = run(api.get_team_worlds('TEAM_A', T1)) self.assertEqual([w['id'] for w in worlds], [1]) self.assertEqual(worlds[0]['team_id'], 'TEAM_A') self.assertEqual(run(api.get_team_worlds('TEAM_A', T2)), [], '跨租户不得返回他租户团队世界') def test_get_team_worlds_requires_team_id(self): self.assertPblError(E_VALIDATION, api.get_team_worlds('', T1)) class TestThinExtensionInvariants(BaseCase): """薄扩展铁律：不改基表 + 表总账 + 契约齐备""" def test_base_tables_untouched(self): before = {t: fake_db.scalar(self.conn, "SELECT COUNT(*) FROM %s" % t) for t in ('world', 'scene', 'entity')} run(api.bind_ref('world', 1, class_id='C', team_id='T', tenant_id=T1)) run(api.bind_ref('scene', 11, tenant_id=T1)) run(api.bind_ref('entity', 101, tenant_id=T1)) run(api.update_ref('world', 1, {'class_id': 'C2'}, tenant_id=T1)) run(api.bind_team_to_world(2, 'TEAM_B', T1)) # 读路径（联合查询）——必须在解绑前跑，解绑后 scene 11 对本租户即 403 run(api.list_worlds_by_tenant(T1)) run(api.list_scenes_by_world(1, T1)) run(api.list_entities_by_scene(11, T1)) run(api.get_world_with_pbl_context(1, T1)) run(api.check_ref_access('world', 1, T1)) run(api.list_teams_by_class('C', T1)) run(api.get_team_worlds('T', T1)) run(api.list_refs({}, 1, 20, tenant_id=T1)) # 写路径收尾：解绑（软删，只动自有表） run(api.unbind_ref('scene', 11, tenant_id=T1)) run(api.unbind_ref('entity', 101, tenant_id=T1)) after = {t: fake_db.scalar(self.conn, "SELECT COUNT(*) FROM %s" % t) for t in ('world', 'scene', 'entity')} self.assertEqual(before, after, '基表行数不得变化（零写入）') # 基表结构零 ALTER：列集合不变 for table in ('world', 'scene', 'entity'): cols = [r[1] for r in fake_db.raw_sql( self.conn, "PRAGMA table_info(%s)" % table)] self.assertNotIn('tenant_id', cols, '禁止给基表加 tenant_id 列') def test_own_tables_single_and_matches_models(self): self.assertEqual(init_mod.OWN_TABLES, ['pbl_domain_ref']) models_dir = os.path.join(ROOT, 'models') files = sorted(f for f in os.listdir(models_dir) if f.endswith('.json')) self.assertEqual(files, ['pbl_domain_ref.json'], 'models/ 必须与 OWN_TABLES 一一对应（QC #4）') def test_model_json_four_sections(self): path = os.path.join(ROOT, 'models', 'pbl_domain_ref.json') model = json.load(open(path, encoding='utf-8')) for key in ('summary', 'fields', 'indexes', 'codes'): self.assertIn(key, model, '表定义四段式缺 %s' % key) self.assertIsInstance(model['summary'], list, 'summary 必须是数组（QC #3）') self.assertTrue(all(isinstance(s, str) for s in model['summary'])) self.assertEqual(model['summary'][0], 'pbl_domain_ref') primaries = [f for f in model['fields'] if f.get('primary')] self.assertEqual(len(primaries), 1) self.assertEqual(primaries[0]['name'], 'id') uniq = [i for i in model['indexes'] if i.get('unique') and not i.get('primary')] self.assertIn(['tenant_id', 'ref_type', 'ref_id'], [i['fields'] for i in uniq], '必须有 UNIQUE(tenant_id,ref_type,ref_id)（设计 §2）') def test_contract_13_interfaces_all_callable(self): self.assertEqual(len(api.CONTRACT_INTERFACES), 13) for name in api.CONTRACT_INTERFACES: self.assertTrue(callable(getattr(api, name, None)), '缺实现：%s' % name) def test_contract_map_covers_dspy_files(self): cmap = init_mod.get_contract_map() self.assertEqual(len(cmap), 13) api_dir = os.path.join(ROOT, 'wwwroot', 'api') existing = set(os.listdir(api_dir)) for name, (impl, dspy) in cmap.items(): self.assertTrue(callable(getattr(api, name, None)), '缺实现：%s' % name) self.assertIn(os.path.basename(dspy), existing, '契约 %s 缺 dspy 端点 %s（QC #5）' % (name, dspy)) self.assertTrue(impl.startswith('pbl_domain_ext/api.py:')) def test_load_path_registers_all_dspy(self): sys.path.insert(0, os.path.join(ROOT, 'scripts')) import load_path as lp self.assertEqual(lp.selfcheck(), [], 'load_path 自检必须无问题（QC #6）') registered = set(lp.API_PATHS) | set(lp.UI_PATHS) api_dir = os.path.join(ROOT, 'wwwroot', 'api') for fname in os.listdir(api_dir): if fname.endswith('.dspy'): self.assertIn('/pbl_domain_ext/api/%s' % fname, registered, '端点 %s 未在 load_path.py 注册 → 上线 403' % fname) self.assertIn('/pbl_domain_ext/index.ui', registered) self.assertEqual(len(lp.API_PATHS), 13) for path in lp.ALL_PATHS: self.assertNotIn('*', path, '禁止通配符注册') def test_load_module_registers_env_functions(self): env = init_mod.load_pbl_domain_ext() for name in api.CONTRACT_INTERFACES: self.assertTrue(callable(getattr(env, 'pbl_%s' % name, None)), 'env.pbl_%s 未注册' % name) self.assertTrue(callable(getattr(env, name, None)), 'env.%s（设计原名）未注册' % name) info = env.pbl_domain_ext_module_info self.assertEqual(info['own_tables'], ['pbl_domain_ref']) self.assertFalse(info['base_table_altered']) self.assertEqual(len(info['contracts']), 13) def test_package_exports_match_init(self): import pbl_domain_ext as pkg for name in api.CONTRACT_INTERFACES: self.assertTrue(callable(getattr(pkg, name, None)), '__init__.py 未导出 %s（三处同步注册之 ②）' % name) self.assertTrue(callable(pkg.load_pbl_domain_ext)) def test_base_layer_readonly_projection(self): self.assertTrue(run(base_mod.base_exists('world', 1))) self.assertFalse(run(base_mod.base_exists('world', 999))) rows = run(base_mod.fetch_children('scene', 1)) self.assertEqual(sorted(r['id'] for r in rows), [11, 12]) col = run(base_mod.resolve_column('scene', base_mod.SCENE_PARENT_CANDIDATES)) self.assertEqual(col, 'world_id') if __name__ == '__main__': unittest.main(verbosity=2)