# SPDX-FileCopyrightText: 2026 Helmholtz-Zentrum Dresden-Rossendorf e.V (HZDR)
# SPDX-License-Identifier: Apache-2.0
"""Platform-aware helpers for documented `uv run poe ...` workflows."""
from __future__ import annotations
import argparse
import importlib.util
import os
import platform
import shutil
import subprocess
import sys
import time
from dataclasses import dataclass
from pathlib import Path
APP_IMPORT_TARGET = "labfrog:create_app()"
DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 5000
DEFAULT_WORKERS = 4
REPO_ROOT = Path(__file__).resolve().parents[1]
INSTANCE_DIR = REPO_ROOT / "instance"
CUSTOMIZE_DIR = REPO_ROOT / "labfrog" / "customize"
SCRIPT_DIR = Path(__file__).resolve().parent
# Keep package imports working when this helper is executed as
# `python labfrog/dev_tasks.py ...` from the repo root.
sys.path = [entry for entry in sys.path if Path(entry or ".").resolve() != SCRIPT_DIR]
if str(REPO_ROOT) not in sys.path:
sys.path.insert(0, str(REPO_ROOT))
[docs]
@dataclass(frozen=True)
class DoctorCheck:
name: str
status: str
detail: str
hint: str | None = None
def _status_label(status: str) -> str:
return {
"pass": "PASS",
"warn": "WARN",
"fail": "FAIL",
}.get(status, status.upper())
def _quoted_env_value(value: str) -> str:
text = str(value).strip()
if len(text) >= 2 and text[0] == text[-1] and text[0] in {'"', "'"}:
text = text[1:-1].strip()
return text
def _read_dotenv_assignments(path: Path) -> dict[str, str]:
values: dict[str, str] = {}
if not path.is_file():
return values
for raw_line in path.read_text(encoding="utf-8").splitlines():
line = raw_line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
values[key.strip()] = _quoted_env_value(value)
return values
[docs]
def resolve_repo_config_name(
repo_env: dict[str, str] | None = None,
environ: dict[str, str] | None = None,
) -> str:
"""Match the app's config-file selection rules for contributor checks."""
repo_env = repo_env or _read_dotenv_assignments(REPO_ROOT / ".env")
environ = environ or os.environ
return (
repo_env.get("CONFIG_NAME")
or repo_env.get("CONFIG_FILE")
or environ.get("CONFIG_NAME")
or environ.get("CONFIG_FILE")
or "default.cfg"
)
def _load_python_config(path: Path) -> dict[str, object]:
namespace = {"__file__": str(path), "__name__": "__labfrog_config__", "os": os}
compiled = compile(path.read_text(encoding="utf-8"), str(path), "exec")
exec(compiled, namespace)
return {
key: value
for key, value in namespace.items()
if key.isupper() and not key.startswith("__")
}
def _tool_check(
command: str, *, required: bool, label: str | None = None
) -> DoctorCheck:
resolved = shutil.which(command)
tool_label = label or command
if resolved:
return DoctorCheck(tool_label, "pass", f"Found at `{resolved}`.")
status = "fail" if required else "warn"
hint = None
if command == "trunk":
hint = "Install Trunk or use the repo's documented local launcher path before running `uv run poe lint`."
elif command == "docker":
hint = (
"Install Docker Desktop on Windows or Docker Engine plus the CLI on Linux."
)
elif command == "git":
hint = "Git is required for contribution workflows and `git diff --check`."
elif command == "uv":
hint = "Install uv before using the documented `uv run poe ...` workflows."
return DoctorCheck(
tool_label,
status,
f"`{command}` is not on PATH.",
hint=hint,
)
def _playwright_check() -> DoctorCheck:
if importlib.util.find_spec("playwright") is None:
return DoctorCheck(
"Playwright package",
"warn",
"The Playwright Python package is not installed in this environment.",
hint="Run `uv sync` and `uv run poe install-playwright` before browser smoke checks.",
)
return DoctorCheck(
"Playwright package",
"pass",
"The Playwright Python package is importable.",
)
def _config_path_check() -> tuple[list[DoctorCheck], Path | None]:
checks: list[DoctorCheck] = []
env_path = REPO_ROOT / ".env"
if env_path.is_file():
checks.append(
DoctorCheck(
".env",
"pass",
f"Found at `{env_path}`.",
)
)
else:
checks.append(
DoctorCheck(
".env",
"fail",
"The repo .env file is missing.",
hint="Create `.env` and set `CONFIG_NAME=default.cfg` or another instance config file.",
)
)
default_cfg = INSTANCE_DIR / "default.cfg"
if default_cfg.is_file():
checks.append(
DoctorCheck(
"instance/default.cfg",
"pass",
f"Found at `{default_cfg}`.",
)
)
else:
checks.append(
DoctorCheck(
"instance/default.cfg",
"fail",
"The default instance config is missing.",
)
)
config_name = resolve_repo_config_name()
config_path = INSTANCE_DIR / config_name
if config_path.is_file():
checks.append(
DoctorCheck(
"Active instance config",
"pass",
f"`CONFIG_NAME` resolves to `{config_name}`.",
)
)
return checks, config_path
checks.append(
DoctorCheck(
"Active instance config",
"fail",
f"`CONFIG_NAME` resolves to `{config_name}`, but `{config_path}` does not exist.",
hint="Update `.env` or add the missing file under `instance/`.",
)
)
return checks, None
def _config_content_checks(config_path: Path | None) -> list[DoctorCheck]:
if config_path is None:
return []
try:
config = _load_python_config(config_path)
except Exception as exc:
return [
DoctorCheck(
"Config parse",
"fail",
f"Failed to execute `{config_path.name}`: {exc}",
)
]
checks = [
DoctorCheck(
"Config parse",
"pass",
f"Loaded `{config_path.name}` successfully.",
)
]
use_custom_options = bool(config.get("USE_CUSTOM_OPTIONS", False))
use_full_custom = bool(config.get("USE_FULL_CUSTOM", False))
auth_method = str(config.get("AUTH_METHOD", "ldap")).strip().lower()
if use_custom_options:
for key in ("CUSTOM_OPTIONS_CAMPAIGNS", "CUSTOM_OPTIONS_TARGETS"):
relative_name = _quoted_env_value(config.get(key, ""))
target_path = CUSTOMIZE_DIR / relative_name if relative_name else None
if target_path and target_path.is_file():
checks.append(
DoctorCheck(
key,
"pass",
f"Found `{relative_name}` in `labfrog/customize/`.",
)
)
else:
checks.append(
DoctorCheck(
key,
"fail",
f"`{key}` points to `{relative_name or '(missing value)'}`, but the file is not present.",
)
)
if use_full_custom:
relative_name = _quoted_env_value(config.get("USE_FULL_CUSTOM_FILE", ""))
target_path = CUSTOMIZE_DIR / relative_name if relative_name else None
if target_path and target_path.is_file():
checks.append(
DoctorCheck(
"USE_FULL_CUSTOM_FILE",
"pass",
f"Found `{relative_name}` in `labfrog/customize/`.",
)
)
else:
checks.append(
DoctorCheck(
"USE_FULL_CUSTOM_FILE",
"fail",
f"`USE_FULL_CUSTOM_FILE` points to `{relative_name or '(missing value)'}`, but the file is not present.",
)
)
if auth_method == "helmholtz":
client_id = _quoted_env_value(str(config.get("OIDC_CLIENT_ID", "")))
if not client_id or client_id == "set-client-id":
checks.append(
DoctorCheck(
"OIDC client settings",
"warn",
"Helmholtz auth is enabled but `OIDC_CLIENT_ID` still looks like a placeholder.",
hint="Set `OIDC_CLIENT_ID` and `OIDC_CLIENT_SECRET` before testing Helmholtz login.",
)
)
return checks
def _app_import_smoke_check() -> DoctorCheck:
previous = {
key: os.environ.get(key)
for key in (
"LABFROG_TESTING",
"SKIP_MEDIAWIKI",
"SKIP_CUSTOM_OPTIONS",
)
}
os.environ["LABFROG_TESTING"] = "1"
os.environ["SKIP_MEDIAWIKI"] = "1"
os.environ["SKIP_CUSTOM_OPTIONS"] = "1"
try:
from labfrog import create_app
app = create_app({
"TESTING": True,
"AUTH_METHOD": "none",
"SECRET_KEY": "doctor-check",
"USE_WIKI": False,
"USE_CUSTOM_OPTIONS": False,
"USE_FULL_CUSTOM": False,
"DEFAULT_COLLECTION_FOR_SETTINGS": "field_selections",
"MONGODB_COLLECTION_FOR_SETTINGS": "field_selections",
"MONGODB_HOST": "localhost",
"MONGODB_PORT": 27018,
"MONGODB_USERNAME": "root",
"MONGODB_PASSWORD": "mypasswd",
"MONGODB_AUTH_SOURCE": "admin",
"MONGODB_DATABASE": "shotsheet",
})
route_count = len(app.url_map._rules)
return DoctorCheck(
"App import smoke",
"pass",
f"Created a test-safe app instance with {route_count} registered routes.",
)
except Exception as exc:
return DoctorCheck(
"App import smoke",
"fail",
f"Failed to create a test-safe app instance: {exc}",
)
finally:
for key, value in previous.items():
if value is None:
os.environ.pop(key, None)
else:
os.environ[key] = value
[docs]
def collect_doctor_checks() -> list[DoctorCheck]:
checks = [
_tool_check("git", required=True),
_tool_check("uv", required=True),
_tool_check("trunk", required=False),
_tool_check("docker", required=False, label="Docker CLI"),
_playwright_check(),
]
config_checks, config_path = _config_path_check()
checks.extend(config_checks)
checks.extend(_config_content_checks(config_path))
checks.append(_app_import_smoke_check())
return checks
[docs]
def run_doctor() -> int:
"""Print a concise contributor-environment report."""
checks = collect_doctor_checks()
fail_count = 0
warn_count = 0
print("LabFrog collaborator doctor")
print(f"Repo: {REPO_ROOT}")
print(f"Platform: {platform.system()} / Python {platform.python_version()}")
for check in checks:
print(f"- [{_status_label(check.status)}] {check.name}: {check.detail}")
if check.hint:
print(f" Hint: {check.hint}")
if check.status == "fail":
fail_count += 1
elif check.status == "warn":
warn_count += 1
if fail_count:
print(f"Summary: {fail_count} failing checks, {warn_count} warnings.")
return 1
print(f"Summary: 0 failing checks, {warn_count} warnings.")
return 0
def _gunicorn_available() -> bool:
return importlib.util.find_spec("gunicorn.app.wsgiapp") is not None
[docs]
def resolve_server_backend(
preferred: str = "auto",
*,
system_name: str | None = None,
gunicorn_available: bool | None = None,
) -> str:
"""Choose a stable production-style server backend for the current platform."""
normalized = (preferred or "auto").strip().lower()
system_name = system_name or platform.system()
if gunicorn_available is None:
gunicorn_available = _gunicorn_available()
if normalized == "auto":
if system_name == "Windows":
return "waitress"
if gunicorn_available:
return "gunicorn"
return "waitress"
if normalized == "gunicorn":
if not gunicorn_available:
raise RuntimeError(
"Gunicorn is not available in this environment. "
"Use `--backend auto` or `--backend waitress` instead."
)
return "gunicorn"
if normalized == "waitress":
return "waitress"
raise RuntimeError(f"Unsupported server backend: {preferred}")
def _gunicorn_command(
*, host: str, port: int, workers: int, preload: bool
) -> list[str]:
cmd = [
sys.executable,
"-m",
"gunicorn.app.wsgiapp",
"-w",
str(workers),
"-b",
f"{host}:{port}",
APP_IMPORT_TARGET,
]
if preload:
cmd.insert(-1, "--preload")
return cmd
[docs]
def run_serve(
*, host: str, port: int, workers: int, backend: str, preload: bool
) -> int:
"""Run the production-style app server with a platform-aware backend."""
resolved_backend = resolve_server_backend(backend)
if resolved_backend == "gunicorn":
cmd = _gunicorn_command(
host=host,
port=port,
workers=workers,
preload=preload,
)
try:
os.execv(cmd[0], cmd)
except OSError as exc:
print(
f"Failed to execute `{' '.join(cmd)}`: {exc}",
file=sys.stderr,
)
return 1
return 0
from waitress import serve as waitress_serve
from labfrog import create_app
app = create_app()
waitress_serve(app, host=host, port=port)
return 0
def _docker_ready() -> bool:
return (
subprocess.run(
["docker", "info"],
check=False,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
).returncode
== 0
)
def _docker_start_command(system_name: str | None = None) -> list[str] | None:
system_name = system_name or platform.system()
if system_name != "Linux":
return None
if shutil.which("systemctl"):
return ["systemctl", "start", "docker"]
if shutil.which("service"):
return ["service", "docker", "start"]
return None
def _docker_restart_command(system_name: str | None = None) -> list[str] | None:
system_name = system_name or platform.system()
if system_name != "Linux":
return None
if shutil.which("systemctl"):
return ["systemctl", "restart", "docker"]
if shutil.which("service"):
return ["service", "docker", "restart"]
return None
def _wait_for_docker_ready(
*, timeout_seconds: float = 15.0, interval_seconds: float = 0.5
) -> bool:
deadline = time.monotonic() + timeout_seconds
while time.monotonic() < deadline:
if _docker_ready():
return True
time.sleep(interval_seconds)
return _docker_ready()
def _restart_docker_and_wait(restart_cmd: list[str]) -> int:
restarted = subprocess.run(restart_cmd, check=False)
if restarted.returncode != 0:
print(
"Could not restart Docker with "
f"`{' '.join(restart_cmd)}`. Restart Docker manually and rerun the "
"LabFrog task.",
file=sys.stderr,
)
return restarted.returncode or 1
if not _wait_for_docker_ready():
print(
"Docker restart completed, but the daemon did not become ready in "
"time. Wait a moment, then rerun the LabFrog task.",
file=sys.stderr,
)
return 1
# Give Docker a brief moment after readiness so bridge chains settle.
time.sleep(2.0)
return 0
[docs]
def restart_docker_service(*, system_name: str | None = None) -> int:
"""Restart Docker explicitly when the host needs a clean daemon reset."""
system_name = system_name or platform.system()
restart_cmd = _docker_restart_command(system_name)
if restart_cmd is None:
if system_name == "Windows":
print(
"This task does not restart Docker Desktop on Windows. Close "
"active containers or restart Docker Desktop manually if needed.",
file=sys.stderr,
)
return 0
print(
"No supported Docker service restart command is available on this "
"platform. Restart Docker manually and rerun the LabFrog task.",
file=sys.stderr,
)
return 1
print("Restarting Docker...", file=sys.stderr)
return _restart_docker_and_wait(restart_cmd)
[docs]
def ensure_docker_running(*, system_name: str | None = None) -> int:
"""Make Docker-backed tasks fail clearly instead of assuming one platform."""
system_name = system_name or platform.system()
if shutil.which("docker") is None:
print(
"Docker CLI not found. Install Docker Desktop on Windows or "
"Docker Engine and the Docker CLI on Linux first.",
file=sys.stderr,
)
return 1
if _docker_ready():
print("Docker daemon is available.")
return 0
start_cmd = _docker_start_command(system_name)
if start_cmd is None:
if system_name == "Windows":
print(
"Docker Desktop is not running. Start it, wait until it is ready, "
"then rerun the LabFrog Docker task.",
file=sys.stderr,
)
else:
print(
"Docker is installed but the daemon is not reachable. Start Docker "
"and rerun the LabFrog Docker task.",
file=sys.stderr,
)
return 1
started = subprocess.run(start_cmd, check=False)
if started.returncode != 0:
print(
"Tried to start Docker with "
f"`{' '.join(start_cmd)}` but it failed. If your system requires "
"elevated privileges, start Docker first and rerun the LabFrog task.",
file=sys.stderr,
)
return started.returncode or 1
if _docker_ready():
print("Docker daemon is available.")
return 0
print(
"Docker start command completed but the daemon is still unreachable.",
file=sys.stderr,
)
return 1
def _build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=__doc__)
subparsers = parser.add_subparsers(dest="command", required=True)
serve = subparsers.add_parser(
"serve", help="Run the production-style web server with an auto backend."
)
serve.add_argument("--host", default=DEFAULT_HOST)
serve.add_argument("--port", type=int, default=DEFAULT_PORT)
serve.add_argument("--workers", type=int, default=DEFAULT_WORKERS)
serve.add_argument(
"--backend",
choices=["auto", "gunicorn", "waitress"],
default="auto",
help="Auto chooses Gunicorn on non-Windows when available, otherwise Waitress.",
)
serve.add_argument(
"--preload",
action=argparse.BooleanOptionalAction,
default=True,
help="Preload the app before Gunicorn forks workers (enabled by default).",
)
subparsers.add_parser(
"ensure-docker",
help="Check the Docker daemon and try a common Linux start command when possible.",
)
subparsers.add_parser(
"restart-docker",
help="Restart Docker explicitly on Linux hosts before using the regular Docker tasks.",
)
subparsers.add_parser(
"doctor",
help="Report whether the current checkout is ready for shared contributor workflows.",
)
return parser
[docs]
def main(argv: list[str] | None = None) -> int:
parser = _build_parser()
args = parser.parse_args(argv)
if args.command == "serve":
return run_serve(
host=args.host,
port=args.port,
workers=args.workers,
backend=args.backend,
preload=args.preload,
)
if args.command == "ensure-docker":
return ensure_docker_running()
if args.command == "restart-docker":
return restart_docker_service()
if args.command == "doctor":
return run_doctor()
parser.error(f"Unknown command: {args.command}")
return 2
if __name__ == "__main__":
raise SystemExit(main())