Tyk Gateway Testing: API Definitions, Policies, Middleware, and Integration Tests

Tyk Gateway Testing: API Definitions, Policies, Middleware, and Integration Tests

Tyk is an open-source API gateway that manages authentication, rate limiting, analytics, and API versioning through a declarative API definition system. Unlike gateways that use UI-driven configuration, Tyk stores its entire configuration as JSON documents — which means you can version-control it and write automated tests against it.

This guide covers testing Tyk API definitions, policies, middleware plugins, and integration behavior.

Tyk's Configuration Model

Tyk uses three core configuration types:

API Definitions — describe a single API: its upstream URL, authentication method, rate limits, middleware, and routing rules. Stored as JSON in the Tyk Dashboard or as files for Tyk OSS.

Policies — reusable sets of access rights, rate limits, and quota settings that can be applied to multiple API keys. A key can reference one or more policies.

Keys — represent API consumers. Each key has a policy applied and optionally per-key overrides.

Testing Tyk means validating that these three layers work correctly together.

Setting Up a Test Environment

Tyk runs as a Go binary with Redis for session state and optionally MongoDB/PostgreSQL for analytics. For testing, use Docker Compose.

# docker-compose.test.yml
version: "3.8"

services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

  tyk:
    image: tykio/tyk-gateway:v5.3
    ports:
      - "8080:8080"
      - "8081:8081"  # Gateway API
    environment:
      - TYK_GW_SECRET=test-secret
      - TYK_GW_STORAGE_HOST=redis
      - TYK_GW_STORAGE_PORT=6379
      - TYK_GW_LISTENPORT=8080
      - TYK_GW_CONTROLAPIPORT=8081
    volumes:
      - ./tyk/tyk.conf:/opt/tyk-gateway/tyk.conf
      - ./tyk/apps:/opt/tyk-gateway/apps
      - ./tyk/middleware:/opt/tyk-gateway/middleware
    depends_on:
      - redis

  upstream:
    image: kennethreitz/httpbin
    ports:
      - "8082:80"
// tyk/tyk.conf
{
  "listen_port": 8080,
  "secret": "test-secret",
  "node_secret": "test-secret",
  "template_path": "/opt/tyk-gateway/templates",
  "use_db_app_configs": false,
  "app_path": "/opt/tyk-gateway/apps",
  "middleware_path": "/opt/tyk-gateway/middleware",
  "storage": {
    "type": "redis",
    "host": "redis",
    "port": 6379
  },
  "enable_api_segregation": false,
  "control_api_hostname": "",
  "control_api_port": 8081
}

API Definition Testing

Validating API Definitions

Before loading an API definition into Tyk, validate its structure.

import json
import pytest
from jsonschema import validate, ValidationError

# Load Tyk's API definition schema (download from Tyk docs or derive from examples)
with open("schemas/tyk-api-definition.json") as f:
    TYK_SCHEMA = json.load(f)

def load_api_def(filename):
    with open(f"tyk/apps/{filename}") as f:
        return json.load(f)

def test_api_definition_valid_schema():
    """All API definitions pass schema validation."""
    import glob
    for path in glob.glob("tyk/apps/*.json"):
        with open(path) as f:
            defn = json.load(f)
        try:
            validate(defn, TYK_SCHEMA)
        except ValidationError as e:
            pytest.fail(f"{path}: {e.message}")

def test_api_definition_has_required_fields():
    defn = load_api_def("users-api.json")
    assert "api_id" in defn
    assert "name" in defn
    assert "proxy" in defn
    assert defn["proxy"]["target_url"], "target_url must not be empty"
    assert defn["proxy"]["listen_path"], "listen_path must not be empty"
    assert defn["proxy"]["listen_path"].endswith("/"), "listen_path must end with /"

def test_api_requires_auth():
    defn = load_api_def("users-api.json")
    # APIs should not be open (no auth) in production definitions
    assert not defn.get("use_keyless", False), \
        "API definition should not be keyless in production"
    assert defn.get("auth", {}).get("auth_header_name"), \
        "auth header name must be configured"

