quality/testdays-web
6
0
Fork 0
Code Issues 9 Pull requests Projects Releases Packages Wiki Activity Actions

Refactor app to use application factory pattern
All checks were successful
Run tests / test (pull_request) Successful in 43s
Details

Browse source 
Fixes #110

Assisted-By: Claude Code
This commit is contained in:
Jaroslav Groman 2026-05-07 09:32:23 +02:00
commit b9a12b250e
16 changed files with 441 additions and 390 deletions
Download patch file Download diff file Expand all files Collapse all files

4
.gitignore vendored
View file

@ -17,6 +17,10 @@
.devcontainer
.opencode
# Some skills from https://github.com/obra/superpowers might use this folder
# to store plan files
/docs/superpowers
/env
conf/client_secrets.json
conf/*settings.py

17
alembic/env.py
View file

@ -17,7 +17,8 @@ fileConfig(config.config_file_name, disable_existing_loggers=False)
# add your model's MetaData object here
# for 'autogenerate' support
from testdays import db
from testdays.extensions import db
import testdays.models # noqa: F401 pylint: disable=unused-import
target_metadata = db.metadata
#target_metadata = None
@ -53,8 +54,18 @@ def run_migrations_online():
"""
alembic_config = config.get_section(config.config_ini_section)
from testdays import app
alembic_config['sqlalchemy.url'] = app.config['SQLALCHEMY_DATABASE_URI']
# Use the existing app if we're already inside an app context (e.g. when
# called from test fixtures or from create_app's dev auto-migrate).
# Otherwise, create a new app to read the database URI from config.
from flask import current_app
try:
db_uri = current_app.config['SQLALCHEMY_DATABASE_URI']
except RuntimeError:
from testdays.app import create_app
app = create_app()
db_uri = app.config['SQLALCHEMY_DATABASE_URI']
alembic_config['sqlalchemy.url'] = db_uri
engine = engine_from_config(
alembic_config,

4
conf/testdays.wsgi
View file

@ -7,4 +7,6 @@
import os
os.environ['TESTDAYS_CONFIG'] = '/etc/testdays/settings.py'
from testdays import app as application
from testdays import create_app
application = create_app()

14
runapp.py
View file

@ -21,13 +21,13 @@
# Authors:
# Josef Skladanka <jskladan@redhat.com>
import os
import testdays
from testdays import create_app
if __name__ == '__main__':
testdays.app.run(
host = testdays.app.config['RUN_HOST'],
port = testdays.app.config['RUN_PORT'],
debug = testdays.app.config['DEBUG'],
)
app = create_app()
app.run(
host=app.config['RUN_HOST'],
port=app.config['RUN_PORT'],
debug=app.config['DEBUG'],
)

285
testdays/__init__.py
View file

@ -1,286 +1,5 @@
from datetime import datetime
import logging
import logging.handlers
import os
from typing import Optional
import jinjax
from flask import Flask
from flask_login import LoginManager, current_user
from flask_oidc import OpenIDConnect
from flask_sqlalchemy import SQLAlchemy
from flask_simple_captcha import CAPTCHA
from git import Repo
from git.exc import InvalidGitRepositoryError, NoSuchPathError, GitCommandError
from . import config
"""Fedora Testdays web application."""
__version__ = "2.0.0"
# Flask App
app = Flask(
__name__,
static_url_path="",
static_folder="static",
)
app.jinja_env.trim_blocks = True
app.jinja_env.lstrip_blocks = True
# Load default config (Production)
# * update with values from a config file
# * update with values from environment variables
config_obj = config.Production()
config_file = "/etc/testdays/settings.py"
if os.getenv("RUNMODE") == "dev" or os.getenv("DEV") == "true":
config_obj = config.Development()
config_file = f"{os.getcwd()}/conf/settings.py"
elif os.getenv("RUNMODE") == "test" or os.getenv("TEST") == "true":
config_obj = config.Production()
config_file = None
config_file = os.environ.get("FORCE_CONFIG_FILE", config_file)
if config_file and os.path.exists(config_file):
config_obj.update_from_pyfile(config_file)
# Is this an OpenShift deployment?
if os.getenv("IS_OPENSHIFT"):
config.check_required_openshift_envvars()
config_obj.OIDC_CLIENT_SECRETS = "/opt/app-root/secret/client_secrets.json"
# config_obj.OIDC_SCOPES = "openid email profile https://id.fedoraproject.org/scope/fas-attributes"
config_obj.update_from_envvars()
app.config.from_object(config_obj)
# Logging
fmt = "%(asctime)s [%(levelname)8s] (%(filename)s:%(lineno)s) %(message)s"
datefmt = "%Y-%m-%d %H:%M:%S"
loglevel = logging.DEBUG if app.debug else logging.INFO
formatter = logging.Formatter(fmt=fmt, datefmt=datefmt)
def _add_log_handlers(logger, level=logging.INFO):
"""Add stream/syslog/file handlers to a logger based on app config."""
if app.config["STREAM_LOGGING"]:
h = logging.StreamHandler()
h.setLevel(level)
h.setFormatter(formatter)
logger.addHandler(h)
if app.config["SYSLOG_LOGGING"] and os.path.exists("/dev/log"):
h = logging.handlers.SysLogHandler(
address="/dev/log", facility=logging.handlers.SysLogHandler.LOG_LOCAL4
)
h.setLevel(level)
h.setFormatter(formatter)
logger.addHandler(h)
if app.config["FILE_LOGGING"] and app.config["LOGFILE"]:
h = logging.handlers.RotatingFileHandler(
app.config["LOGFILE"], maxBytes=5000000, backupCount=5
)
h.setLevel(level)
h.setFormatter(formatter)
logger.addHandler(h)
# Root logger (for third-party libraries)
root_logger = logging.getLogger()
root_logger.handlers.clear()
root_logger.setLevel(loglevel)
_add_log_handlers(root_logger, loglevel)
# Suppress Werkzeug's per-request log lines -- they embed their own timestamp
# which duplicates our formatter's timestamp. Request logging is handled by
# our own after_request hook below instead. We use a filter rather than
# raising the log level so that startup messages ("Running on ...") and
# warnings ("This is a development server ...") still come through.
class _WerkzeugRequestFilter(logging.Filter):
"""Drop Werkzeug's per-request log lines but keep startup/error messages."""
def filter(self, record):
msg = record.getMessage()
# Werkzeug request lines look like:
# '10.0.0.1 - - [01/Apr/2026 12:00:00] "GET / HTTP/1.1" 200 -'
# The ' - - [' timestamp bracket pattern uniquely identifies request
# lines. Startup messages (' * Running on ...') never contain it.
# Note: we match on ' - - [' alone rather than also checking for HTTP
# methods, because werkzeug injects ANSI color codes around the method
# (e.g. '\x1b[36mGET ... \x1b[0m') which break literal string matching.
if " - - [" in msg:
return False
return True
logging.getLogger("werkzeug").addFilter(_WerkzeugRequestFilter())
# App logger
app.logger.handlers.clear()
app.logger.propagate = False
app.logger.setLevel(loglevel)
_add_log_handlers(app.logger, loglevel)
app.secret_key = app.config["SECRET_KEY"]
if app.config["PRODUCTION"]:
if app.secret_key == "not-really-a-secret":
raise Warning("You need to change the app.secret_key value for production")
oidc = OpenIDConnect(app, prefix="/flask_oidc")
git_short_hash: str = "unknown"
git_date: Optional[datetime] = None
try:
# Non-existent repo may cause InvalidGitRepositoryError or NoSuchPathError
repo = Repo(os.path.dirname(os.path.dirname(__file__)))
# Empty repo may cause ValueError or GitCommandError
git_short_hash = repo.git.rev_parse(repo.head, short=True)
git_date = repo.head.commit.committed_datetime
except GitCommandError as e:
# Check if this is a "dubious ownership" error
if "dubious ownership" in str(e):
try:
repo_path = os.path.dirname(os.path.dirname(__file__))
# Add the directory as a safe directory and retry
repo = Repo(repo_path)
repo.git.config("--global", "--add", "safe.directory", repo_path)
git_short_hash = repo.git.rev_parse(repo.head, short=True)
git_date = repo.head.commit.committed_datetime
except Exception as retry_error: # pylint:disable=broad-exception-caught
app.logger.warning("Failed to read git repo after retry: %r", retry_error)
else:
app.logger.warning("Failed to read git repo data: %r", e)
except Exception as e: # pylint:disable=broad-exception-caught
app.logger.warning("Failed to read git repo data: %r", e)
# JinjaX components
catalog = jinjax.Catalog(jinja_env=app.jinja_env, root_url="/components/")
catalog.add_folder("testdays/components")
catalog.jinja_env.globals.update({
"current_user": current_user,
"git_short_hash": git_short_hash,
"git_date": git_date,
})
# Whitenoise middleware to serve the components' js and css files
app.wsgi_app = catalog.get_middleware(
app.wsgi_app,
autorefresh=app.debug,
max_age=None if app.debug else 60,
allowed_ext=[".css", ".js"],
)
@app.context_processor
def jinjax_components_in_jinja():
def C(name, **kwargs):
return catalog.render(name, **kwargs)
return {"C": C}
@app.after_request
def log_request(response):
"""Log HTTP requests through app.logger with the unified log format.
Synchronous logging is appropriate here -- this is a low-traffic
internal admin tool and the StreamHandler write is sub-microsecond.
"""
from flask import request as req
app.logger.info(
'%s - "%s %s %s" %s',
req.remote_addr,
req.method,
req.path,
req.environ.get("SERVER_PROTOCOL", "HTTP/1.1"),
response.status_code,
)
return response
class ReverseProxied(object):
def __init__(self, app):
self.app = app
def __call__(self, environ, start_response):
scheme = environ.get("HTTP_X_FORWARDED_PROTO")
# For some reason, the current Openshift/Proxy setup is weird, and instead of
# "https" the X-Forwarded-Proto header is set to "https,https"
# This works around it, while (hopefully) not fucking up the other instances
# where the value is either unset, or set properly.
scheme = scheme.split(",")[0].strip() if scheme and "," in scheme else scheme
if scheme:
environ["wsgi.url_scheme"] = scheme
return self.app(environ, start_response)
# Openshift is reverse-proxied. Maybe change this to: https://werkzeug.palletsprojects.com/en/3.0.x/middleware/proxy_fix/
if os.getenv("IS_OPENSHIFT"):
app.wsgi_app = ReverseProxied(app.wsgi_app)
if app.config["SHOW_DB_URI"]:
app.logger.debug("using DBURI: %s" % app.config["SQLALCHEMY_DATABASE_URI"])
CAPTCHA_CONFIG = {
'SECRET_CAPTCHA_KEY': app.config["SECRET_KEY"],
'CAPTCHA_LENGTH': 6,
'CAPTCHA_DIGITS': False,
'EXPIRE_SECONDS': 600,
}
SIMPLE_CAPTCHA = CAPTCHA(config=CAPTCHA_CONFIG)
app = SIMPLE_CAPTCHA.init_app(app)
# database
# app.config['SQLALCHEMY_ECHO'] = True
db = SQLAlchemy(app)
# Auto-apply pending database migrations on startup (dev mode only).
# In production, migrations are handled by gunicorn's on_starting hook
# (see gunicorn.cfg.py). In test mode, test fixtures manage their own DB.
if os.getenv("RUNMODE") == "dev":
_alembic_ini = "./alembic.ini" if os.path.exists("./alembic.ini") else None
if _alembic_ini:
from alembic import command as _al_command
from alembic.config import Config as _AlembicConfig
try:
_alembic_cfg = _AlembicConfig(_alembic_ini)
with app.app_context():
_al_command.upgrade(_alembic_cfg, "head")
app.logger.info("Database schema is up to date")
except Exception as _e:
app.logger.warning("Auto-migration failed: %s", _e)
# Audit logger -- standalone logger with its own handlers, independent of
# Flask's app.logger lifecycle.
audit_logger = logging.getLogger("testdays.audit")
audit_logger.handlers.clear()
audit_logger.setLevel(logging.INFO)
audit_logger.propagate = False
_add_log_handlers(audit_logger, logging.INFO)
# flask-login
login_manager = LoginManager()
login_manager.init_app(app)
login_manager.session_protection = "strong"
login_manager.login_view = "login_page.login"
login_manager.login_message_category = "info"
from .controllers.admin import admin
# register blueprints
from .controllers.main import main
from .controllers.login_page import login_page
app.register_blueprint(main)
app.register_blueprint(login_page)
app.register_blueprint(admin, url_prefix="/admin")
from .app import create_app # noqa: F401

322
testdays/app.py Normal file
View file

@ -0,0 +1,322 @@
"""Application factory for the Fedora Testdays web application."""
import logging
import logging.handlers
import os
from jinjax.catalog import Catalog as JinjaXCatalog
from flask import Flask
from flask_login import current_user
from flask_simple_captcha import CAPTCHA
from git import Repo
from git.exc import GitCommandError
from . import config
from . import extensions
class _WerkzeugRequestFilter(logging.Filter):
"""Drop Werkzeug's per-request log lines but keep startup/error messages."""
def filter(self, record):
msg = record.getMessage()
# Werkzeug request lines look like:
# '10.0.0.1 - - [01/Apr/2026 12:00:00] "GET / HTTP/1.1" 200 -'
# The ' - - [' timestamp bracket pattern uniquely identifies request
# lines. Startup messages (' * Running on ...') never contain it.
# Note: we match on ' - - [' alone rather than also checking for HTTP
# methods, because werkzeug injects ANSI color codes around the method
# (e.g. '\x1b[36mGET ... \x1b[0m') which break literal string matching.
if " - - [" in msg:
return False
return True
class ReverseProxied(object):
def __init__(self, app):
self.app = app
def __call__(self, environ, start_response):
scheme = environ.get("HTTP_X_FORWARDED_PROTO")
# For some reason, the current Openshift/Proxy setup is weird, and instead of
# "https" the X-Forwarded-Proto header is set to "https,https"
# This works around it, while (hopefully) not fucking up the other instances
# where the value is either unset, or set properly.
scheme = scheme.split(",")[0].strip() if scheme and "," in scheme else scheme
if scheme:
environ["wsgi.url_scheme"] = scheme
return self.app(environ, start_response)
def _load_config_from_env():
"""Build a Config object using the existing 3-layer strategy.
Layer 1: Select config class based on RUNMODE/DEV/TEST env vars.
Layer 2: Override from pyfile (path from env or convention).
Layer 3: Override from environment variables.
Returns:
A populated Config instance.
"""
config_obj = config.Production()
config_file = "/etc/testdays/settings.py"
if os.getenv("RUNMODE") == "dev" or os.getenv("DEV") == "true":
config_obj = config.Development()
config_file = f"{os.getcwd()}/conf/settings.py"
elif os.getenv("RUNMODE") == "test" or os.getenv("TEST") == "true":
config_obj = config.Production()
config_file = None
config_file = os.environ.get("FORCE_CONFIG_FILE", config_file)
if config_file and os.path.exists(config_file):
config_obj.update_from_pyfile(config_file)
# OpenShift deployment overrides
if os.getenv("IS_OPENSHIFT"):
config.check_required_openshift_envvars()
config_obj.OIDC_CLIENT_SECRETS = "/opt/app-root/secret/client_secrets.json"
config_obj.update_from_envvars()
return config_obj
def _configure_logging(app):
"""Set up logging handlers for root, app, werkzeug, and audit loggers.
Skipped for the root logger when TESTING is True to avoid clobbering
pytest's root logger handlers.
"""
is_testing = app.config.get("TESTING", False)
fmt = "%(asctime)s [%(levelname)8s] (%(filename)s:%(lineno)s) %(message)s"
datefmt = "%Y-%m-%d %H:%M:%S"
loglevel = logging.DEBUG if app.debug else logging.INFO
formatter = logging.Formatter(fmt=fmt, datefmt=datefmt)
def _add_log_handlers(logger, level=logging.INFO):
if app.config.get("STREAM_LOGGING"):
h = logging.StreamHandler()
h.setLevel(level)
h.setFormatter(formatter)
logger.addHandler(h)
if app.config.get("SYSLOG_LOGGING") and os.path.exists("/dev/log"):
h = logging.handlers.SysLogHandler(
address="/dev/log",
facility=logging.handlers.SysLogHandler.LOG_LOCAL4,
)
h.setLevel(level)
h.setFormatter(formatter)
logger.addHandler(h)
if app.config.get("FILE_LOGGING") and app.config.get("LOGFILE"):
h = logging.handlers.RotatingFileHandler(
app.config["LOGFILE"], maxBytes=5000000, backupCount=5
)
h.setLevel(level)
h.setFormatter(formatter)
logger.addHandler(h)
# Root logger -- skip in testing to preserve pytest's handlers
if not is_testing:
root_logger = logging.getLogger()
root_logger.handlers.clear()
root_logger.setLevel(loglevel)
_add_log_handlers(root_logger, loglevel)
# Suppress Werkzeug per-request log lines
logging.getLogger("werkzeug").addFilter(_WerkzeugRequestFilter())
# App logger
app.logger.handlers.clear()
app.logger.propagate = False
app.logger.setLevel(loglevel)
_add_log_handlers(app.logger, loglevel)
# Audit logger -- standalone logger with its own handlers, independent of
# Flask's app.logger lifecycle.
audit_logger = logging.getLogger("testdays.audit")
audit_logger.handlers.clear()
audit_logger.setLevel(logging.INFO)
audit_logger.propagate = False
_add_log_handlers(audit_logger, logging.INFO)
def _extract_git_info(app):
"""Extract git short hash and commit date for UI footer display."""
git_short_hash = "unknown"
git_date = None
try:
# Non-existent repo may cause InvalidGitRepositoryError or NoSuchPathError
repo = Repo(os.path.dirname(os.path.dirname(__file__)))
# Empty repo may cause ValueError or GitCommandError
git_short_hash = repo.git.rev_parse(repo.head, short=True)
git_date = repo.head.commit.committed_datetime
except GitCommandError as e:
# Check if this is a "dubious ownership" error
if "dubious ownership" in str(e):
try:
repo_path = os.path.dirname(os.path.dirname(__file__))
# Add the directory as a safe directory and retry
repo = Repo(repo_path)
repo.git.config("--global", "--add", "safe.directory", repo_path)
git_short_hash = repo.git.rev_parse(repo.head, short=True)
git_date = repo.head.commit.committed_datetime
except Exception as retry_error: # pylint:disable=broad-exception-caught
app.logger.warning("Failed to read git repo after retry: %r", retry_error)
else:
app.logger.warning("Failed to read git repo data: %r", e)
except Exception as e: # pylint:disable=broad-exception-caught
app.logger.warning("Failed to read git repo data: %r", e)
return git_short_hash, git_date
def create_app(config_obj=None):
"""Application factory.
Args:
config_obj: Optional Config instance. If None, config is determined
from environment variables (RUNMODE, DEV, TEST, etc.)
using the existing 3-layer loading strategy.
Returns:
Configured Flask app instance.
"""
app = Flask(
"testdays",
static_url_path="",
static_folder="static",
)
app.jinja_env.trim_blocks = True
app.jinja_env.lstrip_blocks = True
# --- Configuration ---
if config_obj is None:
config_obj = _load_config_from_env()
app.config.from_object(config_obj)
# --- Logging ---
_configure_logging(app)
# --- Secret key ---
app.secret_key = app.config["SECRET_KEY"]
if app.config.get("PRODUCTION"):
if app.secret_key == "not-really-a-secret":
raise Warning("You need to change the app.secret_key value for production")
# --- Extensions ---
extensions.db.init_app(app)
extensions.login_manager.init_app(app)
extensions.oidc.init_app(app, prefix="/flask_oidc")
# CAPTCHA -- requires SECRET_KEY from config at construction time
captcha_config = {
'SECRET_CAPTCHA_KEY': app.config["SECRET_KEY"],
'CAPTCHA_LENGTH': 6,
'CAPTCHA_DIGITS': False,
'EXPIRE_SECONDS': 600,
}
extensions.SIMPLE_CAPTCHA = CAPTCHA(config=captcha_config)
extensions.SIMPLE_CAPTCHA.init_app(app)
# --- Git info ---
git_short_hash, git_date = _extract_git_info(app)
# --- JinjaX catalog ---
extensions.catalog = JinjaXCatalog(jinja_env=app.jinja_env, root_url="/components/")
extensions.catalog.add_folder("testdays/components")
# Import models to register tables on db.metadata and set up
# login_manager callbacks (@user_loader, anonymous_user)
from .models.user import UserRoles # noqa: F401 -- triggers side effects
import testdays.models # noqa: F401 pylint: disable=unused-import
extensions.catalog.jinja_env.globals.update({
"current_user": current_user,
"git_short_hash": git_short_hash,
"git_date": git_date,
"UserRoles": UserRoles,
})
# Whitenoise middleware to serve the components' js and css files
app.wsgi_app = extensions.catalog.get_middleware(
app.wsgi_app,
autorefresh=app.debug,
max_age=None if app.debug else 60,
allowed_ext=[".css", ".js"],
)
# --- Context processor ---
@app.context_processor
def jinjax_components_in_jinja():
def C(name, **kwargs):
return extensions.catalog.render(name, **kwargs)
return {"C": C}
# --- Request logging ---
@app.after_request
def log_request(response):
"""Log HTTP requests through app.logger with the unified log format.
Synchronous logging is appropriate here -- this is a low-traffic
internal admin tool and the StreamHandler write is sub-microsecond.
"""
from flask import request as req
app.logger.info(
'%s - "%s %s %s" %s',
req.remote_addr,
req.method,
req.path,
req.environ.get("SERVER_PROTOCOL", "HTTP/1.1"),
response.status_code,
)
return response
# --- Reverse proxy middleware ---
# Openshift is reverse-proxied. Maybe change this to:
# https://werkzeug.palletsprojects.com/en/3.0.x/middleware/proxy_fix/
if os.getenv("IS_OPENSHIFT"):
app.wsgi_app = ReverseProxied(app.wsgi_app)
if app.config.get("SHOW_DB_URI"):
app.logger.debug("using DBURI: %s", app.config["SQLALCHEMY_DATABASE_URI"])
# --- Login manager config ---
extensions.login_manager.session_protection = "strong"
extensions.login_manager.login_view = "login_page.login"
extensions.login_manager.login_message_category = "info"
# --- Blueprints ---
from .controllers.main import main
from .controllers.login_page import login_page
from .controllers.admin import admin
from .controllers.api import api
app.register_blueprint(main)
app.register_blueprint(login_page)
app.register_blueprint(admin, url_prefix="/admin")
app.register_blueprint(api, url_prefix="/api/0")
# --- Auto-migrate in dev mode ---
# In production, migrations are handled by gunicorn's on_starting hook
# (see gunicorn.cfg.py). In test mode, test fixtures manage their own DB.
if os.getenv("RUNMODE") == "dev":
_alembic_ini = "./alembic.ini" if os.path.exists("./alembic.ini") else None
if _alembic_ini:
from alembic import command as _al_command
from alembic.config import Config as _AlembicConfig
try:
_alembic_cfg = _AlembicConfig(_alembic_ini)
with app.app_context():
_al_command.upgrade(_alembic_cfg, "head")
app.logger.info("Database schema is up to date")
except Exception as _e: # pylint: disable=broad-exception-caught
app.logger.warning("Auto-migration failed: %s", _e)
return app

8
testdays/cli.py
View file

@ -9,7 +9,10 @@ from alembic import command as al_command
from alembic.config import Config
from alembic.migration import MigrationContext
from . import app, db
from flask import current_app
from .app import create_app
from .extensions import db
from .models import *
@ -34,7 +37,7 @@ def CMD_upgrade_db(destructive, **kwargs):
context = MigrationContext.configure(db.engine.connect())
current_rev = context.get_current_revision()
app.logger.info("Upgrading Database to `head` from `%s`", current_rev)
current_app.logger.info("Upgrading Database to `head` from `%s`", current_rev)
al_command.upgrade(alembic_cfg, "head")
@ -274,6 +277,7 @@ def main():
sys.exit(1)
command = globals()[f"CMD_{args[0]}"]
app = create_app()
with app.app_context():
command(destructive=options.destructive, value=options.value)

17
testdays/controllers/admin.py
View file

@ -19,7 +19,8 @@ from wtforms import (
validators,
)
from .. import catalog, db
from .. import extensions
from ..extensions import db
from ..models import *
from ..utils.authorization import admin_required, creator_required
from ..models.audit import ChangeType, log_audit
@ -32,14 +33,14 @@ admin = Blueprint("admin", __name__)
@login_required
@admin_required
def index():
return catalog.render("Admin.Index")
return extensions.catalog.render("Admin.Index")
@admin.route("/users", methods=["GET", "POST"])
@login_required
@admin_required
def users():
users = User.query.order_by(User.id).all()
return catalog.render("Admin.Users", users=users)
return extensions.catalog.render("Admin.Users", users=users)
@admin.route("/users/<int:id_>", methods=["GET", "PUT", "DELETE"])
@ -92,7 +93,7 @@ def users_id(id_=None):
db.session.add(user)
db.session.commit()
return catalog.render("Admin.Users", users=[user])
return extensions.catalog.render("Admin.Users", users=[user])
@admin.route("/audit_log")
@ -105,7 +106,7 @@ def audit_log():
).order_by(AuditLog.timestamp.desc()).paginate(
page=page, per_page=20, error_out=False
)
return catalog.render("Admin.AuditLog", pagination=pagination)
return extensions.catalog.render("Admin.AuditLog", pagination=pagination)
class TDMetadataField(TextAreaField):
@ -276,7 +277,7 @@ def create_testday():
db.session.commit()
return redirect(url_for("main.show_testday", testday_id=td.id))
return catalog.render("Admin.CreateTestday", form=form)
return extensions.catalog.render("Admin.CreateTestday", form=form)
@ -331,7 +332,7 @@ def edit_testday(td_id):
perform_testcase_changes(td, form.structure)
except ValueError as e:
form.structure.errors.append(str(e))
return catalog.render("Admin.CreateTestday", form=form, testday_id=td_id)
return extensions.catalog.render("Admin.CreateTestday", form=form, testday_id=td_id)
# Capture old values before modification.
# start/end are datetime.datetime in the DB but datetime.date from the
@ -391,4 +392,4 @@ def edit_testday(td_id):
form.end.data = td.end
form.draft.data = td.is_draft
return catalog.render("Admin.CreateTestday", form=form, testday_id=td_id, testday_name=td.name)
return extensions.catalog.render("Admin.CreateTestday", form=form, testday_id=td_id, testday_name=td.name)

13
testdays/controllers/login_page.py
View file

@ -24,7 +24,8 @@ from wtforms import (
validators,
)
from .. import SIMPLE_CAPTCHA, app, catalog, db, oidc
from .. import extensions
from ..extensions import db, oidc
from ..models import *
from ..models.user import user_displayname_validator
@ -60,7 +61,7 @@ class AnonLoginForm(FlaskForm):
@login_page.route("/")
def index():
return catalog.render("Index")
return extensions.catalog.render("Index")
@login_page.route("/http403", methods=["GET", "POST"])
@ -76,7 +77,7 @@ def login():
if not next_page or urllib.parse.urlsplit(next_page).netloc != "":
next_page = url_for("main.index")
return catalog.render("LoginHub", next=next_page)
return extensions.catalog.render("LoginHub", next=next_page)
@login_page.route("/anonymous_login", methods=["GET", "POST"])
def anonymous_login():
@ -98,18 +99,18 @@ def anonymous_login():
db.session.commit()
except:
flash(message="Login unsuccessfull")
return catalog.render("Login", form=form)
return extensions.catalog.render("Login", form=form)
login_user(user)
return redirect(next_page)
captcha = SIMPLE_CAPTCHA.create()
captcha = extensions.SIMPLE_CAPTCHA.create()
print(", ".join(captcha.keys()))
form.captcha_hash.data = generate_password_hash(captcha["text"])
c = {"img": captcha["img"], "text": captcha["text"]}
return catalog.render("Login", form=form, next=next_page, captcha=c)
return extensions.catalog.render("Login", form=form, next=next_page, captcha=c)
@login_page.route("/oidc_login")

23
testdays/controllers/main.py
View file

@ -4,7 +4,7 @@ from collections import namedtuple
from urllib.parse import urlparse
from zoneinfo import ZoneInfo
from flask import Blueprint, Response, after_this_request, redirect, request, url_for
from flask import Blueprint, Response, after_this_request, current_app, redirect, request, url_for
from flask_login import current_user, login_required
from flask_wtf import FlaskForm
from jinja2.nodes import Test
@ -17,7 +17,8 @@ from wtforms import (
validators,
)
from .. import app, catalog, db
from .. import extensions
from ..extensions import db
from ..models import *
from ..models.user import user_displayname_validator
@ -45,7 +46,7 @@ def index():
# Event 'end' datetime is stored in db as <last-day-date> 00:00:00 so we use date only for comparing
current = [t for t in tds if t.end.date() >= server_date_gmt_p12]
past = reversed([t for t in tds if t.end.date() < server_date_gmt_p12])
return catalog.render("EventList", current=current, past=past)
return extensions.catalog.render("EventList", current=current, past=past)
@main.route("/events/<int:event_id>")
@ -56,7 +57,7 @@ def show_event_from_archive(event_id, *args, **kwargs):
@main.route("/events")
def event_archive():
return catalog.render("Archive")
return extensions.catalog.render("Archive")
def validate_bugs_field(form, field):
@ -97,9 +98,9 @@ def validate_bugs_field(form, field):
# Check if tld matches exactly or is a subdomain of any allowed TLD
if not any(tld == allowed or tld.endswith(f".{allowed}")
for allowed in app.config["BUG_ALLOWED_TLDS"]):
for allowed in current_app.config["BUG_ALLOWED_TLDS"]):
raise ValidationError(
f"Line #{ln}: Invalid domain, must be one of {app.config['BUG_ALLOWED_TLDS']}"
f"Line #{ln}: Invalid domain, must be one of {current_app.config['BUG_ALLOWED_TLDS']}"
)
links.append(l)
@ -199,7 +200,7 @@ def submit_result(testday_id, section_id, testcase_ulid):
except (ValueError, IndexError, KeyError):
pass
return catalog.render(
return extensions.catalog.render(
"SubmitResult",
form=form,
testday_id=testday_id,
@ -265,7 +266,7 @@ def edit_result(testday_id, section_id, testcase_ulid, result_id):
form.bugs.data = result.bugs or ""
form.comment.data = html.unescape(result.comment or "")
return catalog.render(
return extensions.catalog.render(
"SubmitResult",
form=form,
testday_id=testday_id,
@ -308,7 +309,7 @@ def edit_profile(user_id):
if request.method == "GET":
form.displayname.data = html.unescape(current_user.displayname)
return catalog.render("EditProfile", form=form)
return extensions.catalog.render("EditProfile", form=form)
def filter_structure_for_display(structure):
@ -418,7 +419,7 @@ def show_testday(testday_id):
"\n".join(
[
l
for l in catalog.render(
for l in extensions.catalog.render(
"WikiExport",
testday=testday,
results=results_by_section,
@ -432,7 +433,7 @@ def show_testday(testday_id):
mimetype="text/plain",
)
return catalog.render(
return extensions.catalog.render(
"Testday",
testday=testday,
results=results_by_section,

33
testdays/extensions.py Normal file
View file

@ -0,0 +1,33 @@
"""Flask extension instances.
Extensions are created here without binding to any Flask app.
They are bound to the app via ``ext.init_app(app)`` inside
:func:`testdays.app.create_app`.
Importing this module does **not** create a Flask application,
which allows models and other code to reference ``db`` at import
time without triggering side effects.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, cast
from flask_login import LoginManager
from flask_oidc import OpenIDConnect
from flask_sqlalchemy import SQLAlchemy
if TYPE_CHECKING:
from jinjax.catalog import Catalog
from flask_simple_captcha import CAPTCHA as _CAPTCHAType
db = SQLAlchemy()
login_manager = LoginManager()
oidc = OpenIDConnect()
# These require the app at creation time and are assigned by create_app().
# Typed as non-optional because they are guaranteed to be set before any
# view function runs. The cast() calls satisfy the type checker while
# the actual values are assigned in create_app().
catalog: Catalog = cast("Catalog", None)
SIMPLE_CAPTCHA: _CAPTCHAType = cast("_CAPTCHAType", None)

2
testdays/models/audit.py
View file

@ -4,7 +4,7 @@ import logging
from sqlalchemy.dialects import postgresql
from testdays import db
from testdays.extensions import db
_audit_logger = logging.getLogger("testdays.audit")

2
testdays/models/testday.py
View file

@ -22,7 +22,7 @@ import json
from ulid import ULID
from testdays import db
from testdays.extensions import db
from flask_login import current_user
from copy import deepcopy

5
testdays/models/user.py
View file

@ -3,7 +3,7 @@ import enum
from flask_login import AnonymousUserMixin, UserMixin
from werkzeug.security import check_password_hash, generate_password_hash
from .. import catalog, db, login_manager
from ..extensions import db, login_manager
from wtforms import ValidationError
import html
@ -36,9 +36,6 @@ class UserRoles(enum.Enum):
except KeyError:
return None
# FIXME - maybe not required, but usefull?
catalog.jinja_env.globals.update({"UserRoles": UserRoles})
class User(UserMixin, db.Model):
__tablename__ = "users"

74
tests/conftest.py
View file

@ -5,7 +5,6 @@ Root conftest.py
import logging
import os
from pathlib import Path
import sys
from typing import Generator, Any
from flask import Flask
@ -43,33 +42,6 @@ logging.getLogger("docker.utils.config").setLevel(logging.WARNING)
logging.getLogger("git").setLevel(logging.WARNING)
# testdays-web app is designed so that merely importing testdays module will
# instantiate the Flask app. We may need to import some testdays modules here
# or in other conftest.py files before pytest fixtures could be executed.
# For this reason we create here some minimal configuration required for successful
# testdays module import.
# Some values (like DB_URI) are not known at this time and dummy value is
# used instead. Settings file will be later overwritten at pytest startup
# by _app_config fixture and populated with valid values.
# testdays module will be then re-imported and it will pick up the updated
# and fully valid configuration.
os.environ["RUNMODE"] = "test"
os.environ["FORCE_CONFIG_FILE"] = str(TEST_SETTINGS_PATH)
# Creating minimal startup configuration file
TEST_SETTINGS_PATH.parent.mkdir(parents=True, exist_ok=True)
with open(TEST_SETTINGS_PATH, "w", encoding="utf-8") as temp_settings_file:
temp_settings_file.write(
"\n".join(
[
"SECRET_KEY = 'top-secret-test-key'",
"DB_URI = 'postgresql+psycopg2://dbuser:dbpassword@dbhost:5432/dbname'",
"OIDC_ENABLED = False",
]
)
)
@pytest.fixture(scope="session")
def _db_container_url() -> Generator[str, Any, None]:
"""
@ -232,42 +204,28 @@ def _db_connection(
@pytest.fixture(name="app", scope="session")
def _app(db_connection) -> Generator[Flask, Any, None]:
def _app(db_connection, _app_config) -> Generator[Flask, Any, None]:
"""
Provides Flask application with initialized database schema.
This session-scoped fixture imports the testdays module, runs Alembic
migrations to create all database tables with the latest schema, manages
pytest logger handlers that get cleared during the import and migration
process, and yields the Flask application within an active app context.
The database schema is dropped during teardown.
Uses the application factory to create a fresh app with test
configuration. Runs Alembic migrations, then yields the app
within an active app context.
Args:
db_connection: SQLAlchemy connection to the test database
_app_config: Ensures settings file exists with valid DB URI
Yields:
Flask: The configured Flask application instance with active context
"""
# In case testdays module was already imported by some conftest.py file
# we force reload of the module so that correct final settings file
# is used by that module
for module in list(sys.modules.keys()):
if module.startswith("testdays"):
del sys.modules[module]
from testdays.app import create_app
from testdays.config import Testing
# Clear Flask app logger handlers if it exists from a previous run
# This prevents accumulating duplicate handlers on module reload
if "testdays" in logging.Logger.manager.loggerDict:
flask_logger = logging.getLogger("testdays")
flask_logger.handlers.clear()
config_obj = Testing()
config_obj.update_from_pyfile(str(TEST_SETTINGS_PATH))
# Save pytest's root logger handlers before testdays import clears them
root_logger = logging.getLogger()
saved_handlers_root_logger = root_logger.handlers.copy()
saved_level_root_logger = root_logger.level
# importing testdays module applies current configuration from settings file
from testdays import app as flask_app # pylint: disable=import-outside-toplevel, unused-import
flask_app = create_app(config_obj)
if ALEMBIC_CONFIG_PATH.is_file():
alembic_cfg = AlembicConfig(ALEMBIC_CONFIG_PATH)
@ -277,7 +235,7 @@ def _app(db_connection) -> Generator[Flask, Any, None]:
ini_section="alembic-packaged",
)
# Save disabled state of ALL existing loggers before alembic disables them
# Save disabled state of existing loggers before alembic disables them
# alembic's fileConfig() disables all loggers not in alembic.ini
saved_loggers_state = {}
for logger_name in list(logging.Logger.manager.loggerDict.keys()):
@ -285,18 +243,14 @@ def _app(db_connection) -> Generator[Flask, Any, None]:
if hasattr(logger_obj, 'disabled'):
saved_loggers_state[logger_name] = logger_obj.disabled
# Create all tables and update them to the latest schema
alembic_cmd.upgrade(alembic_cfg, "head")
with flask_app.app_context():
alembic_cmd.upgrade(alembic_cfg, "head")
# Re-enable ALL loggers that were disabled by alembic's fileConfig
# Re-enable loggers that were disabled by alembic's fileConfig
for logger_name, was_disabled in saved_loggers_state.items():
logger_obj = logging.getLogger(logger_name)
logger_obj.disabled = was_disabled
# Restore pytest's root logger handlers that were cleared testdays import
root_logger.handlers = saved_handlers_root_logger
root_logger.setLevel(saved_level_root_logger)
# Log database version
result_row = db_connection.execute(sqlalchemy.text("select version()")).fetchone()
if result_row:

4
wsgi.py
View file

@ -1 +1,3 @@
from testdays import app as application
from testdays import create_app
application = create_app()