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()