def test_rate_limits_defined():
    defn = load_api_def("users-api.json")
    # Should have global rate limiting configured
    assert defn.get("global_rate_limit", {}).get("rate", 0) > 0, \
        "Global rate limit rate must be positive"
    assert defn.get("global_rate_limit", {}).get("per", 0) > 0, \
        "Global rate limit window must be positive"

Loading and Reloading API Definitions

import httpx
import time

GATEWAY_API = "http://localhost:8081"
PROXY_URL = "http://localhost:8080"
TYK_SECRET = "test-secret"

def tyk_headers():
    return {"x-tyk-authorization": TYK_SECRET}

def reload_apis():
    """Hot reload Tyk to pick up new API definitions."""
    r = httpx.get(f"{GATEWAY_API}/tyk/reload/", headers=tyk_headers())
    assert r.status_code == 200
    time.sleep(1)  # Wait for reload to complete

def create_test_key(policy_id: str, rate: int = 100, per: int = 60) -> str:
    """Create an API key with the given policy."""
    payload = {
        "allowance": rate,
        "rate": rate,
        "per": per,
        "expires": -1,
        "quota_max": -1,
        "access_rights": {
            "test-api-id": {
                "api_name": "Test API",
                "api_id": "test-api-id",
                "versions": ["Default"]
            }
        }
    }
    r = httpx.post(
        f"{GATEWAY_API}/tyk/keys/",
        json=payload,
        headers=tyk_headers()
    )
    assert r.status_code == 200
    return r.json()["key"]

Policy Testing

Policies in Tyk define access rights and quotas for a group of API keys. Test that policies enforce what they claim.

@pytest.fixture(autouse=True)
def setup_test_api():
    """Load a test API definition before each test."""
    api_def = {
        "api_id": "test-api-id",
        "name": "Test API",
        "slug": "test-api",
        "use_keyless": False,
        "auth": {"auth_header_name": "Authorization"},
        "definition": {"location": "header", "key": "x-api-version"},
        "proxy": {
            "target_url": "http://upstream:80",
            "listen_path": "/test-api/",
            "strip_listen_path": True
        },
        "version_data": {
            "not_versioned": True,
            "versions": {
                "Default": {
                    "name": "Default",
                    "use_extended_paths": True
                }
            }
        },
        "active": True
    }
    
    r = httpx.post(
        f"{GATEWAY_API}/tyk/apis/",
        json=api_def,
        headers=tyk_headers()
    )
    assert r.status_code in (200, 201)
    reload_apis()
    yield
    # Cleanup: delete the API
    httpx.delete(f"{GATEWAY_API}/tyk/apis/test-api-id", headers=tyk_headers())

def test_valid_key_accesses_api():
    key = create_test_key(rate=100, per=60)
    r = httpx.get(
        f"{PROXY_URL}/test-api/get",
        headers={"Authorization": key}
    )
    assert r.status_code == 200

def test_invalid_key_rejected():
    r = httpx.get(
        f"{PROXY_URL}/test-api/get",
        headers={"Authorization": "invalid-key-xyz"}
    )
    assert r.status_code == 403

def test_missing_key_rejected():
    r = httpx.get(f"{PROXY_URL}/test-api/get")
    assert r.status_code == 401

def test_rate_limit_enforced():
    """Key with rate=3/per=60 should be blocked after 3 requests."""
    key = create_test_key(rate=3, per=60)
    headers = {"Authorization": key}
    
    for i in range(3):
        r = httpx.get(f"{PROXY_URL}/test-api/get", headers=headers)
        assert r.status_code == 200, f"Request {i+1} should succeed"
    
    # 4th request should be rate limited
    r = httpx.get(f"{PROXY_URL}/test-api/get", headers=headers)
    assert r.status_code == 429

