zheromiku
dcdc8d1ddd
全栈知识治理平台首次入库（P1-P10 + 发布链路模拟验证闭环）： - 治理核心：源目录/变更集/人工审核/发布批次/发布执行（fail-closed 闸口） - 测试通道：WeKnora 测试 KB 上传/轮询/验收；poll 完成时写入 parser_generation/index_generation，修复 manual/seed 路径发布闸口永久拦截 - 发布适配器 v071（本地 source=测试角色 / 云端 target=生产角色 模拟验证通过）： * ES 导出/导入 mapping 自适应（.keyword 子字段、可选 tenant_id） * OSS locator 支持嵌套 scheme；chunks.seq_id 由目标序列分配 * bundle 归档同身份重试原子替换；verifier 按 bundle chunk_type 计数 * 回滚 job payload 支持 prior_execution_id（strict 解码） - 三通道发布+回滚演练实测：publish 4/4 verified，检索命中， rollback 4/4 rolled_back，目标端 PG is_enabled=false、ES is_enabled=false - 前端 A·Data-console 风格全站统一，120 用例通过 - 门禁：go vet / go test 全包通过，gofmt 干净
143 lines
10 KiB
Python
143 lines
10 KiB
Python
import io
|
|
import unittest
|
|
from pathlib import Path
|
|
from source_catalog_import import PP_ROOTS, build_alist_source, build_dingtalk_sources, build_pp_live_source, build_pp_source, catalog_sql, safe_text, validate_source
|
|
|
|
|
|
class SourceCatalogImportTests(unittest.TestCase):
|
|
def test_control_characters_are_visible_instead_of_breaking_the_file_list(self):
|
|
self.assertEqual(safe_text('产品\n手册\t.pdf'),'产品手册␉.pdf')
|
|
|
|
def test_dingtalk_preserves_real_parent_child_nodes(self):
|
|
inventory={
|
|
'roots':{'company':{'configured_root_id':'root','live_space_name':'公司级技术支持知识库','nodes':{
|
|
'folder':{'dentry_uuid':'folder','parent_dentry_uuid':'root','dentry_type':'folder','name':'01｜产品资料','file_size':None,'updated_time':2},
|
|
'file':{'dentry_uuid':'file','parent_dentry_uuid':'folder','dentry_type':'file','name':'手册.pdf','file_size':12,'updated_time':3},
|
|
}}},
|
|
}
|
|
source=build_dingtalk_sources(inventory)[0]
|
|
root=next(node for node in source['nodes'] if node['parent_key'] is None)
|
|
folder=next(node for node in source['nodes'] if node['name']=='01｜产品资料')
|
|
file=next(node for node in source['nodes'] if node['name']=='手册.pdf')
|
|
self.assertEqual(folder['parent_key'],root['key'])
|
|
self.assertEqual(file['parent_key'],folder['key'])
|
|
self.assertEqual(root['subtree_file_count'],1)
|
|
validate_source(source)
|
|
|
|
def test_alist_uses_csv_parent_paths_and_recursive_counts(self):
|
|
directories=io.StringIO('path,parent_path,depth,status,direct_directories,direct_files,direct_bytes,subtree_files,subtree_bytes,last_error\n/, ,0,done,1,0,0,1,10,\n/01 公司资料,/,1,done,0,1,10,1,10,\n')
|
|
files=io.StringIO('path,parent_path,name,size,modified,extension\n/01 公司资料/a.pdf,/01 公司资料,a.pdf,10,2026-01-01T00:00:00Z,.pdf\n')
|
|
source=build_alist_source('domestic-industrial','国内工业',directories,files)
|
|
folder=next(node for node in source['nodes'] if node['name']=='01 公司资料')
|
|
file=next(node for node in source['nodes'] if node['name']=='a.pdf')
|
|
self.assertEqual(file['parent_key'],folder['key'])
|
|
self.assertEqual(source['file_count'],1)
|
|
validate_source(source)
|
|
|
|
def test_pp_uses_eight_real_tag_types_and_keeps_untagged_public_files_at_root(self):
|
|
inventory={'items':[
|
|
{'id':'1','fileName':'CRA手册','fileSize':10,'updateTime':'2026-01-01 00:00:00','potalTags':[{'tagType':'工业产品','tag':'CRA系列'},{'tagType':'解决方案','tag':'焊接'}]},
|
|
{'id':'2','fileName':'未分类文档','fileSize':20,'updateTime':'2026-01-02 00:00:00','potalTags':[]},
|
|
]}
|
|
manifest={'results':[{'url':'https://safe.invalid/a','source_ids':['1']},{'url':'https://safe.invalid/b','source_ids':['2']}]}
|
|
source=build_pp_source(manifest,inventory)
|
|
root=next(node for node in source['nodes'] if node['parent_key'] is None)
|
|
first=[node['name'] for node in source['nodes'] if node['parent_key']==root['key'] and node['node_type']=='folder']
|
|
self.assertEqual(first,list(PP_ROOTS))
|
|
self.assertTrue(any(node['name']=='未分类文档' and node['parent_key']==root['key'] for node in source['nodes']))
|
|
self.assertEqual(root['subtree_file_count'],2)
|
|
self.assertEqual(source['file_count'],2)
|
|
validate_source(source)
|
|
|
|
def test_pp_records_real_content_hash_and_rejects_malformed_hash(self):
|
|
sha='a'*64
|
|
inventory={'items':[{'id':'1','fileName':'带哈希手册','fileSize':10,'updateTime':'2026-01-01 00:00:00','potalTags':[]}]}
|
|
manifest={'results':[{'url':'https://safe.invalid/a','source_ids':['1'],'sha256':sha.upper()}]}
|
|
source=build_pp_source(manifest,inventory)
|
|
file=next(node for node in source['nodes'] if node['node_type']=='file')
|
|
self.assertEqual(file['content_sha256'],sha)
|
|
missing={'results':[{'url':'https://safe.invalid/b','source_ids':['1']}]}
|
|
source2=build_pp_source(missing,inventory)
|
|
self.assertIsNone(next(node for node in source2['nodes'] if node['node_type']=='file')['content_sha256'])
|
|
bad={'results':[{'url':'https://safe.invalid/c','source_ids':['1'],'sha256':'zz'}]}
|
|
with self.assertRaises(ValueError):
|
|
build_pp_source(bad,inventory)
|
|
|
|
def test_pp_public_items_without_file_name_fall_back_to_the_source_url_title(self):
|
|
inventory={'items':[
|
|
{'id':'9','fileName':'','fileSize':10,'updateTime':'2026-01-03 00:00:00','potalTags':[{'tagType':'工业产品','tag':'CRA系列'}],'online':1,'fileSuffix':'pdf','fileUrl':'https://safe.invalid/path/CRA%20User%20Guide_V1.0_en.pdf','potalGroup':{'groupName':'公开'}},
|
|
]}
|
|
source=build_pp_live_source({'items':inventory['items']})
|
|
file=next(node for node in source['nodes'] if node['node_type']=='file')
|
|
self.assertEqual(file['name'],'CRA User Guide_V1.0_en.pdf')
|
|
validate_source(source)
|
|
|
|
def test_live_pp_snapshot_derives_extension_from_url_when_file_suffix_is_empty(self):
|
|
empty_suffix={'id':'1','online':1,'fileName':'无后缀手册','fileSize':10,'fileSuffix':'','fileUrl':'https://safe.invalid/path/%E6%89%8B%E5%86%8C.pdf','potalGroup':{'groupName':'公开'},'potalTags':[{'tagType':'工业产品','tag':'CRA系列'}]}
|
|
media={'id':'2','online':1,'fileName':'培训视频','fileSize':10,'fileSuffix':'','fileUrl':'https://safe.invalid/video.mp4','potalGroup':{'groupName':'公开'},'potalTags':[]}
|
|
source=build_pp_live_source({'items':[empty_suffix,media]})
|
|
self.assertEqual(source['file_count'],1)
|
|
file=next(node for node in source['nodes'] if node['node_type']=='file')
|
|
self.assertEqual(file['name'],'无后缀手册')
|
|
validate_source(source)
|
|
|
|
def test_live_pp_snapshot_merges_duplicate_public_urls_into_one_node(self):
|
|
first={'id':'1','online':1,'fileName':'','fileSize':10,'fileSuffix':'pdf','fileUrl':'https://safe.invalid/dup.pdf','potalGroup':{'groupName':'公开'},'potalTags':[]}
|
|
second={**first,'id':'2','fileName':'同一份手册'}
|
|
source=build_pp_live_source({'items':[first,second]})
|
|
self.assertEqual(source['file_count'],1)
|
|
file=next(node for node in source['nodes'] if node['node_type']=='file')
|
|
self.assertEqual(file['name'],'同一份手册')
|
|
validate_source(source)
|
|
|
|
def test_live_pp_snapshot_backfills_content_hash_from_local_manifest(self):
|
|
public={'id':'1','online':1,'fileName':'公开手册','fileSize':10,'fileSuffix':'pdf','fileUrl':'https://safe.invalid/a','potalGroup':{'groupName':'公开'},'potalTags':[]}
|
|
fresh={**public,'id':'2','fileName':'新上传','fileUrl':'https://safe.invalid/b'}
|
|
manifest={'results':[{'url':'https://safe.invalid/a','sha256':'ab'*32,'bytes':10}]}
|
|
source=build_pp_live_source({'items':[public,fresh]},manifest)
|
|
by_name={node['name']:node for node in source['nodes'] if node['node_type']=='file'}
|
|
self.assertEqual(by_name['公开手册']['content_sha256'],'ab'*32)
|
|
self.assertIsNone(by_name['新上传']['content_sha256'])
|
|
validate_source(source)
|
|
|
|
def test_live_pp_snapshot_keeps_only_online_public_supported_documents(self):
|
|
public={'id':'1','online':1,'fileName':'公开手册','fileSize':10,'fileSuffix':'pdf','fileUrl':'https://safe.invalid/a','potalGroup':{'groupName':'公开'},'potalTags':[{'tagType':'工业产品','tag':'CRA系列'}]}
|
|
private={**public,'id':'2','fileName':'代理商资料','fileUrl':'https://safe.invalid/b','potalGroup':{'groupName':'代理商可查'}}
|
|
offline={**public,'id':'3','fileName':'离线资料','fileUrl':'https://safe.invalid/c','online':0}
|
|
source=build_pp_live_source({'items':[public,private,offline]})
|
|
self.assertEqual(source['file_count'],1)
|
|
self.assertTrue(any(node['name']=='公开手册' for node in source['nodes']))
|
|
self.assertFalse(any(node['name'] in ('代理商资料','离线资料') for node in source['nodes']))
|
|
|
|
def test_validation_rejects_orphan_and_host_paths(self):
|
|
source=build_pp_source({'results':[]},{'items':[]})
|
|
source['nodes'][0]['relative_path']='/home/private'
|
|
with self.assertRaises(ValueError): validate_source(source)
|
|
|
|
def test_catalog_sql_is_append_only_and_activates_an_immutable_snapshot(self):
|
|
source=build_pp_source({'results':[]},{'items':[]})
|
|
sql=catalog_sql(source,'sc-pp',apply=False)
|
|
lowered=sql.lower()
|
|
for required in ('insert into source_catalog_snapshots','insert into source_catalog_nodes','insert into source_catalog_change_sets','select compare_catalog_change_set(','insert into source_catalog_heads','rollback'):
|
|
self.assertIn(required,lowered)
|
|
# Classification actions live in the single-source DB function (P2/S8).
|
|
migration=(Path(__file__).resolve().parents[1]/'internal/platform/postgres/migrations/000022_catalog_change_comparison.up.sql').read_text().lower()+(Path(__file__).resolve().parents[1]/'internal/platform/postgres/migrations/000023_complete_change_set.up.sql').read_text().lower()
|
|
for action in ("'add'", "'update'", "'move_scope'", "'delete'", "'no_change'"):
|
|
self.assertIn(action,migration)
|
|
for reason in ("'content_hash_move'", "'content_hash_conflict'", "'path_change'", "'size_change'", "'source_missing'", "'scope_change'", "'move'", "'applying'", "identity_confidence"):
|
|
self.assertIn(reason,(migration+Path(__file__).resolve().parents[1].joinpath('internal/platform/postgres/migrations/000021_source_identity_evidence.up.sql').read_text().lower()))
|
|
self.assertLess(lowered.index('insert into source_catalog_change_sets'),lowered.index('insert into source_catalog_heads'))
|
|
for forbidden in ('delete from','update source_items','drop table'):
|
|
self.assertNotIn(forbidden,lowered)
|
|
|
|
def test_source_names_may_contain_literal_backslashes_without_becoming_host_paths(self):
|
|
inventory={'roots':{'company':{'configured_root_id':'root','live_space_name':'知识库','nodes':{
|
|
'folder':{'dentry_uuid':'folder','parent_dentry_uuid':'root','dentry_type':'folder','name':'协议\\SDK','updated_time':2},
|
|
}}}}
|
|
source=build_dingtalk_sources(inventory)[0]
|
|
self.assertEqual(next(node for node in source['nodes'] if node['parent_key'] is not None)['name'],'协议\\SDK')
|
|
validate_source(source)
|
|
|
|
|
|
if __name__=='__main__': unittest.main()
|