pbl_evidence/scripts/selftest_m5b.py

469 lines
25 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""M5b 自测(pbl_evidence 证据聚合与查询契约)—— 纯 stdlib,无需 DB / 无需 pytest。
覆盖设计 docs/01-design/modules/pbl_evidence.md §3.2 的 M5b 需求点:
1. **四类证据幂等采集**:``pbl_evidence_collect_idempotent`` 首次 ``created=True``,
重复(同 idempotency_key / 同唯一索引三元组)``created=False`` 且不产生第二行;
2. **idempotency_key 归一**:显式键原样透传;缺省键由 ``tenant|source_event_id|
evidence_type`` 派生且稳定(重放同键);非四类 evidence_type 报 ``PBL_E_VALIDATION``;
3. **聚合契约**:``pbl_evidence_aggregate`` 四类计数 + 覆盖度 + 分组 + 时间水位;
4. **列表查询契约**:``pbl_evidence_query`` 分页 total/rows + payload 解析 + 水位;
5. **主体档案契约**:``pbl_evidence_subject_summary`` 学生 / 团队四类分组明细;
6. **Q-OPEN-10**:预览 / Playtest 数据(payload 标记、显式预览会话、STUDENT 私有可见性)
默认剔除,``include_preview=True`` 才纳入,且 ``preview_filter`` 回显剔除口径;
7. **租户 fail-closed**:无 tenant_id 直接 ``PBL_E_TENANT_MISSING``,不落到 SQL。
内存版 DB 会**校验拼出的 SQL**(列清单、LIMIT 占位、无 ``None`` 泄漏)并按 params 过滤,
因此不只是「函数不报错」,而是验证了查询参数与 SQL 的对应关系。
运行:
cd modules/pbl_evidence && python3 scripts/selftest_m5b.py
# 或: python3 -m pytest scripts/selftest_m5b.py -q (test_* 函数可直接被收集)
"""
import asyncio
import json
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
for sib in ('pbl_common',):
p = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(
os.path.abspath(__file__)))), sib)
if os.path.isdir(p) and p not in sys.path:
sys.path.insert(0, p)
from pbl_common.api import PblError # noqa: E402
import pbl_evidence.aggregation as A # noqa: E402
TENANT = '1'
EVIDENCE_COLS = ['id', 'tenant_id', 'artifact_id', 'evidence_type', 'source_event_id',
'session_id', 'learner_id', 'blueprint_id', 'payload_json',
'occurred_at', 'dedup_key', 'collect_batch_id', 'created_at', 'updated_at']
ARTIFACT_COLS = ['id', 'tenant_id', 'artifact_uid', 'session_id', 'blueprint_id',
'team_id', 'creator_id', 'title', 'artifact_type', 'content_json',
'version_no', 'status', 'submitted_at', 'created_at', 'updated_at']
_CAPTURED = {'sql': [], 'params': []}
def _row(idx, etype, learner='S01', session='101', blueprint='9', artifact=0,
occurred=None, payload=None, source=None):
# 默认值不能在签名里引用形参 idx(定义时求值 -> NameError),挪到函数体内。
if occurred is None:
occurred = '2026-09-20 10:00:%02d' % (idx % 9)
return {
'id': idx, 'tenant_id': TENANT, 'artifact_id': artifact, 'evidence_type': etype,
'source_event_id': source or ('EV%03d' % idx), 'session_id': session,
'learner_id': learner, 'blueprint_id': blueprint,
'payload_json': json.dumps(payload or {'event_type': 'mission.completed'},
ensure_ascii=False),
'occurred_at': occurred, 'dedup_key': 'K%03d' % idx,
}
def seed_rows():
"""正式环境证据 5 条(四类齐)+ 预览/Playtest 证据 4 条(应被默认过滤)。"""
return [
_row(1, 'decision'),
_row(2, 'action'),
_row(3, 'observation', learner='S02'),
_row(4, 'artifact_version', artifact=77, learner='S02'),
_row(5, 'decision', learner='S03', occurred='2026-09-21 08:00:00'),
# ── 预览 / Playtest 族(Q-OPEN-10 默认剔除)────────────────
_row(6, 'action', payload={'pbl_preview': True, 'event_type': 'mission.completed'}),
_row(7, 'observation', payload={'env_mode': 'playtest'}),
_row(8, 'decision', session='999',
payload={'event_type': 'team.decision'}), # 显式预览会话 999
_row(9, 'artifact_version', artifact=88,
payload={'visibility': 'student', 'content_ref': 'draft'}), # STUDENT 私有
]
class FakeDB:
"""内存版证据库:按 params 过滤,同时校验 SQL 文本合法性。"""
def __init__(self, rows):
self.rows = rows
async def table_columns(self, table):
if table == 'pbl_evidence':
return list(EVIDENCE_COLS)
if table == 'pbl_artifact':
return list(ARTIFACT_COLS)
return []
@staticmethod
def _match(row, params):
for key, val in (params or {}).items():
if key in ('scan_limit', 'time_from', 'time_to') or key.startswith('etype'):
continue
if key in row and str(row[key]) != str(val):
return False
if 'time_from' in (params or {}) and row.get('occurred_at'):
if str(row['occurred_at']) < str(params['time_from']):
return False
if 'time_to' in (params or {}) and row.get('occurred_at'):
if str(row['occurred_at']) > str(params['time_to']):
return False
types = [v for k, v in (params or {}).items() if k.startswith('etype')]
if types and str(row.get('evidence_type')) not in [str(t) for t in types]:
return False
return True
async def q_all(self, sql, params=None):
_CAPTURED['sql'].append(sql)
_CAPTURED['params'].append(dict(params or {}))
low = sql.lower()
assert 'none' not in low.replace('is null', ''), 'SQL 泄漏 None:%s' % sql
if 'from `pbl_evidence`' in low:
assert 'limit ${scan_limit}$' in low, '缺少扫描上限(防全表拖库):%s' % sql
assert 'tenant_id' in low, '证据查询必须带租户条件:%s' % sql
limit = int((params or {}).get('scan_limit') or A.MAX_SCAN_ROWS)
rows = [r for r in self.rows if self._match(r, params)]
rows.sort(key=lambda r: (str(r.get('occurred_at') or ''), int(r.get('id') or 0)),
reverse=True)
return rows[:limit]
if 'count(*)' in low and 'pbl_artifact' in low:
return [{'c': 2}]
return []
async def q_one(self, sql, params=None):
rows = await self.q_all(sql, params)
return rows[0] if rows else None
def install_fake(rows, tenant=TENANT):
"""把 aggregation 的 IO 依赖换成内存实现(不碰真实 DB)。"""
db = FakeDB(rows)
A.table_columns = db.table_columns
A.q_all = db.q_all
A.q_one = db.q_one
A.tenant_id = lambda *a, **k: tenant
return db
def run(coro):
return asyncio.get_event_loop().run_until_complete(coro)
# ══════════════════════════════════════════════════════════════════════
# 1. 幂等采集(含 created 布尔 + 重复不产生第二行)
# ══════════════════════════════════════════════════════════════════════
def test_idempotent_collect_created_flag():
rows = seed_rows()
install_fake(rows)
store = {}
async def fake_collect(**kw):
key = (str(kw.get('tenant_id') or TENANT), str(kw.get('source_event_id')),
str(kw.get('evidence_type')))
if key in store:
return {'ok': True, 'deduped': True, 'id': store[key],
'unique_index': 'uk_ev_dedup', 'evidence_type': key[2]}
nid = 1000 + len(store)
store[key] = nid
rows.append(_row(nid, key[2], source=key[1]))
return {'ok': True, 'deduped': False, 'id': nid, 'unique_index': 'uk_ev_dedup',
'evidence_type': key[2]}
import pbl_evidence.api as api
origin = api.pbl_evidence_collect
api.pbl_evidence_collect = fake_collect
try:
first = run(A.pbl_evidence_collect_idempotent(
source_event_id='EV900', evidence_type='decision', learner_id='S01',
blueprint_id='9', session_id='101', payload={'choice': 'A'}))
assert first['created'] is True, first
assert first['deduped'] is False, first
assert first['idempotency_key'].startswith('IK') and len(first['idempotency_key']) == 26
assert first['idempotency_key_derived'] is True, first
second = run(A.pbl_evidence_collect_idempotent(
source_event_id='EV900', evidence_type='decision', learner_id='S01',
blueprint_id='9', session_id='101', payload={'choice': 'A'}))
assert second['created'] is False, second
assert second['deduped'] is True, second
assert second['id'] == first['id'], (first, second)
assert second['idempotency_key'] == first['idempotency_key'], '派生键必须稳定'
assert len(store) == 1, '重复采集不得产生第二行:%s' % store
# 显式 idempotency_key 原样透传
third = run(A.pbl_evidence_collect_idempotent(
source_event_id='EV901', evidence_type='action', idempotency_key='IK-EXTERNAL-01',
learner_id='S01'))
assert third['idempotency_key'] == 'IK-EXTERNAL-01', third
assert third['idempotency_key_derived'] is False, third
finally:
api.pbl_evidence_collect = origin
def test_collect_rejects_non_fourth_type():
install_fake(seed_rows())
try:
run(A.pbl_evidence_collect_idempotent(source_event_id='EV902',
evidence_type='research_upload'))
raise AssertionError('非四类 evidence_type 必须被拒')
except PblError as exc:
assert exc.code == 'PBL_E_VALIDATION', exc.code
def test_tenant_fail_closed():
install_fake(seed_rows(), tenant=None)
for fn, kw in ((A.pbl_evidence_aggregate, {}), (A.pbl_evidence_query, {}),
(A.pbl_evidence_subject_summary, {'subject_id': 'S01'}),
(A.pbl_evidence_collect_idempotent, {'source_event_id': 'EV1'})):
try:
run(fn(**kw))
raise AssertionError('%s 缺租户时必须 fail-closed' % fn.__name__)
except PblError as exc:
assert exc.code == 'PBL_E_TENANT_MISSING', (fn.__name__, exc.code)
# ══════════════════════════════════════════════════════════════════════
# 2. 聚合契约
# ══════════════════════════════════════════════════════════════════════
def test_aggregate_four_types_and_coverage():
install_fake(seed_rows())
res = run(A.pbl_evidence_aggregate(group_by='learner'))
assert res['ok'] is True and res['status'] == 'OK', res
assert res['total'] == 5, res['total'] # 9 行中 4 行预览族被默认剔除
assert res['by_type'] == {'decision': 2, 'action': 1, 'observation': 1,
'artifact_version': 1}, res['by_type']
assert res['coverage']['covered_types'] == 4, res['coverage']
assert res['coverage']['missing_types'] == [], res['coverage']
keys = {g['group_key'] for g in res['groups']}
assert keys == {'S01', 'S02', 'S03'}, keys
assert res['watermark']['max_occurred_at'] == '2026-09-21 08:00:00', res['watermark']
assert res['preview_filter']['excluded_preview'] == 3, res['preview_filter']
assert res['preview_filter']['excluded_student'] == 1, res['preview_filter']
assert res['preview_filter']['applied'] is True, res['preview_filter']
assert res['truncated'] is False, res
# 默认口径下应保留的行(正式环境证据,四类齐)= id 1..5;其余 4 行必须被默认剔除:
# id6 payload 预览标记 / id7 env_mode=playtest / id8 预览保留会话 999 / id9 STUDENT 私有
KEPT_DEFAULT = {1, 2, 3, 4, 5}
PREVIEW_DEFAULT = {6, 7, 8}
STUDENT_DEFAULT = {9}
# 发生在 2026-09-20 当天且默认口径保留的行(occurred_at 由 _row 按 idx 生成):
# id1 10:00:01 / id2 10:00:02 / id3 10:00:03 / id4 10:00:04 / id9 10:00:08(STUDENT 私有
# 被剔除) / id6 10:00:06、id7 10:00:07、id8 10:00:08(预览族被剔除);id5 在 09-21。
KEPT_0920 = {1, 2, 3, 4}
def test_aggregate_group_by_type_and_time_range():
install_fake(seed_rows())
res = run(A.pbl_evidence_aggregate(group_by='type',
time_from='2026-09-21 00:00:00',
time_to='2026-09-21 23:59:59'))
assert res['total'] == 1, res
assert res['groups'][0]['group_key'] == 'decision', res['groups']
res2 = run(A.pbl_evidence_aggregate(group_by='session',
time_range={'from': '2026-09-20 00:00:00',
'to': '2026-09-20 23:59:59'}))
assert res2['group_by'] == 'session_id', res2
# 09-20 时间窗内扫描到 8 行(id5 在 09-21 被时间窗排除),其中默认口径保留 id 1,2,3,4
# (id6/7 payload 预览标记、id8 预览保留会话 999、id9 STUDENT 私有被 Q-OPEN-10 排除);
# 保留行的 session_id 全是 101(id8 属 999 已被剔除),故分组只剩一组
assert res2['total'] == 4, res2['total']
assert [g['group_key'] for g in res2['groups']] == ['101'], res2['groups']
assert res2['groups'][0]['total'] == 4, res2['groups']
assert res2['by_type'] == {'decision': 1, 'action': 1, 'observation': 1,
'artifact_version': 1}, res2['by_type']
assert res2['coverage']['covered_types'] == 4, res2['coverage']
def test_aggregate_evidence_types_filter():
install_fake(seed_rows())
res = run(A.pbl_evidence_aggregate(evidence_types=['decision', 'action']))
assert set(res['by_type']) == {'decision', 'action', 'observation', 'artifact_version'}
# 四类 ∩ {decision, action} 且通过默认过滤的保留行 = id 1、5(decision)+ id 2(action);
# id6(payload 预览标记)/id8(预览会话 999)虽属该两类,也被 Q-OPEN-10 默认口径剔除
assert res['total'] == 3, res['total']
assert res['by_type'] == {'decision': 2, 'action': 1, 'observation': 0,
'artifact_version': 0}, res['by_type']
assert res['coverage']['missing_types'] == ['observation', 'artifact_version'], res['coverage']
try:
run(A.pbl_evidence_aggregate(evidence_types='decision,bogus'))
raise AssertionError('非法 evidence_types 必须被拒')
except PblError as exc:
assert exc.code == 'PBL_E_VALIDATION', exc.code
# ══════════════════════════════════════════════════════════════════════
# 3. 列表查询契约(分页 + payload + 水位)
# ══════════════════════════════════════════════════════════════════════
def test_query_pagination_and_payload():
install_fake(seed_rows())
res = run(A.pbl_evidence_query(page=1, rows=2))
assert res['total'] == 5, res['total']
assert len(res['rows']) == 2, len(res['rows'])
assert res['rows'][0]['payload'].get('event_type'), res['rows'][0]
page2 = run(A.pbl_evidence_query(page=2, rows=2))
ids = {r['id'] for r in res['rows']} ^ {r['id'] for r in page2['rows']}
assert len(ids) == 4, ids # 翻页不重叠
tail = run(A.pbl_evidence_query(page=3, rows=2))
assert len(tail['rows']) == 1, tail['rows']
assert {r['id'] for r in tail['rows']} == {1}, tail['rows'] # occurred_at 倒序末位
assert run(A.pbl_evidence_query(page=9, rows=20))['rows'] == []
def test_query_filters_by_learner_and_type():
install_fake(seed_rows())
res = run(A.pbl_evidence_query(learner_id='S02'))
assert res['total'] == 2, res['total']
assert {r['id'] for r in res['rows']} == {3, 4}, res['rows']
assert {r['learner_id'] for r in res['rows']} == {'S02'}, res['rows']
res2 = run(A.pbl_evidence_query(evidence_type='decision'))
assert res2['total'] == 2, res2['total']
# ══════════════════════════════════════════════════════════════════════
# 4. 主体档案契约(学生 / 团队)
# ══════════════════════════════════════════════════════════════════════
def test_subject_summary_student_and_team():
install_fake(seed_rows())
stu = run(A.pbl_evidence_subject_summary(subject_type='student', subject_id='S02'))
assert stu['ok'] is True, stu
assert stu['evidence_total'] == 2, stu['evidence_total']
# S02 的保留行 = id3 observation + id4 artifact_version;decision/action 恒 0 但键必须在
assert stu['by_type'] == {'decision': 0, 'action': 0, 'observation': 1,
'artifact_version': 1}, stu['by_type']
assert set(stu['evidence']) == {'decision', 'action', 'observation', 'artifact_version'}
assert [e['id'] for e in stu['evidence']['observation']] == [3], stu['evidence']
assert stu['evidence']['decision'] == [], '四类恒有键(缺数据为空数组,前端稳定渲染)'
assert stu['artifact_count'] == 2, stu['artifact_count']
tea = run(A.pbl_evidence_subject_summary(subject_type='team', team_id='101',
blueprint_id='9'))
assert tea['subject_id'] == '101', tea
# 会话 101 的保留行 = id 1,2,3,4,5;id6/id7(payload 预览标记)与 id9(STUDENT 私有)
# 被剔除(id8 属预览保留会话 999,本就不在 session=101 的扫描范围内)
assert tea['evidence_total'] == 5, tea['evidence_total']
assert tea['by_type'] == {'decision': 2, 'action': 1, 'observation': 1,
'artifact_version': 1}, tea['by_type']
assert sorted(i['id'] for i in tea['evidence']['decision']) == [1, 5], tea['evidence']
assert tea['preview_filter']['excluded_preview'] == 2, tea['preview_filter']
assert tea['preview_filter']['excluded_student'] == 1, tea['preview_filter']
try:
run(A.pbl_evidence_subject_summary(subject_type='school', subject_id='X'))
raise AssertionError('非法 subject_type 必须被拒')
except PblError as exc:
assert exc.code == 'PBL_E_PARAM', exc.code
try:
run(A.pbl_evidence_subject_summary(subject_type='student'))
raise AssertionError('缺 subject_id 必须被拒')
except PblError as exc:
assert exc.code == 'PBL_E_PARAM', exc.code
# ══════════════════════════════════════════════════════════════════════
# 5. Q-OPEN-10 预览 / Playtest 默认过滤
# ══════════════════════════════════════════════════════════════════════
def test_qopen10_preview_excluded_by_default():
install_fake(seed_rows())
default = run(A.pbl_evidence_query())
withprev = run(A.pbl_evidence_query(include_preview=True,
include_student_private=True,
preview_session_ids=['999']))
assert default['total'] == 5, default['total']
assert {r['id'] for r in default['rows']} == KEPT_DEFAULT, default['rows']
assert default['preview_filter']['excluded_preview'] == len(PREVIEW_DEFAULT), default['preview_filter']
assert default['preview_filter']['excluded_student'] == len(STUDENT_DEFAULT), default['preview_filter']
assert withprev['total'] == 9, withprev['total']
assert withprev['preview_filter']['applied'] is False, withprev['preview_filter']
assert withprev['by_type']['observation'] == 2, withprev['by_type']
def test_qopen10_default_preview_session_range():
"""Q-OPEN-10 号段口径:预览保留会话默认剔除;显式 False 可关闭;追加号段生效。"""
install_fake(seed_rows())
res = run(A.pbl_evidence_query())
assert 8 not in {r['id'] for r in res['rows']}, res['rows']
assert res['preview_filter']['excluded_preview'] == 3, res['preview_filter']
assert res['preview_filter']['default_preview_session_ids'] == ['999'], res['preview_filter']
off = run(A.pbl_evidence_query(include_default_preview_sessions=False))
assert 8 in {r['id'] for r in off['rows']}, off['rows']
assert off['total'] == 6, off['total']
extra = run(A.pbl_evidence_query(preview_session_ids=['101']))
assert extra['total'] == 0, extra['total'] # 正式会话被声明为预览后全部剔除
assert extra['preview_filter']['excluded_preview'] == 9, extra['preview_filter']
def test_qopen10_rules_pure():
prev, rule = A.is_preview_row({'payload_json': json.dumps({'pbl_preview': True}),
'session_id': '1'})
assert prev and rule == 'payload:pbl_preview', rule
prev, rule = A.is_preview_row({'payload_json': json.dumps({'env_mode': 'playtest'}),
'session_id': '1'})
assert prev and rule == 'payload:env_mode', rule
prev, rule = A.is_preview_row({'payload_json': json.dumps({'mode': 'demo'}),
'session_id': '1'})
assert prev and rule == 'payload:mode', rule
prev, rule = A.is_preview_row({'payload_json': json.dumps({'event_type': 'x'}),
'session_id': '999'}, ['999'])
assert prev and rule == 'preview_session:999', rule
prev, _ = A.is_preview_row({'payload_json': '{bad json', 'session_id': '1'})
assert prev is False, '坏 JSON 不得误判为预览'
priv, rule = A.is_student_private({'payload_json': json.dumps({'visibility': 'STUDENT'})})
assert priv and rule.startswith('payload:visibility'), rule
priv, _ = A.is_student_private({'payload_json': json.dumps({'visibility': 'teacher'})})
assert priv is False
kept, stats = A.apply_default_filters(
[{'id': 1, 'payload_json': '{}', 'session_id': '1', 'learner_id': 'S01'}],
{'subject_type': 'student', 'subject_id': 'S02'})
assert kept == [] and stats['excluded_subject'] == 1, stats
key, derived = A.resolve_idempotency_key({'tenant_id': '1', 'source_event_id': 'EV1',
'evidence_type': 'action'})
key2, _ = A.resolve_idempotency_key({'tenant_id': '1', 'source_event_id': 'EV1',
'evidence_type': 'action'})
assert key == key2 and derived is True, (key, key2)
assert A.resolve_idempotency_key({'idempotency_key': 'abc'})[0] == 'abc'
def test_sql_shape_and_column_pruning():
install_fake(seed_rows())
_CAPTURED['sql'].clear()
run(A.pbl_evidence_aggregate(group_by='learner'))
sql = [s for s in _CAPTURED['sql'] if 'from `pbl_evidence`' in s.lower()][0]
assert 'select `id`, `evidence_type`' in sql.lower(), sql
assert '`source_event_id`' in sql and '`dedup_key`' in sql and '`payload_json`' in sql, sql
assert 'order by `occurred_at` desc' in sql.lower(), sql
assert 'limit ${scan_limit}$' in sql.lower(), sql
def test_watermark_contract():
install_fake(seed_rows())
agg = run(A.pbl_evidence_aggregate(group_by='blueprint'))
wm = agg['watermark']
assert wm['count'] == 5 and wm['max_id'] == 5, wm
assert wm['min_occurred_at'] == '2026-09-20 10:00:01', wm
q = run(A.pbl_evidence_query(learner_id='S03'))
assert q['watermark']['max_occurred_at'] == '2026-09-21 08:00:00', q['watermark']
# 不能直接对 function 对象排序('<' not supported between functions),
# 按函数名排序保证输出顺序稳定可复现。
ALL_TESTS = sorted([v for k, v in list(globals().items())
if k.startswith('test_') and callable(v)],
key=lambda fn: fn.__name__)
def main():
passed, failed = 0, 0
for fn in ALL_TESTS:
try:
fn()
except Exception as exc:
failed += 1
print('FAIL %s: %s' % (fn.__name__, exc))
else:
passed += 1
print('PASS %s' % fn.__name__)
print('\n== M5b selftest: %d passed, %d failed, %d total ==' % (passed, failed, len(ALL_TESTS)))
return 1 if failed else 0
if __name__ == '__main__':
sys.exit(main())