feat(source-rate-limits): run source rate limit tests after others for isolation

Author: Daan Manneke

.github/workflows/test-public-api.yml

@@ -157,14 +157,15 @@ jobs:
           BACKEND_IMAGE: test-backend:latest
         run: |
           cd backend
-          # Exclude rate_limit tests from CI/CD - they consume quota and should run separately
-          # To run rate limit tests locally: pytest tests/e2e/smoke -m rate_limit
+          # Exclude api_rate_limit tests (general API rate limiter) from CI
+          # Source rate_limit tests run sequentially after other tests for proper Redis isolation
+          # To run only rate limit tests locally: pytest tests/e2e/smoke -m rate_limit
 
           MAX_RETRIES=2
 
-          # Run all tests first
+          # Run all non-rate-limit tests in parallel first
           set +e
-          pytest tests/e2e/smoke -m "not rate_limit" -n 3 -v --tb=short 2>&1 | tee pytest_output.txt
+          pytest tests/e2e/smoke -m "not rate_limit and not api_rate_limit" -n 3 -v --tb=short 2>&1 | tee pytest_output.txt
           EXIT_CODE=${PIPESTATUS[0]}
           set -e
 
@@ -209,6 +210,32 @@ jobs:
           echo "Tests failed after $MAX_RETRIES retry attempts"
           exit 1
 
+      - name: Run Source Rate Limit Tests (Sequential)
+        if: success()
+        env:
+          TEST_ENV: local
+          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
+          TEST_COMPOSIO_API_KEY: ${{ secrets.TEST_COMPOSIO_API_KEY }}
+          TEST_COMPOSIO_NOTION_AUTH_CONFIG_ID_1: ${{ secrets.TEST_COMPOSIO_NOTION_AUTH_CONFIG_ID_1 }}
+          TEST_COMPOSIO_NOTION_ACCOUNT_ID_1: ${{ secrets.TEST_COMPOSIO_NOTION_ACCOUNT_ID_1 }}
+          TEST_COMPOSIO_NOTION_AUTH_CONFIG_ID_2: ${{ secrets.TEST_COMPOSIO_NOTION_AUTH_CONFIG_ID_2 }}
+          TEST_COMPOSIO_NOTION_ACCOUNT_ID_2: ${{ secrets.TEST_COMPOSIO_NOTION_ACCOUNT_ID_2 }}
+          TEST_COMPOSIO_GOOGLE_DRIVE_AUTH_CONFIG_ID_1: ${{ secrets.TEST_COMPOSIO_GOOGLE_DRIVE_AUTH_CONFIG_ID_1 }}
+          TEST_COMPOSIO_GOOGLE_DRIVE_ACCOUNT_ID_1: ${{ secrets.TEST_COMPOSIO_GOOGLE_DRIVE_ACCOUNT_ID_1 }}
+          TEST_COMPOSIO_GOOGLE_DRIVE_AUTH_CONFIG_ID_2: ${{ secrets.TEST_COMPOSIO_GOOGLE_DRIVE_AUTH_CONFIG_ID_2 }}
+          TEST_COMPOSIO_GOOGLE_DRIVE_ACCOUNT_ID_2: ${{ secrets.TEST_COMPOSIO_GOOGLE_DRIVE_ACCOUNT_ID_2 }}
+          TEST_PIPEDREAM_RATE_LIMIT_CLIENT_ID: ${{ secrets.TEST_PIPEDREAM_RATE_LIMIT_CLIENT_ID }}
+          TEST_PIPEDREAM_RATE_LIMIT_CLIENT_SECRET: ${{ secrets.TEST_PIPEDREAM_RATE_LIMIT_CLIENT_SECRET }}
+          TEST_PIPEDREAM_RATE_LIMIT_PROJECT_ID: ${{ secrets.TEST_PIPEDREAM_RATE_LIMIT_PROJECT_ID }}
+          TEST_PIPEDREAM_GOOGLE_DRIVE_DEFAULT_OAUTH_ACCOUNT_ID: ${{ secrets.TEST_PIPEDREAM_GOOGLE_DRIVE_DEFAULT_OAUTH_ACCOUNT_ID }}
+          TEST_PIPEDREAM_GOOGLE_DRIVE_DEFAULT_OAUTH_EXTERNAL_USER_ID: ${{ secrets.TEST_PIPEDREAM_GOOGLE_DRIVE_DEFAULT_OAUTH_EXTERNAL_USER_ID }}
+          TEST_PIPEDREAM_NOTION_DEFAULT_OAUTH_ACCOUNT_ID: ${{ secrets.TEST_PIPEDREAM_NOTION_DEFAULT_OAUTH_ACCOUNT_ID }}
+          TEST_PIPEDREAM_NOTION_DEFAULT_OAUTH_EXTERNAL_USER_ID: ${{ secrets.TEST_PIPEDREAM_NOTION_DEFAULT_OAUTH_EXTERNAL_USER_ID }}
+        run: |
+          cd backend
+          echo "Running source rate limit tests sequentially (no parallelization for Redis isolation)..."
+          pytest tests/e2e/smoke -m rate_limit -v --tb=short
+
       - name: Print service logs on failure
         if: failure()
         run: |

