feat: Quart app factory with health endpoint

This commit is contained in:
2026-03-17 18:55:15 -04:00
parent ff8489f758
commit 5c25953929
9 changed files with 161 additions and 0 deletions
+48
View File
@@ -0,0 +1,48 @@
from __future__ import annotations
from pathlib import Path
from quart import Quart
from .config import load_config
from .database import init_db
def create_app(
config_path: Path | str | None = None,
testing: bool = False,
) -> Quart:
app = Quart(__name__, template_folder="templates")
# Load config
cfg = load_config(config_path)
app.config.update(
SECRET_KEY=cfg["secret_key"],
DATABASE_URL=cfg["database"]["url"],
SESSION_LIFETIME_HOURS=cfg.get("session", {}).get("lifetime_hours", 8),
DATA_RETENTION_DAYS=cfg.get("data", {}).get("retention_days", 90),
PLUGIN_DIR=cfg.get("plugin_dir", "plugins"),
PLUGINS=cfg.get("plugins", {}),
SMTP=cfg.get("smtp", {}),
WEBHOOK=cfg.get("webhook", {}),
TESTING=testing,
_RAW_CFG=cfg,
)
# Database
if not testing:
init_db(app)
else:
from unittest.mock import MagicMock
app.db_sessionmaker = MagicMock()
# Register blueprints
from .auth.routes import auth_bp
app.register_blueprint(auth_bp)
from .dashboard.routes import dashboard_bp
app.register_blueprint(dashboard_bp)
# Health check
@app.get("/health")
async def health():
return {"status": "ok"}
return app
View File
+37
View File
@@ -0,0 +1,37 @@
from __future__ import annotations
import functools
from quart import session, redirect, url_for, abort
from fablednetmon.models.users import UserRole
_ROLE_ORDER = [UserRole.viewer, UserRole.operator, UserRole.admin]
def require_role(minimum_role: UserRole):
"""Decorator: requires authenticated user with at least minimum_role."""
def decorator(f):
@functools.wraps(f)
async def wrapper(*args, **kwargs):
user_id = session.get("user_id")
if not user_id:
return redirect(url_for("auth.login"))
user_role = UserRole(session.get("user_role", "viewer"))
if _ROLE_ORDER.index(user_role) < _ROLE_ORDER.index(minimum_role):
abort(403)
return await f(*args, **kwargs)
return wrapper
return decorator
def login_user(user) -> None:
"""Write user identity into the session."""
session["user_id"] = user.id
session["user_role"] = user.role.value
session["username"] = user.username
def logout_user() -> None:
session.clear()
def current_user_id() -> str | None:
return session.get("user_id")
+3
View File
@@ -0,0 +1,3 @@
from quart import Blueprint
auth_bp = Blueprint("auth", __name__)
+14
View File
@@ -0,0 +1,14 @@
from pathlib import Path
import click
from .app import create_app
@click.command()
@click.option("--config", default="config.yaml", help="Path to config.yaml")
@click.option("--host", default="0.0.0.0")
@click.option("--port", default=5000, type=int)
@click.option("--debug", is_flag=True, default=False)
def main(config: str, host: str, port: int, debug: bool) -> None:
"""Start the FabledNetMon server."""
app = create_app(config_path=Path(config))
app.run(host=host, port=port, debug=debug)
View File
+11
View File
@@ -0,0 +1,11 @@
from quart import Blueprint, render_template
from fablednetmon.auth.middleware import require_role
from fablednetmon.models.users import UserRole
dashboard_bp = Blueprint("dashboard", __name__)
@dashboard_bp.get("/")
@require_role(UserRole.viewer)
async def index():
return await render_template("dashboard/index.html")
+27
View File
@@ -0,0 +1,27 @@
import pytest
import pytest_asyncio
from pathlib import Path
import textwrap
from fablednetmon.app import create_app
@pytest.fixture
def config_file(tmp_path):
f = tmp_path / "config.yaml"
f.write_text(textwrap.dedent("""\
secret_key: test-secret-key-32-chars-minimum!!
database:
url: postgresql+asyncpg://test:test@localhost/test
"""))
return f
@pytest_asyncio.fixture
async def app(config_file):
application = create_app(config_path=config_file, testing=True)
return application
@pytest_asyncio.fixture
async def client(app):
return app.test_client()
+21
View File
@@ -0,0 +1,21 @@
import pytest
from fablednetmon.app import create_app
@pytest.mark.asyncio
async def test_create_app_returns_quart_app(config_file):
from quart import Quart
app = create_app(config_path=config_file, testing=True)
assert isinstance(app, Quart)
@pytest.mark.asyncio
async def test_app_has_db_sessionmaker(config_file):
app = create_app(config_path=config_file, testing=True)
assert hasattr(app, "db_sessionmaker")
@pytest.mark.asyncio
async def test_health_endpoint(client):
response = await client.get("/health")
assert response.status_code == 200