omniroute_offload: always-on gateway ops
Some checks failed
dependency freshness / dependencies on newest stable (push) Has been cancelled
file length / file-length (push) Has been cancelled
md-naming / markdown naming convention (push) Has been cancelled
omniroute_offload pre-commit / pre-commit (push) Has been cancelled
palette drift / palette (push) Has been cancelled
repo-contract / repo contract (push) Has been cancelled
shared gate tests / gate unit tests (push) Has been cancelled

omniroute-gateway start|health|update|backup|combo keeps the local
OmniRoute gateway up. systemd --user timers: health every 10 min
(real free-best chat; restart <=1/30 min, reactivate providers whose
cooldown ended, notify once per outage), update weekly Sun 12:00
(backup, newest stable -web tag, rollback of image + volume on a
failed health check).

Failure modes: health/update fail loudly (exit 1 + desktop + Signal);
all providers cooling = degraded (exit 0, one notice/day); a missing
signal-bot .env logs and exits non-zero on alert paths; health skips
while update holds the lock.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WzxdWZSdJmNjGUvLnHJ4WG
This commit is contained in:
Krzysztof Rudnicki 2026-11-05 13:34:17 +01:00
parent c2165e9fb8
commit 7af6afe51a
25 changed files with 2748 additions and 4 deletions

View File

@ -1,13 +1,19 @@
#!/bin/bash #!/bin/bash
# Install free-llm (stdlib-only, system python, editable) and link claude-free into ~/.local/bin. # Install free-llm + omniroute-gateway (system python, editable), claude-free, and the gateway timers.
set -euo pipefail set -euo pipefail
REPO_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" REPO_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
readonly REPO_DIR readonly REPO_DIR
readonly SYSTEM_PYTHON="/usr/bin/python3" readonly SYSTEM_PYTHON="/usr/bin/python3"
readonly BIN_DIR="${HOME}/.local/bin" readonly BIN_DIR="${HOME}/.local/bin"
readonly UNIT_DIR="${HOME}/.config/systemd/user"
"${SYSTEM_PYTHON}" -m pip install --user --no-deps --break-system-packages -q -e "${REPO_DIR}" "${SYSTEM_PYTHON}" -m pip install --user --no-deps --break-system-packages -q -e "${REPO_DIR}"
mkdir -p "${BIN_DIR}" mkdir -p "${BIN_DIR}" "${UNIT_DIR}"
ln -sf "${REPO_DIR}/bin/claude-free" "${BIN_DIR}/claude-free" ln -sf "${REPO_DIR}/bin/claude-free" "${BIN_DIR}/claude-free"
command -v free-llm free-llm-read-hook free-llm-check-dir claude-free command -v free-llm free-llm-read-hook free-llm-check-dir claude-free omniroute-gateway
install -m644 "${REPO_DIR}"/systemd/omniroute-* "${UNIT_DIR}/"
systemctl --user daemon-reload
systemctl --user enable --now omniroute-health.timer omniroute-update.timer
systemctl --user list-timers 'omniroute-*' --no-pager

View File

@ -0,0 +1,157 @@
"""HTTP calls to the gateway: liveness, a real chat probe, admin provider state."""
from __future__ import annotations
import http.cookiejar
import json
import urllib.error
import urllib.request
from datetime import datetime
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from omniroute_offload._gw_config import Config
type Json = Any
COMBO = "free-best"
CHAT_TIMEOUT_S = 90
class ApiError(RuntimeError):
"""The gateway answered with an error or not at all."""
def _request(
url: str, body: object, method: str | None, headers: dict[str, str]
) -> urllib.request.Request:
data = None if body is None else json.dumps(body).encode()
hdrs = {"content-type": "application/json", **headers}
return urllib.request.Request(url, data=data, method=method, headers=hdrs)
def _send(
req: urllib.request.Request,
timeout: float,
opener: urllib.request.OpenerDirector | None = None,
) -> Json:
open_ = opener.open if opener else urllib.request.urlopen
try:
with open_(req, timeout=timeout) as resp:
raw = resp.read()
except (urllib.error.URLError, TimeoutError, OSError) as err:
msg = f"{req.full_url}: {err}"
raise ApiError(msg) from err
try:
return json.loads(raw) if raw else {}
except json.JSONDecodeError as err:
msg = f"{req.full_url}: not JSON"
raise ApiError(msg) from err
def call(
url: str,
body: dict[str, Any] | None = None,
*,
headers: dict[str, str] | None = None,
timeout: float = 15,
) -> Json:
"""Send JSON (POST when body is given), return decoded JSON; ApiError on failure."""
return _send(_request(url, body, None, headers or {}), timeout)
def alive(cfg: Config) -> bool:
"""Return True when the web server answers its auth-status endpoint."""
try:
call(f"{cfg.base_url}/api/auth/status", timeout=5)
except ApiError:
return False
return True
def chat(cfg: Config, key: str) -> str:
"""One real completion through the combo; returns its text or raises ApiError."""
reply = call(
f"{cfg.base_url}/v1/chat/completions",
{
"model": COMBO,
"max_tokens": 300,
"messages": [{"role": "user", "content": "Reply with exactly: OK"}],
},
headers={"authorization": f"Bearer {key}"},
timeout=CHAT_TIMEOUT_S,
)
try:
text = str(reply["choices"][0]["message"]["content"] or "").strip()
except (KeyError, IndexError, TypeError) as err:
msg = f"unexpected reply: {str(reply)[:200]}"
raise ApiError(msg) from err
if not text:
msg = "empty completion"
raise ApiError(msg)
return text
class Admin:
"""Cookie session on the admin API."""
def __init__(self, cfg: Config, password: str) -> None:
"""Log in; ApiError when the password is refused."""
self.base = cfg.base_url
jar = http.cookiejar.CookieJar()
self.opener = urllib.request.build_opener(
urllib.request.HTTPCookieProcessor(jar)
)
self._call("/api/auth/login", {"password": password})
def _call(self, path: str, body: object = None, method: str | None = None) -> Json:
return _send(_request(self.base + path, body, method, {}), 15, self.opener)
def providers(self) -> list[dict[str, Any]]:
"""Return every provider connection."""
got = self._call("/api/providers")
return list(got.get("connections", got) if isinstance(got, dict) else got)
def combo_members(self, name: str = COMBO) -> set[str]:
"""Return provider ids the combo routes to (empty when it is missing)."""
got = self._call("/api/combos")
combos = got.get("combos", got) if isinstance(got, dict) else got
return {
str(m.get("providerId") or str(m.get("model", "")).split("/", 1)[0])
for c in combos
if c.get("name") == name
for m in c.get("models", [])
if isinstance(m, dict)
}
def reactivate(self, conn_id: str) -> None:
"""Mark a connection active again (the PUT merges; keys are kept)."""
self._call(f"/api/providers/{conn_id}", {"testStatus": "active"}, "PUT")
def upsert_combo(self, name: str, models: list[str]) -> None:
"""Create or replace a priority combo."""
got = self._call("/api/combos")
combos = got.get("combos", got) if isinstance(got, dict) else got
body = {"name": name, "models": models, "strategy": "priority"}
ids = [c["id"] for c in combos if c.get("name") == name]
if ids:
self._call(f"/api/combos/{ids[0]}", body, "PUT")
else:
self._call("/api/combos", body)
def cooldown_until(conn: dict[str, Any]) -> datetime | None:
"""Return the latest cooldown end of a connection (top level or per key)."""
stamps = [conn.get("rateLimitedUntil")]
health = (conn.get("providerSpecificData") or {}).get("apiKeyHealth") or {}
stamps += [k.get("cooldownUntil") for k in health.values() if isinstance(k, dict)]
parsed = [_parse(s) for s in stamps if isinstance(s, str)]
valid = [p for p in parsed if p is not None]
return max(valid) if valid else None
def _parse(stamp: str) -> datetime | None:
try:
return datetime.fromisoformat(stamp)
except ValueError:
return None

View File

@ -0,0 +1,52 @@
"""Rebuild the free-best combo from the benchmark's order.tsv in conf_dir.
Keeps tool-capable models with a big enough prompt window and Tetris score, in
the file's rank order (the combo is a priority chain).
"""
from __future__ import annotations
import csv
import sys
from typing import TYPE_CHECKING
from omniroute_offload import _gw_api as api
from omniroute_offload import _gw_health as health
if TYPE_CHECKING:
from pathlib import Path
from omniroute_offload._gw_config import Config
MIN_PROMPT = 16_000
MIN_TETRIS = 0.85
def ranked_models(order: Path) -> list[str]:
"""Return agent-capable models from order.tsv that pass the filters, best first."""
with order.open(newline="") as handle:
rows = list(csv.DictReader(handle, delimiter="\t"))
return [
row["model"]
for row in rows
if row.get("tools") == "yes"
and row.get("tetris", "-") != "-"
and float(row["tetris"]) >= MIN_TETRIS
and float(row.get("max_prompt") or 0) >= MIN_PROMPT
]
def refresh(cfg: Config) -> bool:
"""Upsert the combo; log and return False when order.tsv is missing or empty."""
order = cfg.conf_dir / "order.tsv"
if not order.exists():
print(f"combo kept: no {order}", file=sys.stderr)
return False
models = ranked_models(order)
if not models:
print(f"combo kept: no model in {order} passes the filter", file=sys.stderr)
return False
admin = api.Admin(cfg, health.secret(cfg, "admin_password"))
admin.upsert_combo(api.COMBO, models)
print(f"{api.COMBO}: {' > '.join(models)}", file=sys.stderr)
return True

View File

@ -0,0 +1,100 @@
"""One OmniRoute gateway: container, volume, port, lock and state, read at call time."""
from __future__ import annotations
import fcntl
import json
import os
import re
from contextlib import contextmanager
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from collections.abc import Iterator
REPO = "diegosouzapw/omniroute"
DEFAULT_IMAGE = f"{REPO}:3.8.52-web"
STABLE_TAG = re.compile(r"^(\d+)\.(\d+)\.(\d+)-web$")
@dataclass(frozen=True)
class Config:
"""One gateway instance (live: omniroute on 20128)."""
name: str
volume: str
port: int
base_url: str
conf_dir: Path
state_dir: Path
backup_dir: Path
signal_env: Path
@classmethod
def from_env(cls, env: dict[str, str] | None = None) -> Config:
"""Read OMNIROUTE_NAME / _VOLUME / _PORT / _SIGNAL_ENV and HOME."""
env = dict(os.environ) if env is None else env
home = Path(env.get("HOME", str(Path.home())))
name = env.get("OMNIROUTE_NAME", "omniroute")
port = int(env.get("OMNIROUTE_PORT", "20128"))
signal = env.get("OMNIROUTE_SIGNAL_ENV", str(home / "src/signal-bot/.env"))
return cls(
name=name,
volume=env.get("OMNIROUTE_VOLUME", "omniroute-data"),
port=port,
base_url=f"http://127.0.0.1:{port}",
conf_dir=home / ".config/omniroute",
state_dir=home / ".local/state/omniroute-gateway" / name,
backup_dir=home / "data/backups/omniroute" / name,
signal_env=Path(signal),
)
def image(self) -> str:
"""Return the pinned image, or the repo default before any update."""
try:
pinned = (self.state_dir / "image").read_text().strip()
except FileNotFoundError:
return DEFAULT_IMAGE
return pinned or DEFAULT_IMAGE
def version(tag: str) -> tuple[int, int, int] | None:
"""Return (major, minor, patch) of a stable '-web' tag or image ref, else None."""
match = STABLE_TAG.match(tag.rsplit(":", 1)[-1])
if not match:
return None
major, minor, patch = (int(g) for g in match.groups())
return major, minor, patch
@contextmanager
def locked(cfg: Config, *, blocking: bool) -> Iterator[bool]:
"""Hold the instance lock; yields False when non-blocking and already held."""
cfg.state_dir.mkdir(parents=True, exist_ok=True)
with (cfg.state_dir / "lock").open("w") as handle:
flags = fcntl.LOCK_EX | (0 if blocking else fcntl.LOCK_NB)
try:
fcntl.flock(handle, flags)
except BlockingIOError:
yield False
return
yield True
def load_state(cfg: Config) -> dict[str, Any]:
"""Return alert state; empty when missing or unreadable."""
try:
data = json.loads((cfg.state_dir / "alert.json").read_text())
except OSError, json.JSONDecodeError:
return {}
return data if isinstance(data, dict) else {}
def save_state(cfg: Config, data: dict[str, Any], image: str | None = None) -> None:
"""Persist alert state, and the newly running image when given."""
cfg.state_dir.mkdir(parents=True, exist_ok=True)
(cfg.state_dir / "alert.json").write_text(json.dumps(data, sort_keys=True))
if image:
(cfg.state_dir / "image").write_text(f"{image}\n")

