sorb/threadnet-operating
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
59c75c8a01c889c0fcb8345d4ba46e0c31c333d9
threadnet-operating/monitoring/cve/test_targets.py
T
Thore Cimbal 59c75c8a01 cve: actually test the narrowing — the previous commit shipped it uncovered 
Disabling the filter left all twenty-two assertions green, which means nothing
tested it. The counter-case I had written exercises the skip in the derivation,
not the filter inside the registry selection, so the feature went out with no
coverage at all — the project's own favourite failure, in the change that was
meant to remove noise.

Three assertions now cover it: three repositories of which one runs, the
counter-case showing the same call returns all three without estate knowledge,
and one that pins what the rule must not do — it excludes repositories, not old
tags, so a rollback target of a running service still counts.

Same sabotage as before now turns one of them red.
2026-08-21 12:00:00 +00:00

349 lines
15 KiB
Python
Raw Blame History

#!/usr/bin/env python3
# Zusicherungen fuer targets.py (#0106). Stdlib, laeuft ohne Netz:
# python3 test_targets.py
#
# Die Antworten sind AUFGEZEICHNET, nicht erfunden - abgenommen am 2026-08-21
# von der laufenden Anlage (Prometheus auf dem Betriebs-Host, rohana /v2).
#
# ⚠️ Jede Zusicherung hat ihre GEGENPROBE. Eine Pruefung, die nur je "gleich"
# gesagt hat, ist eine Vermutung - das ist in diesem Projekt die haeufigste
# Fehlerklasse ("meldet Erfolg, ist aber blind"), zuletzt belegt in der AAR vom
# 2026-08-01. Deshalb steht neben jedem "muss zusammenfallen" ein "darf NICHT
# zusammenfallen".
import unittest
from datetime import datetime, timezone
import targets
def ts(jahr, monat, tag):
return datetime(jahr, monat, tag, tzinfo=timezone.utc)
# --- Aufgezeichnete Antworten ------------------------------------------------
# Auszug aus `count by (image) (kube_pod_container_info)`, 2026-08-21.
PROM_CLUSTER = [
{"metric": {"image": "docker.io/library/postgres:17-alpine"}},
{"metric": {"image": "docker.io/rancher/mirrored-library-traefik:3.6.10"}},
{"metric": {"image": "rohana.axion1337.de/sorb/threadnet-web:v0.6.0"}},
{"metric": {"image": "oci.element.io/synapse:v1.151.0-ess.1"}},
]
# Auszug aus `container_last_seen{...}` - BEIDE Jobs, damit die Abgrenzung
# ueberhaupt etwas zu tun hat.
PROM_CADVISOR_GEMISCHT = [
{"metric": {"image": "gitea/gitea:1.27.0", "job": "operating_cadvisor"}},
{"metric": {"image": "traefik:v3.7.9", "job": "operating_cadvisor"}},
{"metric": {"image": "ghcr.io/coollabsio/coolify:4.3.9", "job": "gameserver_cadvisor"}},
{"metric": {"image": "redis:7-alpine", "job": "gameserver_cadvisor"}},
]
class Normalisierung(unittest.TestCase):
def test_docker_hub_schreibweisen_fallen_zusammen(self):
formen = [
"docker.io/library/postgres:17-alpine",
"index.docker.io/library/postgres:17-alpine",
"library/postgres:17-alpine",
"postgres:17-alpine",
]
ergebnisse = {targets.normalisiere(f) for f in formen}
self.assertEqual(ergebnisse, {"postgres:17-alpine"})
def test_gegenprobe_verschiedene_fassungen_fallen_NICHT_zusammen(self):
# Ohne diese Zusicherung koennte normalisiere() alles auf denselben Wert
# abbilden und der Test oben waere trotzdem gruen.
self.assertNotEqual(
targets.normalisiere("postgres:15-alpine"),
targets.normalisiere("postgres:17-alpine"),
)
self.assertNotEqual(
targets.normalisiere("rohana.axion1337.de/sorb/threadnet-web:v0.3.0"),
targets.normalisiere("rohana.axion1337.de/sorb/threadnet-web:v0.6.0"),
)
def test_fremde_registry_bleibt_unangetastet(self):
for ref in (
"rohana.axion1337.de/sorb/threadnet-web:v0.6.0",
"ghcr.io/fluxcd/source-controller:v1.8.2",
"quay.io/jetstack/cert-manager-webhook:v1.14.0",
):
self.assertEqual(targets.normalisiere(ref), ref)
def test_ohne_tag_wird_verworfen_und_NICHT_zu_latest(self):
for ref in ("postgres", "docker.io/library/postgres", "ghcr.io/a/b"):
self.assertIsNone(targets.normalisiere(ref))
def test_digest_wird_abgetrennt_tag_bleibt(self):
self.assertEqual(
targets.normalisiere("postgres:17-alpine@sha256:" + "ab" * 32),
"postgres:17-alpine",
)
# Nur Digest, kein Tag -> kein Ziel.
self.assertIsNone(targets.normalisiere("postgres@sha256:" + "ab" * 32))
def test_registry_mit_port_wird_nicht_als_tag_gelesen(self):
self.assertEqual(targets.normalisiere("reg.lan:5000/a/b:v1"), "reg.lan:5000/a/b:v1")
self.assertIsNone(targets.normalisiere("reg.lan:5000/a/b"))
def test_leeres_und_muell(self):
for ref in ("", " ", None):
self.assertIsNone(targets.normalisiere(ref))
class Quellentrennung(unittest.TestCase):
def setUp(self):
self._echt = targets.promql
def tearDown(self):
targets.promql = self._echt
def test_operating_nimmt_gameserver_nicht_mit(self):
# Der Ausdruck filtert serverseitig; hier wird geprueft, dass er den
# Filter ueberhaupt traegt - sonst faellt die Abgrenzung still weg.
self.assertIn('job="operating_cadvisor"', targets.AUSDRUCK_OPERATING)
gesehen = {}
def fake(basis, ausdruck, timeout=10.0):
gesehen["ausdruck"] = ausdruck
# Prometheus wuerde bereits gefiltert antworten - wir simulieren das,
# indem wir den Filter anwenden.
job = "operating_cadvisor" if 'job="operating_cadvisor"' in ausdruck else None
return [r for r in PROM_CADVISOR_GEMISCHT if r["metric"]["job"] == job]
targets.promql = fake
ziele = targets.ziele_operating("http://x")
self.assertEqual(ziele, {"gitea/gitea:1.27.0", "traefik:v3.7.9"})
def test_gegenprobe_ohne_filter_kaeme_gameserver_mit(self):
# Beweist, dass die Zusicherung oben ueberhaupt etwas prueft.
def fake_ohne_filter(basis, ausdruck, timeout=10.0):
return PROM_CADVISOR_GEMISCHT
targets.promql = fake_ohne_filter
ziele = targets.ziele_operating("http://x")
self.assertIn("redis:7-alpine", ziele)
def test_cluster_normalisiert_beim_einlesen(self):
targets.promql = lambda b, a, timeout=10.0: PROM_CLUSTER
ziele = targets.ziele_cluster("http://x")
self.assertIn("postgres:17-alpine", ziele)
self.assertIn("rancher/mirrored-library-traefik:3.6.10", ziele)
self.assertNotIn("docker.io/library/postgres:17-alpine", ziele)
class RegistryAuswahl(unittest.TestCase):
def setUp(self):
self._repos, self._tags, self._erstellt = (
targets.registry_repos,
targets.registry_tags,
targets.tag_erstellt,
)
targets.registry_token = lambda *a, **k: "egal"
def tearDown(self):
targets.registry_repos, targets.registry_tags, targets.tag_erstellt = (
self._repos,
self._tags,
self._erstellt,
)
def _stelle(self, tags, zeiten):
targets.registry_repos = lambda reg, timeout=15.0: ["sorb/x"]
targets.registry_tags = lambda reg, repo, token="", timeout=15.0: tags
targets.tag_erstellt = lambda reg, repo, tag, token="", timeout=15.0: zeiten.get(tag)
def test_juengste_drei_nach_ZEIT_nicht_nach_name(self):
# ⚠️ Der Kern: lexikografisch stuende v0.9.0 vor v0.10.0.
tags = ["v0.9.0", "v0.10.0", "v0.8.0", "v0.7.0"]
zeiten = {
"v0.7.0": ts(2026, 1, 1),
"v0.8.0": ts(2026, 2, 1),
"v0.9.0": ts(2026, 3, 1),
"v0.10.0": ts(2026, 4, 1),
}
self._stelle(tags, zeiten)
ziele = targets.ziele_registry("reg.example", je_repo=3)
self.assertEqual(
ziele,
{"reg.example/sorb/x:v0.10.0", "reg.example/sorb/x:v0.9.0", "reg.example/sorb/x:v0.8.0"},
)
def test_gegenprobe_namenssortierung_haette_anders_gewaehlt(self):
# Belegt, dass der Test oben nicht zufaellig gruen ist: nach Namen
# sortiert faellt die JUENGSTE Fassung heraus und eine alte kommt mit.
tags = ["v0.9.0", "v0.10.0", "v0.8.0", "v0.7.0"]
nach_name = sorted(tags, reverse=True)[:3]
self.assertNotIn("v0.10.0", nach_name) # die juengste - waere verloren
self.assertIn("v0.7.0", nach_name) # die aelteste - waere mitgekommen
def test_bau_artefakte_fallen_raus(self):
tags = ["v1.0.0", "sha-13547676", "latest-ci", "latest"]
zeiten = {t: ts(2026, 5, 10) for t in tags}
self._stelle(tags, zeiten)
ziele = targets.ziele_registry("reg.example", je_repo=5)
self.assertEqual(ziele, {"reg.example/sorb/x:v1.0.0"})
def _stelle_repos(self, repos):
targets.registry_repos = lambda reg, timeout=15.0: repos
targets.registry_tags = lambda reg, repo, token="", timeout=15.0: ["v1.0.0"]
targets.tag_erstellt = lambda reg, repo, tag, token="", timeout=15.0: ts(2026, 5, 1)
def test_nur_repos_mit_laufendem_tag_zaehlen(self):
# ⚠️ Der Kern der Einengung (#0051): Von drei Repos laeuft nur eines.
# element-desktop-build und windows-vm sind Bau-Artefakte und trugen
# zusammen 61 CRITICAL, davon 1 behebbar.
self._stelle_repos(["sorb/threadnet-web", "sorb/element-desktop-build", "sorb/windows-vm"])
laufend = {"reg.example/sorb/threadnet-web:v0.6.0"}
ziele = targets.ziele_registry("reg.example", je_repo=3, laufend=laufend)
self.assertEqual(ziele, {"reg.example/sorb/threadnet-web:v1.0.0"})
def test_gegenprobe_ohne_einengung_kaemen_die_bau_artefakte_mit(self):
# Belegt, dass der Test darueber ueberhaupt etwas prueft: ohne
# Bestandswissen liefert dieselbe Funktion alle drei Repos.
self._stelle_repos(["sorb/threadnet-web", "sorb/element-desktop-build", "sorb/windows-vm"])
ziele = targets.ziele_registry("reg.example", je_repo=3, laufend=None)
self.assertEqual(len(ziele), 3)
self.assertIn("reg.example/sorb/windows-vm:v1.0.0", ziele)
def test_einengung_trifft_nur_das_repo_nicht_den_tag(self):
# Laeuft v0.6.0, ist v0.5.4 trotzdem ein gueltiges Rollback-Ziel -
# die Einengung schliesst REPOS aus, nicht alte Fassungen.
targets.registry_repos = lambda reg, timeout=15.0: ["sorb/threadnet-web"]
targets.registry_tags = lambda reg, repo, token="", timeout=15.0: ["v0.6.0", "v0.5.4"]
targets.tag_erstellt = lambda reg, repo, tag, token="", timeout=15.0: (
ts(2026, 8, 19) if tag == "v0.6.0" else ts(2026, 8, 17))
laufend = {"reg.example/sorb/threadnet-web:v0.6.0"}
ziele = targets.ziele_registry("reg.example", je_repo=3, laufend=laufend)
self.assertIn("reg.example/sorb/threadnet-web:v0.5.4", ziele)
def test_tag_ohne_zeitstempel_wird_uebersprungen_nicht_geraten(self):
tags = ["v1.0.0", "v2.0.0"]
zeiten = {"v1.0.0": ts(2026, 5, 10)} # v2.0.0 liefert None
self._stelle(tags, zeiten)
ziele = targets.ziele_registry("reg.example", je_repo=5)
self.assertEqual(ziele, {"reg.example/sorb/x:v1.0.0"})
class Ausfallverhalten(unittest.TestCase):
"""⚠️ Der eigentliche Kern (ADR-0026): Ein Ausfall der Herleitung darf nicht
wie vollstaendige Deckung aussehen."""
CFG = {
"prometheus": "http://x",
"registry": "reg.example",
"je_repo": 3,
"timeout": 1.0,
"cluster_an": True,
"operating_an": True,
"registry_an": True,
}
def setUp(self):
self._c, self._o, self._r = (
targets.ziele_cluster,
targets.ziele_operating,
targets.ziele_registry,
)
def tearDown(self):
targets.ziele_cluster, targets.ziele_operating, targets.ziele_registry = (
self._c,
self._o,
self._r,
)
def _stelle(self, cluster=None, operating=None, registry=None):
def macher(wert):
if isinstance(wert, Exception):
def f(*a, **k):
raise wert
return f
return lambda *a, **k: (wert or set())
targets.ziele_cluster = macher(cluster)
targets.ziele_operating = macher(operating)
targets.ziele_registry = macher(registry)
def test_alles_gut(self):
self._stelle({"a:1"}, {"b:1"}, {"c:1"})
h = targets.herleiten(self.CFG)
self.assertEqual(h.ziele, {"a:1", "b:1", "c:1"})
self.assertTrue(h.vollstaendig)
self.assertTrue(h.brauchbar)
self.assertEqual(set(h.stand), {"cluster", "operating", "registry"})
def test_eine_quelle_faellt_aus_die_anderen_liefern_weiter(self):
self._stelle({"a:1"}, OSError("Prometheus weg"), {"c:1"})
h = targets.herleiten(self.CFG)
self.assertEqual(h.ziele, {"a:1", "c:1"})
self.assertFalse(h.vollstaendig)
self.assertTrue(h.brauchbar)
self.assertIn("operating", h.fehler)
self.assertNotIn("operating", h.stand) # Frische bleibt aus
self.assertIn("cluster", h.stand)
def test_stand_der_ausgefallenen_quelle_bleibt_auf_dem_ALTEN_wert(self):
# Damit cve_target_source_stale altert, statt zu verschwinden.
self._stelle(OSError("weg"), {"b:1"}, {"c:1"})
h = targets.herleiten(self.CFG, vorheriger_stand={"cluster": 1000.0})
self.assertEqual(h.stand["cluster"], 1000.0)
self.assertGreater(h.stand["operating"], 1000.0)
def test_alle_quellen_weg_ergibt_leere_und_UNBRAUCHBARE_menge(self):
self._stelle(OSError("a"), OSError("b"), OSError("c"))
h = targets.herleiten(self.CFG)
self.assertEqual(h.ziele, set())
self.assertFalse(h.brauchbar) # -> nichts schreiben, nichts loeschen
self.assertEqual(set(h.fehler), {"cluster", "operating", "registry"})
def test_gegenprobe_leere_menge_ist_NICHT_dasselbe_wie_erfolg(self):
# Der Fall, der ohne diese Unterscheidung wie 100 % Deckung aussaehe.
#
# ⚠️ VERHALTEN GEAENDERT am 2026-08-21 (#0051): Antworten Cluster UND
# Betriebs-Host erfolgreich LEER, wird die Registry-Auswahl jetzt
# uebersprungen und als Fehler gemeldet. Frueher galt das als
# "vollstaendig, nur unbrauchbar". Der Grund fuer die Aenderung: Ohne
# laufenden Bestand kann die Registry-Auswahl Rollback-Ziele nicht von
# Bau-Artefakten unterscheiden - und laeuft wirklich nichts, ist das ein
# Ausfall und kein Normalzustand. Er soll deshalb als Fehler sichtbar
# sein, nicht nur an `brauchbar` haengen.
self._stelle(set(), set(), set())
h = targets.herleiten(self.CFG)
self.assertEqual(h.ziele, set())
self.assertFalse(h.brauchbar) # nicht verwertbar ...
self.assertFalse(h.vollstaendig) # ... UND es wird gesagt
self.assertIn("registry", h.fehler)
self.assertIn("kein laufender Bestand", h.fehler["registry"])
def test_registry_ohne_laufenden_bestand_wird_uebersprungen(self):
# Gegenprobe zur Zeile darueber: Mit Bestand laeuft die Registry sehr wohl.
self._stelle({"a:1"}, set(), {"reg/x:v1"})
h = targets.herleiten(self.CFG)
self.assertIn("reg/x:v1", h.ziele)
self.assertNotIn("registry", h.fehler)
class Deckungsrechnung(unittest.TestCase):
"""Die Mengenarithmetik, auf der die Kriterien 2 und 3 beruhen."""
def test_zahlen_der_handmessung_vom_2026_08_21(self):
soll = {f"i{n}:v1" for n in range(51)}
ist = {f"i{n}:v1" for n in range(27)} | {"alt:v1", "weg:v1"}
fehlend = soll - ist
verwaist = ist - soll
self.assertEqual(len(fehlend), 24)
self.assertEqual(len(verwaist), 2)
self.assertAlmostEqual(len(soll & ist) / len(soll), 27 / 51, places=4)
def test_gegenprobe_die_rechnung_kann_rot_werden(self):
soll = {"a:1", "b:1"}
ist = {"a:1", "b:1"}
self.assertEqual(len(soll - ist), 0)
ist_kaputt = {"a:1"}
self.assertEqual(len(soll - ist_kaputt), 1) # merkt es
if __name__ == "__main__":
unittest.main(verbosity=2)
Reference in New Issue View Git Blame Copy Permalink