Python script to test Kong's ACL plugin caching

kong-cache-test.py

import requests
import time
import uuid

KONG_ADMIN_HOST = "http://localhost:8001"
KONG_PROXY_HOST = "http://localhost:8000"
TEST_UPSTREAM_URL = "https://mockbin.org/bin/800a818b-5fb6-40d4-a342-75a1fb8599db/"
USERNAME = uuid.uuid4().hex
ATTEMPTS = 10

# Create consumer
url = "{}/consumers/?username={}".format(KONG_ADMIN_HOST, USERNAME)
requests.post(url)

# Add key-auth to consumer, retrieve token
url = "{}/consumers/{}/key-auth".format(KONG_ADMIN_HOST, USERNAME)
res = requests.post(url)
token = res.json()['key']
print("Consumer token: {}".format(token))

# Repeatedly create new APIs w/ ACL and test them
for n in range(ATTEMPTS):
    project_id = uuid.uuid4().hex
    api_name = "project-{}".format(project_id)
    acl_group_name = "project-{}-group".format(project_id)
    request_path = "/projects/{}/".format(project_id)

    # Create new API
    url = "{}/apis".format(KONG_ADMIN_HOST)
    requests.post(url, data={
        'name': api_name,
        'upstream_url': TEST_UPSTREAM_URL,
        'request_path': request_path
    })

    # Enable ACL plugin on new API
    url = "{}/apis/{}/plugins".format(KONG_ADMIN_HOST, api_name)
    requests.post(url, data={
        'name': 'acl',
        'config.whitelist': acl_group_name
    })

    # Enable key-auth on new API
    url = "{}/apis/{}/plugins".format(KONG_ADMIN_HOST, api_name)
    requests.post(url, data={
        'name': 'key-auth',
        'config.key_names': 'my-api-token'
    })

    # Add new ACL group to consumer
    url = "{}/consumers/{}/acls".format(KONG_ADMIN_HOST, USERNAME)
    requests.post(url, data={
        'group': acl_group_name
    })

    # Enabling the following code to force a cache refresh fixes the problem:
    # url = "{}/cache".format(KONG_ADMIN_HOST)
    # requests.delete(url)

    # Verify that consumer can authenticate & authorize new API
    url = "{}{}".format(KONG_PROXY_HOST, request_path)
    res = requests.get(url, headers={'my-api-token': token})
    print("Response from {}: {} ({})".format(
        request_path, res.content, res.status_code))

    # Try again in one second
    time.sleep(1)