simple-nanoshare/tests/test_database_error_handler.py

import asyncio
import importlib.util
import sys
import types
from pathlib import Path

import httpx
from quart import Blueprint, Quart, request

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

class FakeLimiter:
  def exempt(self, func):
    return func

  def limit(self, *_args, **_kwargs):
    def decorator(func):
      return func
    return decorator

class FakeLogger:
  def __init__(self):
    self.errors = []

  def error(self, message):
    self.errors.append(message)

  def warning(self, message):
    pass

class FailingConvex:
  async def get_current_favicon(self):
    raise httpx.ConnectError("[Errno -2] Name or service not known")

def load_module(module_name, module_path):
  spec = importlib.util.spec_from_file_location(module_name, module_path)
  module = importlib.util.module_from_spec(spec)
  sys.modules[module_name] = module
  spec.loader.exec_module(module)
  return module

def install_route_test_modules(monkeypatch, app, logger):
  fake_setup = types.ModuleType("my_modules.app.setup")
  fake_setup.app = app
  fake_setup.LIMITER = FakeLimiter()
  monkeypatch.setitem(sys.modules, "my_modules.app.setup", fake_setup)

  fake_constens = types.ModuleType("my_modules.app.constens")
  fake_constens.BLOCKED_IPS_ACCESSING_TIMES = 3
  fake_constens.BLOCKED_IPS_STORED_TIMEFRAME = 60
  monkeypatch.setitem(sys.modules, "my_modules.app.constens", fake_constens)

  fake_logger_module = types.ModuleType("my_modules.app.logger")
  fake_logger_module.logger = logger
  monkeypatch.setitem(sys.modules, "my_modules.app.logger", fake_logger_module)

  fake_functions = types.ModuleType("my_modules.functions")
  fake_functions.get_ip = lambda: "203.0.113.10"
  fake_functions.enforce_custom_limit = lambda *_args, **_kwargs: None
  fake_functions.get_request_context = lambda: types.SimpleNamespace(path=request.path)
  fake_functions.is_valid_uuid = lambda value: True
  monkeypatch.setitem(sys.modules, "my_modules.functions", fake_functions)

def register_template_routes(app):
  side_main = Blueprint("side_main", __name__)

  async def index():
    return "ok"

  side_main.add_url_rule("/", "index", index)
  side_main.add_url_rule("/files", "files_list", index)
  side_main.add_url_rule("/access", "access_list", index)
  app.register_blueprint(side_main)

  auth_login = Blueprint("auth_login", __name__)
  auth_login.add_url_rule("/login", "login", index)
  auth_login.add_url_rule("/logout", "logout", index)
  app.register_blueprint(auth_login)

  link_composer = Blueprint("link_composer", __name__)
  link_composer.add_url_rule("/links", "link_composer_page", index)
  app.register_blueprint(link_composer)


def load_errors_and_basics(monkeypatch, app):
  logger = FakeLogger()
  install_route_test_modules(monkeypatch, app, logger)
  root = Path(__file__).resolve().parents[1]
  errors = load_module("test_routes_handeling_errorsAndBots", root / "routes" / "handeling" / "errorsAndBots.py")
  basics = load_module("test_routes_handeling_basics", root / "routes" / "handeling" / "basics.py")
  return errors, basics, logger

def test_convex_connect_error_is_returned_as_global_database_error(monkeypatch):
  async def run_test():
    app = Quart(__name__, template_folder=str(Path(__file__).resolve().parents[1] / "templates" / "side"))
    register_template_routes(app)
    app.convex = FailingConvex()
    _errors, basics, logger = load_errors_and_basics(monkeypatch, app)
    app.register_blueprint(basics.basic_bp)

    response = await app.test_client().get("/favicon.ico")

    assert response.status_code == 503
    assert response.headers["Retry-After"]
    assert logger.errors == []

  asyncio.run(run_test())

def test_convex_connect_error_uses_legacy_logger_when_wide_events_are_disabled(monkeypatch):
  async def run_test():
    monkeypatch.setenv("WIDE_EVENT_ENABLED", "false")
    app = Quart(__name__, template_folder=str(Path(__file__).resolve().parents[1] / "templates" / "side"))
    register_template_routes(app)
    app.convex = FailingConvex()
    _errors, basics, logger = load_errors_and_basics(monkeypatch, app)
    app.register_blueprint(basics.basic_bp)

    response = await app.test_client().get("/favicon.ico")

    assert response.status_code == 503
    assert any("Name or service not known" in str(error) for error in logger.errors)

  asyncio.run(run_test())

def test_api_convex_connect_error_returns_json_database_error(monkeypatch):
  async def run_test():
    app = Quart(__name__, template_folder=str(Path(__file__).resolve().parents[1] / "templates" / "side"))
    load_errors_and_basics(monkeypatch, app)

    @app.get("/api/failing")
    async def failing_api():
      raise httpx.ConnectError("[Errno -2] Name or service not known")

    response = await app.test_client().get("/api/failing")
    payload = await response.get_json()

    assert response.status_code == 503
    assert payload["error"] == "Database Error"

  asyncio.run(run_test())

def test_api_convex_timeout_returns_gateway_timeout(monkeypatch):
  async def run_test():
    app = Quart(__name__, template_folder=str(Path(__file__).resolve().parents[1] / "templates" / "side"))
    load_errors_and_basics(monkeypatch, app)

    @app.get("/api/slow")
    async def slow_api():
      raise httpx.ReadTimeout("convex took too long")

    response = await app.test_client().get("/api/slow")
    payload = await response.get_json()

    assert response.status_code == 504
    assert payload["error"] == "Database Error"

  asyncio.run(run_test())