HitonabiandClaude Opus 5.5
12cefb427a
- services/lucy_autonomie.py: Stufe je Aktionsart (nur lesen, erst fragen, selbst handeln), Start „mehr Freiheit“;
Meldung danach (POST /api/lucy/gehandelt)
- services/rueckfragen.py: nummerierte Rückfragen, nur Knöpfe der Oberfläche (ERLAUBT), „ja“ führt genau die
gespeicherte Aktion aus, 24 h gültig, hinfällig wenn der Hinweis weg ist, Nachprüfung danach
- notify.sh -k: Antworttasten (Antworttastatur über den Direktweg, ohne Eingriff in Hermes)
- services/ermittlung.py: Wächter meldet rote Hinweise an Lucy; Knopf wird Rückfrage, Lucy sieht über den
api_server nach und schickt Ursache und Weg als eine Nachricht
- services/wochenbericht.py + mc2-wochenbericht.timer (So 07:30): feste Sätze, Absatz vom Modell, offene
App-Updates als Rückfragen
- MCP: Werkzeuge fragen die Freiheiten (auf_bitte), neu: freiheiten, rueckfragen, rueckfrage_beantworten,
sicherheitsupdates_einspielen, snapshot_zurueckspielen, snapshots, geraet_logs, dienst_logs
- Ausführer-Aktion journal (nur lesend), POST /api/homelab/ziele/{id}/sicherheitsupdates
- Läufe des Homelabs mit flock (Oberfläche und Steward schreiben beide)
- Skill homelab-orchestrator, SOUL-Abschnitt als Kopie in deploy/nur-box
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
168 lines
8.2 KiB
Python
168 lines
8.2 KiB
Python
"""Wöchentliches Suchen (services/homelab/pflege.py, 24.09.2026): Der Homelab-Teil erneuert alte Paketlisten in den
|
|
Gästen selbst — nachts, einmal je Gast und Tag, nie neben einem Update. Mit dem echten Bericht vom 24.09.
|
|
(tests/fixtures/pve-bericht.json): AdGuard, NPMplus, NetBird, PVE Scripts Local und Gitea haben alte Listen, die des
|
|
PBS sind frisch."""
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from collections import Counter
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from kern import einstellungen as einstellungen_mod
|
|
from kern.zeit import LOCAL_TZ
|
|
from services import waechter
|
|
from services.homelab import kanal, pflege, updates
|
|
|
|
BERICHT = json.loads((Path(__file__).parent / "fixtures" / "pve-bericht.json").read_text(encoding="utf-8"))
|
|
BACKEND = Path(__file__).resolve().parents[1]
|
|
NACHT = datetime(2026, 9, 25, 3, 0, tzinfo=LOCAL_TZ).timestamp()
|
|
TAG = datetime(2026, 9, 25, 14, 0, tzinfo=LOCAL_TZ).timestamp()
|
|
NAECHSTE_NACHT = datetime(2026, 9, 26, 3, 0, tzinfo=LOCAL_TZ).timestamp()
|
|
ALTE_LISTEN = {100, 101, 102, 103, 104}
|
|
|
|
|
|
@pytest.fixture
|
|
def uhr(tmp_path, monkeypatch):
|
|
"""Daten in tmp; der Ausführer hat zuletzt um uhr["zuletzt"] berichtet."""
|
|
monkeypatch.setenv("MC_DATEN_DIR", str(tmp_path))
|
|
einstellungen_mod.einstellungen.cache_clear()
|
|
stand = {"zuletzt": NACHT}
|
|
monkeypatch.setattr(kanal, "zuletzt", lambda: stand["zuletzt"])
|
|
monkeypatch.setattr(waechter, "TROCKEN", False)
|
|
kanal.bericht_speichern(BERICHT)
|
|
yield stand
|
|
einstellungen_mod.einstellungen.cache_clear()
|
|
|
|
|
|
def _suchen() -> Counter:
|
|
"""Wie viele Aufträge „suchen“ es je Gast gibt."""
|
|
return Counter(a["parameter"]["vmid"] for a in kanal.liste() if a["aktion"] == "suchen")
|
|
|
|
|
|
def _ausfuehrer_antwortet(scheitert: set[int] = frozenset()) -> None:
|
|
while (a := kanal.naechster()):
|
|
vmid = a["parameter"].get("vmid")
|
|
if vmid in scheitert:
|
|
kanal.ergebnis(a["id"], 100, "Hit:1 http://deb.debian.org/debian trixie InRelease\n"
|
|
"E: Failed to fetch http://security.debian.org Temporary failure resolving")
|
|
else:
|
|
kanal.ergebnis(a["id"], 0, "Hit:1 http://deb.debian.org/debian trixie InRelease")
|
|
|
|
|
|
def test_nachts_einmal_je_gast_und_tag(uhr):
|
|
assert pflege.pruefe_paketlisten(jetzt=NACHT) == []
|
|
assert set(_suchen()) == ALTE_LISTEN # PBS: frische Listen; die Arcane-VM ist kein Container
|
|
pflege.pruefe_paketlisten(jetzt=NACHT + 60) # nächster Takt: offen, nichts doppelt
|
|
_ausfuehrer_antwortet()
|
|
pflege.pruefe_paketlisten(jetzt=NACHT + 120) # erledigt, Listen im Bericht noch alt: heute nicht mehr
|
|
assert _suchen() == Counter({v: 1 for v in ALTE_LISTEN})
|
|
zustand = pflege.zustand()
|
|
assert zustand["nacht"] == "2026-09-25"
|
|
assert zustand["gaeste"]["100"]["name"] == "AdGuard Home" and zustand["gaeste"]["100"]["fehlschlaege"] == 0
|
|
uhr["zuletzt"] = NAECHSTE_NACHT # am nächsten Tag wieder, falls die Listen alt bleiben
|
|
pflege.pruefe_paketlisten(jetzt=NAECHSTE_NACHT)
|
|
assert _suchen() == Counter({v: 2 for v in ALTE_LISTEN})
|
|
|
|
|
|
def test_tagsueber_nur_wenn_die_nacht_verpasst_wurde(uhr):
|
|
uhr["zuletzt"] = TAG
|
|
pflege._schreiben({"nacht": "2026-09-25", "gaeste": {}}) # die Instanz war letzte Nacht da
|
|
pflege.pruefe_paketlisten(jetzt=TAG)
|
|
assert _suchen() == Counter() # also bis zur nächsten Nacht warten
|
|
pflege._schreiben({"nacht": "2026-09-24", "gaeste": {}}) # letzte Nacht war sie aus
|
|
pflege.pruefe_paketlisten(jetzt=TAG)
|
|
assert set(_suchen()) == ALTE_LISTEN # dann eben jetzt
|
|
assert pflege.zustand()["nacht"] == "2026-09-24" # tagsüber gilt keine Nacht als gesehen
|
|
|
|
|
|
def test_vor_zwei_uhr_zaehlt_die_nacht_von_gestern(uhr):
|
|
kurz_vor_zwei = datetime(2026, 9, 26, 1, 30, tzinfo=LOCAL_TZ).timestamp()
|
|
uhr["zuletzt"] = kurz_vor_zwei
|
|
pflege._schreiben({"nacht": "2026-09-25", "gaeste": {}})
|
|
pflege.pruefe_paketlisten(jetzt=kurz_vor_zwei)
|
|
assert _suchen() == Counter()
|
|
|
|
|
|
def test_nie_neben_einem_update_oder_einem_offenen_auftrag(uhr):
|
|
updates._merken({"id": "l1", "ziel": "ct-104", "baustein": "app", "name": "Gitea", "status": "laeuft",
|
|
"start": time.time(), "ende": None, "ergebnis": None, "text": None, "schritte": []})
|
|
kanal.anlegen("os_update", {"vmid": 100})
|
|
pflege.pruefe_paketlisten(jetzt=NACHT)
|
|
assert set(_suchen()) == ALTE_LISTEN - {100, 104}
|
|
|
|
|
|
def test_ohne_ausfuehrer_keine_auftraege(uhr):
|
|
uhr["zuletzt"] = NACHT - 3600 # letzter Bericht vor einer Stunde
|
|
pflege.pruefe_paketlisten(jetzt=NACHT)
|
|
assert _suchen() == Counter() and pflege.zustand().get("nacht") is None
|
|
|
|
|
|
def test_ausfuehrer_im_nur_lesen_modus_bekommt_keine_auftraege(uhr):
|
|
nur_lesen = json.loads(json.dumps(BERICHT))
|
|
nur_lesen["host"]["nur_lesen"] = True
|
|
kanal.bericht_speichern(nur_lesen)
|
|
pflege.pruefe_paketlisten(jetzt=NACHT)
|
|
assert _suchen() == Counter() and pflege.zustand().get("nacht") is None
|
|
|
|
|
|
def test_trockenlauf_legt_nichts_an(uhr, monkeypatch):
|
|
monkeypatch.setattr(waechter, "TROCKEN", True)
|
|
pflege.pruefe_paketlisten(jetzt=NACHT)
|
|
assert _suchen() == Counter()
|
|
|
|
|
|
def test_wiederholtes_scheitern_wird_ein_gelber_hinweis(uhr):
|
|
pflege.pruefe_paketlisten(jetzt=NACHT)
|
|
_ausfuehrer_antwortet(scheitert={100})
|
|
assert pflege.pruefe_paketlisten(jetzt=NACHT + 60) == [] # einmal ist noch kein Muster
|
|
uhr["zuletzt"] = NAECHSTE_NACHT
|
|
pflege.pruefe_paketlisten(jetzt=NAECHSTE_NACHT) # neuer Tag, neuer Versuch
|
|
_ausfuehrer_antwortet(scheitert={100})
|
|
befunde = pflege.pruefe_paketlisten(jetzt=NAECHSTE_NACHT + 60)
|
|
assert [(b.id, b.stufe, b.titel, b.sofort) for b in befunde] == [
|
|
("pflege:ct-100", "gelb", "AdGuard Home: Die Suche nach Updates scheitert", True)]
|
|
assert "2-mal hintereinander" in befunde[0].text and "E: Failed to fetch" in befunde[0].text
|
|
# Listen wieder frisch (etwa von Hand gesucht): Hinweis weg, Zähler vergessen.
|
|
frisch = json.loads(json.dumps(BERICHT))
|
|
next(g for g in frisch["gaeste"] if g["vmid"] == 100)["os_updates"]["listen_stand"] = int(NAECHSTE_NACHT)
|
|
kanal.bericht_speichern(frisch)
|
|
assert pflege.pruefe_paketlisten(jetzt=NAECHSTE_NACHT + 120) == []
|
|
assert pflege.zustand()["gaeste"]["100"]["fehlschlaege"] == 0
|
|
|
|
|
|
def test_der_waechter_zeigt_den_hinweis_ohne_telegram(uhr, tmp_path, monkeypatch):
|
|
pflege._schreiben({"nacht": "2026-09-25", "gaeste": {"100": {
|
|
"name": "AdGuard Home", "tag": "2026-09-25", "auftrag": None, "fehlschlaege": 3, "fehler": "E: kaputt"}}})
|
|
uhr["zuletzt"] = TAG
|
|
monkeypatch.setattr(waechter, "STORE_PATH", tmp_path / "mc2-waechter.json")
|
|
monkeypatch.setattr(waechter, "_stand", {"hinweise": {}, "kandidaten": {}, "verlauf": [], "auto": {}, "stand": 0.0})
|
|
monkeypatch.setattr(waechter, "_update_laeuft", lambda: False)
|
|
meldungen: list[str] = []
|
|
monkeypatch.setattr(waechter, "_telegram", lambda betreff, text, knoepfe=None: meldungen.append(text))
|
|
|
|
def pruefe_paketlisten():
|
|
return pflege.pruefe_paketlisten(jetzt=TAG)
|
|
|
|
monkeypatch.setattr(waechter, "PRUEFUNGEN", (pruefe_paketlisten,))
|
|
waechter.takt()
|
|
hinweis = waechter._stand["hinweise"]["pflege:ct-100"]
|
|
assert hinweis["stufe"] == "gelb" and "3-mal" in hinweis["text"] and meldungen == []
|
|
|
|
|
|
@pytest.mark.skipif(os.name != "posix", reason="Die Dateisperre gibt es nur unter Linux (am PC: nur Thread-Sperre).")
|
|
def test_auftraege_aus_zwei_prozessen_gehen_nicht_verloren(tmp_path):
|
|
"""Oberfläche und Steward schreiben beide in die Auftragsliste; ohne Dateisperre überschrieben sie sich."""
|
|
code = ("from services.homelab import kanal\n"
|
|
"for i in range(40):\n"
|
|
" kanal.anlegen('suchen', {'vmid': 100 + i})\n")
|
|
env = {**os.environ, "MC_DATEN_DIR": str(tmp_path), "MC_ROLLE": "homelab"}
|
|
prozesse = [subprocess.Popen([sys.executable, "-c", code], cwd=BACKEND, env=env) for _ in range(3)]
|
|
assert all(p.wait(timeout=120) == 0 for p in prozesse)
|
|
auftraege = json.loads((tmp_path / "ausfuehrer-auftraege.json").read_text(encoding="utf-8"))
|
|
assert len(auftraege) == 120
|