Files
threadnet-operating/monitoring/cve/test_targets.py
T
Thore Cimbal 9a3b1d1dd5 cve: a successful query can still be worthless, and it cost 42 reports
Restarting k3s took kube-state-metrics and alloy's log tailers down with it, so
kube_pod_container_info went empty in prometheus. The derivation asked, got a
clean response with zero rows, and counted it as success. Cluster targets went
from 39 to none, the total from 54 to 12, and the scan loop deleted every report
whose target had vanished. Coverage then read 1.0.

None of the four rules fired, and each for a defensible reason: the set was not
empty because the operating host still answered, and every timestamp was fresh
because an empty success updates it. The gap sat exactly between them.

A source that has delivered before and now delivers nothing is treated as a
failure: its previous targets are kept, its timestamp ages, and the stale rule
takes over. A fifth rule watches the total for a drop of more than 40 percent,
and it deliberately also fires on a deliberate shrink — losing 40 percent of the
checked estate is worth a line either way.

The six rancher decisions are gone too. They described versions that no longer
run: the k3s patch took all fifteen of their criticals with it.
2026-08-21 12:00:00 +00:00

467 lines
21 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 LeereAntwortIstKeinErfolg(unittest.TestCase):
"""⚠️ Nachgetragen am 2026-08-23, nachdem genau dieser Fall die Zielmenge
von 54 auf 12 fallen liess und der Scanner 42 Berichte loeschte.
Der k3s-Neustart legte kube-state-metrics und Alloy lahm; die
Prometheus-Abfrage lief daraufhin sauber durch und lieferte NULL Zeilen.
Das galt als Erfolg: Zeitstempel frisch, Menge nicht leer (der Betriebs-Host
lieferte ja), also schlug keine der drei Regeln an. Deckung 1,0, waehrend
drei Viertel des Bestands unbeobachtet waren.
"""
CFG = {"prometheus": "http://x", "registry": "r", "je_repo": 3, "timeout": 1}
def herleite(self, antworten, vorher=None, stand=None):
echt = (targets.ziele_cluster, targets.ziele_operating, targets.ziele_registry)
targets.ziele_cluster = lambda *a, **k: antworten["cluster"]
targets.ziele_operating = lambda *a, **k: antworten["operating"]
targets.ziele_registry = lambda *a, **k: antworten["registry"]
try:
return targets.herleiten(self.CFG, vorheriger_stand=stand, vorherige_ziele=vorher)
finally:
targets.ziele_cluster, targets.ziele_operating, targets.ziele_registry = echt
VORHER = {"cluster": {"a:1", "b:1"}, "operating": {"c:1"}, "registry": {"d:1"}}
def test_leere_antwort_behaelt_den_alten_stand(self):
h = self.herleite({"cluster": set(), "operating": {"c:1"}, "registry": {"d:1"}},
vorher=self.VORHER, stand={"cluster": 100.0})
self.assertEqual(h.je_quelle["cluster"], {"a:1", "b:1"}) # NICHT verloren
self.assertIn("cluster", h.fehler)
self.assertEqual(h.stand["cluster"], 100.0) # Zeitstempel altert
def test_gegenprobe_eine_ECHT_leere_erstmenge_ist_kein_fehler(self):
# Ohne Vorwissen ist eine leere Antwort schlicht eine leere Antwort -
# sonst koennte die Anlage nie sauber anlaufen.
h = self.herleite({"cluster": set(), "operating": {"c:1"}, "registry": {"d:1"}},
vorher=None, stand=None)
self.assertEqual(h.je_quelle["cluster"], set())
self.assertNotIn("cluster", h.fehler)
def test_gegenprobe_eine_gefuellte_antwort_ersetzt_den_alten_stand(self):
h = self.herleite({"cluster": {"neu:1"}, "operating": {"c:1"}, "registry": {"d:1"}},
vorher=self.VORHER, stand={"cluster": 100.0})
self.assertEqual(h.je_quelle["cluster"], {"neu:1"})
self.assertNotIn("cluster", h.fehler)
self.assertGreater(h.stand["cluster"], 100.0)
def test_ein_ausfall_verliert_die_ziele_der_quelle_NICHT(self):
def kaputt(*a, **k):
raise RuntimeError("Prometheus weg")
echt = targets.ziele_cluster
targets.ziele_cluster = kaputt
try:
h = targets.herleiten(self.CFG, vorheriger_stand={"cluster": 100.0},
vorherige_ziele=self.VORHER)
finally:
targets.ziele_cluster = echt
self.assertEqual(h.je_quelle["cluster"], {"a:1", "b:1"})
self.assertIn("cluster", h.fehler)
def test_die_gesamtmenge_bricht_dadurch_nicht_ein(self):
# Der eigentliche Schaden war die SUMME: 54 -> 12. Genau die darf nicht
# mehr passieren, solange die Quelle schon einmal geliefert hat.
h = self.herleite({"cluster": set(), "operating": {"c:1"}, "registry": set()},
vorher=self.VORHER, stand=None)
self.assertEqual(len(h.ziele), 4) # a,b,c,d - nichts verloren
h_ohne = self.herleite({"cluster": set(), "operating": {"c:1"}, "registry": set()},
vorher=None, stand=None)
self.assertEqual(len(h_ohne.ziele), 1) # Gegenprobe: ohne Vorwissen sehr wohl
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)