def test_quota_enforced():
    """Key with quota_max=5 should stop working after 5 total requests."""
    payload = {
        "allowance": 1000,
        "rate": 1000,
        "per": 60,
        "expires": -1,
        "quota_max": 5,
        "quota_remaining": 5,
        "quota_renewal_rate": 3600,
        "access_rights": {
            "test-api-id": {
                "api_name": "Test API",
                "api_id": "test-api-id",
                "versions": ["Default"]
            }
        }
    }
    r = httpx.post(f"{GATEWAY_API}/tyk/keys/", json=payload, headers=tyk_headers())
    key = r.json()["key"]
    headers = {"Authorization": key}
    
    for i in range(5):
        r = httpx.get(f"{PROXY_URL}/test-api/get", headers=headers)
        assert r.status_code == 200
    
    # Quota exhausted
    r = httpx.get(f"{PROXY_URL}/test-api/get", headers=headers)
    assert r.status_code == 403
    assert "quota" in r.text.lower()

Middleware Testing

Tyk supports middleware in JavaScript (virtual endpoints), Go plugins, and Python plugins. Test each middleware type.

Virtual Endpoint (JavaScript) Testing

// middleware/add-correlation-id.js
function AddCorrelationId(request, session, config) {
  var correlationId = request.Headers["X-Correlation-Id"];
  
  if (!correlationId || correlationId.length === 0) {
    // Generate a simple ID for testing (in production, use UUID)
    correlationId = "gen-" + Math.random().toString(36).substr(2, 9);
  }
  
  request.SetHeaders["X-Correlation-Id"] = correlationId;
  
  return TykJsResponse({
    Body: request.Body,
    Headers: request.Headers,
    Code: 200
  }, session.meta_data);
}
def test_correlation_id_injected_when_missing():
    """Middleware adds correlation ID when not present in request."""
    key = create_test_key()
    r = httpx.get(
        f"{PROXY_URL}/test-api/echo-headers",
        headers={"Authorization": key}
    )
    assert r.status_code == 200
    # The echo endpoint returns headers it received
    received_headers = r.json()["headers"]
    assert "X-Correlation-Id" in received_headers
    assert received_headers["X-Correlation-Id"].startswith("gen-")

def test_correlation_id_preserved_when_present():
    """Middleware preserves existing correlation ID."""
    key = create_test_key()
    my_id = "my-custom-correlation-id"
    r = httpx.get(
        f"{PROXY_URL}/test-api/echo-headers",
        headers={"Authorization": key, "X-Correlation-Id": my_id}
    )
    received_headers = r.json()["headers"]
    assert received_headers["X-Correlation-Id"] == my_id

Request Transformation Testing

def test_request_header_injection():
    """API definition injects backend-specific headers before forwarding."""
    key = create_test_key()
    r = httpx.get(
        f"{PROXY_URL}/test-api/echo-headers",
        headers={"Authorization": key}
    )
    
    received = r.json()["headers"]
    # These should be injected by the API definition's header transform
    assert received.get("X-Internal-Service") == "users-api"
    assert received.get("X-Api-Version") == "v1"

def test_response_header_stripping():
    """API definition strips internal headers from responses."""
    key = create_test_key()
    r = httpx.get(f"{PROXY_URL}/test-api/get", headers={"Authorization": key})
    
    # These internal headers should not leak to clients
    assert "X-Backend-Server" not in r.headers
    assert "X-Internal-Trace" not in r.headers

API Versioning Tests

Tyk supports URL-based and header-based API versioning.

def test_v1_routes_to_v1_upstream():
    key = create_test_key()
    r = httpx.get(
        f"{PROXY_URL}/api/v1/users",
        headers={"Authorization": key}
    )
    assert r.status_code == 200
    # v1 upstream returns a "version" field
    assert r.json().get("version") == "v1"

