#!/usr/bin/env python3 """Offline unit tests for stack_status_lib (no network).""" from __future__ import annotations import json import os import sys import tempfile import time import unittest from pathlib import Path from unittest import mock ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "site-gen")) import stack_status_lib as L # noqa: E402 def _disable_extra_planes() -> None: """Keep offline unit tests off dig/TLS/catalog network work.""" L.SERVICES_ENABLED = False L.MAIL_ENABLED = False L.CERT_PROBES_ENABLED = False L.ORPHAN_PROBES_ENABLED = False L.MAIL_PORTS_ENABLED = False L.ORPHAN_HTTP_ENABLED = False L.PORT_SCAN_ENABLED = False L.DELIVER_ENABLED = False _disable_extra_planes() class AggregateTests(unittest.TestCase): def test_all_green_ok(self): members = [{"light": "green"}] * 3 light, intensity = L.aggregate_stack(members) self.assertEqual(light, "green") self.assertEqual(intensity, "ok") def test_taler_software_from_config(self): body = json.dumps( {"name": "taler-exchange", "build_version": "1.6.15-dev.1", "version": "39:0:5"} ).encode() items = L.software_from_taler_config(body, host="exchange.demo.taler.net", role="exchange") self.assertEqual(len(items), 1) self.assertEqual(items[0]["package"], "taler-exchange") self.assertEqual(items[0]["version"], "1.6.15-dev.1") colored = L.colorize_software( items + L.software_from_server_header("nginx/1.24.0", host="exchange.demo.taler.net"), vulns={"hits": [{"package": "nginx", "version": "1.24.0", "severity": "HIGH"}]}, ) by = {c["package"]: c for c in colored} self.assertEqual(by["taler-exchange"]["light"], "green") self.assertEqual(by["nginx"]["light"], "orange") def test_deliverability_stale_and_fail(self): now = 1_000_000.0 old_en = L.DELIVER_ENABLED L.DELIVER_ENABLED = True try: state = { "domains": { "example.test": { "ok": True, "light": "green", "reason": "rcpt_accepted", "ts": now - 100, "mx": "mx.example.test", "rcpt": "postmaster@example.test", } } } got = L.ensure_deliverability( "example.test", mx_hosts=["mx.example.test"], now=now, state=state, force=False ) self.assertTrue(got["ok"]) self.assertEqual(got["light"], "green") # stale success → yellow (prevent re-probe by raising interval) state["domains"]["example.test"]["ts"] = now - (L.DELIVER_STALE_S + 10) old_i = L.DELIVER_INTERVAL_S L.DELIVER_INTERVAL_S = L.DELIVER_STALE_S + 1000 try: got = L.ensure_deliverability( "example.test", mx_hosts=["mx.example.test"], now=now, state=state, force=False ) finally: L.DELIVER_INTERVAL_S = old_i self.assertFalse(got["ok"]) self.assertEqual(got["light"], "yellow") self.assertTrue(got["stale"]) # hard fail stored state["domains"]["example.test"] = { "ok": False, "light": "orange", "reason": "rcpt_reject:550", "ts": now - 60, "mx": "mx.example.test", "rcpt": "postmaster@example.test", } old_i = L.DELIVER_INTERVAL_S L.DELIVER_INTERVAL_S = 10_000 try: got = L.ensure_deliverability( "example.test", mx_hosts=["mx.example.test"], now=now, state=state, force=False ) finally: L.DELIVER_INTERVAL_S = old_i self.assertEqual(got["light"], "orange") self.assertIn("rcpt_reject", got["reason"]) finally: L.DELIVER_ENABLED = old_en def test_published_currency_behind(self): self.assertEqual(L.normalize_build_version("1.6.15-dev.1"), "1.6.15~dev1") pub = { "enabled": True, "packages": {"taler-merchant": "1.6.22-0+trixie", "taler-exchange": "1.6.15~dev1-0+trixie"}, } items = L.apply_published_currency( [ {"package": "taler-merchant", "version": "1.6.21", "light": "green"}, {"package": "taler-exchange", "version": "1.6.15-dev.1", "light": "green"}, ], pub, ) by = {i["package"]: i for i in items} self.assertEqual(by["taler-merchant"]["currency_status"], "behind") self.assertEqual(by["taler-merchant"]["behind"], "1.6.22") self.assertEqual(by["taler-exchange"]["currency_status"], "current") self.assertGreater(L.deb_version_cmp("1.6.22-0+trixie", "1.6.21-0+trixie"), 0) def test_port_whitelist_and_suspicious(self): rules = L.parse_ports_whitelist # noqa: ensure exists self.assertTrue(callable(rules)) wl = [{"pattern": "*.hacktivism.ch", "ports": [22, 8080, 9200]}, {"pattern": "*", "ports": [8080]}] self.assertIn(22, L.whitelist_ports_for_host("taler.hacktivism.ch", wl)) self.assertIn(9200, L.whitelist_ports_for_host("git.hacktivism.ch", wl)) self.assertIn(8080, L.whitelist_ports_for_host("anything.example", wl)) # Intentional koopa public ports must not be suspicious / orange koopa = L.assess_host_ports( "git.hacktivism.ch", ports=[22, 80, 443, 8080, 9200, 3306], whitelist_rules=wl, open_ports=[22, 80, 443, 8080, 9200], ) self.assertEqual(koopa["suspicious"], []) self.assertTrue(koopa["ok"]) self.assertEqual(koopa["light"], "green") got = L.assess_host_ports( "edge.example", ports=[22, 80, 443, 3306], whitelist_rules=[], open_ports=[80, 443, 3306], ) self.assertEqual([s["port"] for s in got["suspicious"]], [3306]) self.assertEqual(got["suspicious"][0]["severity"], "hard") self.assertEqual(got["light"], "orange") got2 = L.assess_host_ports( "edge.example", ports=[22, 80, 443], whitelist_rules=[{"pattern": "edge.example", "ports": [22]}], open_ports=[22, 443], ) self.assertEqual(got2["suspicious"], []) self.assertTrue(got2["ok"]) soft = L.assess_host_ports( "edge.example", ports=[22, 80, 443], whitelist_rules=[], open_ports=[22, 443], ) self.assertEqual(soft["suspicious"][0]["severity"], "soft") self.assertTrue(soft["ok"]) # soft does not fail ok def test_orphan_skips_siblings_and_wildcards(self): known = {"demo.taler.net", "backend.demo.taler.net"} orphans = L.collect_orphan_hosts( known_hosts=known, tls_sans=["*.demo.taler.net", "donations.demo.taler.net", "gone.example.invalid"], probe_fn=lambda url: {"ok": False, "code": 0, "error": "URLError"}, ) hosts = [o["host"] for o in orphans] self.assertNotIn("demo.taler.net", hosts) self.assertNotIn("donations.demo.taler.net", hosts) # sibling # gone.example.invalid may appear only if dig says no DNS — skip assert on network self.assertTrue(all(not h.startswith("*.") for h in hosts)) def test_total_score_thresholds(self): # 10 green of 10 → green s = L.compute_total_score([{"light": "green"}] * 10) self.assertEqual(s["green"], 10) self.assertEqual(s["total"], 10) self.assertEqual(s["light"], "green") # 9/10 → blue (≥0.85) s = L.compute_total_score([{"light": "green"}] * 9 + [{"light": "orange"}]) self.assertEqual(s["light"], "blue") # 6/10 → yellow s = L.compute_total_score([{"light": "green"}] * 6 + [{"light": "red"}] * 4) self.assertEqual(s["light"], "yellow") # 4/10 → orange s = L.compute_total_score([{"light": "green"}] * 4 + [{"light": "red"}] * 6) self.assertEqual(s["light"], "orange") # 1/10 → red s = L.compute_total_score([{"light": "green"}] + [{"light": "orange"}] * 9) self.assertEqual(s["light"], "red") # 0 green mixed → white (gone) s = L.compute_total_score([{"light": "red"}, {"light": "orange"}]) self.assertEqual(s["green"], 0) self.assertEqual(s["light"], "white") # all black → black s = L.compute_total_score([{"light": "black"}] * 3) self.assertEqual(s["light"], "black") def test_breakdown_counts(self): members = [ {"role": "bank", "host": "a", "light": "green", "reason": "ok"}, {"role": "exchange", "host": "b", "light": "green", "reason": "ok"}, {"role": "merchant", "host": "c", "light": "orange", "reason": "http_501"}, {"role": "bank", "host": "IBAN", "light": "blue", "reason": "ebics_iban"}, ] bd = L.aggregate_breakdown(members) self.assertEqual(bd["by_light"].get("green"), 2) self.assertEqual(bd["by_light"].get("orange"), 1) self.assertNotIn("blue", bd["by_light"]) # skipped informational self.assertEqual(bd["worst"], "orange") self.assertEqual(bd["counted"], 3) def test_blue_present_ok(self): members = [{"light": "green"}, {"light": "blue"}, {"light": "blue"}] light, intensity = L.aggregate_stack(members) self.assertEqual(light, "blue") self.assertEqual(intensity, "ok") def test_one_yellow_muted(self): members = [ {"light": "green"}, {"light": "yellow"}, {"light": "green"}, ] light, intensity = L.aggregate_stack(members) self.assertEqual(light, "yellow") self.assertEqual(intensity, "muted") def test_one_red_muted(self): members = [ {"light": "green"}, {"light": "red"}, {"light": "green"}, ] light, intensity = L.aggregate_stack(members) self.assertEqual(light, "red") self.assertEqual(intensity, "muted") def test_all_red_full(self): members = [{"light": "red"}] * 3 light, intensity = L.aggregate_stack(members) self.assertEqual(light, "red") self.assertEqual(intensity, "full") def test_all_black_full(self): members = [{"light": "black"}] * 3 light, intensity = L.aggregate_stack(members) self.assertEqual(light, "black") self.assertEqual(intensity, "full") def test_mixed_orange_green_muted(self): members = [ {"light": "orange"}, {"light": "green"}, {"light": "green"}, ] light, intensity = L.aggregate_stack(members) self.assertEqual(light, "orange") self.assertEqual(intensity, "muted") def test_all_yellow_still_muted(self): # Full punch only for red/black total outage members = [{"light": "yellow"}] * 3 light, intensity = L.aggregate_stack(members) self.assertEqual(light, "yellow") self.assertEqual(intensity, "muted") def test_weakest_wins_black_over_red(self): members = [ {"light": "red"}, {"light": "black"}, {"light": "green"}, ] light, intensity = L.aggregate_stack(members) self.assertEqual(light, "black") self.assertEqual(intensity, "muted") class ClassifyTests(unittest.TestCase): def test_ok_green(self): c = L.classify_probe({"ok": True, "code": 200, "elapsed": 0.2, "error": None}, now=1000.0) self.assertEqual(c["light"], "green") self.assertIsNone(c["down_since"]) def test_slow_yellow(self): c = L.classify_probe({"ok": True, "code": 200, "elapsed": 3.0, "error": None}, now=1000.0) self.assertEqual(c["light"], "yellow") def test_acute_red(self): c = L.classify_probe( {"ok": False, "code": 0, "elapsed": 1.0, "error": "URLError"}, prev={"ok": True}, now=1000.0, acute_s=90, ) self.assertEqual(c["light"], "red") self.assertEqual(c["down_since"], 1000.0) def test_long_black(self): c = L.classify_probe( {"ok": False, "code": 0, "elapsed": 1.0, "error": "URLError"}, prev={"ok": False, "down_since": 1000.0}, now=1200.0, acute_s=90, ) self.assertEqual(c["light"], "black") self.assertEqual(c["down_since"], 1000.0) def test_timeout_orange_when_fresh(self): c = L.classify_probe( {"ok": False, "code": 0, "elapsed": 5.0, "error": "TimeoutError"}, prev={"ok": True}, now=1000.0, acute_s=90, ) self.assertEqual(c["light"], "orange") class EndpointTests(unittest.TestCase): def test_legal_missing_is_medium_orange(self): for label in ("privacy", "terms"): c = L.classify_endpoint_probe( {"ok": False, "code": 404, "elapsed": 0.1, "error": "http_404", "body": b""}, expect=(200,), severity="medium", now=1000.0, ) self.assertEqual(c["light"], "orange", label) self.assertFalse(c["expected"], label) def test_summarize_dots(self): eps = [ {"light": "green", "skipped": False}, {"light": "yellow", "skipped": False}, {"light": "blue", "skipped": True}, ] s = L.summarize_endpoints(eps) self.assertEqual(s["ok"], 1) self.assertEqual(s["soft"], 1) self.assertEqual(s["skipped"], 1) self.assertEqual(s["light"], "yellow") self.assertEqual(s["dots"], ["green", "yellow"]) # skips omitted from strip def test_probe_role_endpoints_intro_skip(self): calls = [] def probe(url: str): calls.append(url) return {"ok": True, "code": 200, "elapsed": 0.05, "error": None, "body": b"{}", "url": url} eps = L.probe_role_endpoints( "bank", "bank.example.test", has_intro=False, probe_fn=probe, now=1000.0, ) ids = {e["id"]: e for e in eps} self.assertTrue(ids["intro"]["skipped"]) self.assertEqual(ids["intro"]["reason"], "no_intro") self.assertFalse(ids["config"]["skipped"]) self.assertTrue(any(u.endswith("/config") for u in calls)) self.assertFalse(any("/intro/" in u for u in calls)) def test_custom_intro_stacks_only(self): self.assertTrue( L.stack_has_custom_intro({"currency": "GOA", "id": "hacktivism.ch", "kind": "taler"}) ) self.assertTrue( L.stack_has_custom_intro( {"currency": "TESTPAYSAN", "id": "stage.lefrancpaysan.ch", "kind": "taler"} ) ) self.assertFalse( L.stack_has_custom_intro({"currency": "KUDOS", "id": "demo.taler.net", "kind": "taler"}) ) self.assertFalse( L.stack_has_custom_intro({"currency": "CHF", "id": "taler-ops.ch", "kind": "taler"}) ) self.assertFalse( L.stack_has_custom_intro( {"currency": "", "id": "lefrancpaysan.ch", "kind": "static_pending"} ) ) def test_server_header_parse(self): self.assertEqual(L.guess_http_package("nginx/1.22.1"), "nginx") self.assertEqual(L.extract_http_version("nginx/1.22.1"), "1.22.1") self.assertEqual(L.guess_http_package("Caddy"), "caddy") def test_vuln_raises_box_colour(self): members = [ {"role": "exchange", "host": "ex.example", "light": "green", "reason": "ok"}, {"role": "merchant", "host": "m.example", "light": "green", "reason": "ok"}, ] vulns = { "enabled": True, "worst": "CRITICAL", "light": "red", "hits": [ { "id": "CVE-2099-0001", "severity": "CRITICAL", "package": "nginx", "version": "1.0.0", "host": "ex.example", } ], } light, intensity, members2 = L.apply_vuln_to_stack("green", "ok", members, vulns) self.assertEqual(light, "red") self.assertEqual(intensity, "muted") self.assertEqual(members2[-1]["role"], "cve") def test_network_zones_and_ssh(self): self.assertEqual( L.classify_network_zone( {"id": "hacktivism.ch", "currency": "GOA", "local": True, "kind": "taler"} )["zone"], "goa-local", ) self.assertEqual( L.classify_network_zone( {"id": "demo.taler.net", "currency": "KUDOS", "local": False, "kind": "taler"} )["zone"], "public-demo", ) targets = L.ssh_targets_for_stack( { "id": "hacktivism.ch", "currency": "GOA", "local": True, "kind": "taler", "merchant": "taler.hacktivism.ch", } ) self.assertTrue(targets["expected"]) self.assertTrue(any("hacktivism" in h for h in targets["hosts"])) def fake_ssh(host: str): return { "ok": True, "light": "green", "reason": "SSH-2.0-OpenSSH_test", "host": host, "elapsed": 0.05, "banner": "SSH-2.0-OpenSSH_test", } net = L.build_network_status( { "id": "hacktivism.ch", "currency": "GOA", "local": True, "kind": "taler", "merchant": "taler.hacktivism.ch", }, members=[{"light": "green", "reason": "ok"}], ssh_probe_fn=fake_ssh, ) self.assertEqual(net["zone"], "goa-local") self.assertTrue(net["ssh"]["ok"]) self.assertIn("hacktivism", net["ssh"]["host"]) def test_tops_stage_ssh_is_env_mapped_hidden(self): st = { "id": "stage.taler-ops.ch", "currency": "CHF", "kind": "taler", "local": False, "merchant": "stage.my.taler-ops.ch", "exchange": "exchange.stage.taler-ops.ch", } self.assertEqual(L.classify_network_zone(st)["zone"], "tops-stage") with mock.patch.dict( os.environ, {"TALER_MONITORD_SSH_TOPS": "betel.taler-systems.com"} ): targets = L.ssh_targets_for_stack(st) self.assertEqual(targets["hosts"], ["betel.taler-systems.com"]) self.assertTrue(targets.get("hide_host")) self.assertNotIn("stage.taler-ops.ch", targets["hosts"]) def fake_ssh(host: str): self.assertEqual(host, "betel.taler-systems.com") return { "ok": True, "light": "green", "reason": "SSH-2.0-OpenSSH_test", "host": host, "elapsed": 0.05, "banner": "SSH-2.0-OpenSSH_test", } net = L.build_network_status( st, members=[{"light": "green", "reason": "ok"}], ssh_probe_fn=fake_ssh ) self.assertTrue(net["ssh"]["ok"]) # UI/JSON must not advertise env-mapped host self.assertEqual(net["ssh"]["host"], "") self.assertEqual(net["ssh"]["candidates"], []) def test_tops_stage_ssh_without_env_uses_public_fqdn(self): st = { "id": "stage.taler-ops.ch", "currency": "CHF", "kind": "taler", "local": False, "merchant": "stage.my.taler-ops.ch", "exchange": "exchange.stage.taler-ops.ch", } cleared = { k: "" for k in ( "TALER_MONITORD_SSH_TOPS", "TALER_MONITORD_SSH_FP", "TALER_MONITORD_SSH_GOA", "TALER_MONITORD_SSH_TOPS_PROD", ) } with mock.patch.dict(os.environ, cleared): targets = L.ssh_targets_for_stack(st) self.assertEqual(targets["hosts"][0], "stage.taler-ops.ch") self.assertFalse(targets.get("hide_host")) def test_fp_stage_ssh_is_env_mapped_hidden(self): st = { "id": "stage.lefrancpaysan.ch", "currency": "TESTPAYSAN", "kind": "taler", "local": False, "merchant": "stage.monnaie.lefrancpaysan.ch", "exchange": "stage.exchange.lefrancpaysan.ch", "bank": "stage.bank.lefrancpaysan.ch", } self.assertEqual(L.classify_network_zone(st)["zone"], "fp-stage") with mock.patch.dict( os.environ, {"TALER_MONITORD_SSH_FP": "ov-1a5ee8.infomaniak.ch"} ): targets = L.ssh_targets_for_stack(st) self.assertEqual(targets["hosts"], ["ov-1a5ee8.infomaniak.ch"]) self.assertTrue(targets.get("hide_host")) self.assertNotIn("lefrancpaysan", " ".join(targets["hosts"])) def fake_ssh(host: str): self.assertEqual(host, "ov-1a5ee8.infomaniak.ch") return { "ok": True, "light": "green", "reason": "SSH-2.0-OpenSSH_test", "host": host, "elapsed": 0.05, "banner": "SSH-2.0-OpenSSH_test", } net = L.build_network_status( st, members=[{"light": "green", "reason": "ok"}], ssh_probe_fn=fake_ssh ) self.assertTrue(net["ssh"]["ok"]) self.assertEqual(net["ssh"]["host"], "") self.assertEqual(net["ssh"]["candidates"], []) def test_load_monitoring_env_expands_home(self): with tempfile.TemporaryDirectory() as td: p = Path(td) / "env" p.write_text( "TALER_MONITORD_SSH_TOPS=betel.example\n" "SUITE_DIR=$HOME/git/taler-monitoring\n" "# comment\n" "export TALER_MONITORD_SSH_FP=fp.example\n", encoding="utf-8", ) with mock.patch.dict(os.environ, {}, clear=False): for k in ( "TALER_MONITORD_SSH_TOPS", "TALER_MONITORD_SSH_FP", "SUITE_DIR", ): os.environ.pop(k, None) got = L.load_monitoring_env(path=p) self.assertEqual(got, p) self.assertEqual(os.environ["TALER_MONITORD_SSH_TOPS"], "betel.example") self.assertEqual(os.environ["TALER_MONITORD_SSH_FP"], "fp.example") self.assertTrue(os.environ["SUITE_DIR"].endswith("/git/taler-monitoring")) def test_commercial_prod_tops_only(self): self.assertTrue( L.stack_is_commercial_prod( {"id": "taler-ops.ch", "currency": "CHF", "kind": "taler", "local": False} ) ) self.assertEqual( L.legal_doc_severity( {"id": "taler-ops.ch", "currency": "CHF", "kind": "taler", "local": False} ), "medium", ) self.assertFalse( L.stack_is_commercial_prod( {"id": "stage.taler-ops.ch", "currency": "CHF", "kind": "taler", "local": False} ) ) self.assertEqual( L.legal_doc_severity( {"id": "stage.taler-ops.ch", "currency": "CHF", "kind": "taler", "local": False} ), "soft", ) self.assertFalse( L.stack_is_commercial_prod( {"id": "hacktivism.ch", "currency": "GOA", "kind": "taler", "local": True} ) ) self.assertFalse( L.stack_is_commercial_prod( {"id": "demo.taler.net", "currency": "KUDOS", "kind": "taler", "local": False} ) ) class CurrencySpecTests(unittest.TestCase): def test_summarize_chf(self): cs = { "name": "Swiss francs", "currency": "CHF", "alt_unit_names": {"0": "Fr.", "-2": "Rp."}, "common_amounts": ["CHF:5", "CHF:10", "CHF:25", "CHF:50"], } info = L.summarize_currency_spec(cs) self.assertEqual(info["currency"], "CHF") self.assertEqual(info["symbol"], "Fr.") self.assertEqual(info["units_label"], "Fr. / Rp.") self.assertEqual(info["denoms"], ["5", "10", "25", "50"]) def test_denoms_from_keys_when_common_empty(self): cs = {"currency": "FLODOS", "alt_unit_names": {"0": "FLODOS"}, "common_amounts": []} body = b'{"denominations":[{"value":"FLODOS:1"},{"value":"FLODOS:5"},{"value":"FLODOS:0"}]}' info = L.summarize_currency_spec(cs, keys_body=body) self.assertEqual(info["denoms"], ["1", "5"]) class WireBackendTests(unittest.TestCase): def test_payto_methods(self): blob = b'{"accounts":[{"payto_uri":"payto://iban/CH9300762011623852957"},{"payto_uri":"payto://x-taler-bank/bank.example/ex"}]}' self.assertEqual(L.payto_wire_methods(blob), {"iban", "x-taler-bank"}) def test_classify_ebics_iban_only(self): self.assertEqual(L.classify_bank_backend({"iban"}), "ebics") def test_classify_xtaler_wins_mixed(self): self.assertEqual(L.classify_bank_backend({"iban", "x-taler-bank"}), "x-taler-bank") def test_classify_conf_skip(self): self.assertEqual(L.classify_bank_backend(set(), bank_configured=False), "ebics") self.assertFalse(L.bank_host_configured("-")) self.assertFalse(L.bank_host_configured("ebics")) self.assertTrue(L.bank_host_configured("bank.example.test")) class ParseAndBuildTests(unittest.TestCase): def setUp(self): _disable_extra_planes() def test_parse_stage_fp(self): conf = ROOT / "domains.conf" stacks = L.parse_domains_conf(conf) ids = {s["id"] for s in stacks} self.assertIn("stage.lefrancpaysan.ch", ids) stage = next(s for s in stacks if s["id"] == "stage.lefrancpaysan.ch") self.assertEqual(stage["bank"], "stage.bank.lefrancpaysan.ch") self.assertEqual(stage["merchant"], "stage.monnaie.lefrancpaysan.ch") self.assertEqual(stage["currency"], "TESTPAYSAN") self.assertEqual(stage["kind"], "taler") prod = next(s for s in stacks if s["id"] == "lefrancpaysan.ch") self.assertEqual(prod["kind"], "static_pending") self.assertEqual(prod["currency"], "") def test_ebics_skips_bank_probe(self): stacks = [ { "id": "tops.example", "bank": "bank.tops.example", "exchange": "exchange.tops.example", "merchant": "my.tops.example", "currency": "CHF", "kind": "taler", "local": False, } ] probed = [] def probe(url: str): probed.append(url) if "/keys" in url: body = b'{"accounts":[{"payto_uri":"payto://iban/CH6808573105529100001?receiver-name=X"}]}' return {"ok": True, "code": 200, "elapsed": 0.1, "error": None, "body": body, "url": url} if "bank." in url: return {"ok": False, "code": 0, "elapsed": 0.1, "error": "URLError", "body": b"", "url": url} return {"ok": True, "code": 200, "elapsed": 0.1, "error": None, "body": b"{}", "url": url} doc = L.build_stack_status(stacks, probe_fn=probe, now=5000.0, public_only=False) st = doc["stacks"][0] self.assertEqual(st["bank_backend"], "ebics") self.assertEqual(st["light"], "green") self.assertEqual(st["intensity"], "ok") roles = {m["role"]: m for m in st["members"]} self.assertEqual(roles["bank"]["reason"], "ebics_iban") self.assertEqual(roles["bank"]["light"], "blue") self.assertEqual(roles["bank"]["host"], "IBAN") self.assertTrue(all("bank.tops" not in u for u in probed)) def test_xtaler_bank_still_probed(self): stacks = [ { "id": "demo.example", "bank": "bank.demo.example", "exchange": "exchange.demo.example", "merchant": "backend.demo.example", "currency": "KUDOS", "kind": "taler", "local": False, } ] def probe(url: str): if "/keys" in url: body = b'{"accounts":[{"payto_uri":"payto://x-taler-bank/bank.demo.example/exchange"}]}' return {"ok": True, "code": 200, "elapsed": 0.1, "error": None, "body": body, "url": url} if "bank." in url: return {"ok": False, "code": 0, "elapsed": 0.1, "error": "URLError", "body": b"", "url": url} return {"ok": True, "code": 200, "elapsed": 0.1, "error": None, "body": b"{}", "url": url} doc = L.build_stack_status(stacks, probe_fn=probe, now=5000.0, public_only=False) st = doc["stacks"][0] self.assertEqual(st["bank_backend"], "x-taler-bank") self.assertEqual(st["light"], "red") self.assertEqual(st["intensity"], "muted") roles = {m["role"]: m for m in st["members"]} self.assertEqual(roles["bank"]["light"], "red") def test_domains_conf_bank_dash(self): with tempfile.TemporaryDirectory() as td: conf = Path(td) / "domains.conf" conf.write_text( "ebics.example - exchange.ebics.example my.ebics.example CHF 0 0 en de-CH\n", encoding="utf-8", ) stacks = L.parse_domains_conf(conf) self.assertEqual(len(stacks), 1) self.assertFalse(stacks[0]["bank_configured"]) self.assertEqual(stacks[0]["bank"], "") def test_landing_stats_summary(self): stacks = [ { "id": "stage.example", "bank": "bank.stage.example", "exchange": "exchange.stage.example", "merchant": "monnaie.stage.example", "currency": "TESTPAYSAN", "kind": "taler", "local": False, } ] stats_body = json.dumps( { "ok": True, "currency": "TESTPAYSAN", "generated_at_unix": 5000.0 - 120, "generated_at_human": "just now", "source": "stage-shared", "bank_accounts": {"total": 36, "users": 34}, "withdraws": {"count": 45, "total_amount": "TESTPAYSAN:12"}, } ).encode() cfg_body = json.dumps( { "currency": "TESTPAYSAN", "currency_specification": { "name": "TESTPAYSAN stage currency", "currency": "TESTPAYSAN", "alt_unit_names": {"0": "TESTPAYSAN", "-2": "ct"}, "common_amounts": [ "TESTPAYSAN:5", "TESTPAYSAN:10", "TESTPAYSAN:20", "TESTPAYSAN:50", "TESTPAYSAN:100", ], }, } ).encode() def probe(url: str): if url.endswith("/intro/stats.json") and "bank.stage" in url: return { "ok": True, "code": 200, "elapsed": 0.05, "error": None, "body": stats_body, "url": url, } if "/keys" in url: body = b'{"accounts":[{"payto_uri":"payto://x-taler-bank/bank.stage.example/exchange"}]}' return {"ok": True, "code": 200, "elapsed": 0.1, "error": None, "body": body, "url": url} if url.endswith("/config") and "exchange.stage" in url: return { "ok": True, "code": 200, "elapsed": 0.05, "error": None, "body": cfg_body, "url": url, } return {"ok": True, "code": 200, "elapsed": 0.1, "error": None, "body": b"{}", "url": url} doc = L.build_stack_status(stacks, probe_fn=probe, now=5000.0, public_only=False) st = doc["stacks"][0] self.assertIsNotNone(st.get("stats")) self.assertTrue(st["stats"]["present"]) self.assertEqual(st["stats"]["age_s"], 120) self.assertIn("accounts 36", st["stats"]["lines"]) self.assertFalse(st["stats"]["stale"]) self.assertEqual(st["currency"], "TESTPAYSAN") self.assertEqual(st["currency_info"]["symbol"], "TESTPAYSAN") self.assertEqual(st["currency_info"]["units_label"], "TESTPAYSAN / ct") self.assertEqual(st["currency_info"]["denoms"], ["5", "10", "20", "50", "100"]) def test_static_pending_softens_api_down(self): stacks = [ { "id": "lefrancpaysan.ch", "bank": "bank.lefrancpaysan.ch", "exchange": "exchange.lefrancpaysan.ch", "merchant": "monnaie.lefrancpaysan.ch", "currency": "", "kind": "static_pending", "local": False, } ] def probe(url: str): if url.rstrip("/").endswith("lefrancpaysan.ch"): return {"ok": True, "code": 200, "elapsed": 0.1, "error": None, "url": url} return {"ok": False, "code": 502, "elapsed": 0.1, "error": "http_502", "url": url} doc = L.build_stack_status(stacks, probe_fn=probe, now=5000.0, public_only=False) st = doc["stacks"][0] self.assertEqual(st["kind"], "static_pending") self.assertEqual(st["light"], "blue") self.assertEqual(st["intensity"], "ok") roles = {m["role"]: m for m in st["members"]} self.assertEqual(roles["site"]["light"], "green") self.assertEqual(roles["bank"]["reason"], "no_prod_yet") self.assertEqual(roles["bank"]["light"], "blue") def test_build_with_mock_probe(self): stacks = [ { "id": "example.test", "bank": "bank.example.test", "exchange": "exchange.example.test", "merchant": "merch.example.test", "currency": "X", "local": False, } ] def probe(url: str): if "exchange" in url: return {"ok": False, "code": 0, "elapsed": 0.1, "error": "URLError", "url": url} return {"ok": True, "code": 200, "elapsed": 0.1, "error": None, "url": url} doc = L.build_stack_status(stacks, probe_fn=probe, now=5000.0, public_only=False) self.assertEqual(len(doc["stacks"]), 1) st = doc["stacks"][0] self.assertEqual(st["light"], "red") self.assertEqual(st["intensity"], "muted") self.assertIn("breakdown", st) self.assertEqual(doc.get("schema"), L.SCHEMA) def test_atomic_write_roundtrip(self): with tempfile.TemporaryDirectory() as td: path = Path(td) / "stack-status.json" doc = {"schema": 1, "stacks": []} L.atomic_write_json(path, doc) self.assertTrue(path.is_file()) got = L.read_status(path) self.assertEqual(got["schema"], 1) def test_parse_services_conf(self): with tempfile.TemporaryDirectory() as td: conf = Path(td) / "services-catalog.conf" conf.write_text( "bugs.gnunet.org bugs.gnunet.org MantisBT firefly\n" "erp.stage.taler-systems.com erp.stage.taler-systems.com ERPnext pixel\n", encoding="utf-8", ) rows = L.parse_services_conf(conf) self.assertEqual(len(rows), 2) self.assertEqual(rows[0]["service_label"], "MantisBT") self.assertEqual(rows[1]["zone"], "pixel") class HtmlRenderTests(unittest.TestCase): def test_degraded_why_and_cve_link(self): import stack_status_html as H doc = { "schema": L.SCHEMA, "ts": time.time(), "acute_down_s": 90, "stacks": [ { "id": "my.taler-ops.ch", "light": "orange", "intensity": "muted", "currency": "CHF", "bank_backend": "ebics", "kind": "taler", "breakdown": {"by_light": {"green": 4, "orange": 3}, "worst": "orange", "counted": 7}, "members": [ {"role": "bank", "host": "IBAN", "light": "blue", "reason": "ebics_iban"}, {"role": "exchange", "host": "exchange.taler-ops.ch", "light": "green", "reason": "ok"}, { "role": "merchant", "host": "my.taler-ops.ch", "light": "orange", "reason": "http_501", }, ], "vulns": { "enabled": True, "worst": "HIGH", "hits": [ { "id": "CVE-2024-1234", "severity": "HIGH", "package": "nginx", "version": "1.24.0", "summary": "test vuln", "host": "my.taler-ops.ch", } ], "servers": [ {"host": "my.taler-ops.ch", "package": "nginx", "version": "1.24.0", "server": "nginx/1.24.0"} ], }, "network": {"zone": "tops-prod", "ssh": {"expected": False}}, "tls": {"days_left": 60, "sans": ["my.taler-ops.ch"]}, } ], "services": [], "mail": [], "orphans": [], } html = H.render_html(doc) self.assertIn("4× OK", html) self.assertIn("3× Degraded", html) self.assertIn("osv.dev/vulnerability/CVE-2024-1234", html) self.assertIn("openStack", html) self.assertIn("aria-label=\"nginx\"", html) self.assertIn("0/1 green", html) # orange stack → not green self.assertIn("Gone", html) # or Degraded label in score — 0 green → white/Gone def test_git_ls_remote_soft_probe_mocked(self): calls: list[list[str]] = [] def fake_run(cmd, **kwargs): calls.append(list(cmd)) class R: returncode = 0 stdout = b"abc123\trefs/heads/main\n" stderr = b"" return R() with tempfile.TemporaryDirectory() as td: with mock.patch.dict( os.environ, {"TALER_MONITORD_STATE": td, "TALER_MONITORD_GIT": "1"} ): L.refresh_runtime_flags_from_env() with mock.patch("stack_status_lib.subprocess.run", side_effect=fake_run): got = L.collect_git_probe("git.example", now=1000.0, force=True) self.assertTrue(got["ok"]) self.assertEqual(got["light"], "green") self.assertGreaterEqual(len(calls), 2) # https + ssh n = len(calls) with mock.patch("stack_status_lib.subprocess.run", side_effect=fake_run): cached = L.collect_git_probe("git.example", now=1100.0, force=False) self.assertTrue(cached["ok"]) self.assertEqual(len(calls), n) # interval cache hit def test_result_index_optional_by_colour_and_cmd(self): import stack_status_html as H doc = { "schema": L.SCHEMA, "ts": time.time(), "acute_down_s": 90, "stacks": [ { "id": "stage.example", "light": "yellow", "intensity": "muted", "currency": "CHF", "kind": "taler", "members": [ { "role": "exchange", "host": "exchange.stage.example", "light": "green", "reason": "ok", "endpoints": [ { "id": "config", "path": "/config", "url": "https://exchange.stage.example/config", "light": "green", "reason": "ok", "skipped": False, }, { "id": "keys", "path": "/keys", "url": "https://exchange.stage.example/keys", "light": "yellow", "reason": "slow", "skipped": False, }, ], }, { "role": "bank", "host": "IBAN", "light": "blue", "reason": "ebics_iban", }, ], "network": { "zone": "tops-stage", "ssh": { "expected": True, "ok": True, "light": "green", "host": "", "candidates": [], "reason": "SSH-2.0-OpenSSH_test", }, }, } ], "services": [], "mail": [], } off = H.render_html(doc) self.assertNotIn('id="result-index"', off) colors = H._render_result_index(doc, mode="colors") self.assertIn('id="result-index"', colors) self.assertIn("all probes by colour", colors) self.assertIn("exchange/config", colors) self.assertIn("exchange/keys", colors) self.assertIn("bank", colors) self.assertIn("ssh", colors) self.assertNotIn("ri-cmd", colors) cmd = H._render_result_index(doc, mode="cmd") self.assertIn("with command", cmd) self.assertIn("GET https://exchange.stage.example/config", cmd) self.assertIn("GET https://exchange.stage.example/keys", cmd) self.assertIn("TCP", cmd) if __name__ == "__main__": unittest.main()