tests: Add Docker integration tests for uWSGI protocol with nginx

Add comprehensive integration tests verifying gunicorn's uWSGI binary protocol works correctly with nginx's uwsgi_pass directive. Test categories: - Basic GET/POST requests with query strings and large bodies - Header preservation (custom headers, Host, Content-Type) - HTTP keep-alive connections - Error responses (400-503 status codes) - WSGI environ variables - Large response streaming (1MB) - Concurrent request handling - Edge cases (binary data, unicode, long headers) Architecture: pytest -> nginx:8080 -> uwsgi_pass -> gunicorn:8000 Also adds GitHub Actions workflow that runs on changes to uwsgi module or docker test files.
2026-07-01 18:21:30 +08:00 · 2026-01-22 19:06:30 +01:00 · 2026-01-22 19:06:30 +01:00 · ecc471f3b4
commit ecc471f3b4
parent ac7296ec49
10 changed files with 973 additions and 0 deletions
--- a/.github/workflows/docker-integration.yml
+++ b/.github/workflows/docker-integration.yml
@ -0,0 +1,45 @@
+name: Docker Integration Tests
+
+on:
+  push:
+    branches: [master]
+    paths:
+      - 'gunicorn/uwsgi/**'
+      - 'tests/docker/uwsgi/**'
+      - '.github/workflows/docker-integration.yml'
+  pull_request:
+    paths:
+      - 'gunicorn/uwsgi/**'
+      - 'tests/docker/uwsgi/**'
+      - '.github/workflows/docker-integration.yml'
+
+permissions:
+  contents: read
+
+env:
+  FORCE_COLOR: 1
+
+jobs:
+  uwsgi-nginx:
+    name: uWSGI Protocol with nginx
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+
+    steps:
+      - uses: actions/checkout@v6
+
+      - name: Set up Python
+        uses: actions/setup-python@v6
+        with:
+          python-version: "3.12"
+          cache: pip
+          cache-dependency-path: requirements_test.txt
+
+      - name: Install test dependencies
+        run: |
+          python -m pip install --upgrade pip
+          python -m pip install pytest requests
+
+      - name: Run uWSGI integration tests
+        run: |
+          pytest tests/docker/uwsgi/ -v --tb=short
--- a/tests/docker/uwsgi/Dockerfile.gunicorn
+++ b/tests/docker/uwsgi/Dockerfile.gunicorn
@ -0,0 +1,16 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+# Copy gunicorn source
+COPY . /app/gunicorn-src/
+
+# Install gunicorn from source
+RUN pip install --no-cache-dir /app/gunicorn-src/
+
+# Copy test application
+COPY tests/docker/uwsgi/app.py /app/
+
+EXPOSE 8000
+
+CMD ["gunicorn", "--protocol", "uwsgi", "--uwsgi-allow-from", "*", "--bind", "0.0.0.0:8000", "--workers", "2", "--log-level", "debug", "app:application"]
--- a/tests/docker/uwsgi/Dockerfile.nginx
+++ b/tests/docker/uwsgi/Dockerfile.nginx
@ -0,0 +1,12 @@
+FROM nginx:alpine
+
+# Remove default config
+RUN rm /etc/nginx/conf.d/default.conf
+
+# Copy custom config
+COPY nginx.conf /etc/nginx/nginx.conf
+COPY uwsgi_params /etc/nginx/uwsgi_params
+
+EXPOSE 8080
+
+CMD ["nginx", "-g", "daemon off;"]
--- a/tests/docker/uwsgi/README.md
+++ b/tests/docker/uwsgi/README.md
@ -0,0 +1,154 @@
+# uWSGI Protocol Docker Integration Tests
+
+This directory contains Docker-based integration tests that verify gunicorn's
+uWSGI binary protocol implementation works correctly with nginx's `uwsgi_pass`
+directive.
+
+## Architecture
+
+```
+[pytest] --HTTP--> [nginx:8080] --uwsgi_pass--> [gunicorn:8000]
+```
+
+The tests make HTTP requests to nginx, which proxies them to gunicorn using the
+uWSGI binary protocol. This validates the complete request/response cycle through
+the protocol.
+
+## Prerequisites
+
+- Docker
+- Docker Compose (v2)
+- Python 3.8+
+- pytest
+- requests
+
+## Running Tests
+
+### From repository root:
+
+```bash
+# Run all uWSGI integration tests
+pytest tests/docker/uwsgi/ -v
+
+# Run specific test class
+pytest tests/docker/uwsgi/ -v -k TestBasicRequests
+
+# Skip Docker tests (for CI environments without Docker)
+pytest tests/ -v -m "not docker"
+```
+
+### Manual testing:
+
+```bash
+cd tests/docker/uwsgi
+
+# Start services
+docker compose up -d
+
+# Wait for services to be healthy
+docker compose ps
+
+# Test endpoints
+curl http://localhost:8080/
+curl -X POST -d "test body" http://localhost:8080/echo
+curl http://localhost:8080/headers
+curl "http://localhost:8080/query?foo=bar"
+curl http://localhost:8080/environ
+curl http://localhost:8080/error/404
+curl http://localhost:8080/large > /dev/null  # 1MB response
+
+# View logs
+docker compose logs gunicorn
+docker compose logs nginx
+
+# Stop services
+docker compose down -v
+```
+
+## Test Categories
+
+| Category | Description |
+|----------|-------------|
+| `TestBasicRequests` | GET, POST, query strings, large bodies |
+| `TestHeaderPreservation` | Custom headers, Host, Content-Type, User-Agent |
+| `TestKeepAlive` | Multiple requests per connection |
+| `TestErrorResponses` | HTTP error codes (400, 404, 500, etc.) |
+| `TestEnvironVariables` | WSGI environ: REQUEST_METHOD, PATH_INFO, etc. |
+| `TestLargeResponses` | 1MB response body streaming |
+| `TestConcurrency` | Parallel request handling |
+| `TestSpecialCases` | Edge cases: binary data, unicode, long headers |
+
+## Files
+
+| File | Purpose |
+|------|---------|
+| `docker-compose.yml` | Orchestrates nginx + gunicorn containers |
+| `Dockerfile.gunicorn` | Builds gunicorn image with test app |
+| `Dockerfile.nginx` | Builds nginx with uwsgi config |
+| `nginx.conf` | nginx configuration using `uwsgi_pass` |
+| `uwsgi_params` | Standard uwsgi parameter mappings |
+| `app.py` | Test WSGI application with multiple endpoints |
+| `conftest.py` | pytest fixtures for Docker lifecycle |
+| `test_uwsgi_integration.py` | Test cases |
+
+## Test App Endpoints
+
+| Endpoint | Method | Description |
+|----------|--------|-------------|
+| `/` | GET | Basic hello response |
+| `/echo` | POST | Echo request body |
+| `/headers` | GET/POST | Return received headers as JSON |
+| `/environ` | GET/POST | Return WSGI environ as JSON |
+| `/query` | GET | Return query params as JSON |
+| `/json` | POST | Parse and echo JSON body |
+| `/error/{code}` | GET | Return specified HTTP error |
+| `/large` | GET | Return 1MB response |
+
+## Gunicorn Configuration
+
+The gunicorn container runs with:
+
+```bash
+gunicorn \
+  --protocol uwsgi \
+  --uwsgi-allow-from "*" \
+  --bind 0.0.0.0:8000 \
+  --workers 2 \
+  --log-level debug \
+  app:application
+```
+
+Key settings:
+- `--protocol uwsgi`: Enable uWSGI binary protocol
+- `--uwsgi-allow-from "*"`: Accept connections from Docker network IPs
+
+## Troubleshooting
+
+### Services won't start
+
+Check Docker logs:
+```bash
+docker compose logs
+```
+
+### Connection refused
+
+Wait for health checks:
+```bash
+docker compose ps  # Check health status
+```
+
+### Tests timing out
+
+Increase `STARTUP_TIMEOUT` in `conftest.py` or check if ports are in use:
+```bash
+lsof -i :8080
+lsof -i :8000
+```
+
+### Rebuild after code changes
+
+```bash
+docker compose build --no-cache
+docker compose up -d
+```
--- a/tests/docker/uwsgi/app.py
+++ b/tests/docker/uwsgi/app.py
@ -0,0 +1,222 @@
+"""
+Test WSGI application for uWSGI protocol integration tests.
+
+This application provides various endpoints to test different aspects
+of the uWSGI binary protocol when proxied through nginx.
+"""
+
+import json
+
+
+def application(environ, start_response):
+    """Main WSGI application entry point."""
+    path = environ.get('PATH_INFO', '/')
+    method = environ.get('REQUEST_METHOD', 'GET')
+
+    # Route to appropriate handler
+    if path == '/':
+        return handle_root(environ, start_response)
+    elif path == '/echo':
+        return handle_echo(environ, start_response)
+    elif path == '/headers':
+        return handle_headers(environ, start_response)
+    elif path == '/environ':
+        return handle_environ(environ, start_response)
+    elif path.startswith('/error/'):
+        return handle_error(environ, start_response, path)
+    elif path == '/large':
+        return handle_large(environ, start_response)
+    elif path == '/json':
+        return handle_json(environ, start_response)
+    elif path == '/query':
+        return handle_query(environ, start_response)
+    else:
+        return handle_not_found(environ, start_response)
+
+
+def handle_root(environ, start_response):
+    """Basic root endpoint."""
+    status = '200 OK'
+    headers = [('Content-Type', 'text/plain')]
+    start_response(status, headers)
+    return [b'Hello from gunicorn uWSGI!\n']
+
+
+def handle_echo(environ, start_response):
+    """Echo back the request body."""
+    try:
+        content_length = int(environ.get('CONTENT_LENGTH', 0))
+    except (ValueError, TypeError):
+        content_length = 0
+
+    body = b''
+    if content_length > 0:
+        body = environ['wsgi.input'].read(content_length)
+
+    status = '200 OK'
+    headers = [
+        ('Content-Type', 'application/octet-stream'),
+        ('Content-Length', str(len(body)))
+    ]
+    start_response(status, headers)
+    return [body]
+
+
+def handle_headers(environ, start_response):
+    """Return received HTTP headers as JSON."""
+    headers_dict = {}
+    for key, value in environ.items():
+        if key.startswith('HTTP_'):
+            # Convert HTTP_X_CUSTOM_HEADER to X-Custom-Header
+            header_name = key[5:].replace('_', '-').title()
+            headers_dict[header_name] = value
+
+    # Also include some special headers
+    if 'CONTENT_TYPE' in environ:
+        headers_dict['Content-Type'] = environ['CONTENT_TYPE']
+    if 'CONTENT_LENGTH' in environ:
+        headers_dict['Content-Length'] = environ['CONTENT_LENGTH']
+
+    body = json.dumps(headers_dict, indent=2).encode('utf-8')
+    status = '200 OK'
+    headers = [
+        ('Content-Type', 'application/json'),
+        ('Content-Length', str(len(body)))
+    ]
+    start_response(status, headers)
+    return [body]
+
+
+def handle_environ(environ, start_response):
+    """Return WSGI environ variables as JSON."""
+    # Filter to serializable values
+    safe_environ = {}
+    skip_keys = {'wsgi.input', 'wsgi.errors', 'wsgi.file_wrapper'}
+
+    for key, value in environ.items():
+        if key in skip_keys:
+            continue
+        try:
+            # Test if value is JSON serializable
+            json.dumps(value)
+            safe_environ[key] = value
+        except (TypeError, ValueError):
+            safe_environ[key] = str(value)
+
+    body = json.dumps(safe_environ, indent=2).encode('utf-8')
+    status = '200 OK'
+    headers = [
+        ('Content-Type', 'application/json'),
+        ('Content-Length', str(len(body)))
+    ]
+    start_response(status, headers)
+    return [body]
+
+
+def handle_error(environ, start_response, path):
+    """Return specified HTTP error code."""
+    try:
+        code = int(path.split('/')[-1])
+    except ValueError:
+        code = 500
+
+    status_messages = {
+        400: 'Bad Request',
+        401: 'Unauthorized',
+        403: 'Forbidden',
+        404: 'Not Found',
+        500: 'Internal Server Error',
+        502: 'Bad Gateway',
+        503: 'Service Unavailable',
+    }
+
+    message = status_messages.get(code, 'Error')
+    status = f'{code} {message}'
+    body = json.dumps({'error': message, 'code': code}).encode('utf-8')
+
+    headers = [
+        ('Content-Type', 'application/json'),
+        ('Content-Length', str(len(body)))
+    ]
+    start_response(status, headers)
+    return [body]
+
+
+def handle_large(environ, start_response):
+    """Return a 1MB response body for testing large responses."""
+    # Generate 1MB of data (1024 * 1024 bytes)
+    chunk_size = 1024
+    num_chunks = 1024
+    chunk = b'X' * chunk_size
+
+    status = '200 OK'
+    headers = [
+        ('Content-Type', 'application/octet-stream'),
+        ('Content-Length', str(chunk_size * num_chunks))
+    ]
+    start_response(status, headers)
+
+    # Return as generator for streaming
+    def generate():
+        for _ in range(num_chunks):
+            yield chunk
+
+    return generate()
+
+
+def handle_json(environ, start_response):
+    """Handle JSON POST requests."""
+    try:
+        content_length = int(environ.get('CONTENT_LENGTH', 0))
+    except (ValueError, TypeError):
+        content_length = 0
+
+    if content_length > 0:
+        body = environ['wsgi.input'].read(content_length)
+        try:
+            data = json.loads(body.decode('utf-8'))
+            response = {'received': data, 'status': 'ok'}
+        except json.JSONDecodeError:
+            response = {'error': 'Invalid JSON', 'status': 'error'}
+    else:
+        response = {'error': 'No body', 'status': 'error'}
+
+    body = json.dumps(response).encode('utf-8')
+    status = '200 OK'
+    headers = [
+        ('Content-Type', 'application/json'),
+        ('Content-Length', str(len(body)))
+    ]
+    start_response(status, headers)
+    return [body]
+
+
+def handle_query(environ, start_response):
+    """Return query string parameters as JSON."""
+    from urllib.parse import parse_qs
+    query_string = environ.get('QUERY_STRING', '')
+    params = parse_qs(query_string)
+
+    # Convert lists to single values where appropriate
+    simple_params = {k: v[0] if len(v) == 1 else v for k, v in params.items()}
+
+    body = json.dumps(simple_params).encode('utf-8')
+    status = '200 OK'
+    headers = [
+        ('Content-Type', 'application/json'),
+        ('Content-Length', str(len(body)))
+    ]
+    start_response(status, headers)
+    return [body]
+
+
+def handle_not_found(environ, start_response):
+    """Handle 404 for unknown paths."""
+    body = json.dumps({'error': 'Not Found', 'path': environ.get('PATH_INFO')}).encode('utf-8')
+    status = '404 Not Found'
+    headers = [
+        ('Content-Type', 'application/json'),
+        ('Content-Length', str(len(body)))
+    ]
+    start_response(status, headers)
+    return [body]
--- a/tests/docker/uwsgi/conftest.py
+++ b/tests/docker/uwsgi/conftest.py
@ -0,0 +1,121 @@
+"""
+pytest fixtures for uWSGI Docker integration tests.
+"""
+
+import os
+import subprocess
+import time
+
+import pytest
+import requests
+
+
+COMPOSE_FILE = os.path.join(os.path.dirname(__file__), 'docker-compose.yml')
+NGINX_URL = 'http://127.0.0.1:8080'
+STARTUP_TIMEOUT = 60  # seconds
+
+
+def is_docker_available():
+    """Check if Docker is available."""
+    try:
+        result = subprocess.run(
+            ['docker', 'info'],
+            capture_output=True,
+            timeout=10
+        )
+        return result.returncode == 0
+    except (subprocess.TimeoutExpired, FileNotFoundError):
+        return False
+
+
+def is_compose_available():
+    """Check if docker compose is available."""
+    try:
+        result = subprocess.run(
+            ['docker', 'compose', 'version'],
+            capture_output=True,
+            timeout=10
+        )
+        return result.returncode == 0
+    except (subprocess.TimeoutExpired, FileNotFoundError):
+        return False
+
+
+docker_available = pytest.mark.skipif(
+    not is_docker_available() or not is_compose_available(),
+    reason="Docker or docker compose not available"
+)
+
+
+@pytest.fixture(scope='session')
+def docker_services():
+    """
+    Start Docker Compose services for the test session.
+
+    This fixture builds and starts the gunicorn and nginx containers,
+    waits for them to be healthy, and tears them down after all tests.
+    """
+    if not is_docker_available() or not is_compose_available():
+        pytest.skip("Docker or docker compose not available")
+
+    # Build and start services
+    subprocess.run(
+        ['docker', 'compose', '-f', COMPOSE_FILE, 'build'],
+        check=True,
+        capture_output=True
+    )
+
+    subprocess.run(
+        ['docker', 'compose', '-f', COMPOSE_FILE, 'up', '-d'],
+        check=True,
+        capture_output=True
+    )
+
+    # Wait for services to be healthy
+    start_time = time.time()
+    while time.time() - start_time < STARTUP_TIMEOUT:
+        try:
+            response = requests.get(f'{NGINX_URL}/', timeout=2)
+            if response.status_code == 200:
+                break
+        except requests.RequestException:
+            pass
+        time.sleep(1)
+    else:
+        # Get logs for debugging
+        logs = subprocess.run(
+            ['docker', 'compose', '-f', COMPOSE_FILE, 'logs'],
+            capture_output=True,
+            text=True
+        )
+        subprocess.run(
+            ['docker', 'compose', '-f', COMPOSE_FILE, 'down', '-v'],
+            capture_output=True
+        )
+        pytest.fail(
+            f"Services did not become healthy within {STARTUP_TIMEOUT}s.\n"
+            f"Logs:\n{logs.stdout}\n{logs.stderr}"
+        )
+
+    yield
+
+    # Teardown
+    subprocess.run(
+        ['docker', 'compose', '-f', COMPOSE_FILE, 'down', '-v'],
+        capture_output=True
+    )
+
+
+@pytest.fixture
+def nginx_url(docker_services):
+    """Return the nginx base URL."""
+    return NGINX_URL
+
+
+@pytest.fixture
+def session(docker_services):
+    """Return a requests Session with keep-alive enabled."""
+    with requests.Session() as s:
+        # Enable keep-alive
+        s.headers['Connection'] = 'keep-alive'
+        yield s
--- a/tests/docker/uwsgi/docker-compose.yml
+++ b/tests/docker/uwsgi/docker-compose.yml
@ -0,0 +1,29 @@
+services:
+  gunicorn:
+    build:
+      context: ../../..
+      dockerfile: tests/docker/uwsgi/Dockerfile.gunicorn
+    expose:
+      - "8000"
+    healthcheck:
+      test: ["CMD", "python", "-c", "import socket; s=socket.socket(); s.connect(('localhost', 8000)); s.close()"]
+      interval: 2s
+      timeout: 5s
+      retries: 10
+      start_period: 5s
+
+  nginx:
+    build:
+      context: .
+      dockerfile: Dockerfile.nginx
+    ports:
+      - "8080:8080"
+    depends_on:
+      gunicorn:
+        condition: service_healthy
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://localhost:8080/"]
+      interval: 2s
+      timeout: 5s
+      retries: 10
+      start_period: 5s
--- a/tests/docker/uwsgi/nginx.conf
+++ b/tests/docker/uwsgi/nginx.conf
@ -0,0 +1,46 @@
+worker_processes 1;
+
+events {
+    worker_connections 1024;
+}
+
+http {
+    include       /etc/nginx/mime.types;
+    default_type  application/octet-stream;
+
+    log_format main '$remote_addr - $remote_user [$time_local] "$request" '
+                    '$status $body_bytes_sent "$http_referer" '
+                    '"$http_user_agent"';
+
+    access_log /var/log/nginx/access.log main;
+    error_log /var/log/nginx/error.log debug;
+
+    sendfile on;
+    keepalive_timeout 65;
+
+    upstream gunicorn {
+        server gunicorn:8000;
+    }
+
+    server {
+        listen 8080;
+        server_name localhost;
+
+        # Increase buffer sizes for large headers
+        uwsgi_buffer_size 32k;
+        uwsgi_buffers 8 32k;
+        uwsgi_busy_buffers_size 64k;
+
+        # Read timeout for large responses
+        uwsgi_read_timeout 300s;
+
+        location / {
+            uwsgi_pass gunicorn;
+            include uwsgi_params;
+
+            # Pass additional headers
+            uwsgi_param HTTP_X_FORWARDED_FOR $proxy_add_x_forwarded_for;
+            uwsgi_param HTTP_X_REAL_IP $remote_addr;
+        }
+    }
+}
--- a/tests/docker/uwsgi/test_uwsgi_integration.py
+++ b/tests/docker/uwsgi/test_uwsgi_integration.py
@ -0,0 +1,312 @@
+"""
+Integration tests for gunicorn's uWSGI binary protocol with nginx.
+
+These tests verify that gunicorn correctly implements the uWSGI binary
+protocol by running actual requests through nginx's uwsgi_pass directive.
+"""
+
+import concurrent.futures
+import json
+
+import pytest
+import requests
+
+from conftest import docker_available
+
+
+@docker_available
+class TestBasicRequests:
+    """Test basic HTTP request handling through uWSGI protocol."""
+
+    def test_get_root(self, nginx_url):
+        """Test basic GET request to root endpoint."""
+        response = requests.get(f'{nginx_url}/')
+        assert response.status_code == 200
+        assert b'Hello from gunicorn uWSGI!' in response.content
+
+    def test_get_with_query_string(self, nginx_url):
+        """Test GET request with query string parameters."""
+        response = requests.get(f'{nginx_url}/query?foo=bar&baz=qux')
+        assert response.status_code == 200
+        data = response.json()
+        assert data['foo'] == 'bar'
+        assert data['baz'] == 'qux'
+
+    def test_post_echo(self, nginx_url):
+        """Test POST request with body echo."""
+        test_body = b'This is a test body content'
+        response = requests.post(f'{nginx_url}/echo', data=test_body)
+        assert response.status_code == 200
+        assert response.content == test_body
+
+    def test_post_json(self, nginx_url):
+        """Test POST request with JSON body."""
+        test_data = {'key': 'value', 'number': 42, 'nested': {'a': 1}}
+        response = requests.post(
+            f'{nginx_url}/json',
+            json=test_data,
+            headers={'Content-Type': 'application/json'}
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data['status'] == 'ok'
+        assert data['received'] == test_data
+
+    def test_post_large_body(self, nginx_url):
+        """Test POST with large request body (100KB)."""
+        large_body = b'X' * (100 * 1024)
+        response = requests.post(f'{nginx_url}/echo', data=large_body)
+        assert response.status_code == 200
+        assert len(response.content) == len(large_body)
+        assert response.content == large_body
+
+
+@docker_available
+class TestHeaderPreservation:
+    """Test that headers are correctly passed through uWSGI protocol."""
+
+    def test_custom_headers(self, nginx_url):
+        """Test custom headers are passed to the application."""
+        custom_headers = {
+            'X-Custom-Header': 'custom-value',
+            'X-Another-Header': 'another-value'
+        }
+        response = requests.get(f'{nginx_url}/headers', headers=custom_headers)
+        assert response.status_code == 200
+        data = response.json()
+        assert data.get('X-Custom-Header') == 'custom-value'
+        assert data.get('X-Another-Header') == 'another-value'
+
+    def test_host_header(self, nginx_url):
+        """Test Host header is passed correctly."""
+        response = requests.get(
+            f'{nginx_url}/headers',
+            headers={'Host': 'test.example.com'}
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data.get('Host') == 'test.example.com'
+
+    def test_content_type_header(self, nginx_url):
+        """Test Content-Type header is passed correctly."""
+        response = requests.post(
+            f'{nginx_url}/headers',
+            data='test',
+            headers={'Content-Type': 'application/x-custom-type'}
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data.get('Content-Type') == 'application/x-custom-type'
+
+    def test_user_agent_header(self, nginx_url):
+        """Test User-Agent header is passed correctly."""
+        response = requests.get(
+            f'{nginx_url}/headers',
+            headers={'User-Agent': 'TestAgent/1.0'}
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data.get('User-Agent') == 'TestAgent/1.0'
+
+
+@docker_available
+class TestKeepAlive:
+    """Test HTTP keep-alive with multiple requests per connection."""
+
+    def test_multiple_requests_same_session(self, session, nginx_url):
+        """Test multiple requests using same session/connection."""
+        for i in range(5):
+            response = session.get(f'{nginx_url}/')
+            assert response.status_code == 200
+
+    def test_mixed_requests_same_session(self, session, nginx_url):
+        """Test mixed GET and POST requests using same session."""
+        # GET request
+        response = session.get(f'{nginx_url}/')
+        assert response.status_code == 200
+
+        # POST request
+        response = session.post(f'{nginx_url}/echo', data=b'test')
+        assert response.status_code == 200
+        assert response.content == b'test'
+
+        # Another GET
+        response = session.get(f'{nginx_url}/headers')
+        assert response.status_code == 200
+
+        # JSON POST
+        response = session.post(f'{nginx_url}/json', json={'test': 1})
+        assert response.status_code == 200
+
+
+@docker_available
+class TestErrorResponses:
+    """Test HTTP error responses through uWSGI protocol."""
+
+    @pytest.mark.parametrize('code', [400, 401, 403, 404, 500, 502, 503])
+    def test_error_codes(self, nginx_url, code):
+        """Test various HTTP error codes are returned correctly."""
+        response = requests.get(f'{nginx_url}/error/{code}')
+        assert response.status_code == code
+        data = response.json()
+        assert data['code'] == code
+
+    def test_not_found(self, nginx_url):
+        """Test 404 for non-existent path."""
+        response = requests.get(f'{nginx_url}/nonexistent/path')
+        assert response.status_code == 404
+        data = response.json()
+        assert data['error'] == 'Not Found'
+        assert data['path'] == '/nonexistent/path'
+
+
+@docker_available
+class TestEnvironVariables:
+    """Test WSGI environ variables are correctly set."""
+
+    def test_request_method(self, nginx_url):
+        """Test REQUEST_METHOD is set correctly."""
+        response = requests.get(f'{nginx_url}/environ')
+        assert response.status_code == 200
+        data = response.json()
+        assert data.get('REQUEST_METHOD') == 'GET'
+
+        response = requests.post(f'{nginx_url}/environ', data='')
+        data = response.json()
+        assert data.get('REQUEST_METHOD') == 'POST'
+
+    def test_path_info(self, nginx_url):
+        """Test PATH_INFO is set correctly."""
+        response = requests.get(f'{nginx_url}/environ')
+        assert response.status_code == 200
+        data = response.json()
+        assert data.get('PATH_INFO') == '/environ'
+
+    def test_query_string(self, nginx_url):
+        """Test QUERY_STRING is set correctly."""
+        response = requests.get(f'{nginx_url}/environ?foo=bar&test=123')
+        assert response.status_code == 200
+        data = response.json()
+        assert data.get('QUERY_STRING') == 'foo=bar&test=123'
+
+    def test_server_protocol(self, nginx_url):
+        """Test SERVER_PROTOCOL is set."""
+        response = requests.get(f'{nginx_url}/environ')
+        assert response.status_code == 200
+        data = response.json()
+        assert 'SERVER_PROTOCOL' in data
+        assert data['SERVER_PROTOCOL'].startswith('HTTP/')
+
+    def test_content_length(self, nginx_url):
+        """Test CONTENT_LENGTH is set for POST requests."""
+        body = 'test body content'
+        response = requests.post(f'{nginx_url}/environ', data=body)
+        assert response.status_code == 200
+        data = response.json()
+        assert data.get('CONTENT_LENGTH') == str(len(body))
+
+
+@docker_available
+class TestLargeResponses:
+    """Test large response handling through uWSGI protocol."""
+
+    def test_1mb_response(self, nginx_url):
+        """Test 1MB response body is received correctly."""
+        response = requests.get(f'{nginx_url}/large')
+        assert response.status_code == 200
+        assert len(response.content) == 1024 * 1024
+        # Verify content is all 'X' characters
+        assert response.content == b'X' * (1024 * 1024)
+
+    def test_large_response_content_length(self, nginx_url):
+        """Test Content-Length header for large response."""
+        response = requests.get(f'{nginx_url}/large')
+        assert response.status_code == 200
+        assert response.headers.get('Content-Length') == str(1024 * 1024)
+
+
+@docker_available
+class TestConcurrency:
+    """Test concurrent request handling."""
+
+    def test_parallel_requests(self, nginx_url):
+        """Test handling multiple parallel requests."""
+        num_requests = 20
+
+        def make_request(i):
+            response = requests.get(f'{nginx_url}/query?id={i}')
+            return response.status_code, response.json().get('id')
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
+            futures = [executor.submit(make_request, i) for i in range(num_requests)]
+            results = [f.result() for f in concurrent.futures.as_completed(futures)]
+
+        # All requests should succeed
+        assert all(status == 200 for status, _ in results)
+        # All IDs should be present
+        ids = set(id_val for _, id_val in results)
+        assert ids == set(str(i) for i in range(num_requests))
+
+    def test_parallel_mixed_requests(self, nginx_url):
+        """Test parallel GET and POST requests."""
+        def get_request():
+            return requests.get(f'{nginx_url}/').status_code
+
+        def post_request(data):
+            response = requests.post(f'{nginx_url}/echo', data=data)
+            return response.status_code, response.content
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
+            get_futures = [executor.submit(get_request) for _ in range(10)]
+            post_futures = [
+                executor.submit(post_request, f'data-{i}'.encode())
+                for i in range(10)
+            ]
+
+            get_results = [f.result() for f in get_futures]
+            post_results = [f.result() for f in post_futures]
+
+        assert all(status == 200 for status in get_results)
+        assert all(status == 200 for status, _ in post_results)
+
+
+@docker_available
+class TestSpecialCases:
+    """Test edge cases and special scenarios."""
+
+    def test_empty_body_post(self, nginx_url):
+        """Test POST with empty body."""
+        response = requests.post(f'{nginx_url}/echo', data=b'')
+        assert response.status_code == 200
+        assert response.content == b''
+
+    def test_binary_body(self, nginx_url):
+        """Test POST with binary body containing null bytes."""
+        binary_data = bytes(range(256))
+        response = requests.post(f'{nginx_url}/echo', data=binary_data)
+        assert response.status_code == 200
+        assert response.content == binary_data
+
+    def test_unicode_in_query_string(self, nginx_url):
+        """Test unicode characters in query string."""
+        response = requests.get(f'{nginx_url}/query', params={'name': 'test'})
+        assert response.status_code == 200
+        data = response.json()
+        assert data.get('name') == 'test'
+
+    def test_special_characters_in_path(self, nginx_url):
+        """Test handling of special path that triggers 404."""
+        # This should return 404 since the path doesn't exist
+        response = requests.get(f'{nginx_url}/path/with/slashes')
+        assert response.status_code == 404
+
+    def test_long_header_value(self, nginx_url):
+        """Test handling of long header values."""
+        long_value = 'X' * 4096  # 4KB header value
+        response = requests.get(
+            f'{nginx_url}/headers',
+            headers={'X-Long-Header': long_value}
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data.get('X-Long-Header') == long_value
--- a/tests/docker/uwsgi/uwsgi_params
+++ b/tests/docker/uwsgi/uwsgi_params
@ -0,0 +1,16 @@
+uwsgi_param  QUERY_STRING       $query_string;
+uwsgi_param  REQUEST_METHOD     $request_method;
+uwsgi_param  CONTENT_TYPE       $content_type;
+uwsgi_param  CONTENT_LENGTH     $content_length;
+
+uwsgi_param  REQUEST_URI        $request_uri;
+uwsgi_param  PATH_INFO          $document_uri;
+uwsgi_param  DOCUMENT_ROOT      $document_root;
+uwsgi_param  SERVER_PROTOCOL    $server_protocol;
+uwsgi_param  REQUEST_SCHEME     $scheme;
+uwsgi_param  HTTPS              $https if_not_empty;
+
+uwsgi_param  REMOTE_ADDR        $remote_addr;
+uwsgi_param  REMOTE_PORT        $remote_port;
+uwsgi_param  SERVER_PORT        $server_port;
+uwsgi_param  SERVER_NAME        $server_name;