From c3a28a8578fe906f7962d04cf020605e892a6199 Mon Sep 17 00:00:00 2001 From: Akseli Lehtniemi <60463103+Akaseli@users.noreply.github.com> Date: Wed, 29 Jul 2026 13:08:22 +0300 Subject: [PATCH 1/3] fix: healthcheck requires all workers --- docker/entrypoint.sh | 3 +- src/matrixrmapi/api/healthcheck.py | 13 ++++- src/matrixrmapi/config.py | 1 + src/matrixrmapi/utils/startup.py | 86 ++++++++++++++++++++++++++---- tests/conftest.py | 12 +++++ tests/test_app.py | 37 +++++++++++++ tests/test_matrixrmapi.py | 23 ++++++++ 7 files changed, 161 insertions(+), 14 deletions(-) diff --git a/docker/entrypoint.sh b/docker/entrypoint.sh index fd07374..3b619a6 100755 --- a/docker/entrypoint.sh +++ b/docker/entrypoint.sh @@ -12,8 +12,9 @@ fi set -e if [ "$#" -eq 0 ]; then + export WEB_CONCURRENCY="${WEB_CONCURRENCY:-4}" # FIXME: can we know the traefik/nginx internal docker ip easily ? - exec gunicorn "matrixrmapi.app:get_app()" --bind 0.0.0.0:8012 --forwarded-allow-ips='*' -w 4 -k uvicorn.workers.UvicornWorker + exec gunicorn "matrixrmapi.app:get_app()" --bind 0.0.0.0:8012 --forwarded-allow-ips='*' -w "$WEB_CONCURRENCY" -k uvicorn.workers.UvicornWorker else exec "$@" fi diff --git a/src/matrixrmapi/api/healthcheck.py b/src/matrixrmapi/api/healthcheck.py index 0644605..c03dbbf 100644 --- a/src/matrixrmapi/api/healthcheck.py +++ b/src/matrixrmapi/api/healthcheck.py @@ -6,7 +6,8 @@ from fastapi import APIRouter, Request from libpvarki.schemas.product import ProductHealthCheckResponse -from ..config import MAS_HEALTH_URL, SYNAPSE_URL +from ..config import MAS_HEALTH_URL, SYNAPSE_URL, WEB_CONCURRENCY +from ..utils.startup import ready_workers LOGGER = logging.getLogger(__name__) @@ -15,13 +16,21 @@ @router.get("") async def request_healthcheck(request: Request) -> ProductHealthCheckResponse: - """Check that the Matrix integration is initialised and Synapse and MAS respond""" + """Check that the Matrix integration is initialised and Synapse and MAS respond + + Ensures all workers are ready to handle UserCRUD (mas + synapse credentials ) + """ if getattr(request.app.state, "synapse", None) is None or not getattr( request.app.state, "rooms", None ): return ProductHealthCheckResponse( healthy=False, extra="Matrix integration not initialised" ) + ready = ready_workers() + if ready < WEB_CONCURRENCY: + return ProductHealthCheckResponse( + healthy=False, extra=f"only {ready}/{WEB_CONCURRENCY} workers initialised" + ) async with httpx.AsyncClient() as client: for name, url in (("Synapse", SYNAPSE_URL), ("MAS", MAS_HEALTH_URL)): try: diff --git a/src/matrixrmapi/config.py b/src/matrixrmapi/config.py index dd70e6b..ef249ac 100644 --- a/src/matrixrmapi/config.py +++ b/src/matrixrmapi/config.py @@ -26,6 +26,7 @@ MAS_ADMIN_CLIENT_ID: str = cfg("MAS_ADMIN_CLIENT_ID", default="") MAS_ADMIN_CLIENT_SECRET: str = cfg("MAS_ADMIN_CLIENT_SECRET", default="") SYNAPSE_BOT_USERNAME: str = cfg("SYNAPSE_BOT_USERNAME", default="matrixrmapi-bot") +WEB_CONCURRENCY: int = cfg("WEB_CONCURRENCY", default=1, cast=int) @functools.cache diff --git a/src/matrixrmapi/utils/startup.py b/src/matrixrmapi/utils/startup.py index e02f2d5..1416cdd 100644 --- a/src/matrixrmapi/utils/startup.py +++ b/src/matrixrmapi/utils/startup.py @@ -4,6 +4,7 @@ import asyncio import logging +import os import tempfile from pathlib import Path from typing import Dict, List, Optional, Tuple @@ -28,6 +29,11 @@ LOGGER = logging.getLogger(__name__) +INIT_RETRY_BACKOFF = 5.0 +INIT_RETRY_BACKOFF_MAX = 60.0 + +READY_DIR = Path(tempfile.gettempdir()) / "matrixrmapi_ready" + # (key, alias_suffix, display_name, is_space, is_private) ROOMS_CONFIG: List[Tuple[str, str, str, bool, bool]] = [ ("space", "{d}-space", "{d}", True, False), @@ -44,6 +50,34 @@ } +def mark_ready() -> None: + """Record that this worker has its bot session and rooms.""" + READY_DIR.mkdir(parents=True, exist_ok=True) + (READY_DIR / str(os.getpid())).touch() + + +def clear_ready() -> None: + """Drop this worker's marker.""" + (READY_DIR / str(os.getpid())).unlink(missing_ok=True) + + +def ready_workers() -> int: + """How many live workers finished init, pruning markers left behind by dead ones.""" + if not READY_DIR.is_dir(): + return 0 + alive = 0 + for marker in READY_DIR.iterdir(): + try: + os.kill(int(marker.name), 0) # signal 0 checks + except (ProcessLookupError, ValueError): + marker.unlink(missing_ok=True) + continue + except PermissionError: + pass + alive += 1 + return alive + + async def wait_for_service( name: str, url: str, retries: int = 60, interval: float = 5.0 ) -> bool: @@ -240,17 +274,12 @@ def setup_mas_admin(app: FastAPI) -> Optional[MasAdmin]: return mas -async def connect_to_matrix(app: FastAPI) -> None: - """Background task: connect to MAS and Synapse, create bot and rooms.""" - mas = setup_mas_admin(app) - if mas is None: - LOGGER.error("No MAS admin client — Matrix integration disabled") - return - +async def init_matrix_once(app: FastAPI, mas: MasAdmin) -> bool: + """Init attempt: wait for services, get a bot session, ensure rooms.""" if not await wait_for_service("MAS", MAS_HEALTH_URL): - return + return False if not await wait_for_service("Synapse", SYNAPSE_URL): - return + return False manifest = get_manifest() deployment = str(manifest.get("deployment", "pvarki")) @@ -261,7 +290,7 @@ async def connect_to_matrix(app: FastAPI) -> None: ok, is_init = await acquire_bot_token(synapse, mas) if not ok: await synapse.close() - return + return False app.state.synapse = synapse @@ -269,7 +298,8 @@ async def connect_to_matrix(app: FastAPI) -> None: room_ids = await ensure_rooms(synapse, deployment, domain) except Exception as exc: LOGGER.error("Room setup failed: %s", exc) - return + await synapse.close() + return False if is_init: # Only the init worker applies state configuration to avoid redundant @@ -287,6 +317,7 @@ async def connect_to_matrix(app: FastAPI) -> None: # configure_rooms_state's power-level read-modify-write on the space. # Set even if configuration partially failed: rooms exist and are usable. app.state.rooms = room_ids + mark_ready() LOGGER.info("Synapse rooms ready: %s", room_ids) # Apply any promotions/demotions that arrived while rooms were not yet set. @@ -297,3 +328,36 @@ async def connect_to_matrix(app: FastAPI) -> None: if pending: LOGGER.info("Processing %d deferred promotion(s)/demotion(s)", len(pending)) await apply_pending(synapse, room_ids, pending) + return True + + +async def connect_to_matrix(app: FastAPI) -> None: + """Background task: connect to MAS and Synapse, create bot and rooms. + + Retries with backoff until it succeeds. + """ + clear_ready() + mas = setup_mas_admin(app) + if mas is None: + # Missing credentials + LOGGER.error("No MAS admin client; Matrix integration disabled!") + return + + backoff = INIT_RETRY_BACKOFF + attempt = 0 + while True: + attempt += 1 + try: + if await init_matrix_once(app, mas): + return + reason = "init did not complete" + except Exception as exc: + reason = f"{type(exc).__name__}: {exc}" + LOGGER.error( + "Matrix init attempt %d failed (%s), retrying in %.0fs", + attempt, + reason, + backoff, + ) + await asyncio.sleep(backoff) + backoff = min(backoff * 2, INIT_RETRY_BACKOFF_MAX) diff --git a/tests/conftest.py b/tests/conftest.py index c23b4a5..fa345b5 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,6 +1,7 @@ """pytest automagics""" from typing import Generator, Dict +from pathlib import Path import logging import os import uuid @@ -11,6 +12,7 @@ from matrixrmapi.app import get_app from matrixrmapi.config import get_manifest +from matrixrmapi.utils import startup # Default is "ecs" and it's not great for tests os.environ["LOG_CONSOLE_FORMATTER"] = "local" @@ -19,6 +21,16 @@ APP = get_app() +@pytest.fixture(autouse=True) +def ready_dir( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> Generator[Path, None, None]: + """Worker ready markers per-test""" + target = tmp_path / "ready" + monkeypatch.setattr(startup, "READY_DIR", target) + yield target + + @pytest.fixture def mtlsclient() -> Generator[TestClient, None, None]: """Fake the NGinx header""" diff --git a/tests/test_app.py b/tests/test_app.py index 1d17cfd..340f0eb 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -2,11 +2,13 @@ from __future__ import annotations +import os from typing import Dict, cast from unittest.mock import AsyncMock import pytest +from matrixrmapi.utils import startup from matrixrmapi.utils.startup import apply_pending, ensure_room from matrixrmapi.utils.synapse_admin import SynapseAdmin from matrixrmapi.types import AdminAction @@ -139,3 +141,38 @@ async def testensure_room_creates_new_room() -> None: ) assert result == "!new:example.test" synapse.create_room.assert_called_once() + + +# --------------------------------------------------------------------------- +# ready markers +# --------------------------------------------------------------------------- + + +def testready_workers_counts_marked_workers() -> None: + """A marked worker counts; the count is what the healthcheck compares against""" + assert startup.ready_workers() == 0 + startup.mark_ready() + assert startup.ready_workers() == 1 + startup.mark_ready() # one marker per pid + assert startup.ready_workers() == 1 + + +def testready_workers_prunes_dead_worker() -> None: + """A SIGKILLed worker must not keep counting as ready, or 3/4 reads as 4/4""" + pid = os.fork() + if pid == 0: + os._exit(0) + os.waitpid(pid, 0) + startup.READY_DIR.mkdir(parents=True, exist_ok=True) + stale = startup.READY_DIR / str(pid) + stale.touch() + + assert startup.ready_workers() == 0 + assert not stale.exists() + + +def testclear_ready_drops_own_marker() -> None: + """Called at init so a recycled pid cannot inherit the last owner's marker""" + startup.mark_ready() + startup.clear_ready() + assert startup.ready_workers() == 0 diff --git a/tests/test_matrixrmapi.py b/tests/test_matrixrmapi.py index ebe4bc4..df36143 100644 --- a/tests/test_matrixrmapi.py +++ b/tests/test_matrixrmapi.py @@ -3,9 +3,11 @@ from unittest.mock import AsyncMock, patch import httpx +import pytest from fastapi.testclient import TestClient from matrixrmapi import __version__ +from matrixrmapi.utils.startup import mark_ready from .conftest import APP @@ -26,6 +28,7 @@ def test_healthcheck_healthy(mtlsclient: TestClient) -> None: """With integration initialised and Synapse/MAS responding the service reports healthy""" APP.state.synapse = AsyncMock() APP.state.rooms = {"space": "!space:x"} + mark_ready() try: with patch( "matrixrmapi.api.healthcheck.httpx.AsyncClient.get", @@ -47,6 +50,7 @@ def test_healthcheck_synapse_down(mtlsclient: TestClient) -> None: """A failing Synapse health endpoint makes the service unhealthy""" APP.state.synapse = AsyncMock() APP.state.rooms = {"space": "!space:x"} + mark_ready() try: with patch( "matrixrmapi.api.healthcheck.httpx.AsyncClient.get", @@ -60,3 +64,22 @@ def test_healthcheck_synapse_down(mtlsclient: TestClient) -> None: finally: del APP.state.synapse del APP.state.rooms + + +def test_healthcheck_unhealthy_when_a_worker_is_missing( + mtlsclient: TestClient, monkeypatch: pytest.MonkeyPatch +) -> None: + """All workers need to be ready with synapse/mas credentials to quarantee functionality""" + monkeypatch.setattr("matrixrmapi.api.healthcheck.WEB_CONCURRENCY", 4) + APP.state.synapse = AsyncMock() + APP.state.rooms = {"space": "!space:x"} + mark_ready() + try: + resp = mtlsclient.get("/api/v1/healthcheck") + assert resp.status_code == 200 + payload = resp.json() + assert payload["healthy"] is False + assert payload["extra"] == "only 1/4 workers initialised" + finally: + del APP.state.synapse + del APP.state.rooms From e4154fcc12b49751be662e84f193df90ba70a88e Mon Sep 17 00:00:00 2001 From: Akseli Lehtniemi <60463103+Akaseli@users.noreply.github.com> Date: Wed, 29 Jul 2026 13:17:23 +0300 Subject: [PATCH 2/3] chore: update ruff version --- .pre-commit-config.yaml | 2 +- src/matrixrmapi/api/__init__.py | 10 +++---- src/matrixrmapi/api/admininfo.py | 2 +- src/matrixrmapi/api/clientinfo.py | 7 +++-- src/matrixrmapi/api/description.py | 9 +++---- src/matrixrmapi/api/instructions.py | 3 +-- src/matrixrmapi/api/usercrud.py | 19 +++++++------ src/matrixrmapi/app.py | 11 ++++---- src/matrixrmapi/config.py | 10 +++---- src/matrixrmapi/console.py | 27 ++++++++++--------- src/matrixrmapi/types.py | 3 +-- src/matrixrmapi/utils/mas_admin.py | 14 +++++----- src/matrixrmapi/utils/startup.py | 37 +++++++++++++------------- src/matrixrmapi/utils/synapse_admin.py | 32 +++++++++++----------- tests/conftest.py | 10 +++---- tests/test_app.py | 6 ++--- tests/test_crud.py | 28 +++++++++---------- tests/test_fragment.py | 14 +++++----- tests/test_mas_admin.py | 10 +++---- tests/test_matrixrmapi.py | 1 + tests/test_synapse_admin.py | 18 ++++++------- 21 files changed, 134 insertions(+), 139 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index bfed49a..03b2c5f 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -77,7 +77,7 @@ repos: hooks: - id: uv-lock - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.14.11 + rev: v0.16.0 hooks: - id: ruff-check types_or: [python, pyi] diff --git a/src/matrixrmapi/api/__init__.py b/src/matrixrmapi/api/__init__.py index 8924a0c..b79a520 100644 --- a/src/matrixrmapi/api/__init__.py +++ b/src/matrixrmapi/api/__init__.py @@ -2,17 +2,15 @@ from fastapi.routing import APIRouter -from .usercrud import router as usercrud_router -from .clientinfo import router as clientinfo_router from .admininfo import router as admininfo_router -from .healthcheck import router as healthcheck_router +from .clientinfo import router as clientinfo_router from .description import router as description_router -from .instructions import router as instructions_router - from .description import router_v2 as description_router_v2 +from .healthcheck import router as healthcheck_router +from .instructions import router as instructions_router +from .usercrud import router as usercrud_router from .userinfo import router as userinfo_router - all_routers = APIRouter() all_routers.include_router(usercrud_router, prefix="/users", tags=["users"]) all_routers.include_router(clientinfo_router, prefix="/clients", tags=["clients"]) diff --git a/src/matrixrmapi/api/admininfo.py b/src/matrixrmapi/api/admininfo.py index aa92a46..a8af0de 100644 --- a/src/matrixrmapi/api/admininfo.py +++ b/src/matrixrmapi/api/admininfo.py @@ -3,9 +3,9 @@ import logging from fastapi import APIRouter, Depends +from jinja2 import Environment, FileSystemLoader from libpvarki.middleware import MTLSHeader from libpvarki.schemas.product import UserInstructionFragment -from jinja2 import Environment, FileSystemLoader from ..config import TEMPLATES_PATH diff --git a/src/matrixrmapi/api/clientinfo.py b/src/matrixrmapi/api/clientinfo.py index 5fb77ae..19bd1b1 100644 --- a/src/matrixrmapi/api/clientinfo.py +++ b/src/matrixrmapi/api/clientinfo.py @@ -1,10 +1,9 @@ """Endpoints for information for the end-user""" -from typing import List, Dict -import logging +import base64 import io +import logging import zipfile -import base64 from fastapi import APIRouter, Depends from libpvarki.middleware import MTLSHeader @@ -24,7 +23,7 @@ def zip_pem(pem: str, filename: str) -> bytes: @router.post("/fragment", deprecated=True) -async def client_instruction_fragment(user: UserCRUDRequest) -> List[Dict[str, str]]: +async def client_instruction_fragment(user: UserCRUDRequest) -> list[dict[str, str]]: """Return user instructions, we use POST because the integration layer might not keep track of callsigns and certs by UUID and will probably need both for the instructions""" zip1_bytes = zip_pem(user.x509cert, f"{user.callsign}_1.pem") diff --git a/src/matrixrmapi/api/description.py b/src/matrixrmapi/api/description.py index d54b88d..071189b 100644 --- a/src/matrixrmapi/api/description.py +++ b/src/matrixrmapi/api/description.py @@ -1,12 +1,11 @@ """Descriptions API""" -from typing import Literal, Optional import logging +from typing import Literal from fastapi import APIRouter, HTTPException -from pydantic import BaseModel, Field, Extra from libpvarki.schemas.product import ProductDescription - +from pydantic import BaseModel, Extra, Field LOGGER = logging.getLogger(__name__) @@ -31,7 +30,7 @@ class ProductDescriptionExtended(BaseModel): description="Short name for the product, used as slug/key in dicts and urls" ) title: str = Field(description="Fancy name for the product") - icon: Optional[str] = Field(description="URL for icon") + icon: str | None = Field(description="URL for icon") description: str = Field(description="Short-ish description of the product") language: str = Field(description="Language of this response") docs: str = Field(description="Link to documentation") @@ -49,7 +48,7 @@ class Config: ) async def return_product_description(language: str) -> ProductDescription: """Fetch description from each product in manifest""" - LOGGER.debug("Got language: {}".format(language)) + LOGGER.debug(f"Got language: {language}") if language == "fi": return ProductDescription( shortname=PRODUCT_SHORTNAME, diff --git a/src/matrixrmapi/api/instructions.py b/src/matrixrmapi/api/instructions.py index 296412c..df0cd43 100644 --- a/src/matrixrmapi/api/instructions.py +++ b/src/matrixrmapi/api/instructions.py @@ -1,6 +1,5 @@ """Instructions endpoints""" -from typing import Dict import logging from fastapi import APIRouter, Depends @@ -13,7 +12,7 @@ @router.post("/{language}") -async def user_intructions(user: UserCRUDRequest) -> Dict[str, str]: +async def user_intructions(user: UserCRUDRequest) -> dict[str, str]: """return user instructions""" return { "callsign": user.callsign, diff --git a/src/matrixrmapi/api/usercrud.py b/src/matrixrmapi/api/usercrud.py index db2c4e0..4d01e19 100644 --- a/src/matrixrmapi/api/usercrud.py +++ b/src/matrixrmapi/api/usercrud.py @@ -3,7 +3,6 @@ from __future__ import annotations import logging -from typing import Dict, Optional import httpx from fastapi import APIRouter, Depends, HTTPException, Request @@ -12,9 +11,9 @@ from libpvarki.schemas.product import UserCRUDRequest from ..config import get_manifest, get_server_domain +from ..types import AdminAction from ..utils.mas_admin import MasAdmin from ..utils.synapse_admin import SynapseAdmin, matrix_user_id -from ..types import AdminAction LOGGER = logging.getLogger(__name__) @@ -29,25 +28,25 @@ def comes_from_rm(request: Request) -> None: raise HTTPException(status_code=403) -def get_synapse(request: Request) -> Optional[SynapseAdmin]: +def get_synapse(request: Request) -> SynapseAdmin | None: """Return SynapseAdmin from app state, or None if not yet ready.""" - val: Optional[SynapseAdmin] = getattr(request.app.state, "synapse", None) + val: SynapseAdmin | None = getattr(request.app.state, "synapse", None) return val -def get_mas(request: Request) -> Optional[MasAdmin]: +def get_mas(request: Request) -> MasAdmin | None: """Return MasAdmin from app state, or None if not yet ready.""" - val: Optional[MasAdmin] = getattr(request.app.state, "mas", None) + val: MasAdmin | None = getattr(request.app.state, "mas", None) return val -def get_rooms(request: Request) -> Optional[Dict[str, str]]: +def get_rooms(request: Request) -> dict[str, str] | None: """Return room IDs dict from app state, or None if not yet ready.""" - val: Optional[Dict[str, str]] = getattr(request.app.state, "rooms", None) + val: dict[str, str] | None = getattr(request.app.state, "rooms", None) return val -def public_room_ids(rooms: Dict[str, str]) -> list[str]: +def public_room_ids(rooms: dict[str, str]) -> list[str]: """Room IDs for the space + the three public rooms (not admin channel).""" return [ rooms[k] for k in ("space", "general", "helpdesk", "offtopic") if k in rooms @@ -132,7 +131,7 @@ async def apply_admin_action( await synapse.force_join(admin_id, uid) else: await synapse.kick(admin_id, uid) - except Exception as exc: + except Exception as exc: # noqa: BLE001 LOGGER.error("Failed to %s %s: %s", action.value, uid, exc) return OperationResultResponse(success=False) LOGGER.info("%sd %s (power level %d)", action.value.capitalize(), uid, level) diff --git a/src/matrixrmapi/app.py b/src/matrixrmapi/app.py index 58201ac..c8d73a5 100644 --- a/src/matrixrmapi/app.py +++ b/src/matrixrmapi/app.py @@ -5,19 +5,20 @@ import asyncio import contextlib import logging +from collections.abc import AsyncGenerator from contextlib import asynccontextmanager -from typing import AsyncGenerator, Optional from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from libpvarki.logging import init_logging from matrixrmapi import __version__ -from .config import LOG_LEVEL, get_manifest + from .api import all_routers, all_routers_v2 +from .config import LOG_LEVEL, get_manifest from .utils.mas_admin import MasAdmin -from .utils.synapse_admin import SynapseAdmin from .utils.startup import connect_to_matrix +from .utils.synapse_admin import SynapseAdmin LOGGER = logging.getLogger(__name__) @@ -32,10 +33,10 @@ async def app_lifespan(app: FastAPI) -> AsyncGenerator[None, None]: task.cancel() with contextlib.suppress(asyncio.CancelledError): await task - synapse: Optional[SynapseAdmin] = getattr(app.state, "synapse", None) + synapse: SynapseAdmin | None = getattr(app.state, "synapse", None) if synapse: await synapse.close() - mas: Optional[MasAdmin] = getattr(app.state, "mas", None) + mas: MasAdmin | None = getattr(app.state, "mas", None) if mas: await mas.close() diff --git a/src/matrixrmapi/config.py b/src/matrixrmapi/config.py index ef249ac..647766b 100644 --- a/src/matrixrmapi/config.py +++ b/src/matrixrmapi/config.py @@ -1,10 +1,10 @@ """Configurations with .env support""" -from typing import Dict, Any, cast -from pathlib import Path -import json import functools +import json import logging +from pathlib import Path +from typing import Any, cast from starlette.config import Config @@ -30,7 +30,7 @@ @functools.cache -def get_manifest() -> Dict[str, Any]: +def get_manifest() -> dict[str, Any]: """Get manifest contents""" pth = Path("/pvarki/kraftwerk-init.json") if not pth.exists(): @@ -51,7 +51,7 @@ def get_manifest() -> Dict[str, Any]: }, } data = json.loads(pth.read_text(encoding="utf-8")) - return cast(Dict[str, Any], data) + return cast(dict[str, Any], data) def get_server_domain() -> str: diff --git a/src/matrixrmapi/console.py b/src/matrixrmapi/console.py index 57e2b6c..82f16ae 100644 --- a/src/matrixrmapi/console.py +++ b/src/matrixrmapi/console.py @@ -1,16 +1,15 @@ """CLI entrypoints for matrix product integration api""" import asyncio -import logging import json +import logging +import aiohttp import click from libadvian.logging import init_logging -import aiohttp from matrixrmapi import __version__ - LOGGER = logging.getLogger(__name__) @@ -54,16 +53,18 @@ async def doit() -> int: nonlocal host, port, timeout if "://" not in host: host = f"http://{host}" - async with aiohttp.ClientSession( - timeout=aiohttp.ClientTimeout(total=timeout) - ) as session: - async with session.get(f"{host}:{port}/api/v1/healthcheck") as resp: - if resp.status != 200: - return resp.status - payload = await resp.json() - click.echo(json.dumps(payload)) - if not payload["healthy"]: - return 1 + async with ( + aiohttp.ClientSession( + timeout=aiohttp.ClientTimeout(total=timeout) + ) as session, + session.get(f"{host}:{port}/api/v1/healthcheck") as resp, + ): + if resp.status != 200: + return resp.status + payload = await resp.json() + click.echo(json.dumps(payload)) + if not payload["healthy"]: + return 1 return 0 ctx.exit(asyncio.get_event_loop().run_until_complete(doit())) diff --git a/src/matrixrmapi/types.py b/src/matrixrmapi/types.py index 9d6f63f..cc9906b 100644 --- a/src/matrixrmapi/types.py +++ b/src/matrixrmapi/types.py @@ -1,7 +1,6 @@ """Shared domain types and constants.""" from enum import Enum -from typing import Dict class AdminAction(Enum): @@ -13,7 +12,7 @@ class AdminAction(Enum): # Call-related event types that regular users (power level 0) must be allowed to send. # Covers both legacy 1:1 calls and MSC3401 group calls (Element Call). -CALL_EVENTS_DEFAULT_LEVEL: Dict[str, int] = { +CALL_EVENTS_DEFAULT_LEVEL: dict[str, int] = { "m.call.invite": 0, "m.call.answer": 0, "m.call.hangup": 0, diff --git a/src/matrixrmapi/utils/mas_admin.py b/src/matrixrmapi/utils/mas_admin.py index ff9df59..851696d 100644 --- a/src/matrixrmapi/utils/mas_admin.py +++ b/src/matrixrmapi/utils/mas_admin.py @@ -4,7 +4,7 @@ import logging import time -from typing import Any, Dict, Optional, Tuple +from typing import Self from urllib.parse import quote import httpx @@ -36,7 +36,7 @@ def __init__(self, mas_url: str, client_id: str, client_secret: str) -> None: self._url = mas_url.rstrip("/") self._client_id = client_id self._client_secret = client_secret - self._token: Optional[str] = None + self._token: str | None = None self._token_expires: float = 0.0 self._client: httpx.AsyncClient = httpx.AsyncClient() @@ -44,10 +44,10 @@ async def close(self) -> None: """Close the underlying HTTP client.""" await self._client.aclose() - async def __aenter__(self) -> "MasAdmin": + async def __aenter__(self) -> Self: return self - async def __aexit__(self, *_: Any) -> None: + async def __aexit__(self, *_: object) -> None: await self.close() async def _ensure_admin_token(self) -> str: @@ -68,10 +68,10 @@ async def _ensure_admin_token(self) -> str: ) return self._token - async def _auth(self) -> Dict[str, str]: + async def _auth(self) -> dict[str, str]: return {"Authorization": f"Bearer {await self._ensure_admin_token()}"} - async def user_ulid_by_username(self, localpart: str) -> Optional[str]: + async def user_ulid_by_username(self, localpart: str) -> str | None: """Return the MAS user ULID for localpart, or None if not found.""" encoded = quote(localpart, safe="") resp = await self._client.get( @@ -109,7 +109,7 @@ async def create_bot_token( human_name: str, device_id: str = BOT_DEVICE_ID, expires_in: int = BOT_TOKEN_EXPIRES_IN, - ) -> Tuple[str, float]: + ) -> tuple[str, float]: """Create an expiring personal session token for the bot user. Returns (access_token, expires_in seconds). diff --git a/src/matrixrmapi/utils/startup.py b/src/matrixrmapi/utils/startup.py index 1416cdd..4ac080f 100644 --- a/src/matrixrmapi/utils/startup.py +++ b/src/matrixrmapi/utils/startup.py @@ -7,7 +7,6 @@ import os import tempfile from pathlib import Path -from typing import Dict, List, Optional, Tuple import filelock import httpx @@ -23,7 +22,7 @@ get_manifest, get_server_domain, ) -from ..types import AdminAction, CALL_EVENTS_DEFAULT_LEVEL +from ..types import CALL_EVENTS_DEFAULT_LEVEL, AdminAction from .mas_admin import MasAdmin from .synapse_admin import SynapseAdmin @@ -35,7 +34,7 @@ READY_DIR = Path(tempfile.gettempdir()) / "matrixrmapi_ready" # (key, alias_suffix, display_name, is_space, is_private) -ROOMS_CONFIG: List[Tuple[str, str, str, bool, bool]] = [ +ROOMS_CONFIG: list[tuple[str, str, str, bool, bool]] = [ ("space", "{d}-space", "{d}", True, False), ("admin", "{d}-admin", "96-Admin channel", False, True), ("general", "{d}-general", "98-General", False, False), @@ -43,7 +42,7 @@ ("offtopic", "{d}-offtopic", "97-Offtopic", False, False), ] -ROOM_TOPICS: Dict[str, str] = { +ROOM_TOPICS: dict[str, str] = { "general": "Work discussion that does not fit any other room.", "helpdesk": "Report issues and get help from here.", "offtopic": "Everything that is not about the topics or work.", @@ -90,8 +89,8 @@ async def wait_for_service( if resp.status_code == 200: LOGGER.info("%s is ready", name) return True - except Exception: # nosec B110 - pass + except Exception as exc: # noqa: BLE001 + LOGGER.debug("%s health check failed: %s", name, exc) if attempt < retries - 1: await asyncio.sleep(interval) LOGGER.error( @@ -100,7 +99,7 @@ async def wait_for_service( return False -async def acquire_bot_token(synapse: SynapseAdmin, mas: MasAdmin) -> Tuple[bool, bool]: +async def acquire_bot_token(synapse: SynapseAdmin, mas: MasAdmin) -> tuple[bool, bool]: """Set up the bot session, using a file lock for worker coordination. Returns ``(success, is_init_worker)``. Only the init worker @@ -129,7 +128,7 @@ async def acquire_bot_token(synapse: SynapseAdmin, mas: MasAdmin) -> Tuple[bool, try: await synapse.setup(SYNAPSE_BOT_USERNAME, mas) return True, is_init - except Exception as exc: + except Exception as exc: # noqa: BLE001 LOGGER.error("Bot session setup failed: %s", exc) return False, False finally: @@ -153,10 +152,10 @@ async def ensure_room( async def ensure_rooms( synapse: SynapseAdmin, deployment: str, domain: str -) -> Dict[str, str]: +) -> dict[str, str]: """Create space and rooms if they don't exist; return room IDs dict.""" - room_ids: Dict[str, str] = {} - space_id: Optional[str] = None + room_ids: dict[str, str] = {} + space_id: str | None = None for key, alias_tpl, name_tpl, is_space, is_private in ROOMS_CONFIG: alias = f"#{alias_tpl.format(d=deployment)}:{domain}" @@ -175,7 +174,7 @@ async def ensure_rooms( async def apply_pending( - synapse: SynapseAdmin, rooms: Dict[str, str], pending: Dict[str, AdminAction] + synapse: SynapseAdmin, rooms: dict[str, str], pending: dict[str, AdminAction] ) -> None: """Apply promotions/demotions that were queued while Synapse was still starting.""" public_ids = [ @@ -194,14 +193,14 @@ async def apply_pending( if admin_id: await synapse.kick(admin_id, uid) LOGGER.info("Applied deferred demotion for %s", uid) - except Exception as exc: + except Exception as exc: # noqa: BLE001 LOGGER.error( "Failed to apply deferred %s for %s: %s", action.value, uid, exc ) async def configure_rooms_state( - synapse: SynapseAdmin, rooms: Dict[str, str], deployment: str + synapse: SynapseAdmin, rooms: dict[str, str], deployment: str ) -> None: """Apply join rules, encryption, history visibility, topics and names to all rooms. @@ -255,7 +254,7 @@ async def configure_rooms_state( LOGGER.info("Room state configuration applied") -def setup_mas_admin(app: FastAPI) -> Optional[MasAdmin]: +def setup_mas_admin(app: FastAPI) -> MasAdmin | None: """Build the MAS admin client from the shared client id and secret.""" if not MAS_ADMIN_CLIENT_SECRET: LOGGER.error( @@ -296,7 +295,7 @@ async def init_matrix_once(app: FastAPI, mas: MasAdmin) -> bool: try: room_ids = await ensure_rooms(synapse, deployment, domain) - except Exception as exc: + except Exception as exc: # noqa: BLE001 LOGGER.error("Room setup failed: %s", exc) await synapse.close() return False @@ -306,7 +305,7 @@ async def init_matrix_once(app: FastAPI, mas: MasAdmin) -> bool: # duplicate PUTs from every worker on every restart. try: await configure_rooms_state(synapse, room_ids, deployment) - except Exception as exc: + except Exception as exc: # noqa: BLE001 LOGGER.error("Room configuration failed (rooms still usable): %s", exc) else: LOGGER.info( @@ -323,7 +322,7 @@ async def init_matrix_once(app: FastAPI, mas: MasAdmin) -> bool: # Apply any promotions/demotions that arrived while rooms were not yet set. # Snapshot and clear atomically (no await between) so any new requests that # arrive during apply_pending go into the now-empty dict, not the snapshot. - pending: Dict[str, AdminAction] = dict(app.state.pending_promotions) + pending: dict[str, AdminAction] = dict(app.state.pending_promotions) app.state.pending_promotions.clear() if pending: LOGGER.info("Processing %d deferred promotion(s)/demotion(s)", len(pending)) @@ -351,7 +350,7 @@ async def connect_to_matrix(app: FastAPI) -> None: if await init_matrix_once(app, mas): return reason = "init did not complete" - except Exception as exc: + except Exception as exc: # noqa: BLE001 reason = f"{type(exc).__name__}: {exc}" LOGGER.error( "Matrix init attempt %d failed (%s), retrying in %.0fs", diff --git a/src/matrixrmapi/utils/synapse_admin.py b/src/matrixrmapi/utils/synapse_admin.py index 9e1c9a7..2e647df 100644 --- a/src/matrixrmapi/utils/synapse_admin.py +++ b/src/matrixrmapi/utils/synapse_admin.py @@ -6,7 +6,7 @@ import logging import re import time -from typing import Any, Dict, List, Optional +from typing import Any, Self from urllib.parse import quote import httpx @@ -41,23 +41,23 @@ class SynapseAdmin: def __init__(self, synapse_url: str, server_domain: str) -> None: self._url = synapse_url.rstrip("/") self._domain = server_domain - self._token: Optional[str] = None + self._token: str | None = None self._token_expires: float = 0.0 self._token_lock: asyncio.Lock = asyncio.Lock() - self._mas: Optional[MasAdmin] = None - self._bot_username: Optional[str] = None - self._bot_ulid: Optional[str] = None - self._bot_user_id: Optional[str] = None + self._mas: MasAdmin | None = None + self._bot_username: str | None = None + self._bot_ulid: str | None = None + self._bot_user_id: str | None = None self._client: httpx.AsyncClient = httpx.AsyncClient() async def close(self) -> None: """Close the underlying HTTP client.""" await self._client.aclose() - async def __aenter__(self) -> "SynapseAdmin": + async def __aenter__(self) -> Self: return self - async def __aexit__(self, *_: Any) -> None: + async def __aexit__(self, *_: object) -> None: await self.close() # ------------------------------------------------------------------ @@ -109,10 +109,10 @@ async def _exempt_bot_from_ratelimit(self, bot_username: str) -> None: ) resp.raise_for_status() LOGGER.info("Rate-limit override applied for %s", user_id) - except Exception as exc: + except Exception as exc: # noqa: BLE001 LOGGER.warning("Failed to override rate limit for %s: %s", user_id, exc) - async def _auth(self) -> Dict[str, str]: + async def _auth(self) -> dict[str, str]: """Return the bearer header, creating a new session when missing or expired.""" if not self._token: raise RuntimeError("SynapseAdmin.setup() has not been called") @@ -126,7 +126,7 @@ async def _auth(self) -> Dict[str, str]: # Room / space management # ------------------------------------------------------------------ - async def room_id_for_alias(self, alias: str) -> Optional[str]: + async def room_id_for_alias(self, alias: str) -> str | None: """Return room_id for alias, or None if not found.""" encoded = quote(alias, safe="") resp = await self._client.get( @@ -149,7 +149,7 @@ async def create_room( ) -> str: """Create a room or space; return room_id.""" local_part = alias.split(":")[0].lstrip("#") - body: Dict[str, Any] = { + body: dict[str, Any] = { "name": name, "room_alias_name": local_part, "preset": "private_chat" if is_private else "public_chat", @@ -190,7 +190,7 @@ async def set_room_state( self, room_id: str, event_type: str, - content: Dict[str, Any], + content: dict[str, Any], state_key: str = "", ) -> None: """Send a room state event.""" @@ -237,7 +237,7 @@ async def force_join(self, room_id: str, user_id: str) -> None: return resp.raise_for_status() - async def get_power_levels(self, room_id: str) -> Dict[str, Any]: + async def get_power_levels(self, room_id: str) -> dict[str, Any]: """Get the m.room.power_levels state for a room.""" resp = await self._client.get( f"{self._url}/_matrix/client/v3/rooms/{room_id}/state/m.room.power_levels", @@ -252,7 +252,7 @@ async def set_user_power_level( ) -> None: """Set a single user's power level in a room.""" levels = await self.get_power_levels(room_id) - users: Dict[str, int] = dict(levels.get("users", {})) + users: dict[str, int] = dict(levels.get("users", {})) if level == 0: users.pop(user_id, None) else: @@ -300,7 +300,7 @@ async def kick(self, room_id: str, user_id: str) -> None: # ------------------------------------------------------------------ async def set_power_level_in_rooms( - self, room_ids: List[str], user_id: str, level: int + self, room_ids: list[str], user_id: str, level: int ) -> None: """Set power level for user across multiple rooms.""" for room_id in room_ids: diff --git a/tests/conftest.py b/tests/conftest.py index fa345b5..59b5da2 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,14 +1,14 @@ """pytest automagics""" -from typing import Generator, Dict -from pathlib import Path import logging import os import uuid +from collections.abc import Generator +from pathlib import Path -from libpvarki.logging import init_logging import pytest from fastapi.testclient import TestClient +from libpvarki.logging import init_logging from matrixrmapi.app import get_app from matrixrmapi.config import get_manifest @@ -57,7 +57,7 @@ def rm_mtlsclient() -> Generator[TestClient, None, None]: yield client -def create_user_dict(callsign: str) -> Dict[str, str]: +def create_user_dict(callsign: str) -> dict[str, str]: """return valid user dict for crud operations""" return { "uuid": str(uuid.uuid4()), @@ -67,6 +67,6 @@ def create_user_dict(callsign: str) -> Dict[str, str]: @pytest.fixture(scope="session") -def norppa11() -> Dict[str, str]: +def norppa11() -> dict[str, str]: """Session scoped user dict (to keep same UUID)""" return create_user_dict("NORPPA11a") diff --git a/tests/test_app.py b/tests/test_app.py index 340f0eb..54d3dd5 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -3,17 +3,17 @@ from __future__ import annotations import os -from typing import Dict, cast +from typing import cast from unittest.mock import AsyncMock import pytest +from matrixrmapi.types import AdminAction from matrixrmapi.utils import startup from matrixrmapi.utils.startup import apply_pending, ensure_room from matrixrmapi.utils.synapse_admin import SynapseAdmin -from matrixrmapi.types import AdminAction -ROOMS: Dict[str, str] = { +ROOMS: dict[str, str] = { "space": "!space:x", "admin": "!admin:x", "general": "!general:x", diff --git a/tests/test_crud.py b/tests/test_crud.py index 24f5737..a039cf2 100644 --- a/tests/test_crud.py +++ b/tests/test_crud.py @@ -1,6 +1,5 @@ """Test the CRUD operations""" -from typing import Dict import logging import uuid from unittest.mock import AsyncMock @@ -10,19 +9,20 @@ from matrixrmapi.config import get_server_domain from matrixrmapi.types import AdminAction + from .conftest import APP LOGGER = logging.getLogger(__name__) -def test_unauth(norppa11: Dict[str, str]) -> None: +def test_unauth(norppa11: dict[str, str]) -> None: """Check that unauth call to auth endpoint fails""" client = TestClient(APP) resp = client.post("/api/v1/users/created", json=norppa11) assert resp.status_code == 403 -def test_create(norppa11: Dict[str, str], rm_mtlsclient: TestClient) -> None: +def test_create(norppa11: dict[str, str], rm_mtlsclient: TestClient) -> None: """Check that adding user works""" resp = rm_mtlsclient.post("/api/v1/users/created", json=norppa11) assert resp.status_code == 200 @@ -31,7 +31,7 @@ def test_create(norppa11: Dict[str, str], rm_mtlsclient: TestClient) -> None: assert payload["success"] -def test_update(norppa11: Dict[str, str], rm_mtlsclient: TestClient) -> None: +def test_update(norppa11: dict[str, str], rm_mtlsclient: TestClient) -> None: """Check that updating user works""" resp = rm_mtlsclient.put("/api/v1/users/updated", json=norppa11) assert resp.status_code == 200 @@ -40,7 +40,7 @@ def test_update(norppa11: Dict[str, str], rm_mtlsclient: TestClient) -> None: assert payload["success"] -def test_revoke(norppa11: Dict[str, str], rm_mtlsclient: TestClient) -> None: +def test_revoke(norppa11: dict[str, str], rm_mtlsclient: TestClient) -> None: """Check that revoking user works (MAS not ready -> success with warning)""" resp = rm_mtlsclient.post("/api/v1/users/revoked", json=norppa11) assert resp.status_code == 200 @@ -50,7 +50,7 @@ def test_revoke(norppa11: Dict[str, str], rm_mtlsclient: TestClient) -> None: def test_revoke_deactivates_in_mas( - norppa11: Dict[str, str], rm_mtlsclient: TestClient + norppa11: dict[str, str], rm_mtlsclient: TestClient ) -> None: """When MAS is ready, /revoked must deactivate the user via the MAS admin API""" mas = AsyncMock() @@ -66,7 +66,7 @@ def test_revoke_deactivates_in_mas( def test_revoke_user_not_in_mas( - norppa11: Dict[str, str], rm_mtlsclient: TestClient + norppa11: dict[str, str], rm_mtlsclient: TestClient ) -> None: """User that never logged in (absent from MAS) still revokes successfully""" mas = AsyncMock() @@ -81,7 +81,7 @@ def test_revoke_user_not_in_mas( def test_revoke_mas_error_fails( - norppa11: Dict[str, str], rm_mtlsclient: TestClient + norppa11: dict[str, str], rm_mtlsclient: TestClient ) -> None: """A MAS API failure must be reported as success=False""" mas = AsyncMock() @@ -95,7 +95,7 @@ def test_revoke_mas_error_fails( del APP.state.mas -def test_promote(norppa11: Dict[str, str], rm_mtlsclient: TestClient) -> None: +def test_promote(norppa11: dict[str, str], rm_mtlsclient: TestClient) -> None: """Check that promoting user works""" resp = rm_mtlsclient.post("/api/v1/users/promoted", json=norppa11) assert resp.status_code == 200 @@ -104,7 +104,7 @@ def test_promote(norppa11: Dict[str, str], rm_mtlsclient: TestClient) -> None: assert payload["success"] -def test_demote(norppa11: Dict[str, str], rm_mtlsclient: TestClient) -> None: +def test_demote(norppa11: dict[str, str], rm_mtlsclient: TestClient) -> None: """Check that demoting user works""" resp = rm_mtlsclient.post("/api/v1/users/demoted", json=norppa11) assert resp.status_code == 200 @@ -118,7 +118,7 @@ def test_demote(norppa11: Dict[str, str], rm_mtlsclient: TestClient) -> None: # --------------------------------------------------------------------------- -def _unique_user() -> Dict[str, str]: +def _unique_user() -> dict[str, str]: """Return a user dict with a unique callsign to avoid state collisions.""" tag = uuid.uuid4().hex[:6] return { @@ -136,7 +136,7 @@ def test_promote_queues_uid_when_synapse_not_ready(rm_mtlsclient: TestClient) -> assert resp.json()["success"] is True uid = f"@{user['callsign'].lower()}:{get_server_domain()}" - pending: Dict[str, AdminAction] = getattr(APP.state, "pending_promotions", {}) + pending: dict[str, AdminAction] = getattr(APP.state, "pending_promotions", {}) assert pending.get(uid) is AdminAction.PROMOTE @@ -148,7 +148,7 @@ def test_demote_queues_uid_when_synapse_not_ready(rm_mtlsclient: TestClient) -> assert resp.json()["success"] is True uid = f"@{user['callsign'].lower()}:{get_server_domain()}" - pending: Dict[str, AdminAction] = getattr(APP.state, "pending_promotions", {}) + pending: dict[str, AdminAction] = getattr(APP.state, "pending_promotions", {}) assert pending.get(uid) is AdminAction.DEMOTE @@ -160,5 +160,5 @@ def test_demote_overwrites_pending_promote(rm_mtlsclient: TestClient) -> None: rm_mtlsclient.post("/api/v1/users/promoted", json=user) rm_mtlsclient.post("/api/v1/users/demoted", json=user) - pending: Dict[str, AdminAction] = getattr(APP.state, "pending_promotions", {}) + pending: dict[str, AdminAction] = getattr(APP.state, "pending_promotions", {}) assert pending.get(uid) is AdminAction.DEMOTE diff --git a/tests/test_fragment.py b/tests/test_fragment.py index 1411647..89f9a09 100644 --- a/tests/test_fragment.py +++ b/tests/test_fragment.py @@ -1,26 +1,26 @@ """Test the HTML fragment""" -from typing import Dict -import logging import base64 +import logging -from fastapi.testclient import TestClient import pytest +from fastapi.testclient import TestClient from matrixrmapi.config import get_manifest + from .conftest import APP LOGGER = logging.getLogger(__name__) -def test_unauth(norppa11: Dict[str, str]) -> None: +def test_unauth(norppa11: dict[str, str]) -> None: """Check that unauth call to auth endpoint fails""" client = TestClient(APP) resp = client.post("/api/v1/clients/fragment", json=norppa11) assert resp.status_code == 403 -def test_get_fragment(norppa11: Dict[str, str], mtlsclient: TestClient) -> None: +def test_get_fragment(norppa11: dict[str, str], mtlsclient: TestClient) -> None: """Check that getting fragment works""" resp = mtlsclient.post("/api/v1/clients/fragment", json=norppa11) assert resp.status_code == 200 @@ -47,7 +47,7 @@ def test_get_admin_fragment(mtlsclient: TestClient) -> None: @pytest.mark.parametrize("lang", ["en", "fi", "sv"]) def test_get_v2_user_markdown( - norppa11: Dict[str, str], rm_mtlsclient: TestClient, lang: str + norppa11: dict[str, str], rm_mtlsclient: TestClient, lang: str ) -> None: """Check that getting v2 user markdown works""" manifest = get_manifest() @@ -59,7 +59,7 @@ def test_get_v2_user_markdown( @pytest.mark.parametrize("lang", ["en", "fi", "sv"]) def test_get_v2_user_markdown_wrongcaller( - norppa11: Dict[str, str], mtlsclient: TestClient, lang: str + norppa11: dict[str, str], mtlsclient: TestClient, lang: str ) -> None: """Check that getting v2 user markdown fails if not coming from RASENMAEHER""" resp = mtlsclient.post(f"/api/v2/clients/{lang}/info.md", json=norppa11) diff --git a/tests/test_mas_admin.py b/tests/test_mas_admin.py index 3ae6beb..6faff91 100644 --- a/tests/test_mas_admin.py +++ b/tests/test_mas_admin.py @@ -7,7 +7,7 @@ from __future__ import annotations import time -from typing import Any, Dict +from typing import Any from unittest.mock import AsyncMock, patch import httpx @@ -23,7 +23,7 @@ BOT_ULID = "01TESTULID0000000000000000" -def _fake(status: int, body: Dict[str, Any]) -> httpx.Response: +def _fake(status: int, body: dict[str, Any]) -> httpx.Response: """Build a minimal fake httpx.Response.""" return httpx.Response(status, json=body, request=FAKE_REQUEST) @@ -55,7 +55,7 @@ async def test_admin_token_request_shape() -> None: token = await mas._ensure_admin_token() assert token == "tok" # nosec B105 assert mock_post.call_args.kwargs["auth"] == ("clientid", "clientsecret") - data: Dict[str, str] = mock_post.call_args.kwargs["data"] + data: dict[str, str] = mock_post.call_args.kwargs["data"] assert data["grant_type"] == "client_credentials" assert data["scope"] == "urn:mas:admin" @@ -116,7 +116,7 @@ async def test_ensure_user_creates_on_404() -> None: mock_post.return_value = _fake(201, {"data": {"id": BOT_ULID}}) ulid = await mas.ensure_user("bot") assert ulid == BOT_ULID - body: Dict[str, Any] = mock_post.call_args.kwargs["json"] + body: dict[str, Any] = mock_post.call_args.kwargs["json"] assert body == {"username": "bot"} @@ -141,7 +141,7 @@ async def test_create_bot_token_scopes_and_expiry() -> None: token, expires_in = await mas.create_bot_token(BOT_ULID, "matrixrmapi bot") assert token == "mpt_bot_token" # nosec B105 assert expires_in == 3600 - body: Dict[str, Any] = mock_post.call_args.kwargs["json"] + body: dict[str, Any] = mock_post.call_args.kwargs["json"] assert body["expires_in"] == 3600 assert body["actor_user_id"] == BOT_ULID scope: str = body["scope"] diff --git a/tests/test_matrixrmapi.py b/tests/test_matrixrmapi.py index df36143..4121992 100644 --- a/tests/test_matrixrmapi.py +++ b/tests/test_matrixrmapi.py @@ -8,6 +8,7 @@ from matrixrmapi import __version__ from matrixrmapi.utils.startup import mark_ready + from .conftest import APP diff --git a/tests/test_synapse_admin.py b/tests/test_synapse_admin.py index 03c620d..e45d455 100644 --- a/tests/test_synapse_admin.py +++ b/tests/test_synapse_admin.py @@ -7,7 +7,7 @@ from __future__ import annotations import time -from typing import Any, Dict +from typing import Any from unittest.mock import AsyncMock, patch import httpx @@ -19,7 +19,7 @@ FAKE_REQUEST = httpx.Request("POST", "http://synapse.test/fake") -def _fake(status: int, body: Dict[str, Any]) -> httpx.Response: +def _fake(status: int, body: dict[str, Any]) -> httpx.Response: """Build a minimal fake httpx.Response.""" return httpx.Response(status, json=body, request=FAKE_REQUEST) @@ -168,8 +168,8 @@ async def test_create_room_sets_bot_at_power_200() -> None: with patch.object(sa._client, "post", new_callable=AsyncMock) as mock_post: mock_post.return_value = _fake(200, {"room_id": "!new:example.test"}) await sa.create_room("TestRoom", "#test-room:example.test") - body: Dict[str, Any] = mock_post.call_args.kwargs["json"] - users: Dict[str, int] = body.get("power_level_content_override", {}).get( + body: dict[str, Any] = mock_post.call_args.kwargs["json"] + users: dict[str, int] = body.get("power_level_content_override", {}).get( "users", {} ) assert users.get("@bot:example.test") == 200 @@ -182,7 +182,7 @@ async def test_create_space_sets_creation_content() -> None: with patch.object(sa._client, "post", new_callable=AsyncMock) as mock_post: mock_post.return_value = _fake(200, {"room_id": "!space:example.test"}) await sa.create_room("MySpace", "#my-space:example.test", is_space=True) - body: Dict[str, Any] = mock_post.call_args.kwargs["json"] + body: dict[str, Any] = mock_post.call_args.kwargs["json"] assert body.get("creation_content", {}).get("type") == "m.space" @@ -193,7 +193,7 @@ async def test_create_private_room_uses_private_preset() -> None: with patch.object(sa._client, "post", new_callable=AsyncMock) as mock_post: mock_post.return_value = _fake(200, {"room_id": "!priv:example.test"}) await sa.create_room("Admin", "#admin:example.test", is_private=True) - body: Dict[str, Any] = mock_post.call_args.kwargs["json"] + body: dict[str, Any] = mock_post.call_args.kwargs["json"] assert body.get("preset") == "private_chat" @@ -301,7 +301,7 @@ async def test_set_user_power_level_nonzero() -> None: mock_get.return_value = _fake(200, initial) mock_put.return_value = _fake(200, {}) await sa.set_user_power_level("!r:example.test", "@user:example.test", 100) - body: Dict[str, Any] = mock_put.call_args.kwargs["json"] + body: dict[str, Any] = mock_put.call_args.kwargs["json"] assert body["users"]["@user:example.test"] == 100 @@ -318,7 +318,7 @@ async def test_set_user_power_level_zero_removes_user() -> None: mock_get.return_value = _fake(200, initial) mock_put.return_value = _fake(200, {}) await sa.set_user_power_level("!r:example.test", "@user:example.test", 0) - body: Dict[str, Any] = mock_put.call_args.kwargs["json"] + body: dict[str, Any] = mock_put.call_args.kwargs["json"] assert "@user:example.test" not in body["users"] @@ -334,7 +334,7 @@ async def test_invite_success() -> None: with patch.object(sa._client, "post", new_callable=AsyncMock) as mock_post: mock_post.return_value = _fake(200, {}) await sa.invite("!r:example.test", "@user:example.test") - body: Dict[str, Any] = mock_post.call_args.kwargs["json"] + body: dict[str, Any] = mock_post.call_args.kwargs["json"] assert body["user_id"] == "@user:example.test" From 71ed97950d0d6b1312d1632c63fb383b6ae74031 Mon Sep 17 00:00:00 2001 From: Akseli Lehtniemi <60463103+Akaseli@users.noreply.github.com> Date: Wed, 29 Jul 2026 15:25:25 +0300 Subject: [PATCH 3/3] fix: allow configuring connection attempts --- src/matrixrmapi/config.py | 5 ++++ src/matrixrmapi/utils/startup.py | 47 +++++++++++++++++++------------- 2 files changed, 33 insertions(+), 19 deletions(-) diff --git a/src/matrixrmapi/config.py b/src/matrixrmapi/config.py index 647766b..d696b7d 100644 --- a/src/matrixrmapi/config.py +++ b/src/matrixrmapi/config.py @@ -28,6 +28,11 @@ SYNAPSE_BOT_USERNAME: str = cfg("SYNAPSE_BOT_USERNAME", default="matrixrmapi-bot") WEB_CONCURRENCY: int = cfg("WEB_CONCURRENCY", default=1, cast=int) +INIT_MAX_ATTEMPTS: int = cfg("INIT_MAX_ATTEMPTS", default=5, cast=int) +INIT_RETRY_WAIT: float = cfg("INIT_RETRY_WAIT", default=5.0, cast=float) +SERVICE_WAIT_RETRIES: int = cfg("SERVICE_WAIT_RETRIES", default=12, cast=int) +SERVICE_WAIT_INTERVAL: float = cfg("SERVICE_WAIT_INTERVAL", default=5.0, cast=float) + @functools.cache def get_manifest() -> dict[str, Any]: diff --git a/src/matrixrmapi/utils/startup.py b/src/matrixrmapi/utils/startup.py index 4ac080f..b1f9c16 100644 --- a/src/matrixrmapi/utils/startup.py +++ b/src/matrixrmapi/utils/startup.py @@ -13,10 +13,14 @@ from fastapi import FastAPI from ..config import ( + INIT_MAX_ATTEMPTS, + INIT_RETRY_WAIT, MAS_ADMIN_CLIENT_ID, MAS_ADMIN_CLIENT_SECRET, MAS_HEALTH_URL, MAS_URL, + SERVICE_WAIT_INTERVAL, + SERVICE_WAIT_RETRIES, SYNAPSE_BOT_USERNAME, SYNAPSE_URL, get_manifest, @@ -28,9 +32,6 @@ LOGGER = logging.getLogger(__name__) -INIT_RETRY_BACKOFF = 5.0 -INIT_RETRY_BACKOFF_MAX = 60.0 - READY_DIR = Path(tempfile.gettempdir()) / "matrixrmapi_ready" # (key, alias_suffix, display_name, is_space, is_private) @@ -78,7 +79,10 @@ def ready_workers() -> int: async def wait_for_service( - name: str, url: str, retries: int = 60, interval: float = 5.0 + name: str, + url: str, + retries: int = SERVICE_WAIT_RETRIES, + interval: float = SERVICE_WAIT_INTERVAL, ) -> bool: """Poll a service's /health until it responds 200. Returns True on success.""" LOGGER.info("Waiting for %s at %s ...", name, url) @@ -93,9 +97,7 @@ async def wait_for_service( LOGGER.debug("%s health check failed: %s", name, exc) if attempt < retries - 1: await asyncio.sleep(interval) - LOGGER.error( - "%s not reachable after %d attempts — integration disabled", name, retries - ) + LOGGER.error("%s not reachable after %d attempts", name, retries) return False @@ -275,9 +277,11 @@ def setup_mas_admin(app: FastAPI) -> MasAdmin | None: async def init_matrix_once(app: FastAPI, mas: MasAdmin) -> bool: """Init attempt: wait for services, get a bot session, ensure rooms.""" - if not await wait_for_service("MAS", MAS_HEALTH_URL): - return False - if not await wait_for_service("Synapse", SYNAPSE_URL): + mas_ok, synapse_ok = await asyncio.gather( + wait_for_service("MAS", MAS_HEALTH_URL), + wait_for_service("Synapse", SYNAPSE_URL), + ) + if not (mas_ok and synapse_ok): return False manifest = get_manifest() @@ -297,6 +301,7 @@ async def init_matrix_once(app: FastAPI, mas: MasAdmin) -> bool: room_ids = await ensure_rooms(synapse, deployment, domain) except Exception as exc: # noqa: BLE001 LOGGER.error("Room setup failed: %s", exc) + app.state.synapse = None await synapse.close() return False @@ -333,7 +338,7 @@ async def init_matrix_once(app: FastAPI, mas: MasAdmin) -> bool: async def connect_to_matrix(app: FastAPI) -> None: """Background task: connect to MAS and Synapse, create bot and rooms. - Retries with backoff until it succeeds. + Retries a bounded number of times with a fixed wait, then gives up. """ clear_ready() mas = setup_mas_admin(app) @@ -342,21 +347,25 @@ async def connect_to_matrix(app: FastAPI) -> None: LOGGER.error("No MAS admin client; Matrix integration disabled!") return - backoff = INIT_RETRY_BACKOFF - attempt = 0 - while True: - attempt += 1 + for attempt in range(1, INIT_MAX_ATTEMPTS + 1): try: if await init_matrix_once(app, mas): return reason = "init did not complete" except Exception as exc: # noqa: BLE001 reason = f"{type(exc).__name__}: {exc}" + if attempt == INIT_MAX_ATTEMPTS: + LOGGER.error( + "Matrix init failed for a worker after %d attempts (%s)", + INIT_MAX_ATTEMPTS, + reason, + ) + return LOGGER.error( - "Matrix init attempt %d failed (%s), retrying in %.0fs", + "Matrix init attempt %d/%d failed (%s), retrying in %.0fs", attempt, + INIT_MAX_ATTEMPTS, reason, - backoff, + INIT_RETRY_WAIT, ) - await asyncio.sleep(backoff) - backoff = min(backoff * 2, INIT_RETRY_BACKOFF_MAX) + await asyncio.sleep(INIT_RETRY_WAIT)