| 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 |
| |
| 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), |
| |
| |
| 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"] |
|
|
| |
| import time |
| time.sleep(0.5) |
|
|
| stored = _memory_cache.get(f"bulk_result:{task_id}") |
| self.assertIsNotNone(stored, "results must be stored for polling") |
| |
| 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 |
|
|
| 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() |
