Merge branch 'wartung/einstellungen-stand-altreste' into wartung/restarbeiten
# Conflicts: # deploy/deploy.sh
This commit is contained in:
Hitonabi
@@ -0,0 +1,197 @@
|
||||
"""Altreste neben dem Betrieb (24.09.2026): nur aus der festen Liste, nur was da ist, nur auf Knopfdruck — und die
|
||||
Löschfunktion nimmt eine Kennung, nie einen Pfad. Befehle (systemctl, git, sudo) laufen über eine Schicht, die hier
|
||||
ersetzt ist; kein Test fasst die Box an."""
|
||||
|
||||
import os
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
from routers import boxwart
|
||||
from services import aufraeumen
|
||||
from services.aufraeumen import Altrest
|
||||
|
||||
|
||||
class Befehle:
|
||||
"""Ersatz für aufraeumen._befehl: schreibt mit, und Units laufen nur, wenn der Test es sagt."""
|
||||
|
||||
def __init__(self):
|
||||
self.aufrufe: list[list[str]] = []
|
||||
self.laufend: set[str] = set()
|
||||
self.eingetragen: set[str] = set()
|
||||
self.sudo_ok = True
|
||||
|
||||
def __call__(self, args: list[str]) -> tuple[bool, str]:
|
||||
self.aufrufe.append(args)
|
||||
if args[:3] == ["systemctl", "--user", "show"]:
|
||||
unit = args[3]
|
||||
return True, (f"ActiveState={'active' if unit in self.laufend else 'inactive'}\n"
|
||||
f"UnitFileState={'enabled' if unit in self.eingetragen else 'disabled'}")
|
||||
if args[0] == "sudo":
|
||||
return self.sudo_ok, "" if self.sudo_ok else "sudo: a password is required"
|
||||
return True, ""
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def heim(tmp_path, monkeypatch):
|
||||
"""Ein Home-Ordner mit Resten, eine Liste darauf und daneben etwas, das bleiben muss."""
|
||||
home = tmp_path / "home"
|
||||
for pfad, groesse in (("zonos/modell.bin", 3000), ("zonos/sub/x.txt", 500), (".venv-a/lib/a.py", 1000),
|
||||
("skript.sh", 50), ("wt/datei", 10), ("bleibt/wichtig.txt", 5)):
|
||||
datei = home / pfad
|
||||
datei.parent.mkdir(parents=True, exist_ok=True)
|
||||
datei.write_bytes(b"x" * groesse)
|
||||
liste = (
|
||||
Altrest("zonos", "TTS-Test", ("~/zonos",), "Test."),
|
||||
Altrest("venvs", "Alte venvs", ("~/.venv-a", "~/.venv-gibt-es-nicht"), "Alt."),
|
||||
Altrest("skript", "Altes Skript", ("~/skript.sh",), "Tot."),
|
||||
Altrest("fehlt", "Längst weg", ("~/gibt-es-nicht",), "Weg."),
|
||||
Altrest("wt", "Arbeitskopie", ("~/wt",), "Git.", art="worktree"),
|
||||
Altrest("mem", "mem0", ("~/.venv-a",), "Sidecar.", sperre=("mem0-service.service",)),
|
||||
)
|
||||
befehle = Befehle()
|
||||
monkeypatch.setenv("MC_ALTRESTE_HOME", str(home))
|
||||
monkeypatch.setattr(aufraeumen, "ALTRESTE", liste)
|
||||
monkeypatch.setattr(aufraeumen, "AUF_DER_BOX", True)
|
||||
monkeypatch.setattr(aufraeumen, "_befehl", befehle)
|
||||
monkeypatch.setitem(aufraeumen._altreste_cache, "liste", None)
|
||||
return home, befehle
|
||||
|
||||
|
||||
def test_liste_zeigt_nur_was_da_ist(heim):
|
||||
liste = aufraeumen.altreste(frisch=True)
|
||||
assert [a["id"] for a in liste] == ["zonos", "venvs", "mem", "skript", "wt"] # größte zuerst
|
||||
zonos = liste[0]
|
||||
assert zonos["groesse_bytes"] == 3500 and zonos["groesse"] == "4 KB" and zonos["pfade"] == ["~/zonos"]
|
||||
venvs = next(a for a in liste if a["id"] == "venvs")
|
||||
assert venvs["pfade"] == ["~/.venv-a"] # nur, was da ist
|
||||
assert next(a for a in liste if a["id"] == "skript")["groesse"] == "unter 1 KB"
|
||||
|
||||
|
||||
def test_was_ein_dienst_braucht_wird_nicht_angeboten(heim):
|
||||
_, befehle = heim
|
||||
befehle.laufend.add("mem0-service.service")
|
||||
assert "mem" not in {a["id"] for a in aufraeumen.altreste(frisch=True)}
|
||||
ergebnis = aufraeumen.altrest_loeschen("mem")
|
||||
assert not ergebnis["ok"] and "läuft gerade" in ergebnis["detail"]
|
||||
befehle.laufend.clear()
|
||||
befehle.eingetragen.add("mem0-service.service")
|
||||
assert "Autostart" in aufraeumen.altrest_loeschen("mem")["detail"]
|
||||
assert (heim[0] / ".venv-a").exists()
|
||||
|
||||
|
||||
def test_loeschen_nur_ueber_die_kennung(heim):
|
||||
home, _ = heim
|
||||
for kennung in ("../../etc", "~/bleibt", str(home / "bleibt"), ""):
|
||||
ergebnis = aufraeumen.altrest_loeschen(kennung)
|
||||
assert not ergebnis["ok"] and "kennt die Liste nicht" in ergebnis["detail"]
|
||||
ergebnis = aufraeumen.altrest_loeschen("zonos")
|
||||
assert ergebnis == {"ok": True, "text": "TTS-Test gelöscht, 4 KB frei."}
|
||||
assert not (home / "zonos").exists() and (home / "bleibt" / "wichtig.txt").exists()
|
||||
assert "zonos" not in {a["id"] for a in aufraeumen.altreste()} # die Liste ist danach frisch
|
||||
assert "schon weg" in aufraeumen.altrest_loeschen("zonos")["detail"]
|
||||
|
||||
|
||||
def test_nacharbeit_je_art(heim):
|
||||
home, befehle = heim
|
||||
assert aufraeumen.altrest_loeschen("wt")["ok"] and not (home / "wt").exists()
|
||||
assert ["git", "-C", str(aufraeumen.REPO), "worktree", "prune"] in befehle.aufrufe
|
||||
assert aufraeumen.altrest_loeschen("skript")["ok"]
|
||||
assert ["systemctl", "--user", "daemon-reload"] not in befehle.aufrufe # nur bei Units
|
||||
|
||||
|
||||
def test_units_werden_neu_eingelesen(heim, monkeypatch):
|
||||
home, befehle = heim
|
||||
unit = home / ".config" / "systemd" / "user" / "alt.timer"
|
||||
unit.parent.mkdir(parents=True)
|
||||
unit.write_text("[Timer]\n", encoding="utf-8")
|
||||
monkeypatch.setattr(aufraeumen, "ALTRESTE", (Altrest("units", "Alt-Units", ("~/.config/systemd/user/alt.timer",),
|
||||
"Aus.", art="unit", sperre=("alt.timer",)),))
|
||||
befehle.eingetragen.add("alt.timer")
|
||||
assert not aufraeumen.altrest_loeschen("units")["ok"] and unit.exists()
|
||||
befehle.eingetragen.clear()
|
||||
assert aufraeumen.altrest_loeschen("units")["ok"] and not unit.exists()
|
||||
assert befehle.aufrufe[-1] == ["systemctl", "--user", "daemon-reload"]
|
||||
|
||||
|
||||
def test_verknuepfung_nimmt_nie_ihr_ziel_mit(heim, monkeypatch):
|
||||
home, _ = heim
|
||||
try:
|
||||
os.symlink(home / "bleibt", home / "link", target_is_directory=True)
|
||||
except (OSError, NotImplementedError):
|
||||
pytest.skip("Verknüpfungen gehen hier nicht (Windows ohne Rechte)")
|
||||
monkeypatch.setattr(aufraeumen, "ALTRESTE", (Altrest("link", "Verknüpfung", ("~/link",), "Test."),))
|
||||
assert aufraeumen.altrest_loeschen("link")["ok"]
|
||||
assert not (home / "link").is_symlink() and (home / "bleibt" / "wichtig.txt").exists()
|
||||
|
||||
|
||||
def test_root_ordner_mit_sudo_nur_wo_erlaubt(heim, monkeypatch):
|
||||
home, befehle = heim
|
||||
|
||||
def verweigert(pfad, *args, **kwargs):
|
||||
raise PermissionError(13, "Keine Berechtigung", str(pfad))
|
||||
|
||||
monkeypatch.setattr(aufraeumen.shutil, "rmtree", verweigert)
|
||||
monkeypatch.setattr(aufraeumen, "ALTRESTE", (Altrest("rocm", "ROCm", ("~/zonos",), "Alt.", sudo=True),
|
||||
Altrest("ohne", "Ohne sudo", ("~/.venv-a",), "Alt.")))
|
||||
assert aufraeumen.altrest_loeschen("rocm")["ok"]
|
||||
assert ["sudo", "-n", "rm", "-rf", "--", str(home / "zonos")] in befehle.aufrufe
|
||||
ergebnis = aufraeumen.altrest_loeschen("ohne")
|
||||
assert not ergebnis["ok"] and "Keine Berechtigung" in ergebnis["detail"]
|
||||
assert sum(1 for a in befehle.aufrufe if a[0] == "sudo") == 1
|
||||
befehle.sudo_ok = False
|
||||
assert "sudo -n rm ging nicht" in aufraeumen.altrest_loeschen("rocm")["detail"]
|
||||
|
||||
|
||||
def test_am_pc_wird_nichts_geloescht(heim, monkeypatch):
|
||||
monkeypatch.setattr(aufraeumen, "AUF_DER_BOX", False)
|
||||
assert not aufraeumen.altrest_loeschen("zonos")["ok"] and (heim[0] / "zonos").exists()
|
||||
|
||||
|
||||
def test_schnittstelle_nimmt_nur_eine_kennung(heim):
|
||||
app = FastAPI()
|
||||
app.include_router(boxwart.router)
|
||||
c = TestClient(app)
|
||||
assert c.get("/api/aufraeumen/altreste").json()["altreste"][0]["id"] == "zonos"
|
||||
assert c.post("/api/aufraeumen/altreste/loeschen", json={"pfad": "/"}).status_code == 422
|
||||
r = c.post("/api/aufraeumen/altreste/loeschen", json={"id": "../bleibt"})
|
||||
assert r.status_code == 409 and "kennt die Liste nicht" in r.json()["detail"]
|
||||
assert c.post("/api/aufraeumen/altreste/loeschen", json={"id": "skript"}).json()["ok"] is True
|
||||
|
||||
|
||||
# --- Die echte Liste ----------------------------------------------------------------------------------
|
||||
|
||||
GESCHUETZT = ("~", "~/.hermes", "~/.hermes/hermes-agent", "~/.hermes/hermes-agent/hermes_cli/web_dist",
|
||||
"~/.hermes/config.yaml", "~/.hermes/.env", "~/.hermes/memories", "~/.hermes/skills",
|
||||
"~/.hermes/kanban.db", "~/.hermes/scripts/news-melden.sh", "~/.hermes/scripts/sonntags-update.sh",
|
||||
"~/.hermes/scripts/stack-radar.sh", "~/.hermes/plugins/mc2-web-lesen", "~/mission-control-v2",
|
||||
"~/mission-control-v2/frontend/dist/index.html", "~/.config/systemd/user",
|
||||
"~/.config/systemd/user/mc2-radar.timer", "~/.config/systemd/user/lucy-stimme.service",
|
||||
"~/.lucy-stimme", "~/.voice", "~/.cache", "~/.opencode", "~/projekte", "~/.local/bin",
|
||||
"/srv/models", "/opt", "/opt/llamacpp-vulkan", "/opt/llamacpp-vulkan.bak")
|
||||
|
||||
|
||||
def test_feste_liste_ist_eng():
|
||||
kennungen = [a.id for a in aufraeumen.ALTRESTE]
|
||||
assert len(kennungen) == len(set(kennungen))
|
||||
for a in aufraeumen.ALTRESTE:
|
||||
assert a.art in ("ordner", "worktree", "unit") and a.hinweis and a.name
|
||||
for pfad in a.pfade:
|
||||
assert pfad.startswith("~/") or pfad == "/opt/llamacpp", pfad
|
||||
assert not any(z in pfad for z in ("*", "?", "..", "//")) and not pfad.endswith("/"), pfad
|
||||
for heilig in GESCHUETZT:
|
||||
assert pfad != heilig and not heilig.startswith(pfad + "/"), (pfad, heilig)
|
||||
assert a.sudo == (a.pfade == ("/opt/llamacpp",)) # sudo nur für den ROCm-Ordner unter /opt
|
||||
|
||||
|
||||
def test_feste_liste_enthaelt_die_bekannten_reste():
|
||||
alle = {p for a in aufraeumen.ALTRESTE for p in a.pfade}
|
||||
for pfad in ("~/zonos2-test", "/opt/llamacpp", "~/.mem0", "~/.hermes/kanban", "~/.venv-audit",
|
||||
"~/.hermes/worktrees/orch-graph-verknuepfer", "~/projekte/mc2-referenz",
|
||||
"~/.hermes/plugins/mc2-memory", "~/.hermes/PINNED_VERSION",
|
||||
"~/mission-control-v2/frontend/dist/avatar.vrm"):
|
||||
assert pfad in alle, pfad
|
||||
# Was noch jemand nutzt, steht nicht darin (Prüfung vom 24.09.2026).
|
||||
assert not alle & {"~/.hermes/kanban.db", "~/.voice", "~/.cache", "~/.opencode",
|
||||
"~/.hermes/scripts/ampel-waechter.sh", "~/.hermes/scripts/pc_path_lookup.sh"}
|
||||
assert "~/.hermes/scripts/curator-idle-watchdog.sh" not in alle # steht in der crontab
|
||||
@@ -0,0 +1,494 @@
|
||||
"""Einstellungen der Oberfläche (24.09.2026): Update-Zeitfenster, Radar-Schalter, Telegram-Test.
|
||||
|
||||
Kein Test setzt echte systemctl-Befehle ab: services/einstellungen spricht systemd nur über _systemctl, und das
|
||||
ersetzt hier ein kleines Abbild von systemd — samt der Nachhol-Regel von Persistent=true, die die Box am 24.09.
|
||||
mitten am Tag neu startete. Der Telegram-Test fährt das echte notify.sh, aber mit einem Ersatz für `hermes send`,
|
||||
einer Tmp-Datei als Telegram-Zugang und einer lokalen Bot-API: Es geht nie eine echte Nachricht raus."""
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
from datetime import datetime, timedelta
|
||||
from datetime import time as uhr
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from pathlib import Path
|
||||
from urllib.parse import parse_qs
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
from kern.zeit import LOCAL_TZ
|
||||
from routers import einstellungen as einstellungen_router
|
||||
from services import einstellungen, waechter
|
||||
|
||||
UPDATE, RADAR = "mc2-autoupdate.timer", "mc2-radar.timer"
|
||||
|
||||
|
||||
class FakeSystemd:
|
||||
"""Genug systemd für die Einstellungen: Zustände, Aufrufe — und die Nachhol-Regel eines Timers mit
|
||||
Persistent=true. Beim Start gilt der Stempel als letzter Lauf; liegt ein Termin des geladenen Plans zwischen ihm
|
||||
und jetzt, läuft der Dienst sofort. Beim daemon-reload rechnet ein laufender Timer mit seinem letzten Lauf im
|
||||
Speicher und dem NEUEN Plan (so verhält sich systemd beim Deserialisieren) — hat sich sein Plan nicht geändert,
|
||||
ist nichts nachzuholen, sonst wäre er schon gelaufen."""
|
||||
|
||||
def __init__(self, stempel: Path, dropin: Path, letzter_lauf: datetime):
|
||||
self.stempel, self.dropin = stempel, dropin
|
||||
self.aufrufe: list[tuple[str, ...]] = []
|
||||
self.nachgeholt: list[str] = []
|
||||
self.scheitert: set[str] = set() # Befehle, die beim nächsten Aufruf einmal scheitern
|
||||
self.aktiv = {UPDATE: True, RADAR: True, "mc2-autoupdate.service": False}
|
||||
self.eingetragen = {UPDATE: True, RADAR: True}
|
||||
self.im_speicher = {UPDATE: letzter_lauf, RADAR: letzter_lauf}
|
||||
self.geladen = {UPDATE: self._plan_aus_datei(UPDATE), RADAR: self._plan_aus_datei(RADAR)}
|
||||
stempel.mkdir(parents=True, exist_ok=True)
|
||||
for timer in (UPDATE, RADAR):
|
||||
datei = stempel / f"stamp-{timer}"
|
||||
datei.touch()
|
||||
os.utime(datei, (letzter_lauf.timestamp(), letzter_lauf.timestamp()))
|
||||
|
||||
def _plan_aus_datei(self, timer: str) -> tuple[int | None, int, int]:
|
||||
if timer == RADAR:
|
||||
return None, 0, 30 # täglich 00:30
|
||||
text = self.dropin.read_text(encoding="utf-8") if self.dropin.exists() else ""
|
||||
m = re.search(r"^OnCalendar=(\w{3}) \*-\*-\* (\d\d):(\d\d):00$", text, re.MULTILINE)
|
||||
if not m:
|
||||
return 7, 4, 30 # Plan der Repo-Unit: So 04:30
|
||||
return einstellungen.SYSTEMD_TAGE.index(m.group(1)) + 1, int(m.group(2)), int(m.group(3))
|
||||
|
||||
@staticmethod
|
||||
def termin_dazwischen(plan: tuple[int | None, int, int], von: datetime, bis: datetime) -> bool:
|
||||
wochentag, stunde, minute = plan
|
||||
tag = von.date()
|
||||
while tag <= bis.date():
|
||||
termin = datetime.combine(tag, uhr(stunde, minute), LOCAL_TZ)
|
||||
if (wochentag is None or termin.isoweekday() == wochentag) and von < termin <= bis:
|
||||
return True
|
||||
tag += timedelta(days=1)
|
||||
return False
|
||||
|
||||
def __call__(self, *args: str) -> tuple[bool, str]:
|
||||
self.aufrufe.append(args)
|
||||
befehl, jetzt = args[0], datetime.now(LOCAL_TZ)
|
||||
if befehl in self.scheitert:
|
||||
self.scheitert.discard(befehl)
|
||||
return False, f"{befehl} ging schief"
|
||||
if befehl == "show":
|
||||
unit = args[1]
|
||||
eingetragen = self.eingetragen.get(unit)
|
||||
datei = "static" if eingetragen is None else ("enabled" if eingetragen else "disabled")
|
||||
return True, (f"LoadState=loaded\nActiveState={'active' if self.aktiv.get(unit) else 'inactive'}\n"
|
||||
f"UnitFileState={datei}")
|
||||
if befehl == "list-timers":
|
||||
unit = args[-1]
|
||||
naechster = int((jetzt + timedelta(days=1)).timestamp() * 1_000_000) if self.aktiv.get(unit) else None
|
||||
return True, json.dumps([{"unit": unit, "next": naechster, "last": None}])
|
||||
if befehl == "stop":
|
||||
self.aktiv[args[1]] = False
|
||||
elif befehl == "daemon-reload":
|
||||
for timer in (UPDATE, RADAR):
|
||||
neu = self._plan_aus_datei(timer)
|
||||
if self.aktiv[timer] and neu != self.geladen[timer] \
|
||||
and self.termin_dazwischen(neu, self.im_speicher[timer], jetzt):
|
||||
self.nachgeholt.append(timer)
|
||||
self.geladen[timer] = neu
|
||||
elif befehl == "start":
|
||||
timer = args[1]
|
||||
if not self.aktiv[timer]:
|
||||
datei = self.stempel / f"stamp-{timer}"
|
||||
letzter = datetime.fromtimestamp(datei.stat().st_mtime, LOCAL_TZ) if datei.exists() else None
|
||||
if letzter is not None and letzter < jetzt:
|
||||
self.im_speicher[timer] = letzter
|
||||
if self.termin_dazwischen(self.geladen[timer], letzter, jetzt):
|
||||
self.nachgeholt.append(timer)
|
||||
self.aktiv[timer] = True
|
||||
elif befehl == "enable":
|
||||
self.eingetragen[args[1]] = True
|
||||
elif befehl == "disable":
|
||||
self.eingetragen[args[-1]] = False
|
||||
if "--now" in args:
|
||||
self.aktiv[args[-1]] = False
|
||||
return True, ""
|
||||
|
||||
def befehle(self, *arten: str) -> list[str]:
|
||||
return [a[0] for a in self.aufrufe if a[0] in arten]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def box(tmp_path, monkeypatch):
|
||||
"""Einstellungsdatei, Units und Stempel in tmp; systemd ist das Abbild oben (letzter Lauf vor vier Tagen)."""
|
||||
monkeypatch.setenv("MC_EINSTELLUNGEN", str(tmp_path / "mc2-einstellungen.json"))
|
||||
monkeypatch.setenv("MC_SYSTEMD_USER_DIR", str(tmp_path / "units"))
|
||||
monkeypatch.setenv("MC_TIMER_STEMPEL_DIR", str(tmp_path / "timers"))
|
||||
monkeypatch.setattr(einstellungen, "AUF_DER_BOX", True)
|
||||
fake = FakeSystemd(tmp_path / "timers", einstellungen.dropin(), datetime.now(LOCAL_TZ) - timedelta(days=4))
|
||||
monkeypatch.setattr(einstellungen, "_systemctl", fake)
|
||||
return fake
|
||||
|
||||
|
||||
def _vorgestern() -> tuple[int, str]:
|
||||
"""Ein Plan, dessen Termin vorgestern war — zwischen dem letzten Lauf (vor vier Tagen) und jetzt."""
|
||||
d = datetime.now(LOCAL_TZ) - timedelta(days=2)
|
||||
if (d.isoweekday(), d.strftime("%H:%M")) == (7, "04:30"): # wäre der alte Plan — dann nichts „Neues“
|
||||
d -= timedelta(minutes=1)
|
||||
return d.isoweekday(), d.strftime("%H:%M")
|
||||
|
||||
|
||||
# --- Datei und Regeln ------------------------------------------------------------------------------
|
||||
|
||||
def test_ohne_datei_gilt_der_standard(box):
|
||||
assert einstellungen.lesen() == {"update_fenster": {"tag": 7, "uhrzeit": "04:30"}, "radar": {"an": True}}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("inhalt", [
|
||||
"kein json", "[]", '{"update_fenster": {"tag": 8, "uhrzeit": "04:30"}, "radar": {"an": "nein"}}',
|
||||
'{"update_fenster": {"tag": true, "uhrzeit": "4:30"}}', '{"update_fenster": {"tag": 3, "uhrzeit": "24:00"}}',
|
||||
])
|
||||
def test_unsinn_in_der_datei_faellt_auf_den_standard(box, inhalt):
|
||||
einstellungen.datei().write_text(inhalt, encoding="utf-8")
|
||||
assert einstellungen.lesen() == {"update_fenster": {"tag": 7, "uhrzeit": "04:30"}, "radar": {"an": True}}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("tag, uhrzeit, text", [
|
||||
(7, "04:30", "Sonntag 04:00–06:59"),
|
||||
(6, "05:00", "Samstag 05:00–07:59"),
|
||||
(6, "23:30", "Samstag 23:00 – Sonntag 01:59"),
|
||||
(7, "22:00", "Sonntag 22:00 – Montag 00:59"),
|
||||
(1, "00:00", "Montag 00:00–02:59"),
|
||||
])
|
||||
def test_neustart_fenster(tag, uhrzeit, text):
|
||||
assert einstellungen.neustart_fenster(tag, uhrzeit) == text
|
||||
|
||||
|
||||
def test_dropin_ersetzt_den_plan_statt_einen_zweiten_daneben():
|
||||
text = einstellungen.dropin_text(6, "05:00")
|
||||
assert "[Timer]\nOnCalendar=\nOnCalendar=Sat *-*-* 05:00:00\n" in text
|
||||
|
||||
|
||||
@pytest.mark.parametrize("zustand", [
|
||||
{"ActiveState": "inactive", "UnitFileState": "disabled"},
|
||||
{"ActiveState": "inactive", "UnitFileState": "enabled"},
|
||||
{"ActiveState": "active", "UnitFileState": "disabled"},
|
||||
{"ActiveState": "failed", "UnitFileState": "disabled"},
|
||||
{},
|
||||
])
|
||||
def test_schlaeft_wie_der_waechter(zustand):
|
||||
assert einstellungen.schlaeft(zustand) == waechter.schlaeft(zustand)
|
||||
|
||||
|
||||
# --- Update-Zeitfenster ------------------------------------------------------------------------------
|
||||
|
||||
def test_zeitfenster_holt_nicht_nach(box):
|
||||
"""Die Falle vom 24.09.: Ein Plan, dessen Termin seit dem letzten Lauf schon vorbei ist, darf nicht sofort laufen."""
|
||||
tag, uhrzeit = _vorgestern()
|
||||
ergebnis = einstellungen.update_fenster_setzen(tag, uhrzeit)
|
||||
assert ergebnis["ok"], ergebnis
|
||||
assert box.nachgeholt == []
|
||||
# Erst anhalten, dann neu einlesen, dann (mit Stempel) starten — sonst rechnet systemd mit dem alten Lauf.
|
||||
assert box.befehle("stop", "daemon-reload", "start") == ["stop", "daemon-reload", "start"]
|
||||
assert einstellungen.dropin().read_text(encoding="utf-8") == einstellungen.dropin_text(tag, uhrzeit)
|
||||
assert einstellungen.lesen()["update_fenster"] == {"tag": tag, "uhrzeit": uhrzeit}
|
||||
assert einstellungen.neustart_fenster(tag, uhrzeit) in ergebnis["text"]
|
||||
|
||||
|
||||
def test_gegenprobe_das_abbild_erkennt_die_falle(box):
|
||||
"""Ohne die Vorsicht holt das Abbild nach — sonst bewiese der Test oben nichts."""
|
||||
tag, uhrzeit = _vorgestern()
|
||||
einstellungen._dropin_schreiben(einstellungen.dropin_text(tag, uhrzeit))
|
||||
box("daemon-reload") # Reload bei laufendem Timer
|
||||
assert box.nachgeholt == [UPDATE]
|
||||
|
||||
|
||||
def test_ohne_stempel_wuerde_der_timer_nachholen(box, monkeypatch):
|
||||
monkeypatch.setattr(einstellungen, "_stempel_setzen", lambda timer: None)
|
||||
tag, uhrzeit = _vorgestern()
|
||||
assert einstellungen.update_fenster_setzen(tag, uhrzeit)["ok"]
|
||||
assert box.nachgeholt == [UPDATE]
|
||||
|
||||
|
||||
def test_zeitfenster_nicht_waehrend_eines_updates(box):
|
||||
box.aktiv["mc2-autoupdate.service"] = True
|
||||
ergebnis = einstellungen.update_fenster_setzen(6, "05:00")
|
||||
assert not ergebnis["ok"] and "läuft das Update" in ergebnis["detail"]
|
||||
assert box.befehle("stop", "daemon-reload", "start") == []
|
||||
assert not einstellungen.dropin().exists()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("tag, uhrzeit", [(0, "04:30"), (8, "04:30"), (7, "4:30"), (7, "24:00"), (7, "04:60")])
|
||||
def test_zeitfenster_prueft_die_eingabe(box, tag, uhrzeit):
|
||||
ergebnis = einstellungen.update_fenster_setzen(tag, uhrzeit)
|
||||
assert not ergebnis["ok"] and ergebnis["detail"]
|
||||
assert box.aufrufe == []
|
||||
|
||||
|
||||
def test_zeitfenster_rollt_bei_fehler_zurueck(box):
|
||||
einstellungen._dropin_schreiben(einstellungen.dropin_text(6, "05:00"))
|
||||
vorher = einstellungen.dropin().read_text(encoding="utf-8")
|
||||
box.scheitert.add("daemon-reload")
|
||||
ergebnis = einstellungen.update_fenster_setzen(3, "04:00")
|
||||
assert not ergebnis["ok"] and "nicht übernommen" in ergebnis["detail"]
|
||||
assert einstellungen.dropin().read_text(encoding="utf-8") == vorher
|
||||
assert box.aktiv[UPDATE] is True # ohne Timer gäbe es keine Updates mehr
|
||||
assert not einstellungen.datei().exists()
|
||||
assert box.nachgeholt == []
|
||||
|
||||
|
||||
def test_gestoppter_timer_bleibt_aus(box):
|
||||
box.aktiv[UPDATE] = False
|
||||
assert einstellungen.update_fenster_setzen(6, "05:00")["ok"]
|
||||
assert box.befehle("stop", "start") == [] and box.aktiv[UPDATE] is False
|
||||
assert einstellungen.stand()["update_fenster"]["timer_aktiv"] is False
|
||||
|
||||
|
||||
def test_stand_zeigt_ob_der_timer_zum_wunsch_passt(box):
|
||||
stand = einstellungen.stand()
|
||||
assert stand["schaltbar"] is True
|
||||
assert stand["update_fenster"]["wirksam"] is True and stand["update_fenster"]["tag_name"] == "Sonntag"
|
||||
assert stand["update_fenster"]["naechster_lauf"]
|
||||
# Wunsch in der Datei, aber kein Drop-in (etwa nach einem Neuaufbau): nicht wirksam.
|
||||
einstellungen.datei().write_text('{"update_fenster": {"tag": 6, "uhrzeit": "05:00"}}', encoding="utf-8")
|
||||
assert einstellungen.stand()["update_fenster"]["wirksam"] is False
|
||||
assert einstellungen.update_fenster_setzen(6, "05:00")["ok"]
|
||||
assert einstellungen.stand()["update_fenster"]["wirksam"] is True
|
||||
|
||||
|
||||
def test_am_pc_wird_nichts_geschaltet(box, monkeypatch):
|
||||
monkeypatch.setattr(einstellungen, "AUF_DER_BOX", False)
|
||||
assert not einstellungen.update_fenster_setzen(6, "05:00")["ok"]
|
||||
assert not einstellungen.radar_schalten(False)["ok"]
|
||||
assert box.befehle("stop", "start", "enable", "disable", "daemon-reload") == []
|
||||
|
||||
|
||||
# --- Radar-Schalter -----------------------------------------------------------------------------------
|
||||
|
||||
def test_radar_aus_und_wieder_an_ohne_nachholen(box):
|
||||
ergebnis = einstellungen.radar_schalten(False)
|
||||
assert ergebnis["ok"] and "schläft" in ergebnis["text"]
|
||||
assert ("disable", "--now", RADAR) in box.aufrufe
|
||||
assert einstellungen.lesen()["radar"] == {"an": False}
|
||||
radar = einstellungen.stand()["radar"]
|
||||
assert radar["zustand"] == "schlaeft" and radar["wirksam"] is True and radar["naechster_lauf"] is None
|
||||
|
||||
ergebnis = einstellungen.radar_schalten(True)
|
||||
assert ergebnis["ok"]
|
||||
assert box.befehle("enable", "start") == ["enable", "start"]
|
||||
assert box.nachgeholt == [] # der letzte Lauf war vor vier Tagen, 00:30 lag dazwischen
|
||||
assert einstellungen.lesen()["radar"] == {"an": True}
|
||||
assert einstellungen.stand()["radar"]["zustand"] == "an"
|
||||
|
||||
|
||||
def test_radar_nur_gestoppt_ist_nicht_aus(box):
|
||||
"""Gestoppt, aber noch im Autostart: nach einem Neustart der Box liefe das Radar wieder."""
|
||||
einstellungen.datei().write_text('{"radar": {"an": false}}', encoding="utf-8")
|
||||
box.aktiv[RADAR] = False
|
||||
radar = einstellungen.stand()["radar"]
|
||||
assert radar["zustand"] == "aus" and radar["wirksam"] is False
|
||||
box.eingetragen[RADAR] = False
|
||||
assert einstellungen.stand()["radar"]["wirksam"] is True
|
||||
|
||||
|
||||
def test_radar_ohne_stempel_wuerde_nachholen(box, monkeypatch):
|
||||
monkeypatch.setattr(einstellungen, "_stempel_setzen", lambda timer: None)
|
||||
einstellungen.radar_schalten(False)
|
||||
einstellungen.radar_schalten(True)
|
||||
assert box.nachgeholt == [RADAR]
|
||||
|
||||
|
||||
def test_radar_fehler_wird_nicht_gespeichert(box):
|
||||
box.scheitert.add("disable")
|
||||
ergebnis = einstellungen.radar_schalten(False)
|
||||
assert not ergebnis["ok"] and "nicht ausgeschaltet" in ergebnis["detail"]
|
||||
assert not einstellungen.datei().exists()
|
||||
|
||||
|
||||
def test_waechter_laesst_den_schlafenden_radar_in_ruhe(monkeypatch):
|
||||
"""Ein alter Fehlschlag des Radars ist kein Hinweis mehr, sobald sein Timer schläft (Radar aus)."""
|
||||
zustaende = {
|
||||
"mc2-radar": {"LoadState": "loaded", "ActiveState": "failed", "Result": "exit-code"},
|
||||
"mc2-radar.timer": {"LoadState": "loaded", "ActiveState": "inactive", "UnitFileState": "disabled"},
|
||||
}
|
||||
monkeypatch.setattr(waechter, "_systemctl_show", lambda name, system: zustaende.get(
|
||||
name, {"LoadState": "loaded", "ActiveState": "inactive", "Result": "success"}))
|
||||
monkeypatch.setattr(waechter, "_journal_fehlerzeile", lambda name, system=False: "")
|
||||
assert "timer:mc2-radar" not in {b.id for b in waechter.pruefe_timer_dienste()}
|
||||
zustaende["mc2-radar.timer"] = {"LoadState": "loaded", "ActiveState": "active", "UnitFileState": "enabled"}
|
||||
assert "timer:mc2-radar" in {b.id for b in waechter.pruefe_timer_dienste()}
|
||||
|
||||
|
||||
# --- Schnittstellen -----------------------------------------------------------------------------------
|
||||
|
||||
def _client() -> TestClient:
|
||||
app = FastAPI()
|
||||
app.include_router(einstellungen_router.router)
|
||||
app.include_router(einstellungen_router.homelab_router)
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
def test_schnittstellen(box):
|
||||
c = _client()
|
||||
assert c.get("/api/einstellungen").json()["radar"]["an"] is True
|
||||
r = c.post("/api/einstellungen/update-fenster", json={"tag": 9, "uhrzeit": "04:30"})
|
||||
assert r.status_code == 409 and "Wochentag" in r.json()["detail"]
|
||||
r = c.post("/api/einstellungen/update-fenster", json={"tag": 6, "uhrzeit": "05:00"})
|
||||
assert r.status_code == 200 and r.json()["einstellungen"]["update_fenster"]["tag"] == 6
|
||||
r = c.post("/api/einstellungen/radar", json={"an": False})
|
||||
assert r.status_code == 200 and r.json()["einstellungen"]["radar"]["zustand"] == "schlaeft"
|
||||
|
||||
|
||||
BACKEND = Path(__file__).resolve().parent.parent
|
||||
|
||||
|
||||
@pytest.mark.parametrize("rolle, da, nicht_da", [
|
||||
("box", "/api/einstellungen/telegram-test", "/api/homelab/einstellungen/telegram-test"),
|
||||
("homelab", "/api/homelab/einstellungen/telegram-test", "/api/einstellungen"),
|
||||
])
|
||||
def test_jede_rolle_hat_ihren_test_knopf(tmp_path, rolle, da, nicht_da):
|
||||
"""Der Homelab-Teil bekommt nur den Telegram-Test (und lädt dafür nichts von der KI-Box, siehe test_partner)."""
|
||||
env = {**os.environ, "MC_ROLLE": rolle, "MC_MODELS_DIR": str(tmp_path), "MC_DATEN_DIR": str(tmp_path),
|
||||
"MC_PARTNER_URL": ""}
|
||||
lauf = subprocess.run([sys.executable, "-c", "import json, app; print(json.dumps(sorted(app.app.openapi()['paths'])))"],
|
||||
cwd=BACKEND, env=env, capture_output=True, text=True, timeout=120)
|
||||
assert lauf.returncode == 0, lauf.stderr[-2000:]
|
||||
pfade = json.loads(lauf.stdout.strip().splitlines()[-1])
|
||||
assert da in pfade and nicht_da not in pfade
|
||||
|
||||
|
||||
# --- Telegram-Test ------------------------------------------------------------------------------------
|
||||
|
||||
BASH = shutil.which("bash")
|
||||
braucht_bash = pytest.mark.skipif(
|
||||
BASH is None or (sys.platform == "win32" and "system32" in BASH.lower()),
|
||||
reason="braucht eine bash (Git-Bash oder Linux)",
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def melden(tmp_path, monkeypatch):
|
||||
"""notify.sh mit einem Ersatz für `hermes send`, der nur mitschreibt, und ohne echten Telegram-Zugang."""
|
||||
protokoll = tmp_path / "gesendet.txt"
|
||||
stub = tmp_path / "hermes-stub"
|
||||
stub.write_text(
|
||||
"#!/usr/bin/env bash\n"
|
||||
'printf "%s\\n" "$*" >> "$STUB_PROTOKOLL"\n'
|
||||
'[ -n "${STUB_AUSGABE:-}" ] && echo "$STUB_AUSGABE"\n'
|
||||
'exit "${STUB_EXIT:-0}"\n',
|
||||
encoding="utf-8", newline="\n")
|
||||
stub.chmod(0o755)
|
||||
for name, wert in {"MC_NOTIFY_LOG": tmp_path / "notify.log", "MC_NIGHT_QUEUE": tmp_path / "night-queue.txt",
|
||||
"MC_NOTIFY_HERMES": stub, "MC_TELEGRAM_ENV": tmp_path / "keine.env",
|
||||
"STUB_PROTOKOLL": protokoll}.items():
|
||||
monkeypatch.setenv(name, Path(wert).as_posix())
|
||||
monkeypatch.setenv("MC_TELEGRAM_API", "http://127.0.0.1:9")
|
||||
monkeypatch.setenv("MC_NOTIFY_STUNDE", "03") # mitten in der Nacht: -d muss trotzdem sofort senden
|
||||
monkeypatch.delenv("STUB_EXIT", raising=False)
|
||||
monkeypatch.delenv("STUB_AUSGABE", raising=False)
|
||||
monkeypatch.setattr(einstellungen, "AUF_DER_BOX", True)
|
||||
monkeypatch.setitem(einstellungen._letzter_test, "ts", 0.0)
|
||||
return tmp_path, protokoll
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def bot_api():
|
||||
"""Ein Ersatz für api.telegram.org, der mitschreibt, was ankommt."""
|
||||
empfangen: list[dict] = []
|
||||
antwort = {"code": 200}
|
||||
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
def do_POST(self):
|
||||
laenge = int(self.headers.get("Content-Length", 0))
|
||||
felder = parse_qs(self.rfile.read(laenge).decode("utf-8"))
|
||||
empfangen.append({"pfad": self.path, **{k: v[0] for k, v in felder.items()}})
|
||||
self.send_response(antwort["code"])
|
||||
self.end_headers()
|
||||
self.wfile.write(b'{"ok": true}')
|
||||
|
||||
def log_message(self, *args):
|
||||
pass
|
||||
|
||||
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
||||
threading.Thread(target=server.serve_forever, daemon=True).start()
|
||||
yield f"http://127.0.0.1:{server.server_port}", empfangen, antwort
|
||||
server.shutdown()
|
||||
|
||||
|
||||
@braucht_bash
|
||||
def test_telegram_test_ueber_hermes_auch_nachts(melden):
|
||||
tmp, protokoll = melden
|
||||
ergebnis = einstellungen.telegram_test()
|
||||
assert ergebnis["ok"] and ergebnis["gesendet"] and ergebnis["weg"] == "hermes", ergebnis
|
||||
gesendet = protokoll.read_text(encoding="utf-8")
|
||||
assert "send --to telegram --subject [Test]" in gesendet and f"Kennung {ergebnis['kennung']}" in gesendet
|
||||
assert not (tmp / "night-queue.txt").exists()
|
||||
assert "OK telegram: Testmeldung aus den Einstellungen" in (tmp / "notify.log").read_text(encoding="utf-8")
|
||||
|
||||
|
||||
@braucht_bash
|
||||
def test_telegram_test_ueber_den_zweitweg(melden, bot_api, monkeypatch):
|
||||
tmp, _ = melden
|
||||
url, empfangen, _ = bot_api
|
||||
env = tmp / "telegram.env"
|
||||
env.write_text("TELEGRAM_BOT_TOKEN=1:test\nTELEGRAM_HOME_CHANNEL=42\n", encoding="utf-8", newline="\n")
|
||||
monkeypatch.setenv("MC_TELEGRAM_ENV", env.as_posix())
|
||||
monkeypatch.setenv("MC_TELEGRAM_API", url)
|
||||
monkeypatch.setenv("STUB_EXIT", "1") # wie im Homelab-Teil: kein Hermes
|
||||
ergebnis = einstellungen.telegram_test()
|
||||
assert ergebnis["gesendet"] and ergebnis["weg"] == "direkt", ergebnis
|
||||
assert len(empfangen) == 1 and empfangen[0]["text"].startswith("[Test]\nTestmeldung aus den Einstellungen")
|
||||
|
||||
|
||||
@braucht_bash
|
||||
def test_telegram_test_nennt_den_grund(melden, bot_api, monkeypatch):
|
||||
tmp, _ = melden
|
||||
monkeypatch.setenv("STUB_EXIT", "1")
|
||||
monkeypatch.setenv("STUB_AUSGABE", "Fehler: Telegram-Plattform nicht verbunden")
|
||||
ergebnis = einstellungen.telegram_test()
|
||||
assert not ergebnis["ok"] and not ergebnis["gesendet"]
|
||||
assert ergebnis["grund"].startswith("Fehler: Telegram-Plattform nicht verbunden; Zweitweg: Zugangsdatei")
|
||||
assert "fehlt oder ist nicht lesbar" in ergebnis["grund"] and ergebnis["detail"].startswith("Die Testmeldung ging")
|
||||
|
||||
# Telegram lehnt ab (etwa ein falsches Token): der Grund kommt aus curls Fehlercode.
|
||||
url, _, antwort = bot_api
|
||||
antwort["code"] = 401
|
||||
env = tmp / "telegram.env"
|
||||
env.write_text("TELEGRAM_BOT_TOKEN=1:falsch\nTELEGRAM_HOME_CHANNEL=42\n", encoding="utf-8", newline="\n")
|
||||
monkeypatch.setenv("MC_TELEGRAM_ENV", env.as_posix())
|
||||
monkeypatch.setenv("MC_TELEGRAM_API", url)
|
||||
monkeypatch.setitem(einstellungen._letzter_test, "ts", 0.0)
|
||||
assert "Zweitweg: Telegram lehnt ab" in einstellungen.telegram_test()["grund"]
|
||||
|
||||
|
||||
@braucht_bash
|
||||
def test_telegram_test_nicht_im_sekundentakt(melden):
|
||||
assert einstellungen.telegram_test()["gesendet"]
|
||||
zweiter = einstellungen.telegram_test()
|
||||
assert not zweiter["ok"] and "kurz warten" in zweiter["detail"]
|
||||
|
||||
|
||||
def test_telegram_test_am_pc(monkeypatch):
|
||||
monkeypatch.setattr(einstellungen, "AUF_DER_BOX", False)
|
||||
monkeypatch.setitem(einstellungen._letzter_test, "ts", 0.0)
|
||||
ergebnis = einstellungen.telegram_test()
|
||||
assert not ergebnis["ok"] and "nur auf der Box" in ergebnis["detail"]
|
||||
|
||||
|
||||
def test_melde_log_lesen(tmp_path):
|
||||
"""Eine mehrzeilige Ausgabe von `hermes send` landet mitten im FALLBACK-Eintrag (notify.sh)."""
|
||||
text = "Testmeldung aus den Einstellungen (Box-Wart, 24.09. 21:00 Uhr, Kennung abc123). Kommt sie an …"
|
||||
log = tmp_path / "notify.log"
|
||||
log.write_text(
|
||||
"2026-09-24 20:00:00 OK telegram: Frühere Meldung\n"
|
||||
"2026-09-24 21:00:00 FALLBACK (telegram fehlgeschlagen: Traceback (most recent call last):\n"
|
||||
" ConnectionError: gateway down; Zweitweg: Telegram antwortet nicht rechtzeitig): " + text + "\n"
|
||||
"2026-09-24 21:00:05 OK telegram: Spätere Meldung\n", encoding="utf-8")
|
||||
eintrag = einstellungen.log_eintrag(log, "Kennung abc123")
|
||||
assert eintrag.startswith("2026-09-24 21:00:00 FALLBACK") and eintrag.endswith(text)
|
||||
gedeutet = einstellungen.deute_eintrag(eintrag, text)
|
||||
assert gedeutet == {"gesendet": False, "weg": None, "grund": "Traceback (most recent call last): ConnectionError: "
|
||||
"gateway down; Zweitweg: Telegram antwortet nicht rechtzeitig"}
|
||||
assert einstellungen.log_eintrag(log, "Kennung fehlt") is None
|
||||
assert einstellungen.log_eintrag(tmp_path / "gibt-es-nicht.log", "x") is None
|
||||
assert einstellungen.deute_eintrag("2026-09-24 21:00:00 OK telegram direkt: x", "x")["weg"] == "direkt"
|
||||
assert einstellungen.deute_eintrag("2026-09-24 03:00:00 QUEUED für Morgen-Digest: x", "x")["gesendet"] is False
|
||||
@@ -14,7 +14,7 @@ from fastapi.testclient import TestClient
|
||||
from kern import einstellungen as einstellungen_mod
|
||||
from kern import partner
|
||||
from routers import partner as partner_router
|
||||
from services import waechter
|
||||
from services import software_stand, waechter
|
||||
|
||||
BACKEND = Path(__file__).resolve().parent.parent
|
||||
|
||||
@@ -110,10 +110,12 @@ def test_durchreiche_get_und_post(mit_partner, partner_server):
|
||||
assert json.loads(letzte["koerper"]) == {"ziel": "ct:101"}
|
||||
|
||||
|
||||
def test_instanz_ohne_netz(mit_partner):
|
||||
def test_instanz_ohne_netz(mit_partner, monkeypatch):
|
||||
mit_partner("http://127.0.0.1:9")
|
||||
monkeypatch.setattr(software_stand, "stand", lambda: {"quelle": "unbekannt"}) # seit 24.09.2026 dabei
|
||||
assert _client().get("/api/instanz").json() == {
|
||||
"rolle": "box", "instanz": "Box-Wart", "partner": {"eingerichtet": True, "name": "Homelab"}}
|
||||
"rolle": "box", "instanz": "Box-Wart", "partner": {"eingerichtet": True, "name": "Homelab"},
|
||||
"stand": {"quelle": "unbekannt"}}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("pfad", ["partner/health", "partner", "stream"])
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
"""Software-Stand (24.09.2026): welcher Commit seit wann live ist — Stand-Datei, git-Rückfall, /api/instanz,
|
||||
und wie deploy.sh und ausrollen.sh die Datei schreiben (deploy/stand-schreiben.py)."""
|
||||
|
||||
import json
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
from routers import partner as partner_router
|
||||
from services import software_stand
|
||||
|
||||
DEPLOY = Path(__file__).resolve().parents[2] / "deploy"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def stand_datei(tmp_path, monkeypatch):
|
||||
datei = tmp_path / "mc2-stand.json"
|
||||
monkeypatch.setenv("MC_STAND_DATEI", str(datei))
|
||||
monkeypatch.setitem(software_stand._git_cache, "ts", 0.0)
|
||||
return datei
|
||||
|
||||
|
||||
def _schreiben(ziel: Path, eingabe: str) -> subprocess.CompletedProcess:
|
||||
return subprocess.run([sys.executable, str(DEPLOY / "stand-schreiben.py"), str(ziel)],
|
||||
input=eingabe.encode("utf-8"), capture_output=True, timeout=60)
|
||||
|
||||
|
||||
def test_stand_schreiben_wie_der_deploy(stand_datei):
|
||||
lauf = _schreiben(stand_datei, "40d2840\n2026-09-24T20:46:03+02:00\ndoku: Phase 4 ehrlich, Rückweg über Nacht\n")
|
||||
assert lauf.returncode == 0, lauf.stderr
|
||||
daten = json.loads(stand_datei.read_text(encoding="utf-8"))
|
||||
assert daten["commit"] == "40d2840" and daten["commit_datum"] == "2026-09-24T20:46:03+02:00"
|
||||
assert daten["betreff"] == "doku: Phase 4 ehrlich, Rückweg über Nacht"
|
||||
assert re.fullmatch(r"\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d[+-]\d\d:\d\d", daten["deploy_zeit"])
|
||||
assert not list(stand_datei.parent.glob(".mc2-stand-*")) # keine Reste der atomaren Schreibweise
|
||||
assert software_stand.stand() == {"quelle": "deploy", **daten}
|
||||
|
||||
|
||||
def test_stand_schreiben_ohne_commit_schreibt_nichts(stand_datei):
|
||||
assert _schreiben(stand_datei, "").returncode == 1
|
||||
assert not stand_datei.exists()
|
||||
|
||||
|
||||
def test_ohne_datei_aus_git(stand_datei, monkeypatch):
|
||||
monkeypatch.setattr(software_stand, "_git_log", lambda: "abc1234\n2026-09-24T19:00:00+02:00\nhomelab: Ausführer\n")
|
||||
assert software_stand.stand() == {"quelle": "git", "commit": "abc1234", "betreff": "homelab: Ausführer",
|
||||
"commit_datum": "2026-09-24T19:00:00+02:00", "deploy_zeit": None}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("datei_inhalt", [None, "kaputt", '{"commit": ""}', "[1, 2]"])
|
||||
def test_ohne_datei_und_ohne_git_unbekannt(stand_datei, monkeypatch, datei_inhalt):
|
||||
if datei_inhalt is not None:
|
||||
stand_datei.write_text(datei_inhalt, encoding="utf-8")
|
||||
monkeypatch.setattr(software_stand, "_git_log", lambda: None)
|
||||
assert software_stand.stand() == {"quelle": "unbekannt", "commit": None, "betreff": None,
|
||||
"commit_datum": None, "deploy_zeit": None}
|
||||
|
||||
|
||||
def test_git_wird_nicht_bei_jeder_anfrage_gefragt(stand_datei, monkeypatch):
|
||||
aufrufe = []
|
||||
monkeypatch.setattr(software_stand, "_git_log", lambda: aufrufe.append(1) or "abc\n\nx\n")
|
||||
software_stand.stand()
|
||||
software_stand.stand()
|
||||
assert len(aufrufe) == 1
|
||||
|
||||
|
||||
def test_instanz_liefert_den_stand(monkeypatch):
|
||||
monkeypatch.setattr(software_stand, "stand", lambda: {"quelle": "deploy", "commit": "40d2840"})
|
||||
app = FastAPI()
|
||||
app.include_router(partner_router.router)
|
||||
assert TestClient(app).get("/api/instanz").json()["stand"] == {"quelle": "deploy", "commit": "40d2840"}
|
||||
|
||||
|
||||
def test_deploy_und_ausrollen_schreiben_den_stand_erst_nach_dem_erfolg():
|
||||
deploy = (DEPLOY / "deploy.sh").read_text(encoding="utf-8")
|
||||
stufe2 = deploy[deploy.index("stufe2() {"):deploy.index("homelab_mitziehen() {")]
|
||||
assert stufe2.index("trap - ERR") < stufe2.index("stand_schreiben") < stufe2.index("ist live")
|
||||
assert "stand-schreiben.py \"$STAND_DATEI\"" in deploy and 'STAND_DATEI="/srv/models/mc2-stand.json"' in deploy
|
||||
|
||||
ausrollen = (DEPLOY / "homelab" / "ausrollen.sh").read_text(encoding="utf-8")
|
||||
assert ausrollen.index("einrichten.sh") < ausrollen.index("stand-schreiben.py /var/lib/mc2/mc2-stand.json")
|
||||
# Scheitert nur der Stand, gilt das Ausrollen trotzdem (sonst käme ein Alarm für eine Nebensache).
|
||||
assert "|| echo \"Hinweis: Die Stand-Datei im Container" in ausrollen
|
||||
@@ -0,0 +1,150 @@
|
||||
"""Wartungsfenster für den Neustart nach Updates (deploy/wartungsfenster.sh, seit 24.09.2026 einstellbar) und wie
|
||||
deploy.sh mit den Einstellungen umgeht: Radar-Schalter respektieren, das Zeitfenster-Drop-in nie anfassen.
|
||||
|
||||
Die Tests fahren die echten Shell-Funktionen mit vorgetäuschter Zeit (MC_FENSTER_JETZT)."""
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from services import einstellungen
|
||||
|
||||
DEPLOY = Path(__file__).resolve().parents[2] / "deploy"
|
||||
BASH = shutil.which("bash")
|
||||
|
||||
pytestmark = pytest.mark.skipif(
|
||||
BASH is None or (sys.platform == "win32" and "system32" in BASH.lower()),
|
||||
reason="braucht eine bash (Git-Bash oder Linux)",
|
||||
)
|
||||
|
||||
|
||||
def _fenster(befehl: str, **env: str) -> subprocess.CompletedProcess:
|
||||
skript = f'. "{(DEPLOY / "wartungsfenster.sh").as_posix()}"; {befehl}'
|
||||
umgebung = {k: v for k, v in os.environ.items() if k not in ("MC_WARTUNGSFENSTER", "MC_FENSTER_JETZT")}
|
||||
umgebung.update({"MC_EINSTELLUNGEN": "/gibt/es/nicht.json", **env})
|
||||
return subprocess.run([BASH, "-c", skript], env=umgebung, capture_output=True, text=True, encoding="utf-8",
|
||||
timeout=60)
|
||||
|
||||
|
||||
def _im_fenster(jetzt: str, fenster: str | None = None) -> bool:
|
||||
env = {"MC_FENSTER_JETZT": jetzt}
|
||||
if fenster:
|
||||
env["MC_WARTUNGSFENSTER"] = fenster
|
||||
return _fenster("im_wartungsfenster", **env).returncode == 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize("jetzt, erwartet", [
|
||||
("7 04:00", True), ("7 04:40", True), ("7 06:59", True),
|
||||
("7 03:59", False), ("7 07:00", False),
|
||||
("4 15:00", False), # Donnerstagnachmittag: der Neustart vom 24.09.
|
||||
("6 05:00", False),
|
||||
])
|
||||
def test_standard_ist_sonntag_frueh(jetzt, erwartet):
|
||||
assert _im_fenster(jetzt) is erwartet
|
||||
|
||||
|
||||
@pytest.mark.parametrize("fenster, jetzt, erwartet", [
|
||||
("6 05:00", "6 05:00", True), ("6 05:00", "6 07:59", True), ("6 05:00", "6 08:00", False),
|
||||
("6 05:00", "7 05:00", False),
|
||||
("6 23:30", "6 23:00", True), ("6 23:30", "7 01:59", True), ("6 23:30", "7 02:00", False),
|
||||
("6 23:30", "6 22:59", False),
|
||||
("7 23:00", "1 01:30", True), ("7 23:00", "1 02:00", False), # über das Wochenende hinaus
|
||||
("9 25:00", "7 04:30", True), ("9 25:00", "6 05:00", False), # Unsinn → Standard So 04:30
|
||||
])
|
||||
def test_eingestelltes_fenster(fenster, jetzt, erwartet):
|
||||
assert _im_fenster(jetzt, fenster) is erwartet
|
||||
|
||||
|
||||
@pytest.mark.parametrize("tag, uhrzeit", [(7, "04:30"), (6, "05:00"), (6, "23:30"), (7, "22:00"), (1, "00:00"),
|
||||
(3, "21:30")])
|
||||
def test_text_wie_in_der_oberflaeche(tag, uhrzeit):
|
||||
lauf = _fenster("fenster_text", MC_WARTUNGSFENSTER=f"{tag} {uhrzeit}")
|
||||
assert lauf.returncode == 0, lauf.stderr
|
||||
assert lauf.stdout == einstellungen.neustart_fenster(tag, uhrzeit)
|
||||
|
||||
|
||||
@pytest.mark.skipif(shutil.which("jq") is None, reason="jq fehlt hier (auf der Box ist es da)")
|
||||
def test_fenster_aus_der_einstellungsdatei(tmp_path):
|
||||
datei = tmp_path / "mc2-einstellungen.json"
|
||||
datei.write_text(json.dumps({"update_fenster": {"tag": 6, "uhrzeit": "05:00"}}), encoding="utf-8")
|
||||
lauf = _fenster("fenster_text", MC_EINSTELLUNGEN=datei.as_posix())
|
||||
assert lauf.stdout == "Samstag 05:00–07:59"
|
||||
|
||||
|
||||
def test_ohne_datei_gilt_sonntag():
|
||||
assert _fenster("fenster_text").stdout == "Sonntag 04:00–06:59"
|
||||
|
||||
|
||||
def test_autoupdate_nutzt_das_fenster():
|
||||
skript = (DEPLOY / "autoupdate.sh").read_text(encoding="utf-8")
|
||||
assert '. "$SRC_DIR/wartungsfenster.sh"' in skript
|
||||
assert "! im_wartungsfenster" in skript and "$(fenster_text)" in skript
|
||||
assert '[ "$tag" != "7" ]' not in skript # die alte, feste Sonntagsregel ist raus
|
||||
|
||||
|
||||
# --- deploy.sh und die Einstellungen ------------------------------------------------------------------
|
||||
|
||||
def _befehlszeilen(skript: str) -> list[str]:
|
||||
return [z for z in skript.splitlines() if z.strip() and not z.strip().startswith("#")]
|
||||
|
||||
|
||||
def test_deploy_fasst_das_zeitfenster_nicht_an():
|
||||
"""Das Drop-in gehört der Oberfläche: kein Befehl in deploy.sh nennt es, keiner räumt Drop-in-Ordner weg."""
|
||||
for zeile in _befehlszeilen((DEPLOY / "deploy.sh").read_text(encoding="utf-8")):
|
||||
assert "mc2-autoupdate.timer.d" not in zeile and "zeitfenster" not in zeile, zeile
|
||||
assert not re.search(r"\brm\b.*(\.d\b|systemd/user)", zeile), zeile
|
||||
|
||||
|
||||
def test_deploy_respektiert_den_radar_schalter():
|
||||
skript = (DEPLOY / "deploy.sh").read_text(encoding="utf-8")
|
||||
stufe2 = skript[skript.index("stufe2() {"):skript.index("homelab_mitziehen() {")]
|
||||
assert "if radar_aus; then" in stufe2 and 'aktiv="${aktiv//mc2-radar.timer/}"' in stufe2
|
||||
assert "systemctl --user disable --now mc2-radar.timer" in stufe2
|
||||
assert "systemctl --user enable $aktiv" in stufe2 and "enable $AKTIV" not in stufe2
|
||||
# Kein Start des Radars ohne den Schalter und ohne den Stempel gegen das Nachholen.
|
||||
for zeile in _befehlszeilen(stufe2):
|
||||
if "systemctl --user start" in zeile and "mc2-radar.timer" in zeile:
|
||||
assert zeile.strip() == "systemctl --user start mc2-radar.timer"
|
||||
start = stufe2.index("systemctl --user start mc2-radar.timer")
|
||||
assert stufe2.rindex('touch "$HOME/.local/share/systemd/timers/stamp-mc2-radar.timer"', 0, start) > \
|
||||
stufe2.rindex('[ "$radar_an" = 1 ]', 0, start)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def radar_aus(tmp_path):
|
||||
"""Die echte Funktion radar_aus aus deploy.sh, mit einem python3, das es auch am PC gibt."""
|
||||
skript = (DEPLOY / "deploy.sh").read_text(encoding="utf-8")
|
||||
funktion = re.search(r"^radar_aus\(\) \{\n.*?^\}\n", skript, re.DOTALL | re.MULTILINE).group(0)
|
||||
shim = tmp_path / "bin"
|
||||
shim.mkdir()
|
||||
(shim / "python3").write_text(f'#!/bin/sh\nexec "{Path(sys.executable).as_posix()}" "$@"\n',
|
||||
encoding="utf-8", newline="\n")
|
||||
(shim / "python3").chmod(0o755)
|
||||
|
||||
def pruefen(inhalt: str | None) -> bool:
|
||||
datei = tmp_path / "mc2-einstellungen.json"
|
||||
datei.unlink(missing_ok=True)
|
||||
if inhalt is not None:
|
||||
datei.write_text(inhalt, encoding="utf-8")
|
||||
lauf = subprocess.run([BASH, "-c", f'{funktion}\nradar_aus'], capture_output=True, text=True, timeout=60,
|
||||
env={**os.environ, "EINSTELLUNGEN": datei.as_posix(),
|
||||
"PATH": f"{shim}{os.pathsep}{os.environ.get('PATH', '')}"})
|
||||
return lauf.returncode == 0
|
||||
|
||||
return pruefen
|
||||
|
||||
|
||||
@pytest.mark.parametrize("inhalt, aus", [
|
||||
(None, False), # keine Datei: an, wie vor dem Schalter
|
||||
('{"radar": {"an": false}}', True),
|
||||
('{"radar": {"an": true}}', False),
|
||||
('{"update_fenster": {"tag": 6, "uhrzeit": "05:00"}}', False),
|
||||
("kaputt", False), ("[]", False), ('{"radar": "aus"}', False),
|
||||
])
|
||||
def test_radar_aus_liest_den_schalter(radar_aus, inhalt, aus):
|
||||
assert radar_aus(inhalt) is aus
|
||||
Reference in New Issue
Block a user