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 干净
40 lines
2.1 KiB
Python
40 lines
2.1 KiB
Python
import io,json,unittest,contextlib
|
|
import live_catalog_refresh as refresh
|
|
|
|
|
|
class S4ProgressPayloadTests(unittest.TestCase):
|
|
def collect(self, fn):
|
|
buffer=io.StringIO()
|
|
with contextlib.redirect_stdout(buffer):
|
|
fn()
|
|
return [json.loads(line) for line in buffer.getvalue().splitlines() if line.strip()]
|
|
|
|
def test_progress_lines_are_strict_fields_and_carry_scanned_counts(self):
|
|
lines=self.collect(lambda: refresh.emit('scanning',40,'正在扫描目录：已完成 12 个文件夹，待处理 3 个',12,15,current_directory='/产品资料/CR系列',scanned_directories=12,discovered_directories=15,scanned_files=240,estimated_total_files=520,total_is_estimate=True))
|
|
payload=lines[0]
|
|
self.assertEqual(set(payload),{'stage','progress','message','processed','total','current_directory','scanned_directories','discovered_directories','scanned_files','estimated_total_files','total_is_estimate'})
|
|
self.assertEqual(payload['scanned_files'],240)
|
|
self.assertTrue(payload['total_is_estimate'])
|
|
|
|
def test_sanitizer_strips_control_chars_and_refuses_secrets_urls(self):
|
|
self.assertEqual(refresh.sanitize_directory('/产品\x07资料'),'/产品资料')
|
|
self.assertEqual(refresh.sanitize_directory('https://internal.example/x'),'')
|
|
self.assertEqual(refresh.sanitize_directory('/run/source-secrets/pp'),'')
|
|
self.assertEqual(refresh.sanitize_directory(''),'')
|
|
self.assertEqual(len(refresh.sanitize_directory('目'*2000)),1024)
|
|
|
|
def test_source_name_only_appears_when_set_and_never_paths_with_credentials(self):
|
|
refresh._source_name=''
|
|
plain=self.collect(lambda: refresh.emit('building',80,'正在校验目录结构',5,5))
|
|
self.assertNotIn('source_name',plain[0])
|
|
refresh._source_name='国内工业'
|
|
try:
|
|
named=self.collect(lambda: refresh.emit('building',80,'正在校验目录结构',5,5))
|
|
self.assertEqual(named[0]['source_name'],'国内工业')
|
|
finally:
|
|
refresh._source_name=''
|
|
|
|
|
|
if __name__=='__main__':
|
|
unittest.main()
|