Merge branch 'dev' of https://gitea.maskantech.in/gitea_admin/pdf into furqan
This commit is contained in:
Furqan-14
@@ -0,0 +1,207 @@
|
||||
"""Live API validation script for the PDF engine gateway."""
|
||||
import sys, json, urllib.request, urllib.parse
|
||||
sys.path.insert(0, 'gateway')
|
||||
|
||||
BASE = 'http://localhost:8000'
|
||||
|
||||
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 = {}
|
||||
|
||||
# 1. Health check
|
||||
health, status = http_get(f'{BASE}/health')
|
||||
print(f"[HEALTH] status={status} response={health}")
|
||||
results['health'] = {'status': status, 'passed': status == 200}
|
||||
|
||||
# 2. Upload hello_world.pdf
|
||||
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)
|
||||
|
||||
# 3. GET /documents/{id}/fonts
|
||||
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')}")
|
||||
# Type checks
|
||||
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'}")
|
||||
|
||||
# 4. GET /documents/{id}/pages/0/fonts
|
||||
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}
|
||||
|
||||
# Consistency check: doc-level vs page-level
|
||||
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")
|
||||
|
||||
# 5. GET /documents/{id}/pages/0/text
|
||||
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: {repr(text[:80])}")
|
||||
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():
|
||||
if 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]}")
|
||||
|
||||
# 6. Test 404 for non-existent document
|
||||
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}
|
||||
|
||||
# 7. Test page out of bounds
|
||||
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)}
|
||||
|
||||
# 8. Upload vertical_text.pdf and check vertical detection
|
||||
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']}")
|
||||
|
||||
# 9. Upload latin_extended.pdf and validate metrics
|
||||
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'})")
|
||||
|
||||
# 10. Repeated requests (cache consistency)
|
||||
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}
|
||||
|
||||
# 11. Invalid PDF upload
|
||||
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}")
|
||||
Reference in New Issue
Block a user