90 lines
4.4 KiB
Python
90 lines
4.4 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Verify category migration through authenticated, read-only acceptance APIs."""
|
||
|
|
import argparse
|
||
|
|
import datetime
|
||
|
|
import json
|
||
|
|
import urllib.error
|
||
|
|
import urllib.parse
|
||
|
|
import urllib.request
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
from weknora_category_etl import TENANT, save_json
|
||
|
|
|
||
|
|
|
||
|
|
BASE = "http://127.0.0.1:18080"
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
||
|
|
parser.add_argument("--manifest", type=Path, required=True)
|
||
|
|
parser.add_argument("--credentials", type=Path, required=True)
|
||
|
|
parser.add_argument("--out", type=Path, required=True)
|
||
|
|
args = parser.parse_args()
|
||
|
|
manifest = json.loads(args.manifest.read_text())
|
||
|
|
credentials = json.loads(args.credentials.read_text())
|
||
|
|
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
|
||
|
|
|
||
|
|
def request(path, token=None, payload=None):
|
||
|
|
headers = {"Content-Type": "application/json"}
|
||
|
|
if token:
|
||
|
|
headers["Authorization"] = "Bearer " + token
|
||
|
|
req = urllib.request.Request(BASE + path, data=json.dumps(payload).encode() if payload else None, headers=headers)
|
||
|
|
try:
|
||
|
|
with opener.open(req, timeout=20) as response:
|
||
|
|
return response.status, json.load(response)
|
||
|
|
except urllib.error.HTTPError as error:
|
||
|
|
return error.code, None
|
||
|
|
|
||
|
|
status, auth = request("/api/v1/auth/login", payload={k: credentials[k] for k in ("email", "password")})
|
||
|
|
if status != 200 or auth.get("active_tenant", {}).get("id") != TENANT:
|
||
|
|
raise RuntimeError("Verification login failed or outside migration tenant")
|
||
|
|
token = auth["token"]
|
||
|
|
report = {"checked_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), "tenant_id": TENANT, "checks": []}
|
||
|
|
|
||
|
|
def check(name, passed, **counts):
|
||
|
|
report["checks"].append({"check": name, "passed": bool(passed), **counts})
|
||
|
|
|
||
|
|
for kb in manifest["knowledge_bases"]:
|
||
|
|
files = [f for f in manifest["files"] if f["knowledge_base_id"] == kb]
|
||
|
|
tags = [t for t in manifest["tags"] if t["knowledge_base_id"] == kb]
|
||
|
|
status, response = request(f"/api/v1/knowledge-bases/{kb}/knowledge/folders", token)
|
||
|
|
tree = response.get("data", {}) if response else {}
|
||
|
|
nodes = {}
|
||
|
|
|
||
|
|
def walk(items):
|
||
|
|
for node in items:
|
||
|
|
nodes[node["path"]] = node
|
||
|
|
walk(node.get("children") or [])
|
||
|
|
|
||
|
|
walk(tree.get("folders") or [])
|
||
|
|
counts_match = all(t["folder_path"] in nodes and nodes[t["folder_path"]]["total_count"] == sum(
|
||
|
|
f["folder_path"] == t["folder_path"] or f["folder_path"].startswith(t["folder_path"] + "/") for f in files) for t in tags)
|
||
|
|
check("folder_tree", status == 200 and tree.get("total_document_count") == len(files) and counts_match,
|
||
|
|
knowledge_base_id=kb, folders=len(nodes), documents=tree.get("total_document_count"))
|
||
|
|
status, response = request(f"/api/v1/knowledge-bases/{kb}/tags?page=1&page_size=100", token)
|
||
|
|
data = response.get("data", {}) if response else {}
|
||
|
|
items = data.get("data", [])
|
||
|
|
check("tag_list", status == 200 and {t["id"] for t in items} == {t["id"] for t in tags},
|
||
|
|
knowledge_base_id=kb, tags=len(items))
|
||
|
|
for tag in tags:
|
||
|
|
params = urllib.parse.urlencode({"page": 1, "page_size": 100, "tag_ids": tag["id"]})
|
||
|
|
status, response = request(f"/api/v1/knowledge-bases/{kb}/knowledge?{params}", token)
|
||
|
|
data = response.get("data", []) if response else []
|
||
|
|
if isinstance(data, dict):
|
||
|
|
data = data.get("items", [])
|
||
|
|
expected = {r["knowledge_id"] for r in manifest["relations"] if r["tag_id"] == tag["id"]}
|
||
|
|
actual = {f["id"] for f in data}
|
||
|
|
check("tag_document_scope", status == 200 and actual == expected,
|
||
|
|
legacy_category_id=tag["legacy_id"], expected=len(expected), returned=len(actual))
|
||
|
|
status, _ = request(f"/api/v1/knowledge-bases/{kb}/knowledge/folders")
|
||
|
|
check("anonymous_denied", status in (401, 403), knowledge_base_id=kb)
|
||
|
|
report["passed"] = all(c["passed"] for c in report["checks"])
|
||
|
|
save_json(args.out, report)
|
||
|
|
print(json.dumps({"passed": report["passed"], "checks": len(report["checks"]),
|
||
|
|
"failed": [c for c in report["checks"] if not c["passed"]], "out": str(args.out)}))
|
||
|
|
return 0 if report["passed"] else 1
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
raise SystemExit(main())
|