simple-nanoshare/tests/test_wide_event_nanoshare.py

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