View File

@ -0,0 +1,126 @@
"""Docker operations on one gateway: recreate, restart, backup, restore."""
from __future__ import annotations
import gzip
import os
import shutil
import subprocess
import time
from typing import TYPE_CHECKING, cast
if TYPE_CHECKING:
from pathlib import Path
from typing import IO
from omniroute_offload._gw_config import Config
STOP_TIMEOUT_S = 40
KEEP_BACKUPS = 4
class DockerError(RuntimeError):
"""A docker command failed."""
def docker(*args: str, timeout: float = 120) -> str:
"""Run docker, return stdout; raise DockerError with stderr on failure."""
proc = subprocess.run(
["docker", *args], capture_output=True, text=True, check=False, timeout=timeout
)
if proc.returncode:
msg = f"docker {args[0]} failed: {proc.stderr.strip()}"
raise DockerError(msg)
return proc.stdout.strip()
def status(cfg: Config) -> str | None:
"""Return 'running', 'exited', ... or None when the container is missing."""
try:
return docker("inspect", "-f", "{{.State.Status}}", cfg.name)
except DockerError:
return None
def recreate(cfg: Config, image: str, password: str) -> None:
"""Replace the container with one on `image`, same volume and port."""
subprocess.run(
["docker", "rm", "-f", cfg.name], capture_output=True, check=False, timeout=90
)
docker(
"run", "-d", "--name", cfg.name,
"--restart", "unless-stopped", "--stop-timeout", str(STOP_TIMEOUT_S),
"-e", "OMNIROUTE_MEMORY_MB=8192", "--memory=10g",
"-e", f"INITIAL_PASSWORD={password}",
"--network", "host", "-e", "OMNIROUTE_HOSTNAME=127.0.0.1",
"-e", f"PORT={cfg.port}",
"-v", f"{cfg.volume}:/app/data", image,
) # fmt: skip
def restart(cfg: Config) -> None:
"""Docker restart (starts it too when it is stopped)."""
docker("restart", "-t", str(STOP_TIMEOUT_S), cfg.name)
def pull(image: str) -> None:
"""Fetch an image."""
docker("pull", "-q", image, timeout=900)
def backup(cfg: Config, label: str) -> Path:
"""Stop, tar /app/data (gzip, 0600), start again; keep the newest few."""
cfg.backup_dir.mkdir(parents=True, exist_ok=True)
cfg.backup_dir.chmod(0o700)
stamp = time.strftime("%Y%m%d-%H%M%S")
dest = cfg.backup_dir / f"{stamp}-{label}.tar.gz"
docker("stop", "-t", str(STOP_TIMEOUT_S), cfg.name)
try:
_copy_out(cfg, dest)
finally:
docker("start", cfg.name)
_prune(cfg.backup_dir)
return dest
def _copy_out(cfg: Config, dest: Path) -> None:
# The tar holds API keys and server.env: create it 0600 before writing.
fd = os.open(dest, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
with (
os.fdopen(fd, "wb") as raw,
gzip.GzipFile(fileobj=raw, mode="wb") as out,
subprocess.Popen(
["docker", "cp", f"{cfg.name}:/app/data", "-"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
) as proc,
):
shutil.copyfileobj(cast("IO[bytes]", proc.stdout), out)
err = proc.stderr.read().decode() if proc.stderr else ""
rc = proc.wait()
if rc:
dest.unlink(missing_ok=True)
msg = f"docker cp failed: {err.strip()}"
raise DockerError(msg)
def _prune(directory: Path) -> None:
for old in sorted(directory.glob("*.tar.gz"))[:-KEEP_BACKUPS]:
old.unlink()
def restore(cfg: Config, image: str, archive: Path) -> None:
"""Wipe the volume and unpack `archive` into it (container must be stopped)."""
script = "find /app/data -mindepth 1 -delete && tar -xz -C /app"
with archive.open("rb") as src:
proc = subprocess.run(
[
"docker", "run", "--rm", "-i", "--user", "0",
"-v", f"{cfg.volume}:/app/data", "--entrypoint", "sh", image,
"-c", script,
],
stdin=src, capture_output=True, check=False, timeout=300,
) # fmt: skip
if proc.returncode:
msg = f"restore failed: {proc.stderr.decode().strip()}"
raise DockerError(msg)

View File

@ -0,0 +1,163 @@
"""Health = a real completion through free-best, then the fix ladder.
Ladder on "down": 1) restart (or recreate) the container when it or its API is
dead, at most once per RESTART_GAP; 2) reactivate combo providers marked
failed whose cooldown already ended; 3) notify once per outage. Every provider
cooling down is "degraded", not down: no restart, one notice per day.
Exit: 0 ok / degraded / fixed, 1 still down.
"""
from __future__ import annotations
import sys
import time
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from typing import TYPE_CHECKING, Any
from omniroute_offload import _gw_api as api
from omniroute_offload import _gw_config as gwc
from omniroute_offload import _gw_docker as dock
from omniroute_offload import _gw_notify as note
if TYPE_CHECKING:
from omniroute_offload._gw_config import Config
RESTART_GAP = timedelta(minutes=30)
@dataclass(frozen=True)
class Verdict:
"""ok | degraded | down, with the reason."""
state: str
detail: str
def secret(cfg: Config, name: str) -> str:
"""Read a stripped file from conf_dir (admin_password, api_key)."""
return (cfg.conf_dir / name).read_text().strip()
def wait_up(cfg: Config, seconds: float = 90) -> bool:
"""Poll the API until it answers or `seconds` pass."""
deadline = time.monotonic() + seconds
while time.monotonic() < deadline:
if api.alive(cfg):
return True
time.sleep(2)
return api.alive(cfg)
def _members(conns: list[dict[str, Any]], members: set[str]) -> list[dict[str, Any]]:
return [c for c in conns if c.get("provider") in members and c.get("isActive")]
def _cooling(conn: dict[str, Any], now: datetime) -> bool:
until = api.cooldown_until(conn)
return until is not None and until > now
def check(cfg: Config, now: datetime) -> Verdict:
"""Classify the gateway by one real chat call plus provider state."""
if not api.alive(cfg):
return Verdict("down", "API not answering")
try:
api.chat(cfg, secret(cfg, "api_key"))
except (api.ApiError, OSError) as err:
detail = f"chat failed: {err}"
else:
return Verdict("ok", "chat answered")
try:
admin = api.Admin(cfg, secret(cfg, "admin_password"))
conns = _members(admin.providers(), admin.combo_members())
except (api.ApiError, OSError) as err:
return Verdict("down", f"{detail}; admin API: {err}")
if conns and all(_cooling(c, now) for c in conns):
return Verdict("degraded", f"all {len(conns)} combo providers cooling down")
return Verdict("down", detail)
def _restart(cfg: Config, state: dict[str, Any], now: datetime) -> None:
last = state.get("last_restart")
try:
recent = now - datetime.fromisoformat(str(last)) < RESTART_GAP
except TypeError, ValueError:
recent = False
if recent:
print(f"restart skipped: last one at {last}", file=sys.stderr)
return
state["last_restart"] = now.isoformat()
if dock.status(cfg) is None:
print("container missing: recreating", file=sys.stderr)
dock.recreate(cfg, cfg.image(), secret(cfg, "admin_password"))
else:
print("restarting container", file=sys.stderr)
dock.restart(cfg)
wait_up(cfg)
def _reactivate(cfg: Config, now: datetime) -> int:
admin = api.Admin(cfg, secret(cfg, "admin_password"))
stuck = [
c
for c in _members(admin.providers(), admin.combo_members())
if c.get("testStatus") != "active" and not _cooling(c, now)
]
for conn in stuck:
print(f"reactivating {conn.get('provider')}", file=sys.stderr)
admin.reactivate(str(conn["id"]))
return len(stuck)
def heal(cfg: Config, state: dict[str, Any], now: datetime) -> Verdict:
"""Check, then walk ladder steps 1-2 while still down."""
verdict = check(cfg, now)
print(f"check: {verdict.state}: {verdict.detail}", file=sys.stderr)
if verdict.state == "down" and (
dock.status(cfg) != "running" or not api.alive(cfg)
):
_restart(cfg, state, now)
verdict = check(cfg, now)
print(f"after restart: {verdict.state}: {verdict.detail}", file=sys.stderr)
if verdict.state == "down" and api.alive(cfg) and _reactivate(cfg, now):
verdict = check(cfg, now)
print(f"after reactivate: {verdict.state}: {verdict.detail}", file=sys.stderr)
return verdict
def _report(cfg: Config, state: dict[str, Any], verdict: Verdict, now: datetime) -> int:
if verdict.state == "ok":
if state.pop("down_since", None):
sent = note.notify(cfg, "recovered: free-best answers again", urgency="low")
return 0 if sent else 1
return 0
if verdict.state == "degraded":
today = now.date().isoformat()
if state.get("degraded_day") != today:
state["degraded_day"] = today
sent = note.notify(cfg, f"degraded: {verdict.detail}", urgency="low")
return 0 if sent else 1
return 0
if not state.get("down_since"):
state["down_since"] = now.isoformat()
note.notify(cfg, f"DOWN after auto-fix: {verdict.detail}", urgency="critical")
return 1
def run(cfg: Config, now: datetime | None = None) -> int:
"""One health tick under the instance lock (skips while an update runs)."""
now = now or datetime.now(UTC)
with gwc.locked(cfg, blocking=False) as got:
if not got:
print("update in progress: health skipped", file=sys.stderr)
return 0
state = gwc.load_state(cfg)
try:
verdict = heal(cfg, state, now)
except (api.ApiError, dock.DockerError, OSError) as err:
verdict = Verdict("down", f"fix ladder failed: {err}")
code = _report(cfg, state, verdict, now)
gwc.save_state(cfg, state)
print(f"result: {verdict.state}", file=sys.stderr)
return code

View File

@ -0,0 +1,73 @@
"""Tell kuhy: one replaceable desktop popup plus a Signal DM from the bot account."""
from __future__ import annotations
import subprocess
import sys
from typing import TYPE_CHECKING
from omniroute_offload import _gw_api as api
if TYPE_CHECKING:
from pathlib import Path
from omniroute_offload._gw_config import Config
class NotifyError(RuntimeError):
"""A channel could not be used."""
def _read_env(path: Path) -> dict[str, str]:
pairs = (
line.split("=", 1)
for line in path.read_text().splitlines()
if "=" in line and not line.lstrip().startswith("#")
)
return {k.strip(): v.strip() for k, v in pairs}
def signal(cfg: Config, text: str) -> None:
"""DM the first TRUSTED_INVITERS number from the signal-bot account."""
try:
env = _read_env(cfg.signal_env)
owner = env["TRUSTED_INVITERS"].split(",")[0].strip()
body = {"number": env["PHONE_NUMBER"], "recipients": [owner], "message": text}
api.call(f"{env['API_BASE_URL']}/v2/send", body, timeout=20)
except (OSError, KeyError, api.ApiError) as err:
msg = f"signal: {err!r}"
raise NotifyError(msg) from err
def desktop(cfg: Config, text: str, urgency: str) -> None:
"""notify-send, replacing this instance's previous popup instead of stacking."""
id_file = cfg.state_dir / "notify_id"
previous = id_file.read_text().strip() if id_file.exists() else "0"
proc = subprocess.run(
[
"notify-send", "--app-name=omniroute", "--print-id",
f"--replace-id={previous}", f"--urgency={urgency}",
f"OmniRoute ({cfg.name})", text,
],
capture_output=True, text=True, check=False, timeout=10,
) # fmt: skip
if proc.returncode:
msg = f"notify-send: {proc.stderr.strip()}"
raise NotifyError(msg)
id_file.write_text(proc.stdout.strip())
def notify(cfg: Config, text: str, *, urgency: str = "normal") -> bool:
"""Send on both channels, log each failure; True only when Signal got it."""
print(f"notify: {text}", file=sys.stderr)
cfg.state_dir.mkdir(parents=True, exist_ok=True)
try:
desktop(cfg, text, urgency)
except (NotifyError, OSError, subprocess.SubprocessError) as err:
print(f"desktop notify failed: {err}", file=sys.stderr)
try:
signal(cfg, f"OmniRoute ({cfg.name}): {text}")
except NotifyError as err:
print(f"signal notify failed: {err}", file=sys.stderr)
return False
return True

View File

@ -0,0 +1,89 @@
"""Weekly: back up the data volume, then move to the newest stable '-web' tag.
A new tag must pass the health check (a real chat answer) or the volume is
restored and the old image put back. Exit: 0 current / updated, 1 failed.
"""
from __future__ import annotations
import sys
from datetime import UTC, datetime
from typing import TYPE_CHECKING
from omniroute_offload import _gw_api as api
from omniroute_offload import _gw_combo as combo
from omniroute_offload import _gw_config as gwc
from omniroute_offload import _gw_docker as dock
from omniroute_offload import _gw_health as health
from omniroute_offload import _gw_notify as note
if TYPE_CHECKING:
from pathlib import Path
from omniroute_offload._gw_config import Config
HUB = f"https://hub.docker.com/v2/repositories/{gwc.REPO}/tags?page_size=100&ordering=last_updated"
def latest_image() -> str:
"""Return the newest stable '-web' image on Docker Hub."""
tags = [str(t.get("name", "")) for t in api.call(HUB, timeout=30)["results"]]
stable = [t for t in tags if gwc.version(t)]
if not stable:
msg = "no stable -web tag on Docker Hub"
raise api.ApiError(msg)
return f"{gwc.REPO}:{max(stable, key=lambda t: gwc.version(t) or (0, 0, 0))}"
def _switch(cfg: Config, image: str) -> health.Verdict:
dock.recreate(cfg, image, health.secret(cfg, "admin_password"))
if not health.wait_up(cfg):
return health.Verdict("down", f"{image} API never came up")
return health.check(cfg, datetime.now(UTC))
def _rollback(cfg: Config, image: str, archive: Path) -> None:
if dock.status(cfg) is not None:
dock.docker("stop", "-t", str(dock.STOP_TIMEOUT_S), cfg.name)
dock.restore(cfg, image, archive)
dock.recreate(cfg, image, health.secret(cfg, "admin_password"))
health.wait_up(cfg)
def _upgrade(cfg: Config, current: str, archive: Path) -> int:
latest = latest_image()
have, newest = gwc.version(current), gwc.version(latest)
if have is None or newest is None or newest <= have:
print(f"up to date: {current} (newest {latest})", file=sys.stderr)
return 0
print(f"updating {current} -> {latest}", file=sys.stderr)
try:
dock.pull(latest)
verdict = _switch(cfg, latest)
except (dock.DockerError, api.ApiError, OSError) as err:
verdict = health.Verdict("down", str(err))
if verdict.state == "ok":
gwc.save_state(cfg, gwc.load_state(cfg), image=latest)
combo.refresh(cfg)
sent = note.notify(cfg, f"updated {current} -> {latest}", urgency="low")
return 0 if sent else 1
_rollback(cfg, current, archive)
note.notify(
cfg,
f"update to {latest} failed ({verdict.detail}); rolled back to {current}",
urgency="critical",
)
return 1
def run(cfg: Config) -> int:
"""Back up, then update if a newer tag exists, under the instance lock."""
with gwc.locked(cfg, blocking=True):
current = cfg.image()
try:
archive = dock.backup(cfg, current.rsplit(":", 1)[-1])
print(f"backup: {archive}", file=sys.stderr)
return _upgrade(cfg, current, archive)
except (dock.DockerError, api.ApiError, OSError) as err:
note.notify(cfg, f"update job failed: {err}", urgency="critical")
return 1

View File

@ -0,0 +1,73 @@
"""omniroute-gateway: keep the local OmniRoute gateway up, checked and current.
start (re)create the container on the pinned image and wait for its API
health real chat probe + fix ladder (systemd timer, every 10 min)
update backup + newest stable tag with rollback (systemd timer, weekly)
backup stop, tar the data volume, start
combo rebuild free-best from ~/.config/omniroute/order.tsv
Instance: OMNIROUTE_NAME / OMNIROUTE_VOLUME / OMNIROUTE_PORT (default: live).
"""
from __future__ import annotations
import argparse
import sys
from typing import TYPE_CHECKING
from omniroute_offload import _gw_api as api
from omniroute_offload import _gw_combo as combo
from omniroute_offload import _gw_docker as dock
from omniroute_offload import _gw_health as health
from omniroute_offload import _gw_update as update
from omniroute_offload._gw_config import Config
if TYPE_CHECKING:
from collections.abc import Callable
def _start(cfg: Config) -> int:
dock.recreate(cfg, cfg.image(), health.secret(cfg, "admin_password"))
if health.wait_up(cfg):
print(f"omniroute up: {cfg.base_url} ({cfg.image()})")
return 0
print(f"{cfg.name} did not come up; see: docker logs {cfg.name}", file=sys.stderr)
return 1
def _backup(cfg: Config) -> int:
print(dock.backup(cfg, cfg.image().rsplit(":", 1)[-1]))
return 0
def _combo(cfg: Config) -> int:
return 0 if combo.refresh(cfg) else 1
def main(argv: list[str] | None = None) -> int:
"""Run one subcommand; non-zero on failure."""
parser = argparse.ArgumentParser(
prog="omniroute-gateway",
description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument(
"command", choices=["start", "health", "update", "backup", "combo"]
)
args = parser.parse_args(argv)
cfg = Config.from_env()
actions: dict[str, Callable[[Config], int]] = {
"start": _start,
"health": health.run,
"update": update.run,
"backup": _backup,
"combo": _combo,
}
try:
return actions[args.command](cfg)
except (dock.DockerError, api.ApiError, OSError) as err:
print(f"omniroute-gateway {args.command}: {err}", file=sys.stderr)
return 1
if __name__ == "__main__":
sys.exit(main())

View File

@ -0,0 +1,151 @@
"""Opt-in fake gateway world shared by the health tests (no autouse fixtures)."""
from __future__ import annotations
import io
import json
import urllib.request
from typing import TYPE_CHECKING, Any, Self
import pytest
from omniroute_offload import _gw_api as api
from omniroute_offload import _gw_config as gwc
from omniroute_offload import _gw_docker as dock
from omniroute_offload import _gw_health as health
from omniroute_offload import _gw_notify as note
if TYPE_CHECKING:
from pathlib import Path
class World:
"""Fake docker + gateway; every side effect lands in `events`/`notices`."""
def __init__(self) -> None:
self.status: str | None = "running"
self.alive = True
self.chat_ok = True
self.conns: list[dict[str, Any]] = []
self.members = {"p"}
self.admin_error: Exception | None = None
self.restart_fixes = True
self.restart_error: Exception | None = None
self.heal_by_reactivate = False
self.notify_ok = True
self.events: list[str] = []
self.notices: list[tuple[str, str]] = []
@staticmethod
def conn(cid: str, **over: Any) -> dict[str, Any]:
base: dict[str, Any] = {
"id": cid, "provider": "p", "isActive": True, "testStatus": "active",
} # fmt: skip
return base | over
def chat(self, _cfg: gwc.Config, key: str) -> str:
assert key == "key"
if not self.chat_ok:
msg = "503 all providers failed"
raise api.ApiError(msg)
return "OK"
def admin(self, _cfg: gwc.Config, pw: str) -> World:
assert pw == "pw"
if self.admin_error:
raise self.admin_error
return self
def providers(self) -> list[dict[str, Any]]:
return self.conns
def combo_members(self) -> set[str]:
return self.members
def reactivate(self, conn_id: str) -> None:
self.events.append(f"reactivate:{conn_id}")
self.chat_ok = self.chat_ok or self.heal_by_reactivate
def _bring_up(self, event: str) -> None:
self.events.append(event)
if self.restart_error:
raise self.restart_error
if self.restart_fixes:
self.status, self.alive = "running", True
def restart(self, _cfg: gwc.Config) -> None:
self._bring_up("restart")
def recreate(self, _cfg: gwc.Config, image: str, password: str) -> None:
assert (image, password) == (gwc.DEFAULT_IMAGE, "pw")
self._bring_up("recreate")
def notify(self, _cfg: gwc.Config, text: str, *, urgency: str = "normal") -> bool:
self.notices.append((urgency, text))
return self.notify_ok
@pytest.fixture
def health_cfg(tmp_path: Path) -> gwc.Config:
made = gwc.Config.from_env({"HOME": str(tmp_path)})
made.conf_dir.mkdir(parents=True)
(made.conf_dir / "api_key").write_text("key\n")
(made.conf_dir / "admin_password").write_text("pw\n")
return made
@pytest.fixture
def world(monkeypatch: pytest.MonkeyPatch) -> World:
w = World()
monkeypatch.setattr(api, "alive", lambda _cfg: w.alive)
monkeypatch.setattr(api, "chat", w.chat)
monkeypatch.setattr(api, "Admin", w.admin)
monkeypatch.setattr(dock, "status", lambda _cfg: w.status)
monkeypatch.setattr(dock, "restart", w.restart)
monkeypatch.setattr(dock, "recreate", w.recreate)
monkeypatch.setattr(note, "notify", w.notify)
monkeypatch.setattr(health, "wait_up", lambda _cfg, seconds=90: True)
return w
class FakeResp(io.BytesIO):
def __enter__(self) -> Self:
return self
def __exit__(self, *_exc: object) -> None:
self.close()
def _resp(payload: object) -> FakeResp:
return FakeResp(json.dumps(payload).encode())
class Net:
"""Records requests and replays queued replies (or raises queued errors)."""
def __init__(self) -> None:
self.reqs: list[urllib.request.Request] = []
self.replies: list[Any] = []
self.timeouts: list[float] = []
def open(self, req: urllib.request.Request, timeout: float) -> FakeResp:
self.reqs.append(req)
self.timeouts.append(timeout)
reply = self.replies.pop(0)
if isinstance(reply, BaseException):
raise reply
return reply if isinstance(reply, FakeResp) else _resp(reply)
def bodies(self) -> list[Any]:
out: list[Any] = []
for req in self.reqs:
assert req.data is None or isinstance(req.data, bytes)
out.append(json.loads(req.data) if req.data else None)
return out
@pytest.fixture
def net(monkeypatch: pytest.MonkeyPatch) -> Net:
fake = Net()
monkeypatch.setattr(urllib.request, "urlopen", fake.open)
return fake

View File

@ -0,0 +1,131 @@
from __future__ import annotations
import subprocess
import urllib.error
import urllib.request
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any
import pytest
from omniroute_offload import _gw_api as api
from omniroute_offload import _gw_config as gwc
from omniroute_offload.tests.conftest import FakeResp
if TYPE_CHECKING:
from pathlib import Path
from omniroute_offload.tests.conftest import Net
@pytest.fixture(autouse=True)
def _isolate(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
def blocked(*_args: Any, **_kwargs: Any) -> Any:
msg = "real process/network access is blocked in tests"
raise AssertionError(msg)
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setattr(subprocess, "run", blocked)
monkeypatch.setattr(subprocess, "Popen", blocked)
monkeypatch.setattr(urllib.request, "urlopen", blocked)
monkeypatch.setattr(urllib.request, "build_opener", blocked)
@pytest.fixture
def cfg(tmp_path: Path) -> gwc.Config:
return gwc.Config.from_env({"HOME": str(tmp_path)})
def test_call_get_has_no_body(net: Net) -> None:
net.replies = [{"a": 1}]
assert api.call("http://x/y", timeout=3) == {"a": 1}
assert net.reqs[0].data is None
assert net.reqs[0].get_method() == "GET"
assert net.timeouts == [3]
def test_call_post_sends_json_and_headers(net: Net) -> None:
net.replies = [{}]
api.call("http://x/y", {"k": "v"}, headers={"authorization": "Bearer t"})
req = net.reqs[0]
assert req.get_method() == "POST"
assert net.bodies() == [{"k": "v"}]
assert req.get_header("Content-type") == "application/json"
assert req.get_header("Authorization") == "Bearer t"
@pytest.mark.parametrize(
"err",
[urllib.error.URLError("refused"), TimeoutError("slow"), ConnectionResetError("x")],
)
def test_transport_errors_become_api_error(net: Net, err: BaseException) -> None:
net.replies = [err]
with pytest.raises(api.ApiError, match="http://x"):
api.call("http://x")
def test_empty_body_is_empty_dict_but_non_json_is_an_error(net: Net) -> None:
net.replies = [FakeResp(b""), FakeResp(b"<html>")]
assert api.call("http://x") == {}
with pytest.raises(api.ApiError, match="not JSON"):
api.call("http://x")
def test_alive_true_and_false(cfg: gwc.Config, net: Net) -> None:
net.replies = [{}, urllib.error.URLError("down")]
assert api.alive(cfg) is True
assert net.reqs[0].full_url == f"{cfg.base_url}/api/auth/status"
assert net.timeouts == [5]
assert api.alive(cfg) is False
def test_chat_returns_stripped_text_via_combo(cfg: gwc.Config, net: Net) -> None:
net.replies = [{"choices": [{"message": {"content": " OK \n"}}]}]
assert api.chat(cfg, "sekret") == "OK"
req = net.reqs[0]
assert req.full_url == f"{cfg.base_url}/v1/chat/completions"
assert req.get_header("Authorization") == "Bearer sekret"
assert net.bodies()[0]["model"] == api.COMBO
assert net.timeouts == [api.CHAT_TIMEOUT_S]
@pytest.mark.parametrize(
("message", "match"),
[
({}, "unexpected reply"),
({"choices": []}, "unexpected reply"),
({"choices": None}, "unexpected reply"),
({"choices": [{"message": {}}]}, "unexpected reply"),
({"choices": [{"message": {"content": None}}]}, "empty completion"),
({"choices": [{"message": {"content": " "}}]}, "empty completion"),
],
)
def test_chat_bad_replies(
cfg: gwc.Config, net: Net, message: object, match: str
) -> None:
net.replies = [message]
with pytest.raises(api.ApiError, match=match):
api.chat(cfg, "k")
def test_cooldown_until_variants() -> None:
top = "2026-11-05T10:00:00+00:00"
late = "2026-11-05T12:00:00+00:00"
assert api.cooldown_until({}) is None
assert api.cooldown_until({"rateLimitedUntil": top}) == datetime(
2026, 11, 5, 10, tzinfo=UTC
)
conn = {
"rateLimitedUntil": top,
"providerSpecificData": {
"apiKeyHealth": {
"k1": {"cooldownUntil": late},
"k2": {"cooldownUntil": "garbage"},
"k3": "not-a-dict",
"k4": {"cooldownUntil": 5},
}
},
}
assert api.cooldown_until(conn) == datetime(2026, 11, 5, 12, tzinfo=UTC)
assert api.cooldown_until({"rateLimitedUntil": "garbage"}) is None
assert api.cooldown_until({"providerSpecificData": None}) is None

View File

@ -0,0 +1,127 @@
from __future__ import annotations
import subprocess
import urllib.error
import urllib.request
from typing import TYPE_CHECKING, Any
import pytest
from omniroute_offload import _gw_api as api
from omniroute_offload import _gw_config as gwc
if TYPE_CHECKING:
from pathlib import Path
from omniroute_offload.tests.conftest import Net
@pytest.fixture(autouse=True)
def _isolate(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
def blocked(*_args: Any, **_kwargs: Any) -> Any:
msg = "real process/network access is blocked in tests"
raise AssertionError(msg)
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setattr(subprocess, "run", blocked)
monkeypatch.setattr(subprocess, "Popen", blocked)
monkeypatch.setattr(urllib.request, "urlopen", blocked)
monkeypatch.setattr(urllib.request, "build_opener", blocked)
@pytest.fixture
def cfg(tmp_path: Path) -> gwc.Config:
return gwc.Config.from_env({"HOME": str(tmp_path)})
@pytest.fixture
def admin(
cfg: gwc.Config, net: Net, monkeypatch: pytest.MonkeyPatch
) -> tuple[api.Admin, Net]:
seen: list[Any] = []
def fake_build_opener(*handlers: Any) -> Net:
seen.extend(handlers)
return net
monkeypatch.setattr(urllib.request, "build_opener", fake_build_opener)
net.replies = [{"ok": True}]
adm = api.Admin(cfg, "pw")
assert len(seen) == 1
assert net.reqs[0].full_url == f"{cfg.base_url}/api/auth/login"
assert net.bodies() == [{"password": "pw"}]
return adm, net
def test_admin_login_refused(
cfg: gwc.Config, net: Net, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(urllib.request, "build_opener", lambda *_h: net)
net.replies = [urllib.error.URLError("401")]
with pytest.raises(api.ApiError):
api.Admin(cfg, "bad")
def test_providers_shapes(admin: tuple[api.Admin, Net]) -> None:
adm, net = admin
conns = [{"id": "1"}, {"id": "2"}]
net.replies = [{"connections": conns}, conns]
assert adm.providers() == conns
assert adm.providers() == conns
assert net.reqs[-1].full_url.endswith("/api/providers")
def test_combo_members_filters_by_name_and_falls_back_to_model_prefix(
admin: tuple[api.Admin, Net],
) -> None:
adm, net = admin
combos = [
{
"name": "free-best",
"models": [
{"providerId": "nvidia", "model": "nvidia/a"},
{"model": "groq/b"},
"not-a-dict",
],
},
{"name": "other", "models": [{"providerId": "zzz"}]},
]
net.replies = [{"combos": combos}, combos]
assert adm.combo_members() == {"nvidia", "groq"}
assert adm.combo_members() == {"nvidia", "groq"}
net.replies = [{"combos": []}]
assert adm.combo_members("free-best") == set()
def test_reactivate_puts_active(admin: tuple[api.Admin, Net]) -> None:
adm, net = admin
net.replies = [{}]
adm.reactivate("abc")
req = net.reqs[-1]
assert req.get_method() == "PUT"
assert req.full_url.endswith("/api/providers/abc")
assert net.bodies()[-1] == {"testStatus": "active"}
def test_upsert_combo_updates_existing(admin: tuple[api.Admin, Net]) -> None:
adm, net = admin
net.replies = [{"combos": [{"id": "c1", "name": "free-best"}]}, {}]
adm.upsert_combo("free-best", ["a/x", "b/y"])
req = net.reqs[-1]
assert req.get_method() == "PUT"
assert req.full_url.endswith("/api/combos/c1")
assert net.bodies()[-1] == {
"name": "free-best",
"models": ["a/x", "b/y"],
"strategy": "priority",
}
def test_upsert_combo_creates_when_missing(admin: tuple[api.Admin, Net]) -> None:
adm, net = admin
net.replies = [[{"id": "c9", "name": "other"}], {}]
adm.upsert_combo("free-best", ["a/x"])
req = net.reqs[-1]
assert req.get_method() == "POST"
assert req.full_url.endswith("/api/combos")
assert net.bodies()[-1]["name"] == "free-best"

View File

@ -0,0 +1,120 @@
from __future__ import annotations
import subprocess
import urllib.request
from typing import TYPE_CHECKING, Any
import pytest
from omniroute_offload import _gw_api as api
from omniroute_offload import _gw_combo as combo
from omniroute_offload import _gw_config as gwc
if TYPE_CHECKING:
from pathlib import Path
@pytest.fixture(autouse=True)
def _isolate(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
def blocked(*_args: Any, **_kwargs: Any) -> Any:
msg = "real process/network access is blocked in tests"
raise AssertionError(msg)
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setattr(subprocess, "run", blocked)
monkeypatch.setattr(subprocess, "Popen", blocked)
monkeypatch.setattr(urllib.request, "urlopen", blocked)
monkeypatch.setattr(urllib.request, "build_opener", blocked)
HEADER = "model\ttools\ttetris\tmax_prompt\n"
@pytest.fixture
def cfg(tmp_path: Path) -> gwc.Config:
made = gwc.Config.from_env({"HOME": str(tmp_path)})
made.conf_dir.mkdir(parents=True)
(made.conf_dir / "admin_password").write_text("pw\n")
return made
class FakeAdmin:
def __init__(self, cfg: gwc.Config, password: str) -> None:
self.pw = password
self.upserts: list[tuple[str, list[str]]] = []
def upsert_combo(self, name: str, models: list[str]) -> None:
self.upserts.append((name, models))
@pytest.fixture
def admins(monkeypatch: pytest.MonkeyPatch) -> list[FakeAdmin]:
made: list[FakeAdmin] = []
def factory(cfg: gwc.Config, password: str) -> FakeAdmin:
made.append(FakeAdmin(cfg, password))
return made[-1]
monkeypatch.setattr(api, "Admin", factory)
return made
def _write(cfg: gwc.Config, body: str) -> None:
(cfg.conf_dir / "order.tsv").write_text(body)
def test_ranked_models_applies_every_filter_and_keeps_order(tmp_path: Path) -> None:
order = tmp_path / "order.tsv"
order.write_text(
HEADER
+ "z/best\tyes\t1.0\t128000\n"
+ "a/no-tools\tno\t1.0\t128000\n"
+ "a/no-tetris\tyes\t-\t128000\n"
+ "a/low-tetris\tyes\t0.84\t128000\n"
+ "a/exact\tyes\t0.85\t16000\n"
+ "a/small-window\tyes\t0.9\t15999\n"
+ "a/no-window\tyes\t0.9\t\n"
+ "b/second\tyes\t0.9\t32000\n"
)
assert combo.ranked_models(order) == ["z/best", "a/exact", "b/second"]
def test_ranked_models_without_tetris_column_keeps_nothing(tmp_path: Path) -> None:
order = tmp_path / "order.tsv"
order.write_text("model\ttools\tmax_prompt\na/b\tyes\t99999\n")
assert combo.ranked_models(order) == []
def test_refresh_upserts_filtered_combo(
cfg: gwc.Config, admins: list[FakeAdmin], capsys: pytest.CaptureFixture[str]
) -> None:
_write(cfg, HEADER + "x/one\tyes\t0.9\t20000\nx/two\tno\t0.9\t20000\n")
assert combo.refresh(cfg) is True
assert admins[0].pw == "pw"
assert admins[0].upserts == [(api.COMBO, ["x/one"])]
assert "free-best: x/one" in capsys.readouterr().err
def test_refresh_keeps_combo_when_order_missing(
cfg: gwc.Config, admins: list[FakeAdmin], capsys: pytest.CaptureFixture[str]
) -> None:
assert combo.refresh(cfg) is False
assert admins == []
assert "combo kept: no" in capsys.readouterr().err
@pytest.mark.parametrize(
"body",
["", HEADER, HEADER + "x/one\tno\t0.9\t20000\n"],
ids=["empty", "header", "none"],
)
def test_refresh_keeps_combo_when_nothing_passes(
cfg: gwc.Config,
admins: list[FakeAdmin],
capsys: pytest.CaptureFixture[str],
body: str,
) -> None:
_write(cfg, body)
assert combo.refresh(cfg) is False
assert admins == []
assert "passes the filter" in capsys.readouterr().err

View File

@ -0,0 +1,124 @@
from __future__ import annotations
import subprocess
import urllib.request
from typing import TYPE_CHECKING, Any
import pytest
from omniroute_offload import _gw_config as gwc
if TYPE_CHECKING:
from pathlib import Path
@pytest.fixture(autouse=True)
def _isolate(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
def blocked(*_args: Any, **_kwargs: Any) -> Any:
msg = "real process/network access is blocked in tests"
raise AssertionError(msg)
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setattr(subprocess, "run", blocked)
monkeypatch.setattr(subprocess, "Popen", blocked)
monkeypatch.setattr(urllib.request, "urlopen", blocked)
monkeypatch.setattr(urllib.request, "build_opener", blocked)
@pytest.fixture
def cfg(tmp_path: Path) -> gwc.Config:
return gwc.Config.from_env({"HOME": str(tmp_path)})
def test_from_env_defaults(cfg: gwc.Config, tmp_path: Path) -> None:
assert cfg.name == "omniroute"
assert cfg.volume == "omniroute-data"
assert cfg.port == 20128
assert cfg.base_url == "http://127.0.0.1:20128"
assert cfg.conf_dir == tmp_path / ".config/omniroute"
assert cfg.state_dir == tmp_path / ".local/state/omniroute-gateway/omniroute"
assert cfg.backup_dir == tmp_path / "data/backups/omniroute/omniroute"
assert cfg.signal_env == tmp_path / "src/signal-bot/.env"
def test_from_env_overrides(tmp_path: Path) -> None:
cfg = gwc.Config.from_env(
{
"HOME": str(tmp_path),
"OMNIROUTE_NAME": "tetris",
"OMNIROUTE_VOLUME": "t-data",
"OMNIROUTE_PORT": "20999",
"OMNIROUTE_SIGNAL_ENV": "/x/signal.env",
}
)
assert (cfg.name, cfg.volume, cfg.port) == ("tetris", "t-data", 20999)
assert cfg.base_url == "http://127.0.0.1:20999"
assert cfg.state_dir == tmp_path / ".local/state/omniroute-gateway/tetris"
assert cfg.backup_dir == tmp_path / "data/backups/omniroute/tetris"
assert str(cfg.signal_env) == "/x/signal.env"
def test_from_env_reads_process_env_and_home(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
monkeypatch.setenv("OMNIROUTE_NAME", "fromproc")
assert gwc.Config.from_env().state_dir.parts[-1] == "fromproc"
assert gwc.Config.from_env({}).conf_dir == tmp_path / ".config/omniroute"
def test_image_default_pinned_and_empty(cfg: gwc.Config) -> None:
assert cfg.image() == gwc.DEFAULT_IMAGE
cfg.state_dir.mkdir(parents=True)
(cfg.state_dir / "image").write_text(" \n")
assert cfg.image() == gwc.DEFAULT_IMAGE
(cfg.state_dir / "image").write_text("repo:9.9.9-web\n")
assert cfg.image() == "repo:9.9.9-web"
@pytest.mark.parametrize(
("tag", "expected"),
[
("3.8.52-web", (3, 8, 52)),
("diegosouzapw/omniroute:3.10.0-web", (3, 10, 0)),
("latest", None),
("3.8.52", None),
("3.8.52-rc1-web", None),
("repo:3.8-web", None),
],
)
def test_version(tag: str, expected: tuple[int, int, int] | None) -> None:
assert gwc.version(tag) == expected
def test_locked_blocking_yields_true_and_creates_state_dir(cfg: gwc.Config) -> None:
with gwc.locked(cfg, blocking=True) as got:
assert got is True
assert (cfg.state_dir / "lock").exists()
def test_locked_nonblocking_yields_false_while_held(cfg: gwc.Config) -> None:
with gwc.locked(cfg, blocking=True), gwc.locked(cfg, blocking=False) as got:
assert got is False
with gwc.locked(cfg, blocking=False) as got:
assert got is True
def test_load_state_missing_invalid_and_non_dict(cfg: gwc.Config) -> None:
assert gwc.load_state(cfg) == {}
cfg.state_dir.mkdir(parents=True)
alert = cfg.state_dir / "alert.json"
alert.write_text("{not json")
assert gwc.load_state(cfg) == {}
alert.write_text("[1, 2]")
assert gwc.load_state(cfg) == {}
alert.write_text('{"down_since": "t"}')
assert gwc.load_state(cfg) == {"down_since": "t"}
def test_save_state_roundtrip_and_image_pin(cfg: gwc.Config) -> None:
gwc.save_state(cfg, {"b": 1, "a": 2})
assert gwc.load_state(cfg) == {"a": 2, "b": 1}
assert not (cfg.state_dir / "image").exists()
gwc.save_state(cfg, {}, image="repo:3.9.0-web")
assert (cfg.state_dir / "image").read_text() == "repo:3.9.0-web\n"
assert cfg.image() == "repo:3.9.0-web"

View File

@ -0,0 +1,225 @@
from __future__ import annotations
import gzip
import io
import subprocess
import urllib.request
from types import SimpleNamespace
from typing import TYPE_CHECKING, Any, Self
import pytest
from omniroute_offload import _gw_config as gwc
from omniroute_offload import _gw_docker as dock
if TYPE_CHECKING:
from pathlib import Path
@pytest.fixture(autouse=True)
def _isolate(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
def blocked(*_args: Any, **_kwargs: Any) -> Any:
msg = "real process/network access is blocked in tests"
raise AssertionError(msg)
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setattr(subprocess, "run", blocked)
monkeypatch.setattr(subprocess, "Popen", blocked)
monkeypatch.setattr(urllib.request, "urlopen", blocked)
monkeypatch.setattr(urllib.request, "build_opener", blocked)
@pytest.fixture
def cfg(tmp_path: Path) -> gwc.Config:
return gwc.Config.from_env({"HOME": str(tmp_path), "OMNIROUTE_PORT": "20555"})
class Runner:
"""Fake subprocess.run: records argv, replays queued completed processes."""
def __init__(self) -> None:
self.calls: list[list[str]] = []
self.kwargs: list[dict[str, Any]] = []
self.results: list[Any] = []
def __call__(self, argv: list[str], **kwargs: Any) -> Any:
self.calls.append(argv)
self.kwargs.append(kwargs)
if self.results:
return self.results.pop(0)
return SimpleNamespace(returncode=0, stdout="", stderr="")
@pytest.fixture
def run(monkeypatch: pytest.MonkeyPatch) -> Runner:
fake = Runner()
monkeypatch.setattr(subprocess, "run", fake)
return fake
class FakePopen:
def __init__(self, data: bytes = b"", rc: int = 0, err: bytes | None = b"") -> None:
self.stdout = io.BytesIO(data)
self.stderr = None if err is None else io.BytesIO(err)
self._rc = rc
def __enter__(self) -> Self:
return self
def __exit__(self, *_exc: object) -> None:
return None
def wait(self) -> int:
return self._rc
def test_docker_returns_stripped_stdout(run: Runner) -> None:
run.results = [SimpleNamespace(returncode=0, stdout=" out\n", stderr="")]
assert dock.docker("ps", timeout=7) == "out"
assert run.calls == [["docker", "ps"]]
assert run.kwargs[0]["timeout"] == 7
def test_docker_failure_raises_with_stderr(run: Runner) -> None:
run.results = [SimpleNamespace(returncode=1, stdout="", stderr=" no such \n")]
with pytest.raises(dock.DockerError, match="docker inspect failed: no such"):
dock.docker("inspect", "x")
def test_status_running_and_missing(cfg: gwc.Config, run: Runner) -> None:
run.results = [
SimpleNamespace(returncode=0, stdout="running\n", stderr=""),
SimpleNamespace(returncode=1, stdout="", stderr="Error: No such object"),
]
assert dock.status(cfg) == "running"
assert run.calls[0][:2] == ["docker", "inspect"]
assert run.calls[0][-1] == cfg.name
assert dock.status(cfg) is None
def test_recreate_removes_then_runs_on_image_volume_port(
cfg: gwc.Config, run: Runner
) -> None:
dock.recreate(cfg, "repo:1.2.3-web", "pw")
assert run.calls[0] == ["docker", "rm", "-f", cfg.name]
args = run.calls[1]
assert args[:3] == ["docker", "run", "-d"]
assert args[-1] == "repo:1.2.3-web"
assert "INITIAL_PASSWORD=pw" in args
assert "PORT=20555" in args
assert f"{cfg.volume}:/app/data" in args
assert args[args.index("--stop-timeout") + 1] == str(dock.STOP_TIMEOUT_S)
def test_recreate_propagates_run_failure(cfg: gwc.Config, run: Runner) -> None:
run.results = [
SimpleNamespace(returncode=1, stdout="", stderr=""),
SimpleNamespace(returncode=125, stdout="", stderr="bad image"),
]
with pytest.raises(dock.DockerError, match="bad image"):
dock.recreate(cfg, "img", "pw")
def test_restart_and_pull(cfg: gwc.Config, run: Runner) -> None:
dock.restart(cfg)
dock.pull("repo:2.0.0-web")
assert run.calls[0] == ["docker", "restart", "-t", "40", cfg.name]
assert run.calls[1] == ["docker", "pull", "-q", "repo:2.0.0-web"]
assert run.kwargs[1]["timeout"] == 900
class Docker:
"""Records `docker(...)` calls made by backup()."""
def __init__(self) -> None:
self.calls: list[tuple[str, ...]] = []
def __call__(self, *args: str, timeout: float = 120) -> str:
self.calls.append(args)
return ""
@pytest.fixture
def fake_docker(monkeypatch: pytest.MonkeyPatch) -> Docker:
fake = Docker()
monkeypatch.setattr(dock, "docker", fake)
return fake
def test_backup_writes_0600_gzip_stops_then_starts(
cfg: gwc.Config, fake_docker: Docker, monkeypatch: pytest.MonkeyPatch
) -> None:
popen_args: list[list[str]] = []
def popen(argv: list[str], **_kw: Any) -> FakePopen:
popen_args.append(argv)
return FakePopen(b"TARBYTES")
monkeypatch.setattr(subprocess, "Popen", popen)
dest = dock.backup(cfg, "3.8.52-web")
assert dest.parent == cfg.backup_dir
assert dest.name.endswith("-3.8.52-web.tar.gz")
assert dest.stat().st_mode & 0o777 == 0o600
assert cfg.backup_dir.stat().st_mode & 0o777 == 0o700
assert gzip.decompress(dest.read_bytes()) == b"TARBYTES"
assert popen_args == [["docker", "cp", f"{cfg.name}:/app/data", "-"]]
assert [c[0] for c in fake_docker.calls] == ["stop", "start"]
def test_backup_prunes_to_newest_four(
cfg: gwc.Config, fake_docker: Docker, monkeypatch: pytest.MonkeyPatch
) -> None:
cfg.backup_dir.mkdir(parents=True)
old = [cfg.backup_dir / f"2000010{i}-000000-old.tar.gz" for i in range(1, 6)]
for path in old:
path.write_bytes(b"x")
(cfg.backup_dir / "notes.txt").write_text("keep")
monkeypatch.setattr(subprocess, "Popen", lambda *_a, **_k: FakePopen(b"d"))
dest = dock.backup(cfg, "new")
left = sorted(p.name for p in cfg.backup_dir.glob("*.tar.gz"))
assert len(left) == dock.KEEP_BACKUPS
assert left == [*(p.name for p in old[2:]), dest.name]
assert (cfg.backup_dir / "notes.txt").exists()
def test_failed_copy_removes_partial_and_still_restarts(
cfg: gwc.Config, fake_docker: Docker, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(
subprocess, "Popen", lambda *_a, **_k: FakePopen(b"half", 1, b" cp boom\n")
)
with pytest.raises(dock.DockerError, match="docker cp failed: cp boom"):
dock.backup(cfg, "x")
assert list(cfg.backup_dir.glob("*.tar.gz")) == []
assert [c[0] for c in fake_docker.calls] == ["stop", "start"]
def test_copy_out_without_stderr_pipe_succeeds(
cfg: gwc.Config, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(subprocess, "Popen", lambda *_a, **_k: FakePopen(b"d", 0, None))
dest = tmp_path / "out.tar.gz"
dock._copy_out(cfg, dest)
assert gzip.decompress(dest.read_bytes()) == b"d"
assert dest.stat().st_mode & 0o777 == 0o600
def test_restore_pipes_archive_into_helper_container(
cfg: gwc.Config, run: Runner, tmp_path: Path
) -> None:
archive = tmp_path / "a.tar.gz"
archive.write_bytes(b"zz")
dock.restore(cfg, "repo:1.0.0-web", archive)
argv = run.calls[0]
assert argv[:2] == ["docker", "run"]
assert f"{cfg.volume}:/app/data" in argv
assert "repo:1.0.0-web" in argv
assert "stdin" in run.kwargs[0]
def test_restore_failure_raises(cfg: gwc.Config, run: Runner, tmp_path: Path) -> None:
archive = tmp_path / "a.tar.gz"
archive.write_bytes(b"zz")
run.results = [SimpleNamespace(returncode=2, stdout=b"", stderr=b" tar bad\n")]
with pytest.raises(dock.DockerError, match="restore failed: tar bad"):
dock.restore(cfg, "img", archive)

View File

@ -0,0 +1,156 @@
from __future__ import annotations
import runpy
import subprocess
import sys
import urllib.request
from typing import TYPE_CHECKING, Any
import pytest
from omniroute_offload import _gw_api as api
from omniroute_offload import _gw_combo as combo
from omniroute_offload import _gw_config as gwc
from omniroute_offload import _gw_docker as dock
from omniroute_offload import _gw_health as health
from omniroute_offload import _gw_update as update
from omniroute_offload import gateway
if TYPE_CHECKING:
from pathlib import Path
@pytest.fixture(autouse=True)
def _isolate(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
def blocked(*_args: Any, **_kwargs: Any) -> Any:
msg = "real process/network access is blocked in tests"
raise AssertionError(msg)
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setattr(subprocess, "run", blocked)
monkeypatch.setattr(subprocess, "Popen", blocked)
monkeypatch.setattr(urllib.request, "urlopen", blocked)
monkeypatch.setattr(urllib.request, "build_opener", blocked)
monkeypatch.delenv("OMNIROUTE_NAME", raising=False)
monkeypatch.delenv("OMNIROUTE_PORT", raising=False)
monkeypatch.delenv("OMNIROUTE_VOLUME", raising=False)
(tmp_path / ".config/omniroute").mkdir(parents=True)
(tmp_path / ".config/omniroute/admin_password").write_text("pw\n")
class Calls:
def __init__(self) -> None:
self.events: list[tuple[Any, ...]] = []
@pytest.fixture
def calls(monkeypatch: pytest.MonkeyPatch) -> Calls:
rec = Calls()
def recreate(cfg: gwc.Config, image: str, password: str) -> None:
rec.events.append(("recreate", cfg.name, image, password))
monkeypatch.setattr(dock, "recreate", recreate)
monkeypatch.setattr(health, "wait_up", lambda _cfg: True)
return rec
def test_start_recreates_pinned_image_and_reports_url(
calls: Calls, capsys: pytest.CaptureFixture[str]
) -> None:
assert gateway.main(["start"]) == 0
assert calls.events == [("recreate", "omniroute", gwc.DEFAULT_IMAGE, "pw")]
assert "omniroute up: http://127.0.0.1:20128" in capsys.readouterr().out
def test_start_fails_when_api_never_comes_up(
calls: Calls, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
monkeypatch.setattr(health, "wait_up", lambda _cfg: False)
assert gateway.main(["start"]) == 1
assert "docker logs omniroute" in capsys.readouterr().err
def test_instance_comes_from_environment(
calls: Calls, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("OMNIROUTE_NAME", "tetris")
assert gateway.main(["start"]) == 0
assert calls.events[0][1] == "tetris"
def test_backup_prints_path_with_image_tag_label(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
capsys: pytest.CaptureFixture[str],
) -> None:
labels: list[str] = []
def backup(_cfg: gwc.Config, label: str) -> Path:
labels.append(label)
return tmp_path / "x.tar.gz"
monkeypatch.setattr(dock, "backup", backup)
assert gateway.main(["backup"]) == 0
assert labels == ["3.8.52-web"]
assert capsys.readouterr().out.strip() == str(tmp_path / "x.tar.gz")
@pytest.mark.parametrize("code", [0, 1])
def test_combo_exit_code_follows_refresh(
monkeypatch: pytest.MonkeyPatch, code: int
) -> None:
monkeypatch.setattr(combo, "refresh", lambda _cfg: code == 0)
assert gateway.main(["combo"]) == code
@pytest.mark.parametrize(
("command", "module"), [("health", health), ("update", update)]
)
def test_health_and_update_dispatch_and_propagate_exit_code(
monkeypatch: pytest.MonkeyPatch, command: str, module: Any
) -> None:
seen: list[str] = []
def run(cfg: gwc.Config) -> int:
seen.append(cfg.name)
return 7
monkeypatch.setattr(module, "run", run)
assert gateway.main([command]) == 7
assert seen == ["omniroute"]
@pytest.mark.parametrize(
"error",
[dock.DockerError("daemon down"), api.ApiError("refused"), OSError("no file")],
)
def test_known_errors_become_exit_1_with_message(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
error: Exception,
) -> None:
def boom(_cfg: gwc.Config) -> bool:
raise error
monkeypatch.setattr(combo, "refresh", boom)
assert gateway.main(["combo"]) == 1
assert f"omniroute-gateway combo: {error}" in capsys.readouterr().err
def test_unknown_command_is_rejected() -> None:
with pytest.raises(SystemExit) as exc:
gateway.main(["explode"])
assert exc.value.code == 2
def test_main_reads_sys_argv_and_runs_as_script(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(combo, "refresh", lambda _cfg: True)
monkeypatch.setattr(sys, "argv", ["omniroute-gateway", "combo"])
assert gateway.main() == 0
monkeypatch.delitem(sys.modules, "omniroute_offload.gateway")
with pytest.raises(SystemExit) as exc:
runpy.run_module("omniroute_offload.gateway", run_name="__main__")
assert exc.value.code == 0

View File

@ -0,0 +1,231 @@
from __future__ import annotations
import subprocess
import time
import urllib.request
from datetime import UTC, datetime, timedelta
from typing import TYPE_CHECKING, Any
import pytest
from omniroute_offload import _gw_api as api
from omniroute_offload import _gw_health as health
if TYPE_CHECKING:
from pathlib import Path
from omniroute_offload import _gw_config as gwc
from omniroute_offload.tests.conftest import World
NOW = datetime(2026, 11, 5, 12, 0, tzinfo=UTC)
PAST = (NOW - timedelta(hours=1)).isoformat()
FUTURE = (NOW + timedelta(hours=1)).isoformat()
@pytest.fixture(autouse=True)
def _isolate(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
def blocked(*_args: Any, **_kwargs: Any) -> Any:
msg = "real process/network access is blocked in tests"
raise AssertionError(msg)
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setattr(subprocess, "run", blocked)
monkeypatch.setattr(subprocess, "Popen", blocked)
monkeypatch.setattr(urllib.request, "urlopen", blocked)
monkeypatch.setattr(urllib.request, "build_opener", blocked)
class Clock:
def __init__(self) -> None:
self.now = 0.0
self.sleeps: list[float] = []
def monotonic(self) -> float:
return self.now
def sleep(self, seconds: float) -> None:
self.sleeps.append(seconds)
self.now += seconds
@pytest.fixture
def clock(monkeypatch: pytest.MonkeyPatch) -> Clock:
fake = Clock()
monkeypatch.setattr(time, "monotonic", fake.monotonic)
monkeypatch.setattr(time, "sleep", fake.sleep)
return fake
def test_secret_is_stripped(health_cfg: gwc.Config) -> None:
assert health.secret(health_cfg, "api_key") == "key"
def test_wait_up_returns_at_once_when_alive(
health_cfg: gwc.Config, clock: Clock, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(api, "alive", lambda _cfg: True)
assert health.wait_up(health_cfg) is True
assert clock.sleeps == []
def test_wait_up_polls_every_two_seconds_until_alive(
health_cfg: gwc.Config, clock: Clock, monkeypatch: pytest.MonkeyPatch
) -> None:
answers = iter([False, False, True])
monkeypatch.setattr(api, "alive", lambda _cfg: next(answers))
assert health.wait_up(health_cfg) is True
assert clock.sleeps == [2, 2]
def test_wait_up_gives_up_after_deadline(
health_cfg: gwc.Config, clock: Clock, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(api, "alive", lambda _cfg: False)
assert health.wait_up(health_cfg, seconds=5) is False
assert clock.sleeps == [2, 2, 2]
def test_check_down_when_api_dead(health_cfg: gwc.Config, world: World) -> None:
world.alive = False
verdict = health.check(health_cfg, NOW)
assert verdict == health.Verdict("down", "API not answering")
def test_check_ok_on_real_chat(health_cfg: gwc.Config, world: World) -> None:
assert health.check(health_cfg, NOW).state == "ok"
def test_check_degraded_only_when_all_active_members_cool(
health_cfg: gwc.Config, world: World
) -> None:
world.chat_ok = False
world.conns = [
world.conn("1", rateLimitedUntil=FUTURE),
world.conn("2", rateLimitedUntil=FUTURE),
world.conn("3", provider="outsider"),
world.conn("4", isActive=False),
]
verdict = health.check(health_cfg, NOW)
assert verdict.state == "degraded"
assert "all 2" in verdict.detail
world.conns.append(world.conn("5", rateLimitedUntil=PAST))
assert health.check(health_cfg, NOW).state == "down"
def test_check_down_without_member_connections(
health_cfg: gwc.Config, world: World
) -> None:
world.chat_ok = False
world.conns = [world.conn("3", provider="outsider")]
verdict = health.check(health_cfg, NOW)
assert verdict == health.Verdict("down", "chat failed: 503 all providers failed")
def test_check_admin_failure_is_down_with_both_reasons(
health_cfg: gwc.Config, world: World
) -> None:
world.chat_ok = False
world.admin_error = api.ApiError("login refused")
verdict = health.check(health_cfg, NOW)
assert verdict.state == "down"
assert "chat failed" in verdict.detail
assert "admin API: login refused" in verdict.detail
def test_check_missing_key_file_counts_as_chat_failure(
health_cfg: gwc.Config, world: World
) -> None:
(health_cfg.conf_dir / "api_key").unlink()
(health_cfg.conf_dir / "admin_password").unlink()
verdict = health.check(health_cfg, NOW)
assert verdict.state == "down"
assert "admin API" in verdict.detail
def test_restart_proceeds_on_corrupt_last_restart(
health_cfg: gwc.Config, world: World
) -> None:
state = {"last_restart": "not-a-date"}
health._restart(health_cfg, state, NOW)
assert world.events == ["restart"]
assert state["last_restart"] == NOW.isoformat()
def test_restart_respects_gap_then_proceeds(
health_cfg: gwc.Config, world: World
) -> None:
last = (NOW - timedelta(minutes=29)).isoformat()
recent = {"last_restart": last}
health._restart(health_cfg, recent, NOW)
assert world.events == []
assert recent["last_restart"] == last
old = {"last_restart": (NOW - timedelta(minutes=31)).isoformat()}
health._restart(health_cfg, old, NOW)
assert world.events == ["restart"]
assert old["last_restart"] == NOW.isoformat()
def test_restart_recreates_missing_container(
health_cfg: gwc.Config, world: World
) -> None:
world.status = None
state: dict[str, Any] = {}
health._restart(health_cfg, state, NOW)
assert world.events == ["recreate"]
assert state["last_restart"] == NOW.isoformat()
def test_reactivate_only_stuck_combo_members_with_ended_cooldown(
health_cfg: gwc.Config, world: World
) -> None:
conn = world.conn
world.conns = [
conn("ok"),
conn("cooling", testStatus="error", rateLimitedUntil=FUTURE),
conn("stuck", testStatus="error", rateLimitedUntil=PAST),
conn("stuck2", testStatus="unavailable"),
conn("outsider", provider="other", testStatus="error"),
conn("off", isActive=False, testStatus="error"),
]
assert health._reactivate(health_cfg, NOW) == 2
assert world.events == ["reactivate:stuck", "reactivate:stuck2"]
def test_heal_ok_touches_nothing(health_cfg: gwc.Config, world: World) -> None:
assert health.heal(health_cfg, {}, NOW).state == "ok"
assert world.events == []
@pytest.mark.parametrize("status", ["exited", None])
def test_heal_restarts_dead_container_then_rechecks(
health_cfg: gwc.Config, world: World, status: str | None
) -> None:
world.status, world.alive = status, False
assert health.heal(health_cfg, {}, NOW).state == "ok"
assert world.events == ["restart" if status else "recreate"]
def test_heal_restarts_running_container_with_dead_api(
health_cfg: gwc.Config, world: World
) -> None:
world.alive, world.restart_fixes = False, False
assert health.heal(health_cfg, {}, NOW).state == "down"
assert world.events == ["restart"]
def test_heal_reactivates_instead_of_restarting_when_container_is_fine(
health_cfg: gwc.Config, world: World
) -> None:
world.chat_ok, world.heal_by_reactivate = False, True
world.conns = [world.conn("1", testStatus="error", rateLimitedUntil=PAST)]
assert health.heal(health_cfg, {}, NOW).state == "ok"
assert world.events == ["reactivate:1"]
def test_heal_stays_down_when_nothing_to_reactivate(
health_cfg: gwc.Config, world: World
) -> None:
world.chat_ok = False
world.conns = [world.conn("1")]
assert health.heal(health_cfg, {}, NOW).state == "down"
assert world.events == []

View File

@ -0,0 +1,169 @@
from __future__ import annotations
import subprocess
import urllib.request
from datetime import UTC, datetime, timedelta
from typing import TYPE_CHECKING, Any
import pytest
from omniroute_offload import _gw_config as gwc
from omniroute_offload import _gw_docker as dock
from omniroute_offload import _gw_health as health
if TYPE_CHECKING:
from pathlib import Path
from omniroute_offload.tests.conftest import World
NOW = datetime(2026, 11, 5, 12, 0, tzinfo=UTC)
PAST = (NOW - timedelta(hours=1)).isoformat()
FUTURE = (NOW + timedelta(hours=1)).isoformat()
@pytest.fixture(autouse=True)
def _isolate(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
def blocked(*_args: Any, **_kwargs: Any) -> Any:
msg = "real process/network access is blocked in tests"
raise AssertionError(msg)
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setattr(subprocess, "run", blocked)
monkeypatch.setattr(subprocess, "Popen", blocked)
monkeypatch.setattr(urllib.request, "urlopen", blocked)
monkeypatch.setattr(urllib.request, "build_opener", blocked)
def test_report_ok_recovery_clears_down_since(
health_cfg: gwc.Config, world: World
) -> None:
ok = health.Verdict("ok", "")
assert health._report(health_cfg, {}, ok, NOW) == 0
assert world.notices == []
state: dict[str, Any] = {"down_since": "t"}
assert health._report(health_cfg, state, ok, NOW) == 0
assert state == {}
assert world.notices == [("low", "recovered: free-best answers again")]
def test_report_degraded_notifies_once_per_day(
health_cfg: gwc.Config, world: World
) -> None:
verdict = health.Verdict("degraded", "all 2 combo providers cooling down")
state: dict[str, Any] = {}
assert health._report(health_cfg, state, verdict, NOW) == 0
assert health._report(health_cfg, state, verdict, NOW + timedelta(hours=1)) == 0
assert [u for u, _ in world.notices] == ["low"]
assert health._report(health_cfg, state, verdict, NOW + timedelta(days=1)) == 0
assert len(world.notices) == 2
def test_report_failed_alert_exits_1(health_cfg: gwc.Config, world: World) -> None:
world.notify_ok = False
state: dict[str, Any] = {"down_since": "t"}
assert health._report(health_cfg, state, health.Verdict("ok", ""), NOW) == 1
assert [u for u, _ in world.notices] == ["low"]
degraded = health.Verdict("degraded", "all cooling")
assert health._report(health_cfg, state, degraded, NOW) == 1
assert [u for u, _ in world.notices] == ["low", "low"]
# same day again: nothing is sent, so nothing can fail
assert health._report(health_cfg, state, degraded, NOW) == 0
assert len(world.notices) == 2
def test_report_down_notifies_once_per_outage(
health_cfg: gwc.Config, world: World
) -> None:
verdict = health.Verdict("down", "boom")
state: dict[str, Any] = {}
assert health._report(health_cfg, state, verdict, NOW) == 1
assert health._report(health_cfg, state, verdict, NOW + timedelta(hours=1)) == 1
assert world.notices == [("critical", "DOWN after auto-fix: boom")]
assert state["down_since"] == NOW.isoformat()
def test_run_skips_when_lock_held(health_cfg: gwc.Config, world: World) -> None:
with gwc.locked(health_cfg, blocking=True):
assert health.run(health_cfg, NOW) == 0
assert world.events == []
assert world.notices == []
def test_run_healthy_uses_current_time_and_stays_quiet(
health_cfg: gwc.Config, world: World
) -> None:
assert health.run(health_cfg) == 0
assert world.notices == []
def test_run_dead_container_is_fixed_without_alert(
health_cfg: gwc.Config, world: World
) -> None:
world.status, world.alive = "exited", False
assert health.run(health_cfg, NOW) == 0
assert world.events == ["restart"]
assert world.notices == []
assert gwc.load_state(health_cfg)["last_restart"] == NOW.isoformat()
def test_run_missing_container_is_recreated(
health_cfg: gwc.Config, world: World
) -> None:
world.status, world.alive = None, False
assert health.run(health_cfg, NOW) == 0
assert world.events == ["recreate"]
def test_run_second_restart_within_gap_stays_down_and_alerts_once(
health_cfg: gwc.Config, world: World
) -> None:
world.status, world.alive = "exited", False
gwc.save_state(
health_cfg, {"last_restart": (NOW - timedelta(minutes=5)).isoformat()}
)
assert health.run(health_cfg, NOW) == 1
assert world.events == []
assert [u for u, _ in world.notices] == ["critical"]
assert health.run(health_cfg, NOW + timedelta(minutes=10)) == 1
assert len(world.notices) == 1
def test_run_outage_then_recovery_cycle(health_cfg: gwc.Config, world: World) -> None:
world.alive, world.restart_fixes = False, False
assert health.run(health_cfg, NOW) == 1
assert gwc.load_state(health_cfg)["down_since"] == NOW.isoformat()
world.alive = True
assert health.run(health_cfg, NOW + timedelta(hours=1)) == 0
assert "down_since" not in gwc.load_state(health_cfg)
assert [u for u, _ in world.notices] == ["critical", "low"]
def test_run_degraded_exit_zero_no_restart_one_notice(
health_cfg: gwc.Config, world: World
) -> None:
world.chat_ok = False
world.conns = [world.conn("1", rateLimitedUntil=FUTURE)]
assert health.run(health_cfg, NOW) == 0
assert health.run(health_cfg, NOW + timedelta(minutes=10)) == 0
assert world.events == []
assert [u for u, _ in world.notices] == ["low"]
def test_run_reactivates_only_ended_cooldowns_and_never_restarts(
health_cfg: gwc.Config, world: World
) -> None:
world.chat_ok, world.heal_by_reactivate = False, True
world.conns = [
world.conn("a", testStatus="error", rateLimitedUntil=PAST),
world.conn("b", provider="outsider", testStatus="error"),
]
assert health.run(health_cfg, NOW) == 0
assert world.events == ["reactivate:a"]
def test_run_ladder_failure_becomes_down(health_cfg: gwc.Config, world: World) -> None:
msg = "docker restart failed: daemon gone"
world.restart_error = dock.DockerError(msg)
world.status, world.chat_ok = "exited", False
assert health.run(health_cfg, NOW) == 1
assert "fix ladder failed" in world.notices[0][1]

View File

@ -0,0 +1,185 @@
from __future__ import annotations
import subprocess
import urllib.request
from types import SimpleNamespace
from typing import TYPE_CHECKING, Any
import pytest
from omniroute_offload import _gw_api as api
from omniroute_offload import _gw_config as gwc
from omniroute_offload import _gw_notify as note
if TYPE_CHECKING:
from pathlib import Path
@pytest.fixture(autouse=True)
def _isolate(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
def blocked(*_args: Any, **_kwargs: Any) -> Any:
msg = "real process/network access is blocked in tests"
raise AssertionError(msg)
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setattr(subprocess, "run", blocked)
monkeypatch.setattr(subprocess, "Popen", blocked)
monkeypatch.setattr(urllib.request, "urlopen", blocked)
monkeypatch.setattr(urllib.request, "build_opener", blocked)
@pytest.fixture
def cfg(tmp_path: Path) -> gwc.Config:
return gwc.Config.from_env(
{"HOME": str(tmp_path), "OMNIROUTE_SIGNAL_ENV": str(tmp_path / "sig.env")}
)
@pytest.fixture
def signal_env(cfg: gwc.Config) -> Path:
cfg.signal_env.write_text(
"# comment\n"
"\n"
"no equals sign\n"
" # indented comment=1\n"
"PHONE_NUMBER = +48111\n"
"TRUSTED_INVITERS=+48222, +48333\n"
"API_BASE_URL=http://127.0.0.1:8080\n"
)
return cfg.signal_env
class Sent:
"""Replaces api.call and subprocess.run; records what would go out."""
def __init__(self) -> None:
self.posts: list[tuple[str, dict[str, Any], float]] = []
self.popups: list[list[str]] = []
self.post_error: Exception | None = None
self.popup_error: Exception | None = None
self.popup_rc = 0
self.next_id = 41
def call(self, url: str, body: dict[str, Any], *, timeout: float) -> Any:
if self.post_error:
raise self.post_error
self.posts.append((url, body, timeout))
return {}
def run(self, argv: list[str], **_kw: Any) -> Any:
if self.popup_error:
raise self.popup_error
self.popups.append(argv)
self.next_id += 1
return SimpleNamespace(
returncode=self.popup_rc, stdout=f"{self.next_id}\n", stderr=" no dbus \n"
)
@pytest.fixture
def sent(monkeypatch: pytest.MonkeyPatch) -> Sent:
fake = Sent()
monkeypatch.setattr(api, "call", fake.call)
monkeypatch.setattr(subprocess, "run", fake.run)
return fake
def test_signal_dms_first_trusted_inviter(
cfg: gwc.Config, signal_env: Path, sent: Sent
) -> None:
note.signal(cfg, "hello")
assert sent.posts == [
(
"http://127.0.0.1:8080/v2/send",
{"number": "+48111", "recipients": ["+48222"], "message": "hello"},
20,
)
]
def test_signal_missing_file_missing_key_and_api_error(
cfg: gwc.Config, signal_env: Path, sent: Sent
) -> None:
signal_env.write_text("PHONE_NUMBER=+1\n")
with pytest.raises(note.NotifyError, match="signal"):
note.signal(cfg, "x")
signal_env.unlink()
with pytest.raises(note.NotifyError, match="signal"):
note.signal(cfg, "x")
signal_env.write_text(
"PHONE_NUMBER=+1\nTRUSTED_INVITERS=+2\nAPI_BASE_URL=http://h\n"
)
sent.post_error = api.ApiError("refused")
with pytest.raises(note.NotifyError, match="refused"):
note.signal(cfg, "x")
def test_desktop_stores_id_and_replaces_previous(cfg: gwc.Config, sent: Sent) -> None:
cfg.state_dir.mkdir(parents=True)
note.desktop(cfg, "first", "low")
assert "--replace-id=0" in sent.popups[0]
assert "--urgency=low" in sent.popups[0]
assert sent.popups[0][-2:] == [f"OmniRoute ({cfg.name})", "first"]
assert (cfg.state_dir / "notify_id").read_text() == "42"
note.desktop(cfg, "second", "critical")
assert "--replace-id=42" in sent.popups[1]
assert (cfg.state_dir / "notify_id").read_text() == "43"
def test_desktop_failure_raises_and_keeps_id(cfg: gwc.Config, sent: Sent) -> None:
cfg.state_dir.mkdir(parents=True)
sent.popup_rc = 1
with pytest.raises(note.NotifyError, match="no dbus"):
note.desktop(cfg, "x", "normal")
assert not (cfg.state_dir / "notify_id").exists()
def test_notify_both_channels_true(
cfg: gwc.Config,
signal_env: Path,
sent: Sent,
capsys: pytest.CaptureFixture[str],
) -> None:
assert note.notify(cfg, "all good", urgency="low") is True
assert len(sent.popups) == 1
assert sent.posts[0][1]["message"] == f"OmniRoute ({cfg.name}): all good"
assert "notify: all good" in capsys.readouterr().err
@pytest.mark.parametrize(
"error",
[
FileNotFoundError("notify-send"),
subprocess.TimeoutExpired("notify-send", 10),
],
)
def test_desktop_failure_does_not_block_signal(
cfg: gwc.Config,
signal_env: Path,
sent: Sent,
capsys: pytest.CaptureFixture[str],
error: Exception,
) -> None:
sent.popup_error = error
assert note.notify(cfg, "x") is True
assert len(sent.posts) == 1
assert "desktop notify failed" in capsys.readouterr().err
def test_desktop_notify_error_is_logged_too(
cfg: gwc.Config,
signal_env: Path,
sent: Sent,
capsys: pytest.CaptureFixture[str],
) -> None:
sent.popup_rc = 1
assert note.notify(cfg, "x") is True
assert "desktop notify failed" in capsys.readouterr().err
def test_missing_signal_env_returns_false(
cfg: gwc.Config, sent: Sent, capsys: pytest.CaptureFixture[str]
) -> None:
assert note.notify(cfg, "x") is False
assert len(sent.popups) == 1
assert "signal notify failed" in capsys.readouterr().err

View File

@ -0,0 +1,250 @@
from __future__ import annotations
import subprocess
import urllib.request
from typing import TYPE_CHECKING, Any
import pytest
from omniroute_offload import _gw_api as api
from omniroute_offload import _gw_combo as combo
from omniroute_offload import _gw_config as gwc
from omniroute_offload import _gw_docker as dock
from omniroute_offload import _gw_health as health
from omniroute_offload import _gw_notify as note
from omniroute_offload import _gw_update as update
if TYPE_CHECKING:
from pathlib import Path
OLD = gwc.DEFAULT_IMAGE
NEW = f"{gwc.REPO}:3.9.0-web"
@pytest.fixture(autouse=True)
def _isolate(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
def blocked(*_args: Any, **_kwargs: Any) -> Any:
msg = "real process/network access is blocked in tests"
raise AssertionError(msg)
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setattr(subprocess, "run", blocked)
monkeypatch.setattr(subprocess, "Popen", blocked)
monkeypatch.setattr(urllib.request, "urlopen", blocked)
monkeypatch.setattr(urllib.request, "build_opener", blocked)
@pytest.fixture
def cfg(tmp_path: Path) -> gwc.Config:
made = gwc.Config.from_env({"HOME": str(tmp_path)})
made.conf_dir.mkdir(parents=True)
(made.conf_dir / "admin_password").write_text("pw\n")
return made
class Ops:
"""Fake docker/health/notify layer; `events` keeps the order of side effects."""
def __init__(self, archive: Path) -> None:
self.archive = archive
self.events: list[tuple[str, ...]] = []
self.notices: list[tuple[str, str]] = []
self.latest = NEW
self.verdict = health.Verdict("ok", "chat answered")
self.up = True
self.errors: dict[str, Exception] = {}
self.container: str | None = "running"
self.notify_ok = True
def _step(self, name: str, *detail: str) -> None:
self.events.append((name, *detail))
if name in self.errors:
raise self.errors[name]
def backup(self, _cfg: gwc.Config, label: str) -> Path:
self._step("backup", label)
return self.archive
def latest_image(self) -> str:
self._step("latest")
return self.latest
def pull(self, image: str) -> None:
self._step("pull", image)
def recreate(self, _cfg: gwc.Config, image: str, pw: str) -> None:
assert pw == "pw"
self._step("recreate", image)
def status(self, _cfg: gwc.Config) -> str | None:
self._step("status")
return self.container
def docker(self, *args: str, timeout: float = 120) -> str:
self._step("docker", *args)
return ""
def restore(self, _cfg: gwc.Config, image: str, archive: Path) -> None:
assert archive == self.archive
self._step("restore", image)
def wait_up(self, _cfg: gwc.Config, seconds: float = 90) -> bool:
self._step("wait_up")
return self.up
def check(self, _cfg: gwc.Config, _now: object) -> health.Verdict:
self._step("check")
return self.verdict
def refresh(self, _cfg: gwc.Config) -> bool:
self._step("combo")
return True
def notify(self, _cfg: gwc.Config, text: str, *, urgency: str = "normal") -> bool:
self.notices.append((urgency, text))
return self.notify_ok
def names(self) -> list[str]:
return [e[0] for e in self.events]
@pytest.fixture
def ops(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Ops:
o = Ops(tmp_path / "backup.tar.gz")
monkeypatch.setattr(dock, "backup", o.backup)
monkeypatch.setattr(update, "latest_image", o.latest_image)
monkeypatch.setattr(dock, "pull", o.pull)
monkeypatch.setattr(dock, "recreate", o.recreate)
monkeypatch.setattr(dock, "docker", o.docker)
monkeypatch.setattr(dock, "status", o.status)
monkeypatch.setattr(dock, "restore", o.restore)
monkeypatch.setattr(health, "wait_up", o.wait_up)
monkeypatch.setattr(health, "check", o.check)
monkeypatch.setattr(combo, "refresh", o.refresh)
monkeypatch.setattr(note, "notify", o.notify)
return o
def test_latest_image_picks_highest_stable_web_tag(
monkeypatch: pytest.MonkeyPatch,
) -> None:
names = ["3.9.0-web", "3.10.1-web", "3.11.0-rc1-web", "latest", "3.8.52-web"]
seen: list[str] = []
def fake_call(url: str, *, timeout: float) -> Any:
seen.append(url)
return {"results": [{"name": n} for n in names] + [{}]}
monkeypatch.setattr(api, "call", fake_call)
assert update.latest_image() == f"{gwc.REPO}:3.10.1-web"
assert seen == [update.HUB]
def test_latest_image_without_stable_tag_raises(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(api, "call", lambda *_a, **_k: {"results": [{"name": "dev"}]})
with pytest.raises(api.ApiError, match="no stable"):
update.latest_image()
@pytest.mark.parametrize("latest", [OLD, f"{gwc.REPO}:3.8.1-web"])
def test_up_to_date_changes_nothing(cfg: gwc.Config, ops: Ops, latest: str) -> None:
ops.latest = latest
assert update.run(cfg) == 0
assert ops.names() == ["backup", "latest"]
assert ops.events[0] == ("backup", "3.8.52-web")
assert ops.notices == []
assert not (cfg.state_dir / "image").exists()
def test_unparseable_pinned_image_is_left_alone(cfg: gwc.Config, ops: Ops) -> None:
gwc.save_state(cfg, {}, image="repo:latest")
assert update.run(cfg) == 0
assert ops.names() == ["backup", "latest"]
def test_success_pins_image_refreshes_combo_and_notifies(
cfg: gwc.Config, ops: Ops
) -> None:
gwc.save_state(cfg, {"last_restart": "t"})
assert update.run(cfg) == 0
assert ops.names() == [
"backup",
"latest",
"pull",
"recreate",
"wait_up",
"check",
"combo",
]
assert ("recreate", NEW) in ops.events
assert cfg.image() == NEW
assert gwc.load_state(cfg) == {"last_restart": "t"}
assert ops.notices == [("low", f"updated {OLD} -> {NEW}")]
def test_unsent_success_notice_exits_1_but_keeps_the_update(
cfg: gwc.Config, ops: Ops
) -> None:
ops.notify_ok = False
assert update.run(cfg) == 1
assert cfg.image() == NEW
assert ops.names()[-1] == "combo"
assert ops.notices == [("low", f"updated {OLD} -> {NEW}")]
def test_rollback_skips_stop_when_container_is_gone(cfg: gwc.Config, ops: Ops) -> None:
ops.verdict = health.Verdict("down", "boom")
ops.container = None
assert update.run(cfg) == 1
tail = ops.events[ops.names().index("status") :]
assert tail == [("status",), ("restore", OLD), ("recreate", OLD), ("wait_up",)]
assert ("docker", "stop", "-t", "40", cfg.name) not in ops.events
def test_failed_health_rolls_back_in_order_and_alerts(
cfg: gwc.Config, ops: Ops
) -> None:
ops.verdict = health.Verdict("down", "chat failed: 503")
assert update.run(cfg) == 1
assert ops.events[-6:] == [
("check",),
("status",),
("docker", "stop", "-t", "40", cfg.name),
("restore", OLD),
("recreate", OLD),
("wait_up",),
]
assert cfg.image() == OLD
assert "combo" not in ops.names()
urgency, text = ops.notices[0]
assert urgency == "critical"
assert f"update to {NEW} failed (chat failed: 503); rolled back to {OLD}" == text
def test_api_never_up_rolls_back(cfg: gwc.Config, ops: Ops) -> None:
ops.up = False
assert update.run(cfg) == 1
assert "check" not in ops.names()[: ops.names().index("restore")]
assert "never came up" in ops.notices[0][1]
assert ("restore", OLD) in ops.events
@pytest.mark.parametrize(
"error",
[dock.DockerError("pull denied"), api.ApiError("x"), OSError("disk")],
)
def test_pull_failure_rolls_back(cfg: gwc.Config, ops: Ops, error: Exception) -> None:
ops.errors["pull"] = error
assert update.run(cfg) == 1
assert ("restore", OLD) in ops.events
assert str(error) in ops.notices[0][1]
@pytest.mark.parametrize("step", ["backup", "latest"])
def test_job_failure_notifies_and_exits_1(cfg: gwc.Config, ops: Ops, step: str) -> None:
ops.errors[step] = dock.DockerError("docker cp failed: boom")
assert update.run(cfg) == 1
assert "restore" not in ops.names()
assert ops.notices == [("critical", "update job failed: docker cp failed: boom")]

View File

@ -13,6 +13,7 @@ dependencies = []
free-llm = "omniroute_offload._cli:main" free-llm = "omniroute_offload._cli:main"
free-llm-read-hook = "omniroute_offload._hook:main" free-llm-read-hook = "omniroute_offload._hook:main"
free-llm-check-dir = "omniroute_offload._privacy:check_dir_main" free-llm-check-dir = "omniroute_offload._privacy:check_dir_main"
omniroute-gateway = "omniroute_offload.gateway:main"
[tool.setuptools.packages.find] [tool.setuptools.packages.find]
include = ["omniroute_offload*"] include = ["omniroute_offload*"]
@ -62,7 +63,7 @@ omit = ["*/__pycache__/*", "*/tests/*"]
[tool.coverage.report] [tool.coverage.report]
fail_under = 100 fail_under = 100
show_missing = true show_missing = true
exclude_lines = ["pragma: no cover", 'if __name__ == "__main__":'] exclude_lines = ["pragma: no cover", 'if __name__ == "__main__":', "if TYPE_CHECKING:"]
[tool.pylint.main] [tool.pylint.main]
# The repo root on the path, so pylint resolves the package's own imports the # The repo root on the path, so pylint resolves the package's own imports the

View File

@ -0,0 +1,8 @@
[Unit]
Description=OmniRoute gateway health check and auto-fix
After=docker.service
[Service]
Type=oneshot
ExecStart=%h/.local/bin/omniroute-gateway health
TimeoutStartSec=15min

View File

@ -0,0 +1,9 @@
[Unit]
Description=OmniRoute health check every 10 minutes
[Timer]
OnBootSec=3min
OnUnitActiveSec=10min
[Install]
WantedBy=timers.target

View File

@ -0,0 +1,8 @@
[Unit]
Description=OmniRoute data backup and image update with rollback
After=docker.service
[Service]
Type=oneshot
ExecStart=%h/.local/bin/omniroute-gateway update
TimeoutStartSec=45min

View File

@ -0,0 +1,10 @@
[Unit]
Description=OmniRoute weekly backup and update
[Timer]
# The PC is off at night; Persistent catches a missed slot at the next boot.
OnCalendar=Sun 12:00
Persistent=true
[Install]
WantedBy=timers.target