backend/pytest.ini

@@ -19,3 +19,5 @@ asyncio_mode = auto
 # Register custom markers
 markers =
     integration: marks tests as integration tests (deselect with '-m "not integration"')
+    rate_limit: marks tests that test rate limiting (run sequentially for proper isolation)
+    api_rate_limit: marks tests for API-level rate limiting (excluded from CI)

backend/tests/e2e/smoke/test_rate_limiting.py

@@ -32,8 +32,9 @@
 from config import settings
 
 # Module-level marker - applies to all tests in this file
+# Use api_rate_limit to distinguish from source rate limiting tests
 pytestmark = [
-    pytest.mark.rate_limit,
+    pytest.mark.api_rate_limit,
     pytest.mark.asyncio,
 ]
 

backend/tests/e2e/smoke/test_source_rate_limiting.py

@@ -7,7 +7,7 @@
 - Feature flag behavior
 - Atomic Lua script prevents race conditions
 
-All tests marked with @pytest.mark.rate_limit to run last and avoid quota interference.
+All tests marked with @pytest.mark.rate_limit to run sequentially for proper isolation.
 """
 
 import asyncio
@@ -18,6 +18,10 @@
 import httpx
 import pytest
 
+# Mark all tests in this module to run sequentially (no parallel execution)
+# This ensures proper Redis isolation between tests
+pytestmark = [pytest.mark.rate_limit]
+
 
 # Test rate limit configuration
 RATE_LIMIT = 1  # requests
@@ -423,9 +427,9 @@ async def test_notion_connection_level_rate_limiting_isolated(
         verify_sync_stats_only_inserts(job1, "connection 1")
         verify_sync_stats_only_inserts(job2, "connection 2")
 
-        # Verify Redis has 2 separate keys (connection-level tracking)
+        # Verify Redis has exactly 2 separate keys (connection-level tracking)
         # Use monitoring data since keys expire quickly (6s TTL)
-        assert len(monitoring_data) == 2, f"Expected 2 connection-level keys during sync, got {len(monitoring_data)}: {list(monitoring_data.keys())}"
+        assert len(monitoring_data) == 2, f"Expected exactly 2 connection-level keys during sync, got {len(monitoring_data)}: {list(monitoring_data.keys())}"
         print(f"✅ Found 2 separate connection-level rate limit keys during sync")
 
         # Verify rate limits were enforced during sync (check monitoring data)
@@ -545,9 +549,10 @@ async def test_google_drive_org_level_rate_limiting_aggregated(
         if job2["status"] == "completed":
             verify_sync_stats_only_inserts(job2, "connection 2")
 
-        # Verify Redis has only 1 shared key (org-level tracking)
-        redis_keys = await get_redis_keys("source_rate_limit:*:google_drive:org:*")
-        assert len(redis_keys) == 1, f"Expected 1 org-level key, got {len(redis_keys)}: {redis_keys}"
+        # Verify Redis had exactly 1 shared key during sync (org-level tracking)
+        # Use monitoring data since keys expire quickly (3s TTL)
+        assert len(monitoring_data) == 1, f"Expected exactly 1 org-level key during sync, got {len(monitoring_data)}: {list(monitoring_data.keys())}"
+        print(f"✅ Found 1 shared org-level rate limit key during sync")
 
         # Verify rate limits were enforced during sync (check monitoring data)
         print(f"\n📊 Redis monitoring during sync:")
@@ -556,14 +561,10 @@ async def test_google_drive_org_level_rate_limiting_aggregated(
             print(f"  {key}: max={max_counter}/{RATE_LIMIT}, samples={counters}")
             assert max_counter <= RATE_LIMIT, f"Rate limit exceeded during sync: {max_counter}/{RATE_LIMIT}"
 
-        # Verify the shared counter respected the limit
-        counter = await get_redis_counter(redis_keys[0])
-        print(f"\n📊 Shared org-level counter: {counter}/{RATE_LIMIT}")
-
         # With aggressive limits and concurrent syncs, expect some contention
         # At least one should complete (demonstrates shared quota)
         completed_count = sum(1 for job in [job1, job2] if job["status"] == "completed")
-        print(f"\n✅ Org-level aggregation verified: 1 shared quota, {completed_count}/2 syncs completed")
+        print(f"\n✅ Org-level aggregation verified: Shared quota enforced, {completed_count}/2 syncs completed")
 
     finally:
         # Cleanup