import asyncio import importlib import sys import types from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from quart import Quart, g, session class FakeLogger: def __init__(self): self.records = [] async def info(self, message): self.records.append(("info", message)) async def warning(self, message): self.records.append(("warning", message)) async def error(self, message): self.records.append(("error", message)) class FakeIPBotManager: def is_client_ip_always_allowed(self, ip): return False class FakeSetupApp: def before_request(self, func): return func def context_processor(self, func): return func class FakeConvex: async def is_ip_address_whitelisted_or_blocked(self, ip_address): return {"whiteliste": False, "blocked": False} async def is_path_blocked(self, path): return False class FailingConvex: async def is_ip_address_whitelisted_or_blocked(self, ip_address): raise RuntimeError("convex down") def load_middleware(monkeypatch, client_ip="203.0.113.10", logger=None): fake_errors = types.ModuleType("routes.handeling.errorsAndBots") async def maybe_a_hacker(): return "blocked", 418 fake_errors.maybe_a_hacker = maybe_a_hacker fake_constens = types.ModuleType("my_modules.app.constens") fake_constens.THE_IP_BOT_MANAGER = FakeIPBotManager() fake_constens.SKIP_PATH_PREFIXES = ("/static", "/storage") fake_constens.SKIP_PATHS = ("/favicon.ico",) fake_logger_module = types.ModuleType("my_modules.app.logger") fake_logger_module.logger = logger or FakeLogger() fake_functions = types.ModuleType("my_modules.functions") fake_functions.get_ip = lambda: client_ip fake_setup = types.ModuleType("my_modules.app.setup") fake_setup.app = FakeSetupApp() monkeypatch.setitem(sys.modules, "routes.handeling.errorsAndBots", fake_errors) monkeypatch.setitem(sys.modules, "my_modules.app.constens", fake_constens) monkeypatch.setitem(sys.modules, "my_modules.app.logger", fake_logger_module) monkeypatch.setitem(sys.modules, "my_modules.functions", fake_functions) monkeypatch.setitem(sys.modules, "my_modules.app.setup", fake_setup) sys.modules.pop("my_modules.middleware", None) return importlib.import_module("my_modules.middleware") def test_middleware_adds_user_and_client_context(monkeypatch): async def run_test(): middleware = load_middleware(monkeypatch) monkeypatch.setattr(middleware, "get_ip", lambda: "203.0.113.10") monkeypatch.setattr(middleware, "logger", FakeLogger()) app = Quart(__name__) app.secret_key = "test-secret" app.convex = FakeConvex() async with app.test_request_context("/files"): g.wide_event = {} session["user"] = {"sub": "user_123", "preferred_username": "demo"} session["login_at"] = 1 result = await middleware.custom_middleware() assert result is None assert g.wide_event["user"]["id"] == "user_123" assert g.wide_event["user"]["name"] == "demo" assert g.wide_event["user"]["authenticated"] is True assert "authenticated" not in g.wide_event["session"] assert g.wide_event["session"]["permanent"] is True assert g.wide_event["session"]["login_at_unix"] == 1 assert g.wide_event["session"]["age_seconds"] >= 0 assert g.wide_event["client"]["ip"] == "203.0.113.10" asyncio.run(run_test()) def test_middleware_marks_missing_convex_as_skipped(monkeypatch): async def run_test(): middleware = load_middleware(monkeypatch, client_ip="203.0.113.11") monkeypatch.setattr(middleware, "logger", FakeLogger()) app = Quart(__name__) app.secret_key = "test-secret" async with app.test_request_context("/files"): g.wide_event = {} result = await middleware.custom_middleware() assert result is None assert g.wide_event["client"]["ip"] == "203.0.113.11" assert g.wide_event["security"] == { "convex_missing": True, "middleware_skipped": True, } asyncio.run(run_test()) def test_middleware_records_convex_security_failures(monkeypatch): async def run_test(): fake_logger = FakeLogger() middleware = load_middleware(monkeypatch, client_ip="203.0.113.12", logger=fake_logger) app = Quart(__name__) app.secret_key = "test-secret" app.convex = FailingConvex() async with app.test_request_context("/files"): g.wide_event = {} result = await middleware.custom_middleware() assert result is None assert g.wide_event["client"]["ip"] == "203.0.113.12" assert g.wide_event["security"] == {"ip_lookup_failed": True} assert g.wide_event["error"]["type"] == "RuntimeError" assert fake_logger.records == [("error", "[MIDDLEWARE] Convex ip_lookup failed: convex down")] asyncio.run(run_test())