#!/usr/bin/env python3
"""总表维护的相关行为契约;离线保存隔离工件,不验证功法效果。"""
import argparse
from collections import Counter
from copy import deepcopy
from hashlib import sha256
from html import unescape
import importlib.util
import json
from pathlib import Path
import re
import shutil
import subprocess
import sys
import tempfile
import unittest
import source_archive
ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / '.github/sync-gongfa-catalog.py'
ARTIFACTS = None
catalog = None
TEMPLATE = Path(__file__).resolve().parents[1] / '.github/tool-explanation-template-v0.md'
V1_TEMPLATE = Path(__file__).resolve().parents[1] / '.github/tool-explanation-template-v1.md'
V2_TEMPLATE = Path(__file__).resolve().parents[1] / '.github/tool-explanation-template-v2.md'
STALE_EXPLANATION_LABELS = (
'冻结候选', '候选快照', '尚未独立语义复核', '未独立语义复核',
'尚未独立语义核验', '不代表独立语义批准',
)
def template_rows(path):
rows = []
for line in path.read_text(encoding='utf-8').splitlines():
parts = line.split('|')
if len(parts) == 5 and parts[1].strip().isdigit():
rows.append((int(parts[1].strip()), parts[2].strip()))
return rows
def v0_field_names():
return [name for _number, name in template_rows(TEMPLATE)]
def v2_field_names():
return [name for _number, name in template_rows(V2_TEMPLATE)]
def save(path, value):
with path.open('x', encoding='utf-8') as stream:
json.dump(value, stream, ensure_ascii=False, indent=2, allow_nan=False)
def parse_pipe_row(line):
cells = []
current = []
index = 1
while index < len(line) - 1:
if line[index:index + 2] == r'\|':
current.append('|')
index += 2
elif line[index] == '|':
cells.append(''.join(current).strip())
current = []
index += 1
else:
current.append(line[index])
index += 1
cells.append(''.join(current).strip())
return cells
class CatalogMaintenanceTest(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.documents, cls.binding = catalog.load_inputs(ROOT)
def isolated_project(self):
root = Path(tempfile.mkdtemp(prefix='project-', dir=ARTIFACTS))
target = root / '.github/context/gongfa'
target.mkdir(parents=True)
shutil.copyfile(ROOT / '.github/context/gongfa/data.json', target / 'data.json')
context_dir = root / '.github'
shutil.copyfile(V2_TEMPLATE, context_dir / V2_TEMPLATE.name)
return root
def command(self, root, *options):
argv = [sys.executable, str(SCRIPT), '--root', str(root), *options]
result = subprocess.run(argv, cwd=ARTIFACTS, capture_output=True, text=True, timeout=60)
log = Path(tempfile.mkdtemp(prefix='command-', dir=ARTIFACTS))
save(log / 'result.json', {'argv': argv, 'cwd': str(ARTIFACTS), 'exit_code': result.returncode, 'timeout_seconds': 60})
for name, text in [('stdout.log', result.stdout), ('stderr.log', result.stderr)]:
with (log / name).open('x', encoding='utf-8') as stream:
stream.write(text)
return result
def test_user_requested_framework_and_input_output_tiers_are_separate_provisional_opinions(self):
rows = catalog.merge_rows(deepcopy(self.documents))
by_id = {(row['namespace'], row['original_id']): row for row in rows}
framework = by_id['registry', 'gongfa-input-state-transform-output-cross-domain']
input_output = by_id['registry', 'gongfa-input-state-transform-separation']
state_classification = by_id['registry', 'gongfa-state-classification']
self.assertEqual(framework['grade_id'], 'tian-initial')
self.assertEqual(framework['grading_status'], 'provisional')
self.assertEqual(input_output['grade_id'], 'tian-intermediate')
self.assertEqual(input_output['grading_status'], 'provisional')
self.assertEqual(state_classification['grade_id'], 'xuan-intermediate')
self.assertEqual(self.documents['registry']['assessments'], [])
previous = next(batch for batch in self.documents['registry']['grade_proposals']
if batch['id'] == 'gongfa-author-repo-expansion-20261007')
successor = next(batch for batch in self.documents['registry']['grade_proposals']
if batch['id'] == 'gongfa-user-input-state-transform-separation-tian-20261024')
framework_batch = next(batch for batch in self.documents['registry']['grade_proposals']
if batch['id'] == 'gongfa-user-cross-domain-framework-tian-20261024')
self.assertEqual(successor['supersedes'], previous['id'])
self.assertIsNone(framework_batch['supersedes'])
previous_items = {(item['gongfa_id'], item['content_version']): item for item in previous['items']}
successor_items = {(item['gongfa_id'], item['content_version']): item for item in successor['items']}
framework_items = {(item['gongfa_id'], item['content_version']): item for item in framework_batch['items']}
self.assertEqual(set(successor_items), set(previous_items))
self.assertEqual(set(framework_items), {(framework['original_id'], 'r1')})
for key, item in previous_items.items():
if key[0] != 'gongfa-input-state-transform-separation':
self.assertEqual(successor_items[key], item)
self.assertEqual(previous_items['gongfa-input-state-transform-separation', 'r1']['grade_id'],
'xuan-intermediate')
self.assertEqual(successor_items['gongfa-input-state-transform-separation', 'r1']['grade_id'],
'tian-intermediate')
self.assertEqual(framework_items[framework['original_id'], 'r1']['grade_id'], 'tian-initial')
self.assertIn('用户明确指定', framework['explanation_fields'][8]['content'])
self.assertIn('天阶初级', framework['explanation_fields'][8]['content'])
self.assertIn('用户明确指定', input_output['explanation_fields'][8]['content'])
self.assertIn('天阶中级', input_output['explanation_fields'][8]['content'])
def test_experience_pattern_transfer_uses_provisional_tian_initial_and_keeps_prior_opinion(self):
entry_id = 'gongfa-experience-pattern-transfer-feedback'
entries = [entry for entry in self.documents['registry']['gongfa'] if entry['id'] == entry_id]
self.assertEqual(len(entries), 1)
entry = entries[0]
self.assertEqual(entry['name'], '经验规律迁移与反馈校正')
self.assertEqual(entry['content_version'], 'r1')
self.assertEqual(entry['entry_status'], 'registered')
self.assertEqual(entry['ontology_type'], 'method-specification')
finite_sample = [item for item in self.documents['registry']['gongfa']
if item['id'] == 'gongfa-finite-sample-generative-inference']
self.assertEqual(len(finite_sample), 1)
self.assertEqual(finite_sample[0]['name'], '有限样本生成推断')
source = next(source for source in self.documents['registry']['sources']
if source['id'] == entry['source_ref']['source_id'])
self.assertEqual(source['kind'], 'repository_document')
self.assertEqual(source['path'], '.github/docs/gongfa/experience-pattern-transfer-feedback.md')
self.assertEqual(entry['source_ref']['heading_id'], 'experience-pattern-transfer-feedback')
self.assertEqual(entry['source_ref']['selection'], 'markdown-section')
original_text = entry['content']['original_text']
for required in ('当前任务', '过往经验', '比较情境', '差异', '调整', '实际结果', '因果', '训练模型'):
self.assertIn(required, original_text)
self.assertIn('不保证性能提升或最优解', original_text)
batches = [batch for batch in self.documents['registry']['grade_proposals']
if any(item['gongfa_id'] == entry_id for item in batch['items'])]
self.assertEqual(len(batches), 2)
previous = next(batch for batch in batches
if batch['id'] == 'gongfa-author-experience-pattern-transfer-feedback-20261025')
current = next(batch for batch in batches
if batch['id'] == 'gongfa-user-experience-pattern-transfer-feedback-tian-20261025')
previous_proposal = next(item for item in previous['items'] if item['gongfa_id'] == entry_id)
proposal = next(item for item in current['items'] if item['gongfa_id'] == entry_id)
self.assertIsNone(previous_proposal['grade_id'])
self.assertEqual(previous_proposal['grading_status'], 'undetermined')
self.assertEqual(current['supersedes'], previous['id'])
self.assertEqual(current['application_context'], previous['application_context'])
self.assertEqual(current['baseline_assumption'], previous['baseline_assumption'])
self.assertEqual(proposal['grade_id'], 'tian-initial')
self.assertEqual(proposal['grading_status'], 'provisional')
self.assertEqual(proposal['confidence'], 'low')
self.assertIn('天阶初级', proposal['reason'])
self.assertIn('暂定分类', proposal['reason'])
self.assertEqual(self.documents['registry']['assessments'], [])
key = f'registry:{entry_id}@r1'
explanation = next(item for item in self.documents['explanations_v2']['items']
if item['object_key'] == key)
fields = {field['name']: field['content'] for field in explanation['fields']}
self.assertEqual(list(fields), v2_field_names())
self.assertEqual(fields['概念'], '面向单项任务的跨情境经验迁移与结果反馈校正方法')
self.assertIn('当前单项任务', fields['定义'])
self.assertIn('比较情境差异', fields['怎么做'])
self.assertIn('比较实际结果', fields['怎么做'])
self.assertIn('不要求训练模型', fields['扩展'])
self.assertIn('未核验', fields['相关图书'])
self.assertIn('不证明效果或因果', fields['结论'])
self.assertIn('不据此外推未观察效果、许可或正式评级', fields['连接规则'])
self.assertIn(source['id'], fields['依据'])
rows = catalog.merge_rows(deepcopy(self.documents))
row = next(row for row in rows if row['original_id'] == entry_id)
self.assertEqual(row['entry_status'], '已登记')
self.assertEqual(row['grade_id'], 'tian-initial')
self.assertEqual(row['grading_status'], 'provisional')
self.assertIn('天阶初级', fields['关联信息'])
self.assertIn('不是正式评级', fields['关联信息'])
self.assertEqual(row['explanation_fields'], explanation['fields'])
def test_current_all_classified_content_and_two_bad_summaries_keep_original_opinions(self):
rows = catalog.merge_rows(deepcopy(self.documents))
self.assertEqual(Counter(row['namespace'] for row in rows), {'registry': 413})
self.assertEqual(len(rows), 413)
self.assertEqual(len({(r['namespace'], r['original_id'], r['content_version']) for r in rows}), 413)
undetermined = {r['original_id'] for r in rows if r['grading_status'] == 'undetermined'}
self.assertEqual(undetermined, {
'gongfa-candidate-html211-e001-a663456869',
'gongfa-candidate-html96-vc029-6cca304cec',
})
self.assertEqual(sum(r['grading_status'] == 'provisional' for r in rows), 411)
by_id = {(r['namespace'], r['original_id']): r for r in rows}
self.assertEqual(by_id['registry', 'gongfa-first-principles']['grade_id'], 'di-advanced')
self.assertEqual(by_id['registry', 'gongfa-candidate-solve24-psoa-programming-minimal-reproduction-47a914cf5a']['grade_id'],
'di-initial')
undetermined_html96 = 'gongfa-candidate-html96-vc029-6cca304cec'
self.assertIn('一个分布式系统最多只能保证一致性、可用性和分区容错性中的两项。',
by_id['registry', undetermined_html96]['original_text'])
self.assertEqual(by_id['registry', 'gongfa-candidate-html211-e001-a663456869']['grading_status'], 'undetermined')
self.assertIn('失去50%容量时剩余单位需要承担2倍总流量。',
by_id['registry', 'gongfa-candidate-html211-e001-a663456869']['original_text'])
self.assertEqual(sum(r['entry_status'] == '已登记' for r in rows), 413)
lambda_index = next(index for index, row in enumerate(rows) if row['original_id'] == 'gongfa-lambda-calculus')
glue_index = next(index for index, row in enumerate(rows) if row['original_id'] == 'gongfa-glue-coding')
self.assertEqual(lambda_index + 1, glue_index)
self.assertEqual(rows[lambda_index]['name'], 'Lambda 演算')
self.assertEqual(rows[lambda_index]['grade_id'], 'tian-advanced')
self.assertIn('天级上品', rows[lambda_index]['reason'])
rendered = catalog.book_views.render_sources(ROOT, 'gongfa')
chapters = [line for line in rendered.splitlines() if line.startswith('## ')]
self.assertEqual(chapters[chapters.index('## 拼好码') + 1], '## Lambda 演算')
self.assertEqual(sum(r['entry_status'] == '登记库候选' for r in rows), 0)
self.assertEqual(self.documents['registry']['assessments'], [])
self.assertEqual(len(self.documents['registry']['gongfa']), 413)
self.assertTrue(all(item['entry_status'] == 'registered' for item in self.documents['registry']['gongfa']))
registry_item = next(item for item in self.documents['registry']['gongfa'] if item['id'] == 'gongfa-first-principles')
profile = registry_item['content']['profile']
profile_parts = [profile['summary'], '问题:' + profile['problem'], '机制:' + profile['mechanism']]
for key, label in (('conditions', '条件'), ('limitations', '限制'), ('expected_outcomes', '预期结果')):
if profile[key]:
profile_parts.append(label + ':' + ';'.join(profile[key]))
self.assertEqual(' '.join(profile_parts), by_id['registry', 'gongfa-first-principles']['explanation'])
entries_by_anchor = {entry['source_ref']['heading_id']: entry
for entry in self.documents['registry']['gongfa']
if entry['source_ref']['source_id'] == 'source-classified-candidates-20261007'}
self.assertEqual(len(entries_by_anchor), 331)
self.assertEqual(set(entries_by_anchor), {
*(f"html211:{item['candidate_id']}" for item in self.documents['html211_candidates']['candidates']),
*(f"html96:{item['candidate_id']}" for item in self.documents['html96_candidates']['candidates']),
*(f"solve24:{item['source_id']}" for item in self.documents['solve24_analysis']['items']),
})
proposal_items = {
(item['gongfa_id'], item['content_version']): item
for batch in self.documents['registry']['grade_proposals'] for item in batch['items']
}
candidate_grades = [
(f"html211:{grade['candidate_id']}", grade)
for grade in self.documents['html211_grades']['items']
] + [
(f"html96:{grade['candidate_id']}", grade)
for grade in self.documents['html96_grades']['items']
] + [
(f"solve24:{item['source_id']}", item['editorial_analysis'])
for item in self.documents['solve24_analysis']['items']
]
for anchor, previous in candidate_grades:
entry = entries_by_anchor[anchor]
current = proposal_items[entry['id'], entry['content_version']]
grade = previous.get('provisional_grade_id', previous.get('grade_id'))
self.assertEqual(current['grade_id'], grade, anchor)
self.assertEqual(current['task_family'], previous['task_family'], anchor)
self.assertEqual(current['reason'], previous['reason'], anchor)
self.assertEqual(current['falsifier'], previous['falsifier'], anchor)
self.assertEqual(current['confidence'], previous['confidence'], anchor)
self.assertEqual(current['content_sha256'], entry['content']['text_sha256'], anchor)
for anchor in ('html211:E001', 'html96:VC029'):
entry = entries_by_anchor[anchor]
proposal = proposal_items[entry['id'], entry['content_version']]
self.assertIsNone(proposal['grade_id'])
self.assertEqual(proposal['grading_status'], 'undetermined')
row = by_id['registry', entry['id']]
self.assertEqual(row['grading_status'], 'undetermined')
self.assertEqual(row['explanation_fields'][8]['content'].split('登记状态:', 1)[1].split(';', 1)[0], '已登记')
self.assertIn('.github/metadata/gongfa/source-classified-candidates-20261007.json',
row['explanation_fields'][4]['content'])
self.assertEqual(self.documents['html211_candidates']['registry_updated'], False)
self.assertEqual(self.documents['html96_candidates']['registry_updated'], False)
self.assertEqual(self.documents['explanations_v2']['template_sha256'], catalog.tool_explanations.template_sha256(ROOT))
self.assertEqual(len(self.documents['explanations_v2']['items']), 413)
self.assertTrue(all(item['object_key'].startswith('registry:')
for item in self.documents['explanations_v2']['items']))
views = catalog.render_views(ROOT, rows, self.binding)
markdown = views['README.md']
self.assertIn('- 共**413条内容记录**;已登记413条,待登记0条,未登记来源条目0条。', markdown)
self.assertNotIn('[控制论与科学方法论](#控制论与科学方法论)', markdown)
self.assertIn('同名条目在导航中附稳定对象ID,不表示身份合并。', markdown)
self.assertFalse('catalog.xlsx' in markdown, '活动阅读视图只有Markdown,历史Excel仅作原始载荷保存')
self.assertLess(len(markdown.encode('utf-8')), 512 * 1024, 'README只作为轻量导航,不再展开全部功法表格与文集正文')
self.assertNotIn('| 序号 | 说明项 | 内容 |', markdown)
self.assertIn('## 原文文集', markdown)
self.assertIn('', markdown)
self.assertIn('tian.md#', markdown)
self.assertIn('huang.md#', markdown)
field_names = v2_field_names()
self.assertEqual(20, len(field_names))
total_tables = 0
for tier in ('tian', 'di', 'xuan', 'huang'):
page = views[tier + '.md']
tier_rows = [row for row in rows if row['grade_id'] and row['grade_id'].startswith(tier + '-')]
if tier == 'huang':
tier_rows += [row for row in rows if row['grade_id'] is None]
self.assertIn('## 未明确品级(不归属黄阶)', page)
starts = [index for index, line in enumerate(page.splitlines()) if line == '| 序号 | 说明项 | 内容 |']
self.assertEqual(len(tier_rows), len(starts), f'{tier}页面只包含本阶记录及明确标记的未定级记录')
total_tables += len(starts)
lines = page.splitlines()
for row, start in zip(tier_rows, starts):
table_rows = lines[start:start + 22]
self.assertEqual(22, len(table_rows))
self.assertEqual('|:---:|:---|:---|', table_rows[1])
cells = [parse_pipe_row(line) for line in table_rows]
self.assertTrue(all(len(cell) == 3 for cell in cells), '每张表严格为三列')
self.assertEqual(field_names, [cell[1] for cell in cells[2:]])
display_name = re.sub(r'^\d+(?:\.\d+)+[ \t]+', '', row['name'], count=1)
self.assertEqual(display_name, unescape(cells[3][2]))
anchor = catalog.tool_explanations.anchor_for(row['key'])
self.assertIn(f'{tier}.md#{anchor}', markdown)
self.assertIn(f'', markdown, '旧README对象锚点应保留为跳转入口')
self.assertEqual(len(rows), total_tables, '所有对象恰好进入一个阶级页或黄阶页中的未定级附录')
self.assertIn('第一性原理', views['di.md'])
self.assertIn('天阶高级', views['tian.md'])
self.assertNotIn('天高', views['tian.md'])
self.assertNotIn('地高', views['tian.md'])
def test_numbered_chapter_prefixes_are_removed_from_all_rendered_views(self):
rows = catalog.merge_rows(deepcopy(self.documents))
views = catalog.render_views(ROOT, rows, self.binding)
prefix = re.compile(r'^\d+(?:\.\d+)+[ \t]+')
numbered = [row for row in rows if prefix.match(row['name'])]
self.assertEqual(len(numbered), 101)
self.assertIn('3.5 效应局部化', [row['name'] for row in numbered])
duplicate_display = [row for row in rows
if prefix.sub('', row['name'], count=1) == '信息隐藏']
self.assertEqual(len(duplicate_display), 2)
self.assertEqual(len({catalog.tool_explanations.anchor_for(row['key'])
for row in duplicate_display}), 2)
display_counts = Counter(prefix.sub('', row['name'], count=1) for row in rows)
for row in numbered:
match = prefix.match(row['name'])
display_name = row['name'][match.end():]
navigation_label = (f'{display_name}({row["key"]})'
if display_counts[display_name] > 1 else display_name)
anchor = catalog.tool_explanations.anchor_for(row['key'])
tier = row['grade_id'].split('-', 1)[0] if row['grade_id'] else 'huang'
page = views[tier + '.md']
link = next(line for line in page.splitlines() if line.endswith(f'(#{anchor})'))
decoded_link = unescape(link)
self.assertIn(f'[{navigation_label}](#{anchor})', decoded_link)
self.assertNotIn(row['name'], decoded_link)
section = page.split(f'', 1)[1].split('', 1)[1].split('bad|=2+2'
rows = catalog.merge_rows(documents)
hostile_row = next(row for row in rows if row['namespace'] == 'registry' and row['original_id'] == entry['id'])
if 'explanation_fields' not in hostile_row:
hostile_row['explanation_fields'] = [{'name': label, 'content': item['name'] if index == 1 else '隔离安全测试'}
for index, label in enumerate(v2_field_names())]
hostile_row['explanation_fields'][1]['content'] = item['name']
root = Path(tempfile.mkdtemp(prefix='markdown-', dir=ARTIFACTS))
target = root / '.github/context/gongfa'
target.mkdir(parents=True)
catalog.write_views(root, rows, self.binding)
catalog.check_views(root, rows, self.binding)
markdown = (target / 'README.md').read_text()
self.assertNotIn('', markdown)
self.assertIn('<script>bad</script>|=2+2', markdown)
target_tier = hostile_row['grade_id'].split('-', 1)[0] if hostile_row['grade_id'] else 'huang'
target_page = (target / (target_tier + '.md')).read_text()
self.assertNotIn('', target_page)
self.assertIn('<script>bad</script>|=2+2', target_page)
self.assertEqual({p.name for p in target.iterdir()},
{'README.md', 'tian.md', 'di.md', 'xuan.md', 'huang.md'})
(target / 'README.md').write_text(markdown.replace('天阶高级', '改动品级', 1))
with self.assertRaises(ValueError): catalog.check_views(root, rows, self.binding)
for text in ['超限' * (4 * 1024 * 1024), '非法\x01字符']:
bad = deepcopy(rows)
bad[0]['name'] = text
with self.assertRaises(ValueError): catalog.write_views(root, bad, self.binding)
hand_written = Path(tempfile.mkdtemp(prefix='owned-', dir=ARTIFACTS))
own_target = hand_written / '.github/context/gongfa'
own_target.mkdir(parents=True)
(own_target / 'README.md').write_text('手工内容,不能覆盖。', encoding='utf-8')
with self.assertRaises(ValueError): catalog.write_views(hand_written, rows, self.binding)
self.assertEqual((own_target / 'README.md').read_text(), '手工内容,不能覆盖。')
def main():
global ARTIFACTS, catalog
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--artifacts', type=Path)
args = parser.parse_args()
ARTIFACTS = args.artifacts or Path(tempfile.mkdtemp(prefix='gongfa-catalog-test-'))
if args.artifacts:
ARTIFACTS.mkdir(parents=True, exist_ok=False)
ARTIFACTS = ARTIFACTS.resolve()
spec = importlib.util.spec_from_file_location('gongfa_catalog', SCRIPT)
catalog = importlib.util.module_from_spec(spec)
spec.loader.exec_module(catalog)
result = unittest.TextTestRunner(verbosity=2).run(unittest.defaultTestLoader.loadTestsFromTestCase(CatalogMaintenanceTest))
save(ARTIFACTS / 'result.json', {'tests_run': result.testsRun, 'failures': len(result.failures), 'errors': len(result.errors),
'success': result.wasSuccessful(), 'script_sha256': sha256(SCRIPT.read_bytes()).hexdigest()})
print('测试工件:' + str(ARTIFACTS))
return 0 if result.wasSuccessful() else 1
if __name__ == '__main__':
sys.exit(main())