chore: add Thrift call verification

.github/workflows/proxy-tests.yml

@@ -0,0 +1,138 @@
+# Copyright (c) 2025 ADBC Drivers Contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+name: Proxy Tests
+
+on:
+  push:
+    branches:
+      - main
+    paths:
+      - '.github/workflows/proxy-tests.yml'
+      - 'test-infrastructure/proxy-server/**'
+      - 'test-infrastructure/tests/csharp/**'
+      - 'csharp/src/**'
+  pull_request:
+    # Only runs on PRs from the repo itself, not forks
+    paths:
+      - '.github/workflows/proxy-tests.yml'
+      - 'test-infrastructure/proxy-server/**'
+      - 'test-infrastructure/tests/csharp/**'
+      - 'csharp/src/**'
+  workflow_dispatch:
+    inputs:
+      run_all_tests:
+        description: 'Run all tests (ignores path filters)'
+        required: false
+        default: 'true'
+        type: boolean
+
+concurrency:
+  group: ${{ github.repository }}-${{ github.ref }}-${{ github.workflow }}
+  cancel-in-progress: true
+
+permissions:
+  contents: read
+
+jobs:
+  proxy-tests:
+    name: "Proxy Tests"
+    runs-on: ubuntu-latest
+    if: ${{ !contains(github.event.pull_request.title, 'WIP') }}
+    timeout-minutes: 20
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+        with:
+          ref: ${{ github.event.pull_request.head.sha || github.sha }}
+          fetch-depth: 0
+          submodules: recursive
+          persist-credentials: false
+
+      - name: Setup .NET 8.0
+        uses: actions/setup-dotnet@v4
+        with:
+          dotnet-version: '8.0.x'
+
+      - name: Setup Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.11'
+
+      - name: Install Python dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r test-infrastructure/proxy-server/requirements.txt
+
+      - name: Generate mitmproxy certificates
+        run: |
+          # Start mitmdump in background to generate certificates
+          mitmdump &
+          MITM_PID=$!
+          # Wait for certificate generation (happens on first run)
+          sleep 5
+          # Stop mitmdump forcefully
+          kill -9 $MITM_PID || true
+          wait $MITM_PID 2>/dev/null || true
+          # Ensure no mitmdump processes are running
+          pkill -9 mitmdump || true
+          sleep 2
+          # Verify certificate was created
+          ls -la ~/.mitmproxy/
+          cat ~/.mitmproxy/mitmproxy-ca-cert.pem
+
+      - name: Trust mitmproxy certificate
+        run: |
+          sudo cp ~/.mitmproxy/mitmproxy-ca-cert.pem /usr/local/share/ca-certificates/mitmproxy.crt
+          sudo update-ca-certificates
+
+      - name: Build test project
+        run: |
+          cd test-infrastructure/tests/csharp
+          dotnet build
+
+      - name: Create test configuration
+        env:
+          DATABRICKS_SERVER_HOSTNAME: ${{ secrets.DATABRICKS_HOST }}
+          DATABRICKS_TOKEN: ${{ secrets.DATABRICKS_TOKEN }}
+          DATABRICKS_HTTP_PATH: ${{ secrets.TEST_PECO_WAREHOUSE_HTTP_PATH }}
+        run: |
+          cat > databricks-test-config.json << EOF
+          {
+            "uri": "https://${{ env.DATABRICKS_SERVER_HOSTNAME }}${{ env.DATABRICKS_HTTP_PATH }}",
+            "auth_type": "token",
+            "token": "${{ env.DATABRICKS_TOKEN }}"
+          }
+          EOF
+          echo "Configuration file created:"
+          echo "URI: https://${{ env.DATABRICKS_SERVER_HOSTNAME }}${{ env.DATABRICKS_HTTP_PATH }}"
+          echo "Auth type: token"
+
+      - name: Run proxy infrastructure tests
+        env:
+          DATABRICKS_TEST_CONFIG_FILE: ${{ github.workspace }}/databricks-test-config.json
+        run: |
+          cd test-infrastructure/tests/csharp
+          dotnet test --filter "FullyQualifiedName~ProxyInfrastructureTests" \
+            --logger "console;verbosity=normal" \
+            --no-build
+
+      - name: Run CloudFetch proxy tests
+        env:
+          DATABRICKS_TEST_CONFIG_FILE: ${{ github.workspace }}/databricks-test-config.json
+        run: |
+          cd test-infrastructure/tests/csharp
+          dotnet test --filter "FullyQualifiedName~CloudFetchTests" \
+            --logger "console;verbosity=normal" \
+            --no-build

