Created
April 26, 2023 07:08
-
-
Save sveniu/75161d7d8d55975af8a3a97718eeb1ac to your computer and use it in GitHub Desktop.
httpbin-like client to test CDN caching and stale-while-revalidate
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python3 | |
import sys | |
import time | |
import urllib.request | |
from threading import Thread | |
class fgcolor: | |
green = "\x1b[32;20m" | |
yellow = "\x1b[33;20m" | |
red = "\x1b[31;20m" | |
reset = "\x1b[0m" | |
class IgnoreErrors(urllib.request.HTTPErrorProcessor): | |
def http_response(self, request, response): | |
return response | |
https_response = http_response | |
def fetch(url): | |
t_start = time.time() | |
opener = urllib.request.build_opener(IgnoreErrors) | |
req = urllib.request.Request(url) | |
with opener.open(req) as response: | |
ttfb = time.time() - t_start | |
ttfb_color = fgcolor.green if ttfb < 0.5 else fgcolor.red | |
status = response.status | |
status_color = fgcolor.green if 200 <= status < 300 else fgcolor.red | |
# Extract some headers. | |
age = response.getheader("age", "") | |
x_cache = response.getheader("x-cache", "") | |
# Extract some info from the body. | |
body = response.read() | |
body_len = len(body) | |
generated_by = "origin" | |
if b"Generated by cloudfront" in body: | |
generated_by = "CloudFront" | |
generated_by_color = fgcolor.green if generated_by == "origin" else fgcolor.red | |
print( | |
f"ttfb={ttfb_color}{ttfb:5.3f}{fgcolor.reset}" | |
f" status={status_color}{status}{fgcolor.reset}" | |
f" age={age}" | |
f" bytes={body_len}" | |
f" generated_by={generated_by_color}{generated_by}{fgcolor.reset}" | |
f" x-cache={x_cache}" | |
) | |
url = "https://FIXME.cloudfront.net/?wait_ms=2500&header=cache-control=max-age=10,stale-while-revalidate=27&status_code_after_ms=9000&status_code_after=502&session_ttl_ms=300000&session_id=test001" | |
if len(sys.argv) > 1: | |
url = sys.argv[1] | |
while True: | |
Thread(target=fetch, args=[url]).start() | |
time.sleep(0.1) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment