test: stabilize flaky tests with proper async coordination

Fix timing-sensitive tests that were failing intermittently in CI by:
- Adding retry loops with proper timeout handling for async background operations
- Improving error messages to show actual vs expected counts after timeouts
- Removing unused imports to clean up test files
- Adding proper mocking for app creation to avoid blocking operations during tests

The key changes address race conditions where background workers needed more time to process queued operations, particularly in access logging and storage tests. Tests now wait up to 3-5 seconds with polling rather than fixed delays, making them more reliable across different CI environments.

Author: CaddyGlow

tests/test_access_logger_integration.py

@@ -397,13 +397,20 @@ async def test_high_volume_access_logging(
         # Should complete quickly
         assert log_time < 5.0, f"High-volume logging took too long: {log_time}s"
 
-        # Give background worker time to process
-        await asyncio.sleep(2.0)
-
-        # Verify all were stored
-        with Session(storage_with_db._engine) as session:
-            count = len(session.exec(select(AccessLog)).all())
-            assert count == num_logs, f"Expected {num_logs} logs, got {count}"
+        # Give background worker time to process with retries
+        for _attempt in range(10):
+            await asyncio.sleep(0.5)
+            with Session(storage_with_db._engine) as session:
+                count = len(session.exec(select(AccessLog)).all())
+                if count == num_logs:
+                    break
+        else:
+            # Final check with detailed error
+            with Session(storage_with_db._engine) as session:
+                count = len(session.exec(select(AccessLog)).all())
+                assert count == num_logs, (
+                    f"Expected {num_logs} logs, got {count} after 5s wait"
+                )
 
     @pytest.mark.unit
     async def test_mixed_streaming_and_regular_logs(
@@ -443,15 +450,24 @@ async def test_mixed_streaming_and_regular_logs(
         # Execute all concurrently
         await asyncio.gather(*tasks)
 
-        # Give background worker time to process
-        await asyncio.sleep(0.5)
-
-        # Verify all stored correctly
-        with Session(storage_with_db._engine) as session:
-            results = session.exec(
-                select(AccessLog).order_by(AccessLog.request_id)
-            ).all()
-            assert len(results) == 20
+        # Give background worker time to process with retries
+        for _attempt in range(10):
+            await asyncio.sleep(0.3)
+            with Session(storage_with_db._engine) as session:
+                results = session.exec(
+                    select(AccessLog).order_by(AccessLog.request_id)
+                ).all()
+                if len(results) == 20:
+                    break
+        else:
+            # Final check with detailed error
+            with Session(storage_with_db._engine) as session:
+                results = session.exec(
+                    select(AccessLog).order_by(AccessLog.request_id)
+                ).all()
+                assert len(results) == 20, (
+                    f"Expected 20 logs, got {len(results)} after 3s wait"
+                )
 
             # Verify streaming flags are correct
             for i, result in enumerate(results):

tests/test_api.py

@@ -5,7 +5,6 @@
 """
 
 from typing import Any
-from unittest.mock import AsyncMock
 
 import pytest
 from fastapi.testclient import TestClient
@@ -35,8 +34,6 @@
     STANDARD_OPENAI_REQUEST,
     STREAMING_ANTHROPIC_REQUEST,
     STREAMING_OPENAI_REQUEST,
-    create_anthropic_request,
-    create_openai_request,
 )
 
 

tests/test_queue_duckdb_storage.py

@@ -213,15 +213,23 @@ def mock_store_sync(data: AccessLogPayload) -> bool:
             await initialized_storage.store_request(log1)
             await initialized_storage.store_request(log2)
 
-            # Give time for processing
-            await asyncio.sleep(0.2)
+            # Give time for processing with retries
+            result = None
+            for _attempt in range(10):
+                await asyncio.sleep(0.3)
+                with Session(initialized_storage._engine) as session:
+                    result = session.exec(
+                        select(AccessLog).where(
+                            AccessLog.request_id == "success-request-2"
+                        )
+                    ).first()
+                    if result is not None:
+                        break
 
             # Second request should succeed despite first failing
-            with Session(initialized_storage._engine) as session:
-                result = session.exec(
-                    select(AccessLog).where(AccessLog.request_id == "success-request-2")
-                ).first()
-                assert result is not None
+            assert result is not None, (
+                "Expected success-request-2 to be processed after error recovery"
+            )
 
         await initialized_storage.close()
 

tests/test_scheduler.py

@@ -554,16 +554,31 @@ async def test_app_lifespan_with_scheduler(
 
     def test_scheduler_disabled_app_still_works(self) -> None:
         """Test that app works when scheduler is disabled."""
+        from unittest.mock import patch
+
         from ccproxy.api.app import create_app
 
         settings = Settings()
         settings.scheduler.enabled = False
 
-        app = create_app(settings)
+        # Mock any potential blocking operations during app creation
+        with (
+            patch("ccproxy.observability.metrics.get_metrics") as mock_metrics,
+            patch("ccproxy.config.settings.get_settings") as mock_get_settings,
+            patch(
+                "ccproxy.observability.stats_printer.get_stats_collector"
+            ) as mock_stats,
+            patch("ccproxy.pricing.updater.PricingUpdater") as mock_pricing,
+            patch("ccproxy.services.credentials.CredentialsManager") as mock_creds,
+        ):
+            # Mock settings to return our test configuration
+            mock_get_settings.return_value = settings
 
-        with TestClient(app) as client:
-            response = client.get("/health")
-            assert response.status_code == 200
+            app = create_app(settings)
+
+            with TestClient(app) as client:
+                response = client.get("/health")
+                assert response.status_code == 200
 
 
 class TestSchedulerErrorScenarios:

tests/test_streaming.py

@@ -15,7 +15,6 @@
 
 import json
 from typing import TYPE_CHECKING, Any
-from unittest.mock import AsyncMock
 
 import pytest
 from fastapi.testclient import TestClient