241 lines
8.7 KiB
Python
241 lines
8.7 KiB
Python
"""Live API validation script for the PDF engine gateway."""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
sys.path.insert(0, "gateway")
|
|
|
|
BASE = os.environ.get("GATEWAY_URL", "http://localhost:8765")
|
|
|
|
|
|
def http_get(url):
|
|
try:
|
|
resp = urllib.request.urlopen(url)
|
|
return json.loads(resp.read()), resp.status
|
|
except urllib.error.HTTPError as e:
|
|
return json.loads(e.read()), e.code
|
|
|
|
|
|
def multipart_upload(url, filepath, filename):
|
|
boundary = "Xboundary1234X"
|
|
with open(filepath, "rb") as f:
|
|
file_data = f.read()
|
|
header = (
|
|
"--" + boundary + "\r\n"
|
|
'Content-Disposition: form-data; name="file"; filename="' + filename + '"\r\n'
|
|
"Content-Type: application/pdf\r\n\r\n"
|
|
).encode("utf-8")
|
|
footer = ("\r\n--" + boundary + "--\r\n").encode("utf-8")
|
|
body = header + file_data + footer
|
|
req = urllib.request.Request(
|
|
url,
|
|
data=body,
|
|
headers={"Content-Type": "multipart/form-data; boundary=" + boundary},
|
|
method="POST",
|
|
)
|
|
try:
|
|
resp = urllib.request.urlopen(req)
|
|
return json.loads(resp.read()), resp.status
|
|
except urllib.error.HTTPError as e:
|
|
return json.loads(e.read()), e.code
|
|
|
|
|
|
results = {}
|
|
|
|
health, status = http_get(f"{BASE}/health")
|
|
print(f"[HEALTH] status={status} response={health}")
|
|
results["health"] = {"status": status, "passed": status == 200}
|
|
|
|
print("\n=== UPLOADING hello_world.pdf ===")
|
|
doc, status = multipart_upload(
|
|
f"{BASE}/documents", "corpus/basic/hello_world.pdf", "hello_world.pdf"
|
|
)
|
|
print(f"Upload status={status}")
|
|
print(f"Response: {doc}")
|
|
doc_id = doc.get("id")
|
|
results["upload"] = {"status": status, "passed": status == 201, "doc_id": doc_id}
|
|
|
|
if not doc_id:
|
|
print("FATAL: No doc_id, cannot continue")
|
|
sys.exit(1)
|
|
|
|
print(f"\n=== GET /documents/{doc_id}/fonts ===")
|
|
fonts, status = http_get(f"{BASE}/documents/{doc_id}/fonts")
|
|
print(f"Status={status}, Font count={len(fonts)}")
|
|
results["doc_fonts"] = {"status": status, "count": len(fonts), "passed": status == 200}
|
|
|
|
ALL_FONT_FIELDS = [
|
|
"fontName",
|
|
"type",
|
|
"isEmbedded",
|
|
"isSubset",
|
|
"isVertical",
|
|
"encoding",
|
|
"hasToUnicode",
|
|
"cmapName",
|
|
"cidSystemInfo",
|
|
"subsetTag",
|
|
"sourceType",
|
|
"substitutedFrom",
|
|
"substitutedTo",
|
|
"normalizedFamily",
|
|
"internalFontId",
|
|
"flags",
|
|
"ascent",
|
|
"descent",
|
|
"capHeight",
|
|
]
|
|
|
|
schema_errors = []
|
|
for f in fonts:
|
|
missing = [k for k in ALL_FONT_FIELDS if k not in f]
|
|
if missing:
|
|
schema_errors.append(f"Missing fields: {missing} in font {f.get('fontName')}")
|
|
for boolField in ["isEmbedded", "isSubset", "isVertical", "hasToUnicode"]:
|
|
if not isinstance(f.get(boolField), bool):
|
|
schema_errors.append(f"Field {boolField} should be bool, got {type(f.get(boolField))}")
|
|
for floatField in ["ascent", "descent", "capHeight"]:
|
|
if not isinstance(f.get(floatField), int | float):
|
|
schema_errors.append(f"Field {floatField} should be float")
|
|
if not isinstance(f.get("flags"), int):
|
|
schema_errors.append("Field flags should be int")
|
|
print(
|
|
f" Font: {f.get('fontName'):30s} type={f.get('type'):15s} embedded={f.get('isEmbedded')} subset={f.get('isSubset')} vertical={f.get('isVertical')}"
|
|
)
|
|
print(
|
|
f" encoding={f.get('encoding')} cmapName={f.get('cmapName')} cidSystemInfo={f.get('cidSystemInfo')}"
|
|
)
|
|
print(
|
|
f" subsetTag={f.get('subsetTag')} sourceType={f.get('sourceType')} normalizedFamily={f.get('normalizedFamily')}"
|
|
)
|
|
print(f" internalFontId={f.get('internalFontId')} flags={f.get('flags')}")
|
|
print(f" ascent={f.get('ascent')} descent={f.get('descent')} capHeight={f.get('capHeight')}")
|
|
|
|
results["doc_fonts"]["schema_errors"] = schema_errors
|
|
print(f"Schema errors: {schema_errors or 'None'}")
|
|
|
|
print(f"\n=== GET /documents/{doc_id}/pages/0/fonts ===")
|
|
pg_fonts, status = http_get(f"{BASE}/documents/{doc_id}/pages/0/fonts")
|
|
print(f"Status={status}, Page font count={len(pg_fonts)}")
|
|
results["page_fonts"] = {"status": status, "count": len(pg_fonts), "passed": status == 200}
|
|
|
|
if len(fonts) != len(pg_fonts):
|
|
print(f"WARNING: doc-level fonts ({len(fonts)}) != page-level fonts ({len(pg_fonts)})")
|
|
results["page_fonts"]["consistency_warning"] = True
|
|
else:
|
|
print("Consistency: doc-level and page-level font counts match OK")
|
|
|
|
print(f"\n=== GET /documents/{doc_id}/pages/0/text ===")
|
|
text_data, status = http_get(f"{BASE}/documents/{doc_id}/pages/0/text")
|
|
print(f"Status={status}")
|
|
results["page_text"] = {"status": status, "passed": status == 200}
|
|
|
|
text = text_data.get("text", "")
|
|
glyphs = text_data.get("glyphs", [])
|
|
print(f"Text: {text[:80]!r}")
|
|
print(f"Glyph count: {len(glyphs)}")
|
|
|
|
glyph_errors = []
|
|
for i, g in enumerate(glyphs):
|
|
for field in ["text", "x", "y", "w", "h", "fontSize"]:
|
|
if field not in g:
|
|
glyph_errors.append(f"Glyph {i} missing field {field}")
|
|
if g.get("fontSize", 0) <= 0:
|
|
glyph_errors.append(f"Glyph {i} '{g.get('text')}' has fontSize <= 0: {g.get('fontSize')}")
|
|
if g.get("text", "").strip() and (g.get("w", 0) <= 0 or g.get("h", 0) <= 0):
|
|
glyph_errors.append(
|
|
f"Glyph {i} '{g.get('text')}' has zero bounds w={g.get('w')} h={g.get('h')}"
|
|
)
|
|
if g.get("fontSize", 0) > 200:
|
|
glyph_errors.append(f"Glyph {i} unrealistic fontSize={g.get('fontSize')}")
|
|
|
|
results["page_text"]["glyph_count"] = len(glyphs)
|
|
results["page_text"]["glyph_errors"] = glyph_errors
|
|
print(f"Glyph validation errors: {glyph_errors or 'None'}")
|
|
if glyphs:
|
|
print(f"Sample glyphs: {glyphs[:3]}")
|
|
|
|
print("\n=== TEST 404 ===")
|
|
not_found, status = http_get(f"{BASE}/documents/nonexistent-uuid/fonts")
|
|
print(f"404 test status={status} (expected 404)")
|
|
results["not_found"] = {"status": status, "passed": status == 404}
|
|
|
|
print("\n=== TEST PAGE OUT OF BOUNDS ===")
|
|
oob, status = http_get(f"{BASE}/documents/{doc_id}/pages/99/fonts")
|
|
print(f"OOB test status={status} (expected 4xx)")
|
|
results["oob"] = {"status": status, "passed": status in (400, 404)}
|
|
|
|
print("\n=== UPLOADING vertical_text.pdf ===")
|
|
vert_doc, status = multipart_upload(
|
|
f"{BASE}/documents", "corpus/fonts/vertical_text.pdf", "vertical_text.pdf"
|
|
)
|
|
vert_id = vert_doc.get("id")
|
|
print(f"Upload status={status} doc_id={vert_id}")
|
|
if vert_id:
|
|
vert_fonts, status = http_get(f"{BASE}/documents/{vert_id}/fonts")
|
|
print(f"Vertical PDF fonts ({len(vert_fonts)}):")
|
|
for vf in vert_fonts:
|
|
print(
|
|
f" fontName={vf['fontName']} isVertical={vf['isVertical']} encoding={vf['encoding']}"
|
|
)
|
|
|
|
print("\n=== UPLOADING latin_extended.pdf ===")
|
|
lat_doc, status = multipart_upload(
|
|
f"{BASE}/documents", "corpus/fonts/latin_extended.pdf", "latin_extended.pdf"
|
|
)
|
|
lat_id = lat_doc.get("id")
|
|
if lat_id:
|
|
lat_fonts, st = http_get(f"{BASE}/documents/{lat_id}/fonts")
|
|
print(f"Latin extended fonts ({len(lat_fonts)}):")
|
|
for lf in lat_fonts:
|
|
ascent_ok = lf["ascent"] > 0
|
|
descent_ok = lf["descent"] < 0
|
|
cap_ok = lf["capHeight"] > 0
|
|
print(
|
|
f" {lf['fontName']} ascent={lf['ascent']}({'OK' if ascent_ok else 'FAIL'}) descent={lf['descent']}({'OK' if descent_ok else 'FAIL'}) capHeight={lf['capHeight']}({'OK' if cap_ok else 'FAIL'})"
|
|
)
|
|
|
|
print("\n=== CACHE CONSISTENCY (3 repeated font requests) ===")
|
|
responses = []
|
|
for _ in range(3):
|
|
fonts_rep, st = http_get(f"{BASE}/documents/{doc_id}/fonts")
|
|
responses.append(len(fonts_rep))
|
|
print(f"Font counts on repeated requests: {responses}")
|
|
results["cache_consistency"] = {"counts": responses, "passed": len(set(responses)) == 1}
|
|
|
|
print("\n=== INVALID PDF UPLOAD ===")
|
|
boundary = "Xboundary1234X"
|
|
garbage = b"NOT A PDF FILE AT ALL 12345"
|
|
header = (
|
|
"--"
|
|
+ boundary
|
|
+ '\r\nContent-Disposition: form-data; name="file"; filename="bad.pdf"\r\nContent-Type: application/pdf\r\n\r\n'
|
|
).encode()
|
|
footer = ("\r\n--" + boundary + "--\r\n").encode()
|
|
req = urllib.request.Request(
|
|
f"{BASE}/documents",
|
|
data=header + garbage + footer,
|
|
headers={"Content-Type": "multipart/form-data; boundary=" + boundary},
|
|
method="POST",
|
|
)
|
|
try:
|
|
resp = urllib.request.urlopen(req)
|
|
bad_result = json.loads(resp.read()), resp.status
|
|
print(f"UNEXPECTED SUCCESS: {bad_result}")
|
|
results["invalid_upload"] = {"passed": False}
|
|
except urllib.error.HTTPError as e:
|
|
err_detail = json.loads(e.read())
|
|
print(f"Invalid PDF correctly rejected: status={e.code} detail={err_detail}")
|
|
results["invalid_upload"] = {"status": e.code, "passed": e.code == 400}
|
|
|
|
print("\n" + "=" * 60)
|
|
print("VALIDATION SUMMARY")
|
|
print("=" * 60)
|
|
for test, res in results.items():
|
|
passed = res.get("passed", "?")
|
|
status = res.get("status", "-")
|
|
print(f" {'PASS' if passed else 'FAIL'}: {test:35s} status={status}")
|