281 lines
8.8 KiB
Python
281 lines
8.8 KiB
Python
"""Standalone /convert smoke checks without third-party packages.
|
|
|
|
Run: python3 tests/smoke_cache_routes.py
|
|
These are stubs, not a substitute for an installed Flask/Gunicorn integration test.
|
|
"""
|
|
import hashlib
|
|
import importlib.util
|
|
import logging
|
|
import sys
|
|
import types
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
|
|
class HTTPError(Exception):
|
|
def __init__(self, code, description=None):
|
|
self.code = code
|
|
self.description = description
|
|
|
|
|
|
class Response:
|
|
def __init__(self, data=None, content_type=None, status=200, mimetype=None):
|
|
self.data = data
|
|
self.status_code = status
|
|
self.headers = {}
|
|
self.direct_passthrough = False
|
|
self.close_callbacks = []
|
|
if content_type:
|
|
self.headers["Content-Type"] = content_type
|
|
|
|
def call_on_close(self, callback):
|
|
self.close_callbacks.append(callback)
|
|
|
|
def make_conditional(self, request):
|
|
if request.headers.get("If-None-Match") == self.headers.get("ETag"):
|
|
self.status_code = 304
|
|
return self
|
|
|
|
|
|
class Flask:
|
|
def __init__(self, *args):
|
|
self.config = {}
|
|
self.debug = False
|
|
self.logger = logging.getLogger("adlist-test")
|
|
|
|
def route(self, *args, **kwargs):
|
|
return lambda func: func
|
|
|
|
template_filter = route
|
|
before_request = lambda self, fn: fn
|
|
after_request = lambda self, fn: fn
|
|
errorhandler = route
|
|
|
|
|
|
def abort(code, description=None):
|
|
raise HTTPError(code, description)
|
|
|
|
|
|
flask = types.ModuleType("flask")
|
|
flask.Flask = Flask
|
|
flask.Response = Response
|
|
flask.abort = abort
|
|
flask.render_template = lambda *args, **kwargs: "error-page"
|
|
flask.jsonify = lambda d: d
|
|
flask.g = types.SimpleNamespace()
|
|
flask.request = types.SimpleNamespace(
|
|
args={}, headers={}, method="GET", remote_addr="127.0.0.1",
|
|
full_path="/convert", path="/convert",
|
|
)
|
|
sys.modules["flask"] = flask
|
|
|
|
compress = types.ModuleType("flask_compress")
|
|
compress.Compress = lambda app: None
|
|
sys.modules["flask_compress"] = compress
|
|
|
|
limiter = types.ModuleType("flask_limiter")
|
|
|
|
|
|
class Limiter:
|
|
def __init__(self, **kwargs):
|
|
pass
|
|
|
|
def limit(self, *args):
|
|
return lambda func: func
|
|
|
|
|
|
limiter.Limiter = Limiter
|
|
sys.modules["flask_limiter"] = limiter
|
|
|
|
|
|
class FakeRedis:
|
|
def incr(self, *args):
|
|
pass
|
|
|
|
def incrby(self, *args):
|
|
pass
|
|
|
|
def incrbyfloat(self, *args):
|
|
pass
|
|
|
|
def get(self, *args):
|
|
return None
|
|
|
|
def set(self, *args):
|
|
pass
|
|
|
|
|
|
redis = types.ModuleType("redis")
|
|
redis.Redis = types.SimpleNamespace(from_url=lambda *args, **kwargs: FakeRedis())
|
|
sys.modules["redis"] = redis
|
|
|
|
asgiref = types.ModuleType("asgiref")
|
|
asgi_w = types.ModuleType("asgiref.wsgi")
|
|
asgi_w.WsgiToAsgi = lambda app: app
|
|
sys.modules["asgiref"] = asgiref
|
|
sys.modules["asgiref.wsgi"] = asgi_w
|
|
|
|
requests = types.ModuleType("requests")
|
|
|
|
|
|
class RequestException(Exception):
|
|
pass
|
|
|
|
|
|
requests.exceptions = types.SimpleNamespace(RequestException=RequestException)
|
|
sys.modules["requests"] = requests
|
|
|
|
spec = importlib.util.spec_from_file_location("adlist_cache_test_app", ROOT / "app.py")
|
|
app = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(app)
|
|
app.add_recent_convert = lambda: None
|
|
app.track_url_request = lambda *_: None
|
|
app.add_recent_link = lambda *_: None
|
|
|
|
|
|
class FakeUpstream:
|
|
def __init__(self, body, status=200, content_type="text/plain; charset=utf-8",
|
|
headers=None, fail=False):
|
|
self.body = body
|
|
self.status_code = status
|
|
self.headers = {"Content-Type": content_type, **(headers or {})}
|
|
self.closed = False
|
|
self.fail = fail
|
|
|
|
def iter_content(self, chunk_size):
|
|
for i in range(0, len(self.body), chunk_size):
|
|
yield self.body[i:i+chunk_size]
|
|
if self.fail:
|
|
raise RequestException("upstream disconnected")
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
|
|
called = []
|
|
source = FakeUpstream(b"0.0.0.0 example.test\n")
|
|
|
|
|
|
def fake_get(*args, **kwargs):
|
|
called.append((args, kwargs))
|
|
return source
|
|
|
|
|
|
requests.get = fake_get
|
|
count = 0
|
|
|
|
|
|
def check(test, label):
|
|
global count
|
|
assert test, label
|
|
count += 1
|
|
|
|
|
|
def run(method="GET", client_headers=None, args=None):
|
|
flask.request.method = method
|
|
flask.request.args = args or {
|
|
"url": "https://source.test/list.txt", "ip": "0.0.0.0"}
|
|
flask.request.headers = client_headers or {}
|
|
flask.g.start_time = 0.0
|
|
return app.convert()
|
|
|
|
|
|
converted = b"0.0.0.0 example.test\n"
|
|
content_hash = hashlib.sha256(converted).hexdigest()
|
|
resp = run()
|
|
check(resp.status_code == 200, "valid list = 200")
|
|
check(resp.headers["Cache-Control"] ==
|
|
"public, max-age=3600, s-maxage=604800, stale-while-revalidate=86400, stale-if-error=604800",
|
|
"proxy TTL and staleness")
|
|
check(resp.headers["ETag"] == f'"{content_hash}"', "strong output-body ETag")
|
|
check(resp.headers["Content-Length"] == str(len(converted)), "content length")
|
|
check(b"".join(resp.data) == converted, "response bytes")
|
|
check(source.closed, "upstream closed after success")
|
|
check("Last-Modified" not in resp.headers, "origin validator is not forwarded")
|
|
|
|
source = FakeUpstream(converted, headers={"ETag": '"upstream-pretend"',
|
|
"Last-Modified": "Mon, 01 Jan 2024 00:00:00 GMT"})
|
|
resp = run(client_headers={"If-None-Match": f'"{content_hash}"'})
|
|
check(resp.status_code == 304, "client 304 after full body hash")
|
|
check("headers" not in called[-1][1], "client validators not forwarded to upstream")
|
|
check(source.closed, "conditional upstream closed")
|
|
check(resp.headers["ETag"] == f'"{content_hash}"', "304 keeps content validator")
|
|
|
|
source = FakeUpstream(b"0.0.0.0 another.test\n", headers={"ETag": '"upstream-pretend"'})
|
|
changed = run()
|
|
check(changed.headers["ETag"] != f'"{content_hash}"',
|
|
"changed converted content invalidates ETag even if source ETag stays the same")
|
|
check(b"".join(changed.data) == b"0.0.0.0 another.test\n", "changed content served")
|
|
|
|
source = FakeUpstream(converted, headers={"ETag": '"upstream-pretend"'})
|
|
resp = run(client_headers={"If-None-Match": '"other"'})
|
|
check(resp.status_code == 200, "wrong client ETag yields 200")
|
|
check(b"".join(resp.data) == converted, "wrong client ETag still receives body")
|
|
|
|
source = FakeUpstream(converted)
|
|
head = run("HEAD")
|
|
check(head.status_code == 200, "HEAD status")
|
|
check(head.data is None, "HEAD no data iterator")
|
|
check(head.headers["ETag"] == f'"{content_hash}"', "GET/HEAD same ETag")
|
|
check(head.headers["Content-Length"] == str(len(converted)), "HEAD real content length")
|
|
|
|
source = FakeUpstream(converted)
|
|
other = run(args={"url": "https://source.test/list.txt", "ip": "127.0.0.1"})
|
|
check(other.headers["ETag"] != resp.headers["ETag"], "different target IP yields different ETag")
|
|
check(b"127.0.0.1 example.test\n" == b"".join(other.data), "IP body changed")
|
|
|
|
source = FakeUpstream(converted)
|
|
app.config.DEBUG_ENABLE = True
|
|
app.config.DEBUG_KEY = "not-test-key"
|
|
debug = run(args={"url": "https://source.test/list.txt", "ip": "0.0.0.0", "debug": "1"})
|
|
check(debug.headers["Cache-Control"] == "private, no-store", "debug never cached")
|
|
check("ETag" not in debug.headers, "debug no shared ETag")
|
|
check(debug.status_code == 200, "debug displays full response")
|
|
check(debug.headers["X-Debug-Mode"] == "1", "debug diagnostic header")
|
|
app.config.DEBUG_ENABLE = False
|
|
|
|
source = FakeUpstream(b"<html><body>err</body></html>")
|
|
try:
|
|
run()
|
|
raise AssertionError("invalid body accepted")
|
|
except HTTPError as e:
|
|
check(e.code == 422 and "Unable to decode" in e.description, "unprocessable source English 422")
|
|
|
|
source = FakeUpstream(b"<html></html>", content_type="text/html")
|
|
try:
|
|
run()
|
|
raise AssertionError("html content type accepted")
|
|
except HTTPError as e:
|
|
check(e.code == 415 and "supported" in e.description, "unsupported type English 415")
|
|
|
|
source = FakeUpstream(b"not-found", status=404)
|
|
try:
|
|
run()
|
|
raise AssertionError("bad upstream status accepted")
|
|
except HTTPError as e:
|
|
check(e.code == 502 and "source server" in e.description, "upstream English 502")
|
|
|
|
source = FakeUpstream(b"0.0.0.0 example.test\n", fail=True)
|
|
try:
|
|
run()
|
|
raise AssertionError("interrupted source accepted")
|
|
except HTTPError as e:
|
|
check(e.code == 502 and "interrupted" in e.description, "interrupted download 502")
|
|
|
|
r = Response(status=422)
|
|
r.headers["Cache-Control"] = "public, max-age=90000"
|
|
for error_code in [400, 403, 413, 415, 422, 429, 500, 502]:
|
|
error = Response(status=error_code)
|
|
error.headers["Cache-Control"] = "public, max-age=90000"
|
|
check(app.finalize_response(error).headers["Cache-Control"] == "no-store",
|
|
f"HTTP {error_code} must not be cached")
|
|
|
|
test_cfg = app.config.CACHE_ENABLED
|
|
app.config.CACHE_ENABLED = False
|
|
check(app.cache_headers(content_hash)["Cache-Control"] == "no-store", "cache disabled")
|
|
app.config.CACHE_ENABLED = test_cfg
|
|
|
|
print("Standalone route/cache smoke tests passed:", count)
|