File size: 4,111 Bytes
1b2d52d 6f1b553 1b2d52d 6f1b553 1b2d52d | 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 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 | """Self-check for app.py upload and api_result asynchronous endpoints (CR-10)."""
import importlib
import io
import os
import sys
import tempfile
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.dirname(HERE))
def test_async_api_flow():
fd, db_path = tempfile.mkstemp(suffix=".db")
os.close(fd)
os.environ["LDV_DB_PATH"] = db_path
os.environ["LDV_SECRET_KEY"] = "test-secret-key-123"
os.environ.pop("LDV_ENCRYPTION_KEY", None)
import database
importlib.reload(database)
database.init_db()
import auth
importlib.reload(auth)
# Seed org and users
oid = database.create_org("TestOrg")
user_token = "user-api-token-123"
database.create_user(oid, "user@testorg.com", auth.hash_password("password"), "user", user_token)
other_oid = database.create_org("OtherOrg")
other_token = "other-api-token-123"
database.create_user(other_oid, "other@otherorg.com", auth.hash_password("password"), "user", other_token)
# Mock worker.submit_job
submitted_jobs = []
import worker
worker.submit_job = lambda public_id, text, lang, explain, policy_name=None: submitted_jobs.append((public_id, text, lang, explain, policy_name))
# Import app to initialize routes
import app
importlib.reload(app)
client = app.app.test_client()
try:
# --- Test Upload (POST /upload) ---
data = {
"file": (io.BytesIO(b"This is a dummy contract document."), "contract.txt")
}
headers = {
"Authorization": f"Bearer {user_token}"
}
resp = client.post("/api/v1/upload", data=data, headers=headers, content_type="multipart/form-data")
assert resp.status_code == 202, resp.status_code
res_json = resp.json
assert "id" in res_json
assert res_json["status"] == "queued"
# Verify job was enqueued
analysis_id = res_json["id"]
assert len(submitted_jobs) == 1
assert submitted_jobs[0][0] == analysis_id
assert "dummy contract document" in submitted_jobs[0][1]
# --- Test Result Polling (GET /api/result/<id>) ---
# 1. Queued state
resp = client.get(f"/api/v1/result/{analysis_id}", headers=headers)
assert resp.status_code == 200, resp.status_code
assert resp.json["status"] == "queued"
assert resp.json["result"] is None
# 2. Processing state
database.update_analysis(analysis_id, status="processing")
resp = client.get(f"/api/v1/result/{analysis_id}", headers=headers)
assert resp.status_code == 200, resp.status_code
assert resp.json["status"] == "processing"
assert resp.json["result"] is None
# 3. Completed state
mock_result = {"layer1": {}, "layer2": {}, "layer3": {"score": 5, "label": "LOW"}}
database.update_analysis(analysis_id, status="completed", risk_score=5, risk_label="LOW", result=mock_result)
resp = client.get(f"/api/v1/result/{analysis_id}", headers=headers)
assert resp.status_code == 200, resp.status_code
assert resp.json["status"] == "completed"
assert resp.json["result"] == mock_result
# Verify raw_text / extracted_text is excluded by default
assert "extracted_text" not in resp.json
assert "raw_text" not in resp.json
# Verify raw_text is included when ?debug=1 is provided
resp_debug = client.get(f"/api/v1/result/{analysis_id}?debug=1", headers=headers)
assert resp_debug.status_code == 200
assert resp_debug.json.get("raw_text") == "This is a dummy contract document."
# --- Test Access Control ---
# Accessing result from another organization must fail with 403
other_headers = {
"Authorization": f"Bearer {other_token}"
}
resp = client.get(f"/api/v1/result/{analysis_id}", headers=other_headers)
assert resp.status_code == 403, resp.status_code
finally:
os.remove(db_path)
if __name__ == "__main__":
test_async_api_flow()
print("test_async_api OK")
|