Files
zheromiku dcdc8d1ddd feat: 知识治理平台(含已验证发布/回滚契约)
全栈知识治理平台首次入库(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 干净
2026-09-30 10:54:06 +08:00

63 lines
6.3 KiB
Python

"""Read-only browser acceptance for the real multi-level source catalog."""
import json
from pathlib import Path
import urllib.request
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.firefox.options import Options
from selenium.webdriver.firefox.service import Service
from selenium.webdriver.support.ui import WebDriverWait
BASE='http://127.0.0.1:8088';ROOT=Path(__file__).resolve().parents[1]
def get(path):
with urllib.request.urlopen(BASE+path,timeout=30) as response:return json.load(response)
def source(name):return next(item for item in get('/api/v1/source-catalog/sources')['data'] if item['name']==name)
def page(item,parent):return get(f"/api/v1/source-catalog/nodes?connection_id={item['connection_id']}&parent_id={parent}&limit=100&offset=0")
sources=get('/api/v1/source-catalog/sources')['data'];assert len(sources)==7
options=Options();options.add_argument('--headless');options.binary_location='/snap/firefox/current/usr/lib/firefox/firefox'
driver=webdriver.Firefox(options=options,service=Service('/snap/bin/geckodriver'));driver.set_window_size(1440,1000);wait=WebDriverWait(driver,30,poll_frequency=.1)
def body():return driver.find_element(By.TAG_NAME,'body').text
def fits():assert driver.execute_script('return document.documentElement.scrollWidth<=window.innerWidth'),'page horizontal overflow'
def source_button(name):return driver.find_element(By.XPATH,f"//aside[@aria-label='内容来源']//button[.//b[.='{name}']]")
def folder_button(fragment):return driver.find_element(By.XPATH,f"//table[contains(@class,'catalog-table')]//button[contains(.,'{fragment}')]")
def row_names():return [row.find_element(By.CSS_SELECTOR,'td:first-child').text.replace('📁','').replace('📄','').strip() for row in driver.find_elements(By.CSS_SELECTOR,'.catalog-table tbody tr')]
try:
driver.get(BASE);wait.until(lambda d:d.find_element(By.TAG_NAME,'h1').text=='来源与目录');wait.until(lambda d:len(d.find_elements(By.CSS_SELECTOR,'.catalog-source'))==7)
company=source('公司级技术支持知识库');company_root=page(company,company['root_node_id']);company_folders=[node['name'] for node in company_root['items'] if node['node_type']=='folder']
wait.until(lambda d:len(d.find_elements(By.CSS_SELECTOR,'.catalog-table tbody tr'))==len(company_folders))
assert row_names()==company_folders and len(company_folders)==8
for expected in ['01 产品总览与定位','02 产品资料','03 机器人部署与使用','04 异常处理与售后支持','05 二次开发','06 应用案例与解决方案','07 培训与学习','99 隔离区','另有 5 份根目录文件']:
assert expected in body()
assert all(row.find_element(By.CSS_SELECTOR,'td:nth-child(2)').text=='文件夹' for row in driver.find_elements(By.CSS_SELECTOR,'.catalog-table tbody tr'))
# Enter one level and compare the browser exactly with the API's direct children.
target=next(node for node in company_root['items'] if node['node_type']=='folder' and node['name'].startswith('02 产品资料'))
folder_button('02 产品资料').click();target_page=page(company,target['node_id']);wait.until(lambda d:len(d.find_elements(By.CSS_SELECTOR,'.workspace-context-tabs button'))==2)
wait.until(lambda d:len(d.find_elements(By.CSS_SELECTOR,'.catalog-table tbody tr'))==len(target_page['items']))
assert row_names()==[node['name'] for node in target_page['items']]
assert '02 产品资料' in driver.find_element(By.CSS_SELECTOR,'.workspace-context-tabs').text
nested=next((node for node in target_page['items'] if node['node_type']=='folder'),None)
if nested:
folder_button(nested['name'][:12]).click();wait.until(lambda d:len(d.find_elements(By.CSS_SELECTOR,'.workspace-context-tabs button'))==3)
# Top context returns to the source root.
driver.find_element(By.XPATH,"//div[@aria-label='当前来源目录和文件']//button[.='公司级技术支持知识库']").click();wait.until(lambda d:len(d.find_elements(By.CSS_SELECTOR,'.catalog-table tbody tr'))==8)
# PP root: only the eight approved real tag types, 16 untagged files collapsed.
pp=source('PP资料下载中心');source_button('PP资料下载中心').click();wait.until(lambda d:d.find_element(By.CSS_SELECTOR,'.catalog-summary strong').text=='PP资料下载中心')
expected_pp=['工业产品','教育产品','具身产品','多足产品','解决方案','公司','软件','生态&配件'];wait.until(lambda d:len(d.find_elements(By.CSS_SELECTOR,'.catalog-table tbody tr'))==8)
assert row_names()==expected_pp and '另有 16 份根目录文件' in body()
folder_button('工业产品').click();wait.until(lambda d:len(d.find_elements(By.CSS_SELECTOR,'.workspace-context-tabs button'))==2)
assert 'CRA系列' in body()
folder_button('CRA系列').click();wait.until(lambda d:len(d.find_elements(By.CSS_SELECTOR,'.workspace-context-tabs button'))==3)
assert driver.find_elements(By.CSS_SELECTOR,'.catalog-file')
# AList domestic-industrial root is exactly six first-level folders.
alist=source('国内工业');source_button('国内工业').click();wait.until(lambda d:d.find_element(By.CSS_SELECTOR,'.catalog-summary strong').text=='国内工业')
expected_alist=['01 公司资料','02 产品资料','03 行业洞察','04 销售赋能','05 流程制度','06 通用模板'];wait.until(lambda d:len(d.find_elements(By.CSS_SELECTOR,'.catalog-table tbody tr'))==6)
assert row_names()==expected_alist
folder_button('02 产品资料').click();wait.until(lambda d:len(d.find_elements(By.CSS_SELECTOR,'.workspace-context-tabs button'))==2)
assert '国内工业' in driver.find_element(By.CSS_SELECTOR,'.workspace-context-tabs').text and '02 产品资料' in driver.find_element(By.CSS_SELECTOR,'.workspace-context-tabs').text
for width in (1440,1024,768):driver.set_window_size(width,900);fits()
driver.set_window_size(1440,1000);driver.get_full_page_screenshot_as_file(str(ROOT/'tmp/SOURCE_CATALOG_FIREFOX.png'))
result={'sources':7,'nodes':38986,'company_first_level':8,'company_root_files_collapsed':5,'pp_first_level':8,'pp_root_files_collapsed':16,'alist_domestic_first_level':6,'direct_children_match_api':True,'breadcrumb_depth_tested':3,'context_navigation':'pass','root_overflow':False,'write_requests':0}
(ROOT/'tmp/source-catalog-acceptance.json').write_text(json.dumps(result,ensure_ascii=False,indent=2));print(json.dumps(result,ensure_ascii=False))
finally:driver.quit()