Spaces:
Sleeping
Sleeping
File size: 4,186 Bytes
db4ba8d | 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 | import os
import time
from typing import Any
import pytest
from fastapi.testclient import TestClient
from src.main import app
@pytest.mark.integration
def test_full_e2e_upload_process_review_submit(monkeypatch):
"""Full end-to-end flow: upload -> process -> review -> submit.
This test is intentionally guarded and only runs when the environment
variable `RUN_FULL_E2E` is set to `1`. It also supports shortcuts for
local development:
- Set `SKIP_AUTH=1` to bypass Keycloak by monkeypatching `get_current_user`.
- Set `SKIP_WORKER=1` to invoke processing tasks inline after upload.
Requirements when enabled:
- Keycloak, Supabase (db + storage), Redis, MinIO, ChromaDB all running
- A valid bearer token available via `E2E_BEARER_TOKEN` OR use `SKIP_AUTH`
"""
if os.getenv("RUN_FULL_E2E") != "1":
pytest.skip("Set RUN_FULL_E2E=1 to run full end-to-end tests")
client = TestClient(app)
# Optionally bypass auth for local runs
if os.getenv("SKIP_AUTH") == "1":
from src.dependencies import CurrentUser
def _fake_user(*a: Any, **k: Any) -> CurrentUser:
return CurrentUser(
sub="e2e-user",
email="e2e@example.com",
full_name="E2E User",
roles=["operator"],
tier="enterprise",
company_id="e2e-company",
raw_token="e2e",
)
monkeypatch.setattr("src.dependencies.get_current_user", lambda *a, **k: _fake_user())
# Prepare upload payload (single minimal PDF)
files = [
("files", ("bl.pdf", b"%PDF-1.4\n%EOF\n", "application/pdf")),
]
data = {"doc_types": "bill_of_lading"}
auth_header = {}
token = os.getenv("E2E_BEARER_TOKEN")
if token:
auth_header = {"Authorization": f"Bearer {token}"}
else:
# If a .env.e2e file exists (produced by provisioning script), load it
envfile = os.path.join(os.path.dirname(__file__), "../../..", ".env.e2e")
try:
envfile = os.path.abspath(envfile)
if os.path.exists(envfile):
with open(envfile, encoding="utf-8") as f:
for line in f:
if line.strip().startswith("E2E_BEARER_TOKEN="):
token = line.strip().split("=", 1)[1]
auth_header = {"Authorization": f"Bearer {token}"}
break
except Exception:
pass
resp = client.post("/api/v1/batches", files=files, data=data, headers=auth_header)
assert resp.status_code in (200, 201), f"Upload failed: {resp.text}"
batch_id = resp.json().get("batch_id")
assert batch_id, "No batch_id returned"
# Optionally trigger processing inline when no worker is available
if os.getenv("SKIP_WORKER") == "1":
try:
from src.tasks.ocr_tasks import preprocess_document
try:
# Task is often defined with bind=True -> first arg is `self`
preprocess_document(None, batch_id)
except TypeError:
# Some wrappers accept only (batch_id,)
preprocess_document(batch_id)
except Exception as exc: # pragma: no cover - best-effort
pytest.skip(f"Could not invoke preprocess task inline: {exc}")
# Poll batch status until processing/review complete or timeout
timeout = int(os.getenv("E2E_TIMEOUT_SECONDS", "60"))
deadline = time.time() + timeout
status = None
while time.time() < deadline:
r = client.get(f"/api/v1/batches/{batch_id}", headers=auth_header)
if r.status_code == 200:
data = r.json().get("batch") or {}
status = data.get("status")
if status in ("review_complete", "submitted", "rejected"):
break
time.sleep(2)
assert status is not None, "Timed out waiting for batch to become available"
assert status in ("review_complete", "submitted"), f"Unexpected final status: {status}"
|