def test_v2_routes_to_v2_upstream():
    key = create_test_key()
    r = httpx.get(
        f"{PROXY_URL}/api/v2/users",
        headers={"Authorization": key}
    )
    assert r.status_code == 200
    assert r.json().get("version") == "v2"

def test_deprecated_version_returns_warning_header():
    key = create_test_key()
    r = httpx.get(
        f"{PROXY_URL}/api/v1/users",
        headers={"Authorization": key}
    )
    # v1 is deprecated — should include deprecation warning
    assert "X-Api-Deprecated" in r.headers or "Deprecation" in r.headers

def test_unknown_version_rejected():
    key = create_test_key()
    r = httpx.get(
        f"{PROXY_URL}/api/v99/users",
        headers={"Authorization": key}
    )
    assert r.status_code == 404

URL Rewriting Tests

def test_listen_path_stripped():
    """Tyk strips /listen-path/ before forwarding to upstream."""
    key = create_test_key()
    r = httpx.get(
        f"{PROXY_URL}/test-api/get",
        headers={"Authorization": key}
    )
    
    # httpbin echo shows the URL it received
    assert r.status_code == 200
    # Should be /get, not /test-api/get
    assert r.json()["url"] == "http://upstream/get"

def test_url_rewrite_rule():
    """Tyk rewrites /v1/users/{id} to /users/{id} for upstream."""
    key = create_test_key()
    r = httpx.get(
        f"{PROXY_URL}/api/v1/users/123",
        headers={"Authorization": key}
    )
    
    received_path = r.json()["url"]
    assert "/users/123" in received_path
    assert "/v1/" not in received_path

CI Integration

# .github/workflows/tyk-tests.yml
name: Tyk Gateway Tests

on:
  pull_request:
    paths:
      - 'tyk/**'
      - 'tests/gateway/**'

jobs:
  test:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v4

      - name: Start Tyk
        run: docker compose -f docker-compose.test.yml up -d

      - name: Wait for Tyk
        run: |
          timeout 60 bash -c '
            until curl -sf http://localhost:8081/tyk/apis/ \
              -H "x-tyk-authorization: test-secret"; do
              sleep 2
            done
          '

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: Install dependencies
        run: pip install pytest httpx jsonschema

      - name: Run tests
        run: pytest tests/gateway/ -v

      - name: Collect logs on failure
        if: failure()
        run: docker compose -f docker-compose.test.yml logs tyk

      - name: Teardown
        if: always()
        run: docker compose -f docker-compose.test.yml down -v

Common Tyk Testing Pitfalls

Reload required after API changes — Tyk doesn't auto-reload when you modify API definitions via the Gateway API. Always call /tyk/reload/ and wait a second before running assertions.

Key scope mismatch — a key grants access to specific api_id values. If your test key doesn't include the API you're testing in its access_rights, you'll get a 403 that looks like an auth failure but is actually a scope issue. Log the full key payload to debug.

Rate limit window timing — rate limit windows are tied to calendar time, not request time. A test that fires 5 requests at second 59 of a minute window might see the window reset mid-test. Use short windows (per=10 seconds) in test configurations.

Redis state between tests — rate limit counters, quota usage, and session data all live in Redis. If you reuse the same API key across tests, state leaks between them. Create a fresh key per test that needs rate limit or quota testing.

listen_path trailing slash — Tyk requires listen_path to end with /. Missing it causes routing failures that manifest as 404s on all requests through that API.

Read more

Testing Atlantis Terraform PR Automation: Workflows, Plan Verification, and Policy Enforcement

Testing Atlantis Terraform PR Automation: Workflows, Plan Verification, and Policy Enforcement

Atlantis automates Terraform plan and apply through pull requests. But Atlantis itself needs testing: workflow configuration, plan output validation, policy enforcement, and server health checks. This guide covers testing Atlantis workflows locally with atlantis-local, validating plan outputs with custom scripts, enforcing Terraform policies with OPA and Conftest, and monitoring Atlantis

By HelpMeTest