File size: 5,891 Bytes
0f6b827 d92dd26 0f6b827 | 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 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 | import json
import unittest
from unittest.mock import AsyncMock, MagicMock, patch
from fastapi.testclient import TestClient
from app.main import app
client = TestClient(app)
class BulkAsyncTests(unittest.TestCase):
"""The sync bulk fallback must NOT block the HTTP request: it should
return a task_id immediately, process in the background, and let the
frontend poll /validate/bulk/status/{task_id} for progress + results.
The blocking behavior caused the frontend's 45s fetch timeout, which is
why the Results page showed nothing."""
def setUp(self):
self.captured = {}
async def fake_process(emails, webhook_url=None, deep=True):
self.captured["emails"] = list(emails)
self.captured["deep"] = deep
# Realistic shape: exactly what engine.validate_batch returns.
return [
{
"email": e, "status": "risky", "score": 70,
"syntax": {"passed": True},
"scoring": {"score": 70, "risk_level": "medium",
"breakdown": {}, "warnings": []},
"processing_time_ms": 1.0,
"validated_at": "2026-01-01T00:00:00+00:00",
}
for e in emails
]
self.patchers = [
patch("app.api.routes.process_bulk_sync", new=AsyncMock(side_effect=fake_process)),
patch("app.utils.cache.redis_available", new=AsyncMock(return_value=False)),
patch("app.api.routes.settings.enable_db_persistence", False),
# Skip the Redis connection attempt so the in-memory fallback is
# instant — otherwise the background task blocks ~4s per cache call.
patch("app.utils.cache.get_redis", new=AsyncMock(return_value=None)),
]
for p in self.patchers:
p.start()
def tearDown(self):
for p in self.patchers:
p.stop()
def test_bulk_returns_immediately_with_task_id(self):
import time
t0 = time.time()
r = client.post("/api/v1/validate/bulk", json={"emails": ["a@b.test", "c@d.test"]})
elapsed = time.time() - t0
self.assertEqual(r.status_code, 200)
body = r.json()
self.assertIn("task_id", body)
self.assertNotEqual(body["task_id"], "sync", "must return a REAL task id")
self.assertEqual(body["status"], "processing")
self.assertLess(elapsed, 2.0, f"bulk blocked for {elapsed:.1f}s - must return immediately")
def test_background_results_are_stored_for_polling(self):
import asyncio
from app.utils.cache import _memory_cache
_memory_cache.clear()
r = client.post("/api/v1/validate/bulk", json={"emails": ["a@b.test"]})
task_id = r.json()["task_id"]
# Give the background task a moment to complete.
import time
time.sleep(0.5)
stored = _memory_cache.get(f"bulk_result:{task_id}")
self.assertIsNotNone(stored, "results must be stored for polling")
# _memory_set stores (value, expires) tuples — unwrap to the payload.
if isinstance(stored, tuple):
stored = stored[0]
data = json.loads(stored)
self.assertEqual(data["status"], "completed")
self.assertEqual(len(data["results"]), 1)
def test_status_endpoint_returns_stored_results(self):
from app.utils.cache import _memory_cache
_memory_cache.clear()
r = client.post("/api/v1/validate/bulk", json={"emails": ["a@b.test"]})
task_id = r.json()["task_id"]
import time
time.sleep(0.5)
st = client.get(f"/api/v1/validate/bulk/status/{task_id}")
self.assertEqual(st.status_code, 200)
body = st.json()
self.assertEqual(body["status"], "completed")
self.assertEqual(len(body["results"]), 1)
def test_redis_available_but_flag_off_uses_background_path(self):
"""Redis being reachable must NOT switch bulk to Celery on its own.
The deployed Space runs uvicorn only (Dockerfile CMD) — no Celery
worker. If a live Redis alone selected the Celery branch, every bulk
job would sit in the queue forever and the Results page would show
"processing" indefinitely. The explicit flag (default off) is the
opt-in for hosts that actually run a worker.
"""
with (
patch("app.utils.cache.redis_available",
new=AsyncMock(return_value=True)),
patch("app.api.routes.settings.enable_db_persistence", False),
patch("app.api.routes.settings.celery_bulk_enabled", False),
):
r = client.post("/api/v1/validate/bulk",
json={"emails": ["a@b.test"]})
self.assertEqual(r.status_code, 200)
body = r.json()
self.assertNotEqual(body["task_id"], "sync")
self.assertIn("status", body)
def test_flag_on_and_redis_available_uses_celery(self):
from app.tasks.celery_tasks import bulk_validate_task # noqa: F401
fake_task = MagicMock()
fake_task.id = "celery-123"
with (
patch("app.utils.cache.redis_available",
new=AsyncMock(return_value=True)),
patch("app.api.routes.settings.enable_db_persistence", False),
patch("app.api.routes.settings.celery_bulk_enabled", True),
patch("app.api.routes.bulk_validate_task.delay",
return_value=fake_task) as delay_mock,
):
r = client.post("/api/v1/validate/bulk",
json={"emails": ["a@b.test"]})
self.assertEqual(r.status_code, 200)
self.assertEqual(r.json()["task_id"], "celery-123")
delay_mock.assert_called_once()
if __name__ == "__main__":
unittest.main() |