Spaces:
Sleeping
Sleeping
File size: 3,034 Bytes
8970674 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 | """Deployment smoke test against the LIVE HF endpoint (deterministic pipeline).
Waits for the Space to be up, POSTs a real JD + the default résumé to /api/generate
(v1), decodes the returned PDF, and parses it with BOTH pdftotext and pdfplumber.
No LLM dependency — the deterministic pipeline must return a real PDF.
"""
import base64
import os
import re
import sys
import time
import requests
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
BASE = "https://saitejatirunagari-jaa-ats-tool.hf.space"
OUT = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"data", "acceptance", "out")
os.makedirs(OUT, exist_ok=True)
def _wait_up(tries=30, delay=15):
for i in range(tries):
try:
r = requests.get(f"{BASE}/api/health", timeout=20)
if r.status_code == 200:
print(f"[health] up after {i*delay}s: {r.text[:120]}")
return True
except Exception as e:
print(f"[health] try {i}: {str(e)[:60]}")
time.sleep(delay)
return False
def main():
print("Waiting for deployment (HF rebuild)…")
if not _wait_up():
print("DEPLOY SMOKE: endpoint never healthy"); return
jd = re.sub(r"SOURCE_URL:.*", "",
open(os.path.join(os.path.dirname(OUT), "jd1_stripe_payments.txt"),
encoding="utf-8").read(), count=1).strip()
data = {"jd_text": jd, "job_title": "Product Manager, Payments",
"company": "Stripe", "version": "v1"}
print("POST /api/generate …")
try:
r = requests.post(f"{BASE}/api/generate", data=data, timeout=180)
except Exception as e:
print("DEPLOY SMOKE: request failed:", str(e)[:200]); return
print("HTTP", r.status_code)
try:
j = r.json()
except Exception:
print("non-JSON response:", r.text[:300]); return
print("keys:", sorted(list(j.keys()))[:20])
print("status:", j.get("status"), "| source:", j.get("source"),
"| pct:", j.get("external_coverage_pct") or j.get("pct"))
pdfb = j.get("pdf_b64")
if not pdfb:
print("DEPLOY SMOKE: no pdf_b64. error:", j.get("error"), j.get("detail"))
return
path = os.path.join(OUT, "deploy_smoke.pdf")
with open(path, "wb") as f:
f.write(base64.b64decode(pdfb))
print("saved:", path, os.path.getsize(path), "bytes")
# two-parser parse of the DOWNLOADED PDF
from src.pdf_validate import _extract_pdf_text, _pdftotext_extract
a = _extract_pdf_text(path) or ""
b = _pdftotext_extract(path) or ""
print(f"two-parser: pdfplumber={len(a)}ch pdftotext={len(b)}ch")
for kw in ["stakeholder", "roadmap", "cross-functional", "product"]:
print(f" {kw}: pdfplumber={kw in a.lower()} pdftotext={kw in b.lower()}")
print("DEPLOY SMOKE: PASSED (endpoint returned a real, two-parser-readable PDF)"
if a and b else "DEPLOY SMOKE: PDF not parseable by both")
if __name__ == "__main__":
main()
|