Thore Cimbal
27770b6ec7
The scan loop is untouched and still reads images.txt, so nothing about this deployment behaves differently. What changes is that the exporter now knows what *should* be scanned and can say how much of it is: three sources, none of them new infrastructure. The cluster and the operating host both already sit in the same Prometheus, and the registry answers an anonymous token — the same token dance Trivy performs to pull. The coverage numbers are the point of this slice. Once deployed they have to read 24 missing, 2 orphaned and about 0.52, because that is what was counted by hand on 2026-08-21. A different answer means the derivation is wrong, not the hand count stale. Failure handling is the substance rather than an afterthought. A source that fails costs only its own share; its freshness timestamp keeps ageing instead of disappearing, because a series that vanishes can never fire a rule — the third finding of the 2026-08-01 AAR. When every source fails the target file is left untouched, so a restart during a Prometheus outage cannot clear the estate. Twenty-one assertions cover it, each paired with its counter-proof: names that must collapse and names that must not, a source filter that is shown to matter by removing it, and time-ordered tag selection against the name ordering that would silently drop v0.10.0. Sabotaging the normalisation turns eleven of them red, so the suite demonstrably can fail.
298 lines
12 KiB
Python
298 lines
12 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 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.
|
|
self._stelle(set(), set(), set())
|
|
h = targets.herleiten(self.CFG)
|
|
self.assertEqual(h.ziele, set())
|
|
self.assertTrue(h.vollstaendig) # kein Fehler gemeldet ...
|
|
self.assertFalse(h.brauchbar) # ... aber trotzdem nicht verwertbar
|
|
|
|
|
|
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)
|