"""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)