Thore Cimbal
d5e2995314
The alloy chart's 1.x line pins its config-reloader sidecar by digest. Kubernetes then reports that container's image as a bare sha256 and puts the usable reference in image_spec alone, which the derivation was not reading. An hour after the chart bump the sidecar would have gone unscanned — the exact hole #0106 exists to close, reopened by an upgrade rather than by neglect. Worse than missing: the bare digest passed normalisation as repository 'sha256' with the hex as its tag, so it would have entered targets.txt, failed every pull, and shown up as a permanent coverage gap pointing at nothing. Both ends are closed and both are asserted, including that the query still asks for image_spec — an assertion on the parsing alone would stay green while the data never arrives. Ten more decisions cover what is fixed in the repo but not yet rolled out on the operating host, dated a week out so the alert speaks up if the deploy does not happen.
396 lines
18 KiB
Python
396 lines
18 KiB
Python
#!/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
|
|
|
|
|
|
class DigestFestgenagelt(unittest.TestCase):
|
|
"""⚠️ Nachgetragen am 2026-08-21, nachdem ein Chart-Sprung einen per Digest
|
|
festgenagelten Beiwagen hereingebracht hat (config-reloader im alloy-Chart
|
|
1.x). `kube_pod_container_info` fuehrt dessen `image` dann nur noch als
|
|
nackten sha256 - die brauchbare Referenz steht in `image_spec`. Ohne diese
|
|
Zusicherungen faellt so ein Container still aus der Zielmenge, und zwar
|
|
genau in der Herleitung, die gebaut wurde, damit nichts mehr still
|
|
herausfaellt.
|
|
"""
|
|
|
|
# Aufgezeichnet aus dem laufenden Cluster, 2026-08-21 nach dem Chart-Sprung.
|
|
ECHTE_ANTWORT = [
|
|
{"metric": {"image": "docker.io/grafana/alloy:v1.16.3",
|
|
"image_spec": "docker.io/grafana/alloy:v1.16.3"}},
|
|
{"metric": {"image": "sha256:c92477446e98f056925f69537521dab09b1c57f894bf258b3344f8bce723cdc7",
|
|
"image_spec": "quay.io/prometheus-operator/prometheus-config-reloader:v0.91.0@sha256:7d9e4eea5f1139e602508871f422b0116c60e87c662f3dcd234d5ab60cd0d8c1"}},
|
|
]
|
|
|
|
def test_der_beiwagen_landet_in_der_zielmenge(self):
|
|
ziele = targets._images_aus_ergebnis(self.ECHTE_ANTWORT)
|
|
self.assertIn("quay.io/prometheus-operator/prometheus-config-reloader:v0.91.0", ziele)
|
|
self.assertIn("grafana/alloy:v1.16.3", ziele)
|
|
|
|
def test_der_nackte_digest_landet_NICHT_darin(self):
|
|
# Gegenprobe zur Zusicherung darueber: Ohne die Abweisung waere er als
|
|
# Repo "sha256" mit Tag "<hex>" durchgerutscht - ein Ziel, das sich nie
|
|
# ziehen laesst, das aber dauerhaft als Deckungsluecke zaehlt.
|
|
ziele = targets._images_aus_ergebnis(self.ECHTE_ANTWORT)
|
|
self.assertEqual(len(ziele), 2)
|
|
self.assertFalse([z for z in ziele if z.startswith("sha256:")])
|
|
self.assertIsNone(targets.normalisiere("sha256:" + "c9" * 32))
|
|
# ... aber ein Digest MIT Repository bleibt ein gueltiges Ziel.
|
|
self.assertEqual(targets.normalisiere("quay.io/x/y:v1@sha256:abc"), "quay.io/x/y:v1")
|
|
|
|
def test_ohne_image_spec_wird_image_genommen(self):
|
|
# cAdvisor kennt image_spec nicht - der Rueckfall muss tragen.
|
|
ziele = targets._images_aus_ergebnis(
|
|
[{"metric": {"image": "docker.io/library/nginx:1.31.4-alpine"}}])
|
|
self.assertEqual(ziele, {"nginx:1.31.4-alpine"})
|
|
|
|
def test_die_abfrage_holt_image_spec_ueberhaupt(self):
|
|
# ⚠️ Die Auswertung oben kann image_spec nur lesen, wenn die Abfrage es
|
|
# mitliefert. Ohne diese Zeile waere alles darueber gruen und trotzdem
|
|
# wirkungslos - die Fehlerklasse "meldet Erfolg, ist aber blind".
|
|
self.assertIn("image_spec", targets.AUSDRUCK_CLUSTER)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|