feat(logging): add NanoShare wide event instrumentation
Build and Push Docker Container / build-and-push (push) Failing after 51s
Build and Push Docker Container / build-and-push (push) Failing after 51s
- Register quart_common wide-event logging during app setup so every HTTP request emits one canonical structured event. - Replace the inline security middleware with reusable quart_common security middleware wiring and move skip path configuration into app constants. - Add NanoShare-specific wide-event context for health checks, auth/error handlers, file list/edit/delete/serve flows and upload outcomes. - Rename runtime logging/project metadata from simple-picoshare to nanoshare where it is emitted in service context. - Update my_helpers and quart_common submodules for Convex/wide-event integration and reusable security middleware support. - Add NanoShare middleware tests covering safe user context, client IP enrichment, missing Convex handling and Convex security lookup failures.
This commit is contained in:
@@ -0,0 +1,145 @@
|
||||
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 g.wide_event["user"]["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())
|
||||
Reference in New Issue
Block a user