fix: harden OpenVINO doc triage prototype

This commit is contained in:
William Valentin
2026-06-04 12:10:18 -07:00
parent 6b1cae016c
commit 582e0ee553
4 changed files with 58 additions and 13 deletions
@@ -2,6 +2,7 @@
from __future__ import annotations
import json
import socket
import subprocess
import sys
import tempfile
@@ -42,6 +43,29 @@ def busy() -> int | None:
return None
def choose_free_loopback_port() -> int:
"""Ask the OS for a free localhost port and verify it is not listening yet."""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.bind(("127.0.0.1", 0))
port = int(sock.getsockname()[1])
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
probe.settimeout(0.25)
assert probe.connect_ex(("127.0.0.1", port)) != 0, f"selected port already has a listener: {port}"
return port
def assert_loopback_bind_policy() -> None:
blocked = subprocess.run(
[sys.executable, "server.py", "--host", "0.0.0.0", "--port", "0", "--allowed-root", str(ROOT)],
cwd=ROOT,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
assert blocked.returncode != 0, blocked.stdout + blocked.stderr
assert "loopback" in blocked.stderr.lower(), blocked.stderr
def main() -> int:
run([sys.executable, "make_samples.py"])
invoice = SAMPLES / "synthetic_invoice.png"
@@ -69,20 +93,23 @@ def main() -> int:
assert (emb.get("npu_busy_delta_us") or 0) > 0, emb
assert after > before, {"before": before, "after": after, "embedding": emb}
# HTTP smoke on an ephemeral localhost port so we do not collide with 18820 during tests.
proc = subprocess.Popen([sys.executable, "server.py", "--host", "127.0.0.1", "--port", "18828", "--allowed-root", str(ROOT)], cwd=ROOT, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
# HTTP smoke on a preflighted free localhost port so we do not collide with live/prototype ports.
assert_loopback_bind_policy()
smoke_port = choose_free_loopback_port()
base_url = f"http://127.0.0.1:{smoke_port}"
proc = subprocess.Popen([sys.executable, "server.py", "--host", "127.0.0.1", "--port", str(smoke_port), "--allowed-root", str(ROOT)], cwd=ROOT, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
try:
deadline = time.time() + 5
while time.time() < deadline:
try:
health = urllib.request.urlopen("http://127.0.0.1:18828/healthz", timeout=1).read()
health = urllib.request.urlopen(f"{base_url}/healthz", timeout=1).read()
assert b"openvino-doc-image-triage-npu" in health
break
except Exception:
time.sleep(0.1)
else:
raise AssertionError("server did not become ready")
resp = post_json("http://127.0.0.1:18828/triage", {"path": str(invoice), "options": {"allowed_roots": [str(ROOT)]}})
resp = post_json(f"{base_url}/triage", {"path": str(invoice), "options": {"allowed_roots": [str(ROOT)]}})
assert resp["ok"] is True, resp
assert resp["result"]["source_path_basename"] == "synthetic_invoice.png"
assert "source_path" not in resp["result"]
@@ -92,7 +119,7 @@ def main() -> int:
outside.write(b"sensitive text outside configured artifact root")
outside.flush()
status, blocked = post_json_status(
"http://127.0.0.1:18828/triage",
f"{base_url}/triage",
{"path": outside.name, "options": {"allowed_roots": ["/tmp"], "dry_run": True, "use_embeddings": False}},
)
assert status == 400, blocked
@@ -101,7 +128,7 @@ def main() -> int:
# Request bodies must not redirect extracted text to caller-supplied endpoints.
status, blocked = post_json_status(
"http://127.0.0.1:18828/triage",
f"{base_url}/triage",
{"path": str(invoice), "options": {"embedding_url": "http://198.51.100.1:9/v1/embeddings"}},
)
assert status == 400, blocked