test-infrastructure/proxy-server/mitmproxy_addon.py

@@ -337,20 +337,35 @@ def __init__(self):
 
         # Start Flask control API in background thread
         def run_api():
-            app.run(host='0.0.0.0', port=18081, threaded=True)
+            # Disable reloader and use production mode for faster startup
+            app.run(host='0.0.0.0', port=18081, threaded=True, use_reloader=False, debug=False)
 
         api_thread = threading.Thread(target=run_api, daemon=True, name="ControlAPI")
         api_thread.start()
         ctx.log.info("Control API started on http://0.0.0.0:18081")
 
-    def request(self, flow: http.HTTPFlow) -> None:
+    async def request(self, flow: http.HTTPFlow) -> None:
         """
         Intercept requests and inject failures based on enabled scenarios.
         Called by mitmproxy for each HTTP request.
+        Made async to support non-blocking delays.
         """
         # Detect request type
         if self._is_cloudfetch_download(flow.request):
-            self._handle_cloudfetch_request(flow)
+            # Track cloud fetch download
+            with state_lock:
+                call_record = {
+                    "timestamp": time.time(),
+                    "type": "cloud_download",
+                    "url": flow.request.pretty_url,
+                }
+                call_history.append(call_record)
+
+                # Enforce max history limit
+                if len(call_history) > MAX_CALL_HISTORY:
+                    del call_history[:len(call_history) - MAX_CALL_HISTORY]
+
+            await self._handle_cloudfetch_request(flow)
         elif self._is_thrift_request(flow.request):
             self._handle_thrift_request(flow)
 
@@ -381,7 +396,7 @@ def _is_thrift_request(self, request: http.Request) -> bool:
         # SEA requests use /api/2.0/sql/statements path
         return "/sql/1.0/warehouses/" in request.path or "/sql/1.0/endpoints/" in request.path
 
-    def _handle_cloudfetch_request(self, flow: http.HTTPFlow) -> None:
+    async def _handle_cloudfetch_request(self, flow: http.HTTPFlow) -> None:
         """Handle CloudFetch requests and inject failures if scenario is enabled."""
         with state_lock:
             # Find first enabled CloudFetch scenario
@@ -425,10 +440,12 @@ def _handle_cloudfetch_request(self, flow: http.HTTPFlow) -> None:
             self._disable_scenario(scenario_name)
 
         elif action == "delay":
-            # Inject delay (simulates timeout) - now supports configurable duration
+            # Inject delay using asyncio.sleep() to avoid blocking the event loop
+            import asyncio
             duration_seconds = scenario_config.get("duration_seconds", 5)
             ctx.log.info(f"[INJECT] Delaying {duration_seconds}s for scenario: {scenario_name}")
-            time.sleep(duration_seconds)
+            await asyncio.sleep(duration_seconds)
+            ctx.log.info(f"[INJECT] Delay complete, auto-disabled scenario: {scenario_name}")
             self._disable_scenario(scenario_name)
             # Let request continue after delay
 
@@ -455,9 +472,11 @@ def _handle_thrift_request(self, flow: http.HTTPFlow) -> None:
                 with state_lock:
                     call_record = {
                         "timestamp": time.time(),
+                        "type": "thrift",
                         "method": decoded.get("method", "unknown"),
                         "message_type": decoded.get("message_type", "unknown"),
                         "sequence_id": decoded.get("sequence_id", 0),
+                        "fields": decoded.get("fields", {}),
                     }
                     call_history.append(call_record)