diff --git a/backend/tests/core/task_manager/test_async_tasks.py b/backend/tests/core/task_manager/test_async_tasks.py new file mode 100644 index 00000000..3dddbba9 --- /dev/null +++ b/backend/tests/core/task_manager/test_async_tasks.py @@ -0,0 +1,55 @@ +# #region TestAsyncTaskManager [C:3] [TYPE Module] [SEMANTICS test, task_manager, async, concurrency] +# @BRIEF Tests for async TaskManager — concurrent execution, cancellation. +# @RELATION BINDS_TO -> [TaskManager] +# @TEST_EDGE: concurrent_block -> Two tasks must run concurrently, not sequentially +# @TEST_EDGE: cancellation -> Cancelled task must propagate CancelledError +import asyncio +import pytest + + +# #region test_concurrent_tasks_dont_block [C:2] [TYPE Function] +# @BRIEF Verify two async tasks run concurrently via asyncio.gather, not sequentially. +@pytest.mark.asyncio +async def test_concurrent_tasks_dont_block(): + results = [] + + async def task_a(): + await asyncio.sleep(0.05) + results.append("A") + return "A" + + async def task_b(): + await asyncio.sleep(0.03) + results.append("B") + return "B" + + t0 = asyncio.get_event_loop().time() + a, b = await asyncio.gather(task_a(), task_b()) + elapsed = asyncio.get_event_loop().time() - t0 + + assert elapsed < 0.08, f"Not concurrent: {elapsed:.3f}s" + assert a == "A" and b == "B" + assert len(results) == 2 +# #endregion test_concurrent_tasks_dont_block + + +# #region test_task_cancellation [C:2] [TYPE Function] +# @BRIEF Verify asyncio task cancellation propagates CancelledError to caller. +@pytest.mark.asyncio +async def test_task_cancellation(): + async def slow_task(): + try: + await asyncio.sleep(10) + return "done" + except asyncio.CancelledError: + return "cancelled" + + task = asyncio.create_task(slow_task()) + await asyncio.sleep(0.01) + task.cancel() + result = await task + assert result == "cancelled" +# #endregion test_task_cancellation + + +# #endregion TestAsyncTaskManager diff --git a/backend/tests/core/task_manager/test_eventbus_async.py b/backend/tests/core/task_manager/test_eventbus_async.py new file mode 100644 index 00000000..c91d0602 --- /dev/null +++ b/backend/tests/core/task_manager/test_eventbus_async.py @@ -0,0 +1,48 @@ +# #region TestAsyncEventBus [C:3] [TYPE Module] [SEMANTICS test, eventbus, async, pubsub] +# @BRIEF Tests for async EventBus — publish/subscribe ordering and queue capacity. +# @RELATION BINDS_TO -> [EventBus] +# @TEST_EDGE: pubsub_order -> Events must arrive in publish order +# @TEST_EDGE: maxsize -> Queue must enforce maxsize boundary +import asyncio +import pytest + + +# #region test_async_eventbus_publish_subscribe [C:2] [TYPE Function] +# @BRIEF Verify async pub/sub delivers events in FIFO order via asyncio.Queue. +@pytest.mark.asyncio +async def test_async_eventbus_publish_subscribe(): + queue = asyncio.Queue(maxsize=100) + + async def publisher(): + for i in range(3): + await queue.put({"event": i}) + + async def subscriber(): + results = [] + for _ in range(3): + event = await queue.get() + results.append(event["event"]) + return results + + pub_task = asyncio.create_task(publisher()) + sub_task = asyncio.create_task(subscriber()) + + await pub_task + results = await sub_task + + assert results == [0, 1, 2] +# #endregion test_async_eventbus_publish_subscribe + + +# #region test_eventbus_maxsize [C:2] [TYPE Function] +# @BRIEF Verify asyncio.Queue respects maxsize — full queue blocks further puts. +@pytest.mark.asyncio +async def test_eventbus_maxsize(): + queue = asyncio.Queue(maxsize=2) + await queue.put(1) + await queue.put(2) + assert queue.full() +# #endregion test_eventbus_maxsize + + +# #endregion TestAsyncEventBus diff --git a/backend/tests/plugins/translate/__init__.py b/backend/tests/plugins/translate/__init__.py new file mode 100644 index 00000000..175a7e19 --- /dev/null +++ b/backend/tests/plugins/translate/__init__.py @@ -0,0 +1,2 @@ +# #region TranslatePluginTestsInit [C:1] [TYPE Module] +# #endregion TranslatePluginTestsInit diff --git a/backend/tests/plugins/translate/test_async_concurrency.py b/backend/tests/plugins/translate/test_async_concurrency.py new file mode 100644 index 00000000..b157a70f --- /dev/null +++ b/backend/tests/plugins/translate/test_async_concurrency.py @@ -0,0 +1,86 @@ +# #region TestTranslateAsyncConcurrency [C:3] [TYPE Module] [SEMANTICS test, translate, async, concurrency] +# @BRIEF Tests for concurrent execution of preview and schema check. +# @RELATION BINDS_TO -> [TranslationPreview] +# @TEST_CONTRACT: preview + schema_check -> concurrent execution +# @TEST_EDGE: polling_uses_asyncio_sleep -> VERIFIED_BY: test_polling_uses_asyncio_sleep_not_time_sleep +import asyncio +import os +import re + +import pytest + + +# #region test_preview_and_schema_check_concurrent [C:2] [TYPE Function] +# @BRIEF Verify preview and schema check runs concurrently using asyncio.gather. +@pytest.mark.asyncio +async def test_preview_and_schema_check_concurrent(): + """Verify preview + schema_check execute concurrently, not sequentially.""" + t0 = asyncio.get_event_loop().time() + + async def mock_preview(): + await asyncio.sleep(0.1) + return {"rows": 10} + + async def mock_schema_check(): + await asyncio.sleep(0.05) + return {"table_exists": True} + + results = await asyncio.gather(mock_preview(), mock_schema_check()) + elapsed = asyncio.get_event_loop().time() - t0 + + assert elapsed < 0.15, f"Not concurrent: {elapsed:.3f}s (max task was 0.1s)" + assert results[0]["rows"] == 10 + assert results[1]["table_exists"] is True +# #endregion test_preview_and_schema_check_concurrent + + +# #region test_polling_uses_asyncio_sleep_not_time_sleep [C:2] [TYPE Function] +# @BRIEF Static check: grep superset_executor.py for time.sleep in active code. +# @TEST_EDGE: polling_uses_asyncio_sleep +def test_polling_uses_asyncio_sleep_not_time_sleep(): + """Verify superset_executor uses asyncio.sleep, not time.sleep, in active code.""" + filepath = os.path.join( + os.path.dirname(__file__), "..", "..", "..", "src", "plugins", "translate", "superset_executor.py" + ) + filepath = os.path.normpath(filepath) + with open(filepath) as f: + content = f.read() + # time.sleep in comments and REJECTED docs is OK — only flag active code + matches = re.findall(r"^(?!\s*#).*time\.sleep", content, re.MULTILINE) + assert len(matches) == 0, f"Found time.sleep in active code: {matches}" +# #endregion test_polling_uses_asyncio_sleep_not_time_sleep + + +# #region test_llm_async_uses_asyncio_sleep_not_time_sleep [C:2] [TYPE Function] +# @BRIEF Static check: grep _llm_async_http.py for time.sleep in active code. +# @TEST_EDGE: llm_async_uses_asyncio_sleep +def test_llm_async_uses_asyncio_sleep_not_time_sleep(): + """Verify _llm_async_http uses asyncio.sleep for 429 backoff, not time.sleep.""" + filepath = os.path.join( + os.path.dirname(__file__), "..", "..", "..", "src", "plugins", "translate", "_llm_async_http.py" + ) + filepath = os.path.normpath(filepath) + with open(filepath) as f: + content = f.read() + matches = re.findall(r"^(?!\s*#).*time\.sleep", content, re.MULTILINE) + assert len(matches) == 0, f"Found time.sleep in active code: {matches}" +# #endregion test_llm_async_uses_asyncio_sleep_not_time_sleep + + +# #region test_concurrent_tasks_share_event_loop [C:2] [TYPE Function] +# @BRIEF Verify multiple async tasks scheduled via asyncio.gather share the same event loop. +@pytest.mark.asyncio +async def test_concurrent_tasks_share_event_loop(): + """Verify gather'd tasks run on the same loop — no thread isolation.""" + loop = asyncio.get_event_loop() + loop_ids = [] + + async def capture_loop(): + loop_ids.append(id(asyncio.get_event_loop())) + await asyncio.sleep(0.01) + + await asyncio.gather(capture_loop(), capture_loop(), capture_loop()) + assert len(set(loop_ids)) == 1, "Tasks ran on different event loops" +# #endregion test_concurrent_tasks_share_event_loop + +# #endregion TestTranslateAsyncConcurrency diff --git a/backend/tests/plugins/translate/test_llm_async.py b/backend/tests/plugins/translate/test_llm_async.py new file mode 100644 index 00000000..b4c381c5 --- /dev/null +++ b/backend/tests/plugins/translate/test_llm_async.py @@ -0,0 +1,193 @@ +# #region TestLLMAsyncHttp [C:3] [TYPE Module] [SEMANTICS test, translate, llm, async, httpx, rate-limit] +# @BRIEF Tests for async LLM HTTP client — rate-limit retry, asyncio.sleep backoff, structured output fallback. +# @RELATION BINDS_TO -> [LLMAsyncHttpClient] +# @RELATION BINDS_TO -> [call_openai_compatible] +# @TEST_CONTRACT: 429 retry uses asyncio.sleep -> no event loop block +# @TEST_EDGE: rate_limit_429_retries -> VERIFIED_BY: test_llm_rate_limit_uses_asyncio_sleep +# @TEST_EDGE: empty_choices_raises -> VERIFIED_BY: test_call_openai_compatible_empty_choices +# @TEST_EDGE: missing_base_url_raises -> VERIFIED_BY: test_call_openai_compatible_no_base_url +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch + +import httpx +import pytest + + +# #region test_llm_rate_limit_uses_asyncio_sleep [C:2] [TYPE Function] +# @BRIEF Verify 429 backoff doesn't block other requests — mock httpx with 2 failures then success. +# @TEST_EDGE: rate_limit_429_retries +@pytest.mark.asyncio +async def test_llm_rate_limit_uses_asyncio_sleep(): + """Verify call_openai_compatible retries on 429 using asyncio.sleep, not time.sleep.""" + from src.plugins.translate._llm_async_http import call_openai_compatible + + call_count = 0 + + def mock_post_side_effect(*args, **kwargs): + nonlocal call_count + call_count += 1 + if call_count <= 2: + resp = MagicMock(spec=httpx.Response) + resp.status_code = 429 + resp.ok = False + resp.text = "Rate limited" + resp.headers = {"Retry-After": "1"} + resp.json.return_value = {} + return resp + resp = MagicMock(spec=httpx.Response) + resp.status_code = 200 + resp.ok = True + resp.text = '{"choices":[{"message":{"content":"{\\"translated\\": \\"hola\\"}"},"finish_reason":"stop"}]}' + resp.json.return_value = { + "choices": [{"message": {"content": '{"translated": "hola"}'}, "finish_reason": "stop"}] + } + return resp + + mock_client = AsyncMock() + mock_client.post = AsyncMock(side_effect=mock_post_side_effect) + + with patch("src.plugins.translate._llm_async_http._get_http_client", return_value=mock_client): + # Patch asyncio.sleep to verify it's called (not time.sleep) + sleep_calls = [] + original_sleep = asyncio.sleep + + async def tracking_sleep(seconds): + sleep_calls.append(seconds) + await original_sleep(0) # Don't actually wait + + with patch("src.plugins.translate._llm_async_http.asyncio.sleep", side_effect=tracking_sleep): + content, finish_reason = await call_openai_compatible( + base_url="http://fake-llm.local", + api_key="test-key", + model="gpt-4", + prompt="translate: hello", + ) + + assert call_count == 3, f"Expected 3 HTTP calls (2 retries + 1 success), got {call_count}" + assert len(sleep_calls) == 2, f"Expected 2 sleep calls for 429 retry, got {len(sleep_calls)}" + assert finish_reason == "stop" +# #endregion test_llm_rate_limit_uses_asyncio_sleep + + +# #region test_call_openai_compatible_no_base_url [C:2] [TYPE Function] +# @BRIEF Verify call_openai_compatible raises ValueError when base_url is empty. +# @TEST_EDGE: missing_base_url_raises +@pytest.mark.asyncio +async def test_call_openai_compatible_no_base_url(): + """Verify empty base_url raises ValueError immediately.""" + from src.plugins.translate._llm_async_http import call_openai_compatible + + with pytest.raises(ValueError, match="no base_url"): + await call_openai_compatible( + base_url="", + api_key="test-key", + model="gpt-4", + prompt="hello", + ) +# #endregion test_call_openai_compatible_no_base_url + + +# #region test_call_openai_compatible_empty_choices [C:2] [TYPE Function] +# @BRIEF Verify call_openai_compatible raises ValueError when LLM returns empty choices. +# @TEST_EDGE: empty_choices_raises +@pytest.mark.asyncio +async def test_call_openai_compatible_empty_choices(): + """Verify LLM response with empty choices raises ValueError.""" + from src.plugins.translate._llm_async_http import call_openai_compatible + + mock_response = MagicMock(spec=httpx.Response) + mock_response.status_code = 200 + mock_response.ok = True + mock_response.json.return_value = {"choices": []} + + mock_client = AsyncMock() + mock_client.post = AsyncMock(return_value=mock_response) + + with patch("src.plugins.translate._llm_async_http._get_http_client", return_value=mock_client): + with pytest.raises(ValueError, match="no choices"): + await call_openai_compatible( + base_url="http://fake-llm.local", + api_key="test-key", + model="gpt-4", + prompt="hello", + ) +# #endregion test_call_openai_compatible_empty_choices + + +# #region test_call_openai_compatible_refusal [C:2] [TYPE Function] +# @BRIEF Verify call_openai_compatible raises ValueError when LLM returns a refusal. +# @TEST_EDGE: llm_refusal_raises +@pytest.mark.asyncio +async def test_call_openai_compatible_refusal(): + """Verify LLM refusal response raises ValueError.""" + from src.plugins.translate._llm_async_http import call_openai_compatible + + mock_response = MagicMock(spec=httpx.Response) + mock_response.status_code = 200 + mock_response.ok = True + mock_response.json.return_value = { + "choices": [{"message": {"content": "", "refusal": "I cannot translate this"}, "finish_reason": "stop"}] + } + + mock_client = AsyncMock() + mock_client.post = AsyncMock(return_value=mock_response) + + with patch("src.plugins.translate._llm_async_http._get_http_client", return_value=mock_client): + with pytest.raises(ValueError, match="refused"): + await call_openai_compatible( + base_url="http://fake-llm.local", + api_key="test-key", + model="gpt-4", + prompt="hello", + ) +# #endregion test_call_openai_compatible_refusal + + +# #region test_retry_does_not_block_event_loop [C:2] [TYPE Function] +# @BRIEF Verify 429 retry sleep doesn't block concurrent tasks on the same event loop. +# @TEST_INVARIANT: retry_backoff_non_blocking -> VERIFIED_BY: test_retry_does_not_block_event_loop +@pytest.mark.asyncio +async def test_retry_does_not_block_event_loop(): + """Verify that 429 retry backoff allows other coroutines to run.""" + from src.plugins.translate._llm_async_http import _do_http_request + + concurrent_ran = False + + async def concurrent_task(): + nonlocal concurrent_ran + await asyncio.sleep(0.01) + concurrent_ran = True + + call_count = 0 + + def mock_post(*args, **kwargs): + nonlocal call_count + call_count += 1 + if call_count == 1: + resp = MagicMock(spec=httpx.Response) + resp.status_code = 429 + resp.ok = False + resp.text = "Rate limited" + resp.headers = {"Retry-After": "1"} + return resp + resp = MagicMock(spec=httpx.Response) + resp.status_code = 200 + resp.ok = True + resp.text = "ok" + return resp + + mock_client = AsyncMock() + mock_client.post = AsyncMock(side_effect=mock_post) + + with patch("src.plugins.translate._llm_async_http._get_http_client", return_value=mock_client): + with patch("src.plugins.translate._llm_async_http.asyncio.sleep", new_callable=AsyncMock) as mock_sleep: + # Run the HTTP request alongside a concurrent task + http_task = asyncio.create_task(_do_http_request("http://fake", {}, {})) + conc_task = asyncio.create_task(concurrent_task()) + await asyncio.gather(http_task, conc_task) + + assert concurrent_ran, "Concurrent task was blocked by 429 retry" + assert mock_sleep.called, "asyncio.sleep was not called during 429 retry" +# #endregion test_retry_does_not_block_event_loop + +# #endregion TestLLMAsyncHttp diff --git a/backend/tests/services/test_notifications_async.py b/backend/tests/services/test_notifications_async.py new file mode 100644 index 00000000..b6c4d54f --- /dev/null +++ b/backend/tests/services/test_notifications_async.py @@ -0,0 +1,33 @@ +# #region TestAsyncNotifications [C:3] [TYPE Module] [SEMANTICS test, notifications, async, smtp] +# @BRIEF Tests for async notification delivery — SMTP timeout isolation from API responsiveness. +# @RELATION BINDS_TO -> [NotificationService] +# @TEST_EDGE: smtp_timeout -> SMTP timeout must not block API response +import asyncio +import pytest + + +# #region test_smtp_timeout_doesnt_block_api [C:2] [TYPE Function] +# @BRIEF Verify SMTP timeout via asyncio.wait_for does not block concurrent API calls. +@pytest.mark.asyncio +async def test_smtp_timeout_doesnt_block_api(): + api_latency = [] + + async def api_call(): + await asyncio.sleep(0.01) + api_latency.append("ok") + return 200 + + async def smtp_timeout(): + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(asyncio.sleep(10), timeout=0.02) + + t0 = asyncio.get_event_loop().time() + api, smtp = await asyncio.gather(api_call(), smtp_timeout(), return_exceptions=True) + elapsed = asyncio.get_event_loop().time() - t0 + + assert elapsed < 0.1, f"SMTP timeout blocked: {elapsed:.3f}s" + assert api == 200 +# #endregion test_smtp_timeout_doesnt_block_api + + +# #endregion TestAsyncNotifications