diff --git a/omniroute_offload/install.sh b/omniroute_offload/install.sh index cdbd333..e41d76f 100755 --- a/omniroute_offload/install.sh +++ b/omniroute_offload/install.sh @@ -1,13 +1,19 @@ #!/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 REPO_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" readonly REPO_DIR readonly SYSTEM_PYTHON="/usr/bin/python3" 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}" -mkdir -p "${BIN_DIR}" +mkdir -p "${BIN_DIR}" "${UNIT_DIR}" 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 diff --git a/omniroute_offload/omniroute_offload/_gw_api.py b/omniroute_offload/omniroute_offload/_gw_api.py new file mode 100644 index 0000000..9c09ee6 --- /dev/null +++ b/omniroute_offload/omniroute_offload/_gw_api.py @@ -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 diff --git a/omniroute_offload/omniroute_offload/_gw_combo.py b/omniroute_offload/omniroute_offload/_gw_combo.py new file mode 100644 index 0000000..e601860 --- /dev/null +++ b/omniroute_offload/omniroute_offload/_gw_combo.py @@ -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 diff --git a/omniroute_offload/omniroute_offload/_gw_config.py b/omniroute_offload/omniroute_offload/_gw_config.py new file mode 100644 index 0000000..b4f3aa9 --- /dev/null +++ b/omniroute_offload/omniroute_offload/_gw_config.py @@ -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") diff --git a/omniroute_offload/omniroute_offload/_gw_docker.py b/omniroute_offload/omniroute_offload/_gw_docker.py new file mode 100644 index 0000000..b9ddda0 --- /dev/null +++ b/omniroute_offload/omniroute_offload/_gw_docker.py @@ -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) diff --git a/omniroute_offload/omniroute_offload/_gw_health.py b/omniroute_offload/omniroute_offload/_gw_health.py new file mode 100644 index 0000000..c490356 --- /dev/null +++ b/omniroute_offload/omniroute_offload/_gw_health.py @@ -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 diff --git a/omniroute_offload/omniroute_offload/_gw_notify.py b/omniroute_offload/omniroute_offload/_gw_notify.py new file mode 100644 index 0000000..2b0a885 --- /dev/null +++ b/omniroute_offload/omniroute_offload/_gw_notify.py @@ -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 diff --git a/omniroute_offload/omniroute_offload/_gw_update.py b/omniroute_offload/omniroute_offload/_gw_update.py new file mode 100644 index 0000000..972a20c --- /dev/null +++ b/omniroute_offload/omniroute_offload/_gw_update.py @@ -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 diff --git a/omniroute_offload/omniroute_offload/gateway.py b/omniroute_offload/omniroute_offload/gateway.py new file mode 100644 index 0000000..f867ad1 --- /dev/null +++ b/omniroute_offload/omniroute_offload/gateway.py @@ -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()) diff --git a/omniroute_offload/omniroute_offload/tests/conftest.py b/omniroute_offload/omniroute_offload/tests/conftest.py new file mode 100644 index 0000000..b999256 --- /dev/null +++ b/omniroute_offload/omniroute_offload/tests/conftest.py @@ -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 diff --git a/omniroute_offload/omniroute_offload/tests/test_gw_api.py b/omniroute_offload/omniroute_offload/tests/test_gw_api.py new file mode 100644 index 0000000..0a3db43 --- /dev/null +++ b/omniroute_offload/omniroute_offload/tests/test_gw_api.py @@ -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"")] + 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 diff --git a/omniroute_offload/omniroute_offload/tests/test_gw_api_admin.py b/omniroute_offload/omniroute_offload/tests/test_gw_api_admin.py new file mode 100644 index 0000000..4941f1e --- /dev/null +++ b/omniroute_offload/omniroute_offload/tests/test_gw_api_admin.py @@ -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" diff --git a/omniroute_offload/omniroute_offload/tests/test_gw_combo.py b/omniroute_offload/omniroute_offload/tests/test_gw_combo.py new file mode 100644 index 0000000..730f90e --- /dev/null +++ b/omniroute_offload/omniroute_offload/tests/test_gw_combo.py @@ -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 diff --git a/omniroute_offload/omniroute_offload/tests/test_gw_config.py b/omniroute_offload/omniroute_offload/tests/test_gw_config.py new file mode 100644 index 0000000..c0e4ee5 --- /dev/null +++ b/omniroute_offload/omniroute_offload/tests/test_gw_config.py @@ -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" diff --git a/omniroute_offload/omniroute_offload/tests/test_gw_docker.py b/omniroute_offload/omniroute_offload/tests/test_gw_docker.py new file mode 100644 index 0000000..193d6ae --- /dev/null +++ b/omniroute_offload/omniroute_offload/tests/test_gw_docker.py @@ -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) diff --git a/omniroute_offload/omniroute_offload/tests/test_gw_gateway.py b/omniroute_offload/omniroute_offload/tests/test_gw_gateway.py new file mode 100644 index 0000000..79dcecc --- /dev/null +++ b/omniroute_offload/omniroute_offload/tests/test_gw_gateway.py @@ -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 diff --git a/omniroute_offload/omniroute_offload/tests/test_gw_health.py b/omniroute_offload/omniroute_offload/tests/test_gw_health.py new file mode 100644 index 0000000..ab75f43 --- /dev/null +++ b/omniroute_offload/omniroute_offload/tests/test_gw_health.py @@ -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 == [] diff --git a/omniroute_offload/omniroute_offload/tests/test_gw_health_run.py b/omniroute_offload/omniroute_offload/tests/test_gw_health_run.py new file mode 100644 index 0000000..eb36bc0 --- /dev/null +++ b/omniroute_offload/omniroute_offload/tests/test_gw_health_run.py @@ -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] diff --git a/omniroute_offload/omniroute_offload/tests/test_gw_notify.py b/omniroute_offload/omniroute_offload/tests/test_gw_notify.py new file mode 100644 index 0000000..1360dfe --- /dev/null +++ b/omniroute_offload/omniroute_offload/tests/test_gw_notify.py @@ -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 diff --git a/omniroute_offload/omniroute_offload/tests/test_gw_update.py b/omniroute_offload/omniroute_offload/tests/test_gw_update.py new file mode 100644 index 0000000..b4edb54 --- /dev/null +++ b/omniroute_offload/omniroute_offload/tests/test_gw_update.py @@ -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")] diff --git a/omniroute_offload/pyproject.toml b/omniroute_offload/pyproject.toml index 76e21b6..b1cbe6b 100644 --- a/omniroute_offload/pyproject.toml +++ b/omniroute_offload/pyproject.toml @@ -13,6 +13,7 @@ dependencies = [] free-llm = "omniroute_offload._cli:main" free-llm-read-hook = "omniroute_offload._hook:main" free-llm-check-dir = "omniroute_offload._privacy:check_dir_main" +omniroute-gateway = "omniroute_offload.gateway:main" [tool.setuptools.packages.find] include = ["omniroute_offload*"] @@ -62,7 +63,7 @@ omit = ["*/__pycache__/*", "*/tests/*"] [tool.coverage.report] fail_under = 100 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] # The repo root on the path, so pylint resolves the package's own imports the diff --git a/omniroute_offload/systemd/omniroute-health.service b/omniroute_offload/systemd/omniroute-health.service new file mode 100644 index 0000000..55ba51f --- /dev/null +++ b/omniroute_offload/systemd/omniroute-health.service @@ -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 diff --git a/omniroute_offload/systemd/omniroute-health.timer b/omniroute_offload/systemd/omniroute-health.timer new file mode 100644 index 0000000..4425c07 --- /dev/null +++ b/omniroute_offload/systemd/omniroute-health.timer @@ -0,0 +1,9 @@ +[Unit] +Description=OmniRoute health check every 10 minutes + +[Timer] +OnBootSec=3min +OnUnitActiveSec=10min + +[Install] +WantedBy=timers.target diff --git a/omniroute_offload/systemd/omniroute-update.service b/omniroute_offload/systemd/omniroute-update.service new file mode 100644 index 0000000..f90a6cd --- /dev/null +++ b/omniroute_offload/systemd/omniroute-update.service @@ -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 diff --git a/omniroute_offload/systemd/omniroute-update.timer b/omniroute_offload/systemd/omniroute-update.timer new file mode 100644 index 0000000..7b45dca --- /dev/null +++ b/omniroute_offload/systemd/omniroute-update.timer @@ -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