services/dashboards
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
0271387053fa766fc344b8112f945ed6f614e6fc
dashboards/tests/test_gateway_source.py
T
ducoterra 51772ffef2 phase: 32_gateway_resolved_target_liveness 
Fixture and live stream.conf are byte-identical (modulo header comment). Let me test the parser directly:
2026-10-26 14:03:42 -04:00

2135 lines
124 KiB
Python
Raw Blame History

"""`sources/gateway.py`: the two config files plus the socket table, joined into a connection map.
The shape half of these tests runs the **captured** gateway configs and `/proc` tables through
the real agent `Collector` (the `gateway_snapshot` fixture), so a rename of a section key on
either side of the wire fails here rather than drawing an empty pipe on a live page. The
arithmetic half uses synthetic sections, because the point of those tests is which socket lands
on which pipe and what the source refuses to decide — facts a capture states once and a
re-capture may lawfully change (the same rule `tests/test_agent.py` holds for the socket tables).
Phase 30 task 02 replaced the guessing join with two sound attributions — a **local port** for
inbound, an exact **(address, port)** for outbound — and deleted the rest. So the tests for
`by_port`, for `unmatched`, for the ambiguous-fan-out group list and for the flat `outbound` pipe
list are gone with the code they described, and what replaces them pins the arithmetic the new
join still owes: a socket counted **at most** once, a total that never exceeds the kernel's, and an
endpoint shared by several lines counted once. There is deliberately **no** test that a connection
outside the two configs is *reported*, because it no longer is —
`test_a_socket_neither_config_describes_is_counted_nowhere` pins the silence instead.
"""
from __future__ import annotations
import ipaddress
import json
import socket as socket_module
from collections.abc import Mapping, Sequence
from pathlib import Path
from typing import Any
from dashboards.sources import gateway
#: This repo's recorded captures, for the pins that read a fixture directly.
FIXTURES = Path(__file__).resolve().parent / "fixtures"
# --------------------------------------------------------------------------- #
# synthetic builders: a section exactly as the agent serves it, minus the file
# --------------------------------------------------------------------------- #
def bind(port: int | None, addr: str = "::") -> dict[str, Any]:
return {"addr": addr, "port": port}
def haproxy_section(frontends: Sequence[dict[str, Any]],
backends: Sequence[dict[str, Any]] = (), *,
available: bool | None = True) -> dict[str, Any]:
return {"available": available, "config": "/etc/haproxy/haproxy.cfg",
"frontends": list(frontends), "backends": list(backends)}
def frontend(name: str, ports: Sequence[int | None], *, default: str | None = None,
use: Sequence[str] = ()) -> dict[str, Any]:
return {"name": name, "binds": [bind(port) for port in ports],
"default_backend": default, "backends": list(use)}
def backend(name: str, servers: Sequence[dict[str, Any]]) -> dict[str, Any]:
return {"name": name, "servers": list(servers)}
def server(name: str, target: str | None, port: int | None = None) -> dict[str, Any]:
return {"name": name, "target": target, "port": port}
def nginx_section(servers: Sequence[dict[str, Any]] = (), maps: Sequence[dict[str, Any]] = (), *,
upstreams: Sequence[dict[str, Any]] = (),
available: bool | None = True) -> dict[str, Any]:
return {"available": available, "config": "/etc/nginx/nginx.conf", "files": [],
"servers": list(servers), "upstreams": list(upstreams), "maps": list(maps)}
def upstream(name: str, members: Sequence[tuple[str, int | None]]) -> dict[str, Any]:
"""An `upstream <name> { server …; }` block, as the agent publishes it (`{host, port}`)."""
return {"name": name, "servers": [{"host": host, "port": port} for host, port in members]}
def pool_pass(name: str) -> dict[str, Any]:
"""`proxy_pass http://<name>;` — a target with **no** port, which is what makes it a pool."""
return {"location": "/", "scheme": "http", "target": {"host": name, "port": None},
"variable": None}
def nginx_server(ports: Sequence[int | None], names: Sequence[str] = (),
passes: Sequence[dict[str, Any]] = ()) -> dict[str, Any]:
return {"listen": [bind(port, addr="") for port in ports], "server_names": list(names),
"proxy_passes": list(passes)}
def proxy_pass(target: tuple[str, int] | None, *, variable: str | None = None,
location: str | None = "/") -> dict[str, Any]:
host, port = target if target else (None, None)
return {"location": location, "scheme": "http" if target else None,
"target": None if host is None else {"host": host, "port": port},
"variable": variable}
def mapping_block(variable: str,
entries: Sequence[tuple[str, tuple[str, int | None] | None]]) -> dict[str, Any]:
return {"variable": variable,
"entries": [{"key": key,
"target": None if target is None else {"host": target[0],
"port": target[1]}}
for key, target in entries]}
def sockets_section(listens: Sequence[dict[str, Any]] = (),
established: Sequence[dict[str, Any]] = (), *,
available: bool | None = True) -> dict[str, Any]:
return {"available": available, "listens": list(listens), "established": list(established)}
def listen(port: int, addr: str = "0.0.0.0", family: str = "v4") -> dict[str, Any]:
return {"family": family, "addr": addr, "port": port}
def conn(local: tuple[str, int], remote: tuple[str, int], *,
family: str = "v4") -> dict[str, Any]:
return {"family": family, "local": {"addr": local[0], "port": local[1]},
"remote": {"addr": remote[0], "port": remote[1]}}
def resolver_section(entries: Sequence[dict[str, Any]] = (), *, available: bool | None = True,
via: str | None = "libc", config_directive: str | None = "10.1.0.1",
ts: float = 1_760_000_000.0,
errors: Sequence[str] = ()) -> dict[str, Any]:
"""Agent 1.3.0's fourth section, in the shape `host_agent.Collector.resolver` publishes."""
return {"available": available, "via": via, "config_directive": config_directive, "ts": ts,
"entries": list(entries), "errors": list(errors)}
def resolved(name: str, port: int | None, addresses: Sequence[str], *,
status: str = "ok", ts: float | None = 1_760_000_000.0) -> dict[str, Any]:
"""One resolver entry: what libc answered for `(name, port)`, and when."""
return {"name": name, "port": port, "status": status, "addresses": list(addresses), "ts": ts}
def snapshot(*, haproxy: dict[str, Any] | None = None, nginx: dict[str, Any] | None = None,
sockets: dict[str, Any] | None = None, resolver: dict[str, Any] | None = None,
errors: Sequence[str] = (),
agent_version: str | None = "1.2.0") -> dict[str, Any]:
"""The document one agent `/snapshot` would answer, with only what a test names.
No `resolver` key is the default, because that is the arm most of this file predates: a 1.2.0
agent, or a 1.3.0 one run with `--no-resolve`. A test that wants a name to be countable passes
`resolver=resolver_section([...])` and says so.
"""
doc: dict[str, Any] = {"ok": True, "schema": 1, "ts": 1_760_000_000.0, "host": "gateway",
"errors": list(errors)}
if agent_version is not None:
doc["agent_version"] = agent_version
if haproxy is not None:
doc["haproxy"] = haproxy
if nginx is not None:
doc["nginx"] = nginx
if sockets is not None:
doc["sockets"] = sockets
if resolver is not None:
doc["resolver"] = resolver
return doc
def listener_at(payload: Mapping[str, Any], port: int) -> dict[str, Any]:
"""The trunk row for one port, or a failure naming the ports the payload does have."""
for row in payload["listeners"]:
if row["port"] == port:
return row
raise AssertionError(f"no listener on port {port}: {[r['port'] for r in payload['listeners']]}")
def routes_to(payload: Mapping[str, Any], host: str) -> list[dict[str, Any]]:
"""Every route line whose outbound endpoint is `host`, in the graph's own order."""
return [row for row in payload["connections"] if row["out"]["host"] == host]
def endpoint_count(payload: Mapping[str, Any], host: str, port: int) -> Any:
"""The count the payload attributes to one **(address, port)** — once, not per line.
Asserts the invariant that makes `shared` meaningful while returning the number: every line
reaching one endpoint carries the *same* count, because that count belongs to the endpoint. A
join that split it across the lines would fail here rather than print a route carrying a
fraction nobody measured.
"""
rows = [row for row in payload["connections"]
if row["out"]["host"] == host and row["out"]["port"] == port]
assert rows, f"no route to {host}:{port} in the graph"
counts = {row["connections"] for row in rows}
assert len(counts) == 1, f"lines reaching one endpoint disagree: {counts}"
return counts.pop()
def out_metrics(payload: Mapping[str, Any]) -> list[str]:
"""The outbound gauge names the payload would publish — one per **distinct** endpoint."""
return [row["metric"] for row in payload["connections"] if row["metric"] is not None]
def endpoint_keys(row: Mapping[str, Any]) -> set[tuple[str, int]]:
"""The normalised `(address, port)` keys one route reaches, derived **in the test**.
Read straight off `out.address` / `out.resolved` with `ipaddress`, deliberately not by calling
`gateway.endpoint_of`: a total a test compares against must not come from the function under
test. The normalisation is the same one the join makes, because `::ffff:10.1.2.64` and
`10.1.2.64` are one endpoint and a ledger that counted them apart would reconcile to nothing.
"""
out = row["out"]
port = out.get("port")
if not isinstance(port, int):
return set()
spellings = [out["host"]] if out.get("address") is True else list(out.get("resolved") or [])
keys: set[tuple[str, int]] = set()
for spelling in spellings:
try:
address = ipaddress.ip_address(str(spelling))
except ValueError:
continue
mapped = getattr(address, "ipv4_mapped", None)
keys.add((str(mapped if mapped is not None else address), port))
return keys
def endpoint_groups(payload: Mapping[str, Any]) -> dict[tuple[str, int], list[Mapping[str, Any]]]:
"""The route lines grouped by the **distinct endpoint** they reach, keyed by min(key).
Lines whose key sets are identical share a group, which is every case a capture can produce
(one address per name, or one name written by two vhosts). `gateway._endpoint_groups` merges
*overlapping* sets as well, for a multi-homed name sharing one address with a single-homed one;
no capture has that shape, and if one ever does this helper disagrees by keeping two groups —
which fails the totals test below loudly rather than quietly.
"""
groups: dict[tuple[str, int], list[Mapping[str, Any]]] = {}
for row in payload["connections"]:
keys = endpoint_keys(row)
if keys:
groups.setdefault(min(keys), []).append(row)
return groups
def distinct_endpoint_totals(payload: Mapping[str, Any]) -> dict[tuple[str, int], Any]:
"""One count per distinct endpoint — the figure a total may add, never one per line."""
return {key: rows[0]["connections"] for key, rows in endpoint_groups(payload).items()}
def walk_keys(node: Any) -> set[str]:
"""Every key name anywhere in a payload — the device the "no severity / no window" pins use."""
if isinstance(node, Mapping):
found = set(node.keys())
for value in node.values():
found |= walk_keys(value)
return found
if isinstance(node, Sequence) and not isinstance(node, str | bytes):
return set().union(*(walk_keys(item) for item in node)) if node else set()
return set()
# --------------------------------------------------------------------------- #
# the captured gateway: every configured pipe, from the real agent code path
# --------------------------------------------------------------------------- #
def test_the_captured_configs_yield_every_configured_pipe(gateway_snapshot: Any) -> None:
"""Six haproxy frontends, four nginx listen ports, the stream map's fan-out — all present."""
payload = gateway.parse(gateway_snapshot("full"), "gateway")
assert payload["configured"] is True
assert payload["supported"] is True
assert payload["why"] is None, "a gateway whose every section answered gets no why-line"
assert payload["sections"] == {"haproxy": True, "nginx": True, "sockets": True}
haproxy_ports = [row["port"] for row in payload["haproxy"]["frontends"]]
assert haproxy_ports == [22, 2000, 5201, 8080, 25565, 25566], "config order, one row per bind"
assert {row["backend"] for row in payload["haproxy"]["frontends"]} == {
"gitea-2222", "sec-gitea-2000", "iperf3-5201", "unifi-external-8080",
"testcraft-25565", "nimcraft-25566"}
assert [row["port"] for row in payload["nginx"]["servers"]] == [80, 443, 4443, 8443]
names = {row["port"]: row["server_names"] for row in payload["nginx"]["servers"]}
assert names[80] == ["_"], "the catch-all is the config's own text, kept verbatim"
assert names[443] == [], "the stream server routes by SNI, so it names no domain of its own"
assert "nextcloud.reeseapps.com" in names[4443] and "gitea.reeseapps.com" in names[4443], (
"five vhosts share 4443 and their domains belong to the one pipe for that port")
assert payload["nginx"]["map_targets"], "the stream map is the fan-out the page draws"
assert len(payload["nginx"]["map_targets"]) == 18
homepage = next(row for row in payload["nginx"]["map_targets"]
if row["key"] == "homepage.reeseapps.com")
assert homepage == {"key": "homepage.reeseapps.com",
"host": "reeseapps-homepage.reeselink.com", "port": 443}
def test_the_source_reads_the_keys_the_agent_actually_publishes(gateway_snapshot: Any,
host_agent_module: Any) -> None:
"""The coupling is pinned: `binds`/`port`, `servers`/`target`, `listen`/`server_names`.
The fixture runs the real parser, so this compares the source's output against the agent's
own document rather than against a copy of its field names — which is what makes a rename
fail here instead of printing `n/a` beside a healthy gateway (AGENTS.md rule 7).
"""
snap = gateway_snapshot("full")
payload = gateway.parse(snap, "gateway")
agent_bind_ports = {bind["port"] for frontend in snap["haproxy"]["frontends"]
for bind in frontend["binds"]}
assert {row["port"] for row in payload["haproxy"]["frontends"]} == agent_bind_ports
agent_servers = {(row["name"], row["target"], row["port"])
for block in snap["haproxy"]["backends"] for row in block["servers"]}
assert {(row["out"]["name"], row["out"]["host"], row["out"]["port"])
for row in payload["connections"] if row["source"] == "haproxy"} == agent_servers
agent_listen_ports = {listen["port"] for server in snap["nginx"]["servers"]
for listen in server["listen"]}
assert {row["port"] for row in payload["nginx"]["servers"]} == agent_listen_ports
# and the three section names themselves, as the snapshot publishes them
assert {"haproxy", "nginx", "sockets"} <= set(snap)
# At least 1.2.0, the version that shipped these three keys. It is a floor and not an
# equality on purpose: phase 32 added a fourth section and bumped the constant, and the
# coupling this test pins is that the three the source joins still exist in what an agent
# publishes, whatever the number on it has since grown.
version = tuple(int(part) for part in host_agent_module.AGENT_VERSION.split("."))
assert version >= (1, 2, 0), "the sections these keys belong to"
def test_the_live_counts_reconcile_with_the_socket_table(gateway_snapshot: Any) -> None:
"""The arithmetic phase 29 task 06 pinned as an equality is now an **inequality**, on purpose.
`sum(listeners) + sum(distinct endpoint counts) <= established`. Two things took the equality
away and neither is a leak:
* a socket neither file describes is counted nowhere (the `unmatched` arm is deleted, not
hidden), and this gateway has plenty of that — the agent's own `:9120`, the outbound SSH and
NTP-style sessions, the loopback pairs;
* the endpoints are summed **once each**, so a line reaching an address another line already
reached adds nothing a second time.
What stays pinned is the direction of the error and the absence of a residual: the reported
figures are always **under** the kernel's number, and there is no field anywhere that quietly
absorbs the difference to make the two agree.
Run against the **`resolved`** capture, because that is what the deployed ingress serves since
phase 32 task 05: the widened match set is the one the arithmetic has to hold for. Three
figures, read three ways:
* `in_total` — the sum over the listeners (each socket at most once, inbound);
* `out_total` — the sum over **distinct endpoints**, derived here from `out.address` /
`out.resolved` rather than from `gateway.endpoint_of`;
* `lines_total` — the sum over the **lines**, which is deliberately larger wherever a line
repeats a number (`shared`), and is pinned to be exactly `out_total` plus the repetitions.
The `127.0.0.1:4443` loopback hop is the boundary case: it is booked **once inbound** (the rows
whose local port is 4443, on the `:4443` trunk) and **once outbound** (the rows whose remote is
4443, on the `:443 → 127.0.0.1:4443` pipe) and no single `/proc` row is in both.
"""
snap = gateway_snapshot("resolved")
payload = gateway.parse(snap, "gateway")
established = snap["sockets"]["established"]
configured = {row["port"] for row in payload["listeners"]}
assert configured == {22, 2000, 5201, 8080, 25565, 25566, 80, 443, 4443, 8443}
assert all(row["port"] in configured for row in payload["listeners"]), (
"an inbound count on a port no config names cannot exist in this payload")
in_total = sum(row["connections"] for row in payload["listeners"])
totals = distinct_endpoint_totals(payload)
out_total = sum(value or 0 for value in totals.values())
lines_total = sum(row["connections"] or 0 for row in payload["connections"])
repeated = sum((len(rows) - 1) * (rows[0]["connections"] or 0)
for rows in endpoint_groups(payload).values() if len(rows) > 1)
rows_in_table = len(established)
assert payload["sockets"] == {"listens": len(snap["sockets"]["listens"]),
"established": rows_in_table}
assert in_total + out_total <= rows_in_table
assert in_total + out_total < rows_in_table, (
"this capture has traffic outside both config files; if it ever stops having any, this "
"test is proving nothing and should be re-written against a capture that does")
assert "unmatched" not in payload and "partial" not in payload
assert lines_total == out_total + repeated, (
"the only difference between the lines and the endpoints is a repeated count — no socket "
"is booked twice and no endpoint is booked once per name that reaches it")
# the loopback hop, both sides, from the raw table rather than from either counter
accepted = [row for row in established if (row["local"]["port"]) == 4443]
dialled = [row for row in established
if row["local"]["port"] != 4443 and (row["remote"]["addr"], row["remote"]["port"])
== ("127.0.0.1", 4443)]
assert accepted and dialled, "this capture is the loop the stream map's `default` produces"
assert listener_at(payload, 4443)["connections"] == len(accepted)
assert endpoint_count(payload, "127.0.0.1", 4443) == len(dialled)
assert not [row for row in established
if row["local"]["port"] == 4443
and (row["remote"]["addr"], row["remote"]["port"]) == ("127.0.0.1", 4443)], (
"a row that was both would be counted on two pipes by two rules, and the ≤ above would "
"still hold while every figure in the payload was wrong")
def test_a_socket_neither_config_describes_is_counted_nowhere(gateway_snapshot: Any) -> None:
"""Phase 28's "an unclaimed socket stays visible as one" is **not** a live count.
The agent's own `:9120` is listening and dialled by nobody in either file. It stays in
`sockets.listens` — a raw fact about the host, which is what the page prints beside the pipes —
and appears in no pipe and no total. That is the whole simplification: an unexplained socket is
a reason to look at `/system`, not a number to file under `unmatched`.
"""
snap = gateway_snapshot("full")
payload = gateway.parse(snap, "gateway")
live = {row["port"] for row in snap["sockets"]["listens"]}
configured = {row["port"] for row in payload["listeners"]}
assert 9120 in live - configured, "the agent's own port, unclaimed by either config"
assert all(row["port"] in configured for row in payload["listeners"])
assert "unmatched" not in payload
def test_every_connection_the_captured_configs_declare_is_drawn(gateway_snapshot: Any) -> None:
"""The graph as a **set**, so a missing or invented route fails by name.
The join this task rewrote stamps counts onto these rows, so the rows themselves are pinned
here too: 6 haproxy `server` lines, 18 stream-`map` entries (the 17 `*.reeselink.com:443` fan-
out plus the literal `default` → `127.0.0.1:4443`), and 9 vhost `proxy_pass` targets across
`:4443` and `:8443`. `:80` is the refusal that lives on this fleet: the catch-all vhost declares
no `proxy_pass`, so it is a trunk with no connection and no invented line beside it.
"""
payload = gateway.parse(gateway_snapshot("full"), "gateway")
got = {(row["in"]["port"], row["out"]["host"], row["out"]["port"])
for row in payload["connections"]}
assert got == {
# haproxy: one `server` per frontend, the backend name as the selector
(22, "reeseapps-gitea.reeselink.com", 2222),
(2000, "stackexpected-gitea.reeselink.com", 2222),
(5201, "web.reeselink.com", 5201),
(8080, "unifi-external.reeselink.com", 8080),
(25565, "minecraft.reeselink.com", 25565),
(25566, "minecraft.reeselink.com", 25566),
# nginx `:443`: the stream map's fan-out, `default` kept verbatim
(443, "127.0.0.1", 4443),
(443, "reeseapps-brain-of-code.reeselink.com", 443),
(443, "reeseapps-brain-of-reese.reeselink.com", 443),
(443, "reeseapps-forgejo.reeselink.com", 443),
(443, "reeseapps-gitea.reeselink.com", 443),
(443, "reeseapps-homepage.reeselink.com", 443),
(443, "reeseapps-immich.reeselink.com", 443),
(443, "reeseapps-whiteboard.reeselink.com", 443),
(443, "reeseapps-keycloak.reeselink.com", 443),
(443, "reeseapps-litellm.reeselink.com", 443),
(443, "reeseapps-ntfy.reeselink.com", 443),
(443, "reeseapps-openwebui.reeselink.com", 443),
(443, "stackexpected-gitea.reeselink.com", 443),
(443, "stackexpected-hedgedoc.reeselink.com", 443),
(443, "stackexpected-keycloak.reeselink.com", 443),
(443, "stackexpected-nextcloud.reeselink.com", 443),
(443, "stackexpected.reeselink.com", 443),
(443, "wfc.reeselink.com", 443),
# nginx http vhosts: `:4443` and `:8443`, each with its own `server_name`
(4443, "bricktracker.reeselink.com", 3333),
(4443, "emmaleaf.reeselink.com", 80),
(4443, "gitea.reeselink.com", 3000),
(4443, "jellyfin.reeselink.com", 8096),
(4443, "keycloak.reeselink.com", 8080),
(4443, "nextcloud.reeselink.com", 11000),
(8443, "127.0.0.1", 8081),
(8443, "keycloak.reeselink.com", 8080),
(8443, "nextcloud.reeselink.com", 11001),
}, "the captured configs declare exactly these routes — no more, no fewer"
assert len(payload["connections"]) == len(got), "and no two rows collapse into one key"
# `default` is the config's own word for "everything nobody mapped", and its endpoint is an
# address — which is why it is one of the two lines this capture can count at all.
default = next(row for row in payload["connections"] if row["out"]["selectors"] == ["default"])
assert (default["out"]["host"], default["out"]["port"]) == ("127.0.0.1", 4443)
assert default["out"]["address"] is True and default["out"]["kind"] == gateway.KIND_MAP
# refusal rule 5 on live config text: a trunk with nothing behind it stays a trunk alone
assert 80 in {row["port"] for row in payload["listeners"]}
assert 80 not in {row["in"]["port"] for row in payload["connections"]}, (
"the `server_name _` vhost declares no proxy_pass, so it draws no connection")
def test_the_map_fanout_is_routes_and_never_a_variable_host(gateway_snapshot: Any) -> None:
"""Every stream-map entry is a line the page can draw, and `$var` never becomes a host."""
payload = gateway.parse(gateway_snapshot("full"), "gateway")
map_routes = [row for row in payload["connections"] if row["out"]["kind"] == "map"]
assert len(map_routes) == len(payload["nginx"]["map_targets"])
assert not [row for row in payload["connections"]
if row["out"]["host"] is None or "$" in str(row["out"]["host"])], (
"a proxy_pass $var carries no address, so it must not become a line of its own")
brain = next(row for row in map_routes
if row["out"]["host"] == "reeseapps-brain-of-reese.reeselink.com")
assert brain["out"]["address"] is False, "the config wrote a name, and the config is verbatim"
assert brain["connections"] is None, "and on this arm (no resolver section) nothing placed it"
assert brain["metric"] is None, "so it publishes no series"
# --------------------------------------------------------------------------- #
# the inbound join
# --------------------------------------------------------------------------- #
def test_a_socket_lands_in_the_frontend_that_owns_its_local_port() -> None:
snap = snapshot(
haproxy=haproxy_section([frontend("ssh", [22], default="ssh-be")]),
nginx=nginx_section([nginx_server([443], ["a.example.test"])]),
sockets=sockets_section(
listens=[listen(22), listen(443)],
established=[conn(("10.1.1.27", 443), ("203.0.113.7", 51000)),
conn(("10.1.1.27", 443), ("203.0.113.9", 51001)),
conn(("10.1.1.27", 22), ("198.51.100.4", 33000))]))
payload = gateway.parse(snap, "gateway")
https = listener_at(payload, 443)
assert https["source"] == "nginx"
assert https["connections"] == 2
assert https["clients"] == ["203.0.113.7", "203.0.113.9"], "the client is the remote end"
assert https["domains"] == ["a.example.test"]
assert https["metric"] == "gateway.in_conns@443"
assert listener_at(payload, 22)["source"] == "haproxy"
def test_a_frontend_that_is_idle_is_still_a_pipe_with_zero() -> None:
"""A configured port with no connections is `0`, not `n/a` — the socket table did answer."""
snap = snapshot(haproxy=haproxy_section([frontend("ssh", [22])]),
sockets=sockets_section(listens=[listen(22)]))
payload = gateway.parse(snap, "gateway")
assert listener_at(payload, 22)["connections"] == 0
def test_a_port_no_config_names_is_neither_counted_nor_reported() -> None:
"""The arm that used to be `unmatched.inbound` is now simply **not a number anywhere**.
Two sockets on `:8081`, which neither file binds. They are not attributed to the `:443` pipe,
not summed into it, and not tallied beside it — the listener reads `0` and the payload carries
no residual. That is a deliberate loss of information: an unclaimed socket says something about
the host, not about the two config files, and the honest place to read it is `/system`.
"""
snap = snapshot(
nginx=nginx_section([nginx_server([443], ["a.example.test"])]),
sockets=sockets_section(
listens=[listen(443), listen(8081)],
established=[conn(("10.1.1.27", 8081), ("10.1.0.9", 41000)),
conn(("10.1.1.27", 8081), ("10.1.0.9", 41001))]))
payload = gateway.parse(snap, "gateway")
assert listener_at(payload, 443)["connections"] == 0
assert 8081 not in [row["port"] for row in payload["listeners"]]
assert sum(row["connections"] or 0 for row in payload["listeners"]) < payload["sockets"]["established"]
assert "unmatched" not in payload
def test_one_port_named_by_both_configs_is_one_pipe_carrying_both_domains() -> None:
"""A host cannot bind a port twice, so the socket space is one pipe and the facts merge."""
snap = snapshot(
haproxy=haproxy_section([frontend("http", [80])]),
nginx=nginx_section([nginx_server([80], ["a.example.test"])]),
sockets=sockets_section(listens=[listen(80)],
established=[conn(("10.1.1.27", 80), ("203.0.113.1", 1234))]))
payload = gateway.parse(snap, "gateway")
ports = [row["port"] for row in payload["listeners"]]
assert ports == [80], "no second pipe, and no socket counted down both"
assert listener_at(payload, 80)["connections"] == 1
assert listener_at(payload, 80)["domains"] == ["a.example.test"]
def test_the_client_list_is_unique_sorted_and_bounded() -> None:
ports = [80]
many = [conn(("10.1.1.27", 80), (f"203.0.113.{i % 256}", 40000 + i))
for i in range(gateway.MAX_CLIENTS + 12)]
many.append(conn(("10.1.1.27", 80), ("203.0.113.1", 45678)))
snap = snapshot(nginx=nginx_section([nginx_server(ports, ["a.example.test"])]),
sockets=sockets_section(listens=[listen(80)], established=many))
pipe = listener_at(gateway.parse(snap, "gateway"), 80)
assert len(pipe["clients"]) == gateway.MAX_CLIENTS
assert pipe["clients"] == sorted(pipe["clients"])
assert pipe["connections"] == len(many), "the count stays the truth when the list is capped"
# --------------------------------------------------------------------------- #
# the outbound join
# --------------------------------------------------------------------------- #
def test_an_endpoint_written_as_an_address_is_matched_exactly_and_carries_its_count(
) -> None:
"""`/proc` writes `(remote address, remote port)`, so a config that writes both can be joined."""
snap = snapshot(
nginx=nginx_section([nginx_server([8443], ["app.example.test"],
[proxy_pass(("127.0.0.1", 8081))])]),
sockets=sockets_section(
listens=[listen(8443)],
established=[conn(("10.1.1.27", 39000), ("127.0.0.1", 8081)),
conn(("10.1.1.27", 39002), ("10.9.9.9", 8081))]))
payload = gateway.parse(snap, "gateway")
exact = routes_to(payload, "127.0.0.1")[0]
assert exact["out"]["address"] is True
assert exact["connections"] == 1, "the one socket dialled to that address"
assert exact["metric"] == "gateway.out_conns@127.0.0.1:8081"
assert endpoint_count(payload, "127.0.0.1", 8081) == 1, (
"the same port on a different address is not this target's connection")
assert "by_port" not in json.dumps(payload), (
"the port-only arm of the join is deleted, not merely unused: a payload that grew the field "
"back has grown the ambiguity back with it")
def test_a_name_target_is_never_credited_for_a_socket_on_its_port() -> None:
"""The rule that replaced `by_port`, stated as a fact about the payload and not a chip.
`proxy_pass https://nextcloud.example.test:11000` and the socket to `10.1.0.254:11000` are the
same connection in the operator's head and two unrelated strings **in this stack**. The bridge is
the gateway's own resolution (phase 32, tested below), and the port-only guess the old join made
in its place was a guess about a **port**, not about a host — which is why its count was the
port's and its chip said so. With nothing to bridge them the line has no number, no gauge, and no
apology to render; with the agent's document it is counted through the address, and this test
stays true because the snapshot it builds has no resolver section.
"""
snap = snapshot(
nginx=nginx_section([nginx_server([4443], ["cloud.example.test"],
[proxy_pass(("nextcloud.example.test", 11000))])]),
sockets=sockets_section(
listens=[listen(4443)],
established=[conn(("10.1.1.27", 4443), ("10.1.0.4", 55555)),
conn(("10.1.1.27", 54340), ("10.1.0.254", 11000))]))
payload = gateway.parse(snap, "gateway")
row = routes_to(payload, "nextcloud.example.test")[0]
assert row["out"]["address"] is False
assert row["connections"] is None, "unknown, and not the port's 2 either"
assert row["metric"] is None, "and no series for a number that can never exist"
assert listener_at(payload, 4443)["connections"] == 1, (
"the socket that faces the gateway is still counted where it is measurable")
assert "by_port" not in json.dumps(payload)
# the socket to `10.1.0.254:11000` is credited to **nothing**: the sum over the outbound side
# is unchanged by its presence, which is the difference between "unknown" and "filed somewhere"
assert sum(row["connections"] or 0 for row in payload["connections"]) == 0
assert out_metrics(payload) == [], "and it publishes no gauge under any name"
def test_a_port_shared_by_several_names_still_resolves_none_of_them() -> None:
"""The real `stream.conf` shape: sixteen names on 443, none of them counted **without DNS**.
The answer is the same `null` the old join gave, minus the machinery that produced it and the
`unmatched.outbound` total that used to double-count the same sockets.
"""
entries = [(f"site{i}.example.test", (f"backend{i}.example.test", 443)) for i in range(4)]
snap = snapshot(
nginx=nginx_section([nginx_server([443], [], [proxy_pass(None, variable="targetBackend")])],
maps=[mapping_block("$targetBackend", entries)]),
sockets=sockets_section(
listens=[listen(443)],
established=[conn(("10.1.1.27", 443), ("203.0.113.5", 4000)),
conn(("10.1.1.27", 50000), ("10.2.2.2", 443)),
conn(("10.1.1.27", 50002), ("10.2.2.3", 443))]))
payload = gateway.parse(snap, "gateway")
assert len(payload["nginx"]["map_targets"]) == 4
for _key, (host, _port) in entries:
assert routes_to(payload, host)[0]["connections"] is None, "unknown, never 0 and never 2"
assert out_metrics(payload) == []
assert sum(row["connections"] or 0 for row in payload["listeners"]) == 1
assert "unmatched" not in payload
def test_no_join_ever_asks_a_resolver() -> None:
"""The whole point of `by_port`: this module must not turn a name into an address.
Spied on the live `socket` module the stdlib resolvers use, and run over the captured
gateway as well as a synthetic one, so a "helpful" future `socket.getaddrinfo` call inside
the join fails here rather than making 33 DNS queries every 30 seconds from the poll loop.
"""
calls: list[tuple[Any, ...]] = []
def spy(*args: Any, **kwargs: Any) -> Any: # pragma: no cover - must never run
calls.append((args, kwargs))
raise AssertionError("the gateway join must never resolve a name")
original = socket_module.getaddrinfo
socket_module.getaddrinfo = spy
try:
gateway.parse(snapshot(
haproxy=haproxy_section([frontend("mc", [25565])],
[backend("mc-be", [server("mc1", "minecraft.example.test",
25565)])]),
nginx=nginx_section([nginx_server([443], ["a.example.test"],
[proxy_pass(("app.example.test", 8080))])]),
sockets=sockets_section(listens=[listen(25565), listen(443)],
established=[conn(("10.1.1.27", 443), ("203.0.113.5", 900)),
conn(("10.1.1.27", 1), ("10.2.2.2", 8080))])),
"gateway")
finally:
socket_module.getaddrinfo = original
assert calls == []
def test_the_recorded_resolver_document_reaches_every_name_written_route(
gateway_snapshot: Any) -> None:
"""What phase 32 buys, priced against the captures — and spent, since task 03.
31 of this ingress's 33 routes carried an em dash because this module joins sockets by
**address** and the two config files write those endpoints as **names**. The recorded
`gateway_resolver.json` is the thing that removed that reason, so this pins the two properties
the join needs and no more:
* every name-written route has an entry in the document, for that name **and that port**, with
at least one address — so no route on `/gateway` is dashed forever because the resolver had
nothing to say about its endpoint;
* the addresses are the same currency the socket table is written in: the two captures overlap
on live remote endpoints, which is what a join has to have to join anything.
and, in the last few lines, that spending it is a fact about the **document** rather than a
coincidence of the port table: the name-written routes are counted exactly where the entry puts
an address and nowhere else. This module still resolves nothing — the spy pin above stays exactly
as it was, and is the reason the fixture has to exist.
"""
snap = gateway_snapshot("full")
payload = gateway.parse(snap, "gateway")
document = json.loads((FIXTURES / "gateway_resolver.json").read_text())
answered = {(row["name"], row["port"]) for row in document["entries"] if row["addresses"]}
written = {(row["out"]["host"], row["out"]["port"]) for row in payload["connections"]
if not row["out"]["address"]}
assert written, "this capture writes its targets as names, which is the gap the fixture fills"
unreached = written - answered
assert not unreached, ("name-written routes the resolver document cannot reach: "
f"{sorted(unreached)} — re-run ./scripts/capture-gateway-config.sh")
resolved = {address for row in document["entries"] for address in row["addresses"]}
remote = {row["remote"]["addr"] for row in snap["sockets"]["established"]}
assert resolved & remote, (
"the resolver document and the socket table share no address, so a join built on it could "
"attribute nothing — the two captures are from different boxes or one scrub moved a number")
assert all(row["connections"] is None for row in payload["connections"]
if (row["out"]["host"], row["out"]["port"]) in written), (
"this arm is the snapshot with no `resolver` key at all, so a name-written route still "
"carries no count here — and adding one address to that section is what changes it")
spent = gateway.parse(gateway_snapshot("resolved"), "gateway")
rows = {(row["out"]["host"], row["out"]["port"]): row for row in spent["connections"]}
trunks = {row["port"] for row in spent["listeners"]}
for name, port in sorted(written):
row = rows[(name, port)]
entry = next(item for item in document["entries"]
if (item["name"], item["port"]) == (name, port))
assert row["out"]["resolved"] == entry["addresses"], f"{name}:{port} rewrites the capture"
assert row["connections"] is not None, f"{name}:{port} is placed and still carries no count"
# the exact figure, re-counted from the raw table by this test: the rows that reach this
# endpoint, minus the ones the inbound arm claimed first (a socket is counted at most once)
keys = {(address, port) for address in entry["addresses"]}
live = [item for item in snap["sockets"]["established"]
if item["local"]["port"] not in trunks
and (item["remote"]["addr"], item["remote"]["port"]) in keys]
assert row["connections"] == len(live), (
f"{name}:{port} resolves to {sorted(keys)} and the table holds {len(live)} such "
f"sockets, but the line reads {row['connections']}")
def test_two_services_on_one_host_are_two_pipes_not_one_sum() -> None:
"""The reason the outbound series name carries its port (`minecraft` 25565 and 25566)."""
snap = snapshot(
haproxy=haproxy_section([frontend("mc1", [25565], default="testcraft"),
frontend("mc2", [25566], default="nimcraft")],
[backend("testcraft", [server("mc1", "mc.example.test", 25565)]),
backend("nimcraft", [server("mc2", "mc.example.test", 25566)])]),
sockets=sockets_section(
established=[conn(("10.1.1.27", 40000), ("10.5.5.5", 25565)),
conn(("10.1.1.27", 40002), ("10.5.5.5", 25566)),
conn(("10.1.1.27", 40004), ("10.5.5.5", 25566))]))
payload = gateway.parse(snap, "gateway")
rows = routes_to(payload, "mc.example.test")
assert [row["out"]["port"] for row in rows] == [25565, 25566]
assert [row["connections"] for row in rows] == [None, None], "a name is never counted"
assert [row["metric"] for row in rows] == [None, None]
names = {row["out"]["name"] for row in payload["connections"]}
assert names == {"mc1", "mc2"}, "the config's own server names survive the dedup"
def test_two_services_on_one_address_are_two_endpoints_and_two_gauges() -> None:
"""The same two Minecraft servers written as **addresses**: now both are measurable.
The port belongs in the gauge name because one host answering `:25565` and `:25566` is two
services, and two gauges on `gateway.out_conns@10.5.5.5` would make the second insert
overwrite the first in silence.
"""
snap = snapshot(
haproxy=haproxy_section([frontend("mc1", [25565], default="testcraft"),
frontend("mc2", [25566], default="nimcraft")],
[backend("testcraft", [server("mc1", "10.5.5.5", 25565)]),
backend("nimcraft", [server("mc2", "10.5.5.5", 25566)])]),
sockets=sockets_section(
established=[conn(("10.1.1.27", 40000), ("10.5.5.5", 25565)),
conn(("10.1.1.27", 40002), ("10.5.5.5", 25566)),
conn(("10.1.1.27", 40004), ("10.5.5.5", 25566))]))
payload = gateway.parse(snap, "gateway")
assert endpoint_count(payload, "10.5.5.5", 25565) == 1
assert endpoint_count(payload, "10.5.5.5", 25566) == 2
assert out_metrics(payload) == ["gateway.out_conns@10.5.5.5:25565",
"gateway.out_conns@10.5.5.5:25566"]
def test_a_target_with_no_port_is_a_line_with_no_count() -> None:
"""`server x host` names no port, so there is nothing to join on and nothing to invent."""
snap = snapshot(
haproxy=haproxy_section([frontend("be", [8080], default="be")],
[backend("be", [server("solo", "solo.example.test")])]),
sockets=sockets_section(established=[conn(("10.1.1.27", 40000), ("10.5.5.5", 8080))]))
row = routes_to(gateway.parse(snap, "gateway"), "solo.example.test")[0]
assert row["out"]["port"] is None
assert row["out"]["address"] is False
assert row["connections"] is None
assert row["metric"] is None, "an endpoint with no port has no gauge either"
def test_an_address_with_no_port_is_a_line_with_no_count() -> None:
"""`out.address` says the string is an address; the **join needs the pair**, not one half."""
snap = snapshot(
haproxy=haproxy_section([frontend("be", [8080], default="be")],
[backend("be", [server("solo", "10.5.5.5")])]),
sockets=sockets_section(established=[conn(("10.1.1.27", 40000), ("10.5.5.5", 8080))]))
row = routes_to(gateway.parse(snap, "gateway"), "10.5.5.5")[0]
assert row["out"]["address"] is True
assert row["connections"] is None, "the socket's port is nobody's guess (rule 2)"
assert row["metric"] is None
def test_two_connections_sharing_an_endpoint_say_so_and_repeat_the_count_never_split_it(
) -> None:
"""`:4443 -> 10.5.5.5:8080` and `:8443 -> 10.5.5.5:8080` are two lines and **one** count.
Each line carries the endpoint's whole number — a line showing half of it would be a split
nobody measured — and `shared: true` is what tells the totals, the page and any future notifier
that the two are the same sockets. Summing the lines doubles the traffic; summing the distinct
gauges is the number the kernel has.
"""
snap = snapshot(
nginx=nginx_section([nginx_server([4443], ["a.example.test"],
[proxy_pass(("10.5.5.5", 8080))]),
nginx_server([8443], ["admin.a.example.test"],
[proxy_pass(("10.5.5.5", 8080))])]),
sockets=sockets_section(
established=[conn(("10.1.1.27", 40000), ("10.5.5.5", 8080)),
conn(("10.1.1.27", 40002), ("10.5.5.5", 8080))]))
payload = gateway.parse(snap, "gateway")
rows = routes_to(payload, "10.5.5.5")
assert [row["in"]["port"] for row in rows] == [4443, 8443]
assert [row["connections"] for row in rows] == [2, 2], "the endpoint's total, twice"
assert [row["shared"] for row in rows] == [True, True]
assert sum(row["connections"] or 0 for row in rows) == 4, "the lines double-count on purpose"
outbound = [g for g in gateway.samples(payload) if g.metric.startswith("gateway.out_conns@")]
assert [g.metric for g in outbound] == ["gateway.out_conns@10.5.5.5:8080"], "published once"
assert outbound[0].value == 2.0
def test_two_names_one_target_one_port_is_one_connection() -> None:
"""The merge rule, which is what `shared` and the page's line count both depend on.
One `(source, in_port, host, port)` is **one** line: a vhost with two `server_name`s reaching
one `proxy_pass` is one pipe with two selectors, and a second vhost on the same port arriving
at the same endpoint adds its name to that line rather than drawing a second one. Selectors
union in **first-seen order**, because the config's order is the only ordering there is.
"""
snap = snapshot(nginx=nginx_section([
nginx_server([4443], ["a.example.test", "www.a.example.test"],
[proxy_pass(("10.5.5.5", 8080))]),
nginx_server([4443], ["b.example.test"], [proxy_pass(("10.5.5.5", 8080))])]))
rows = gateway.parse(snap, "gateway")["connections"]
assert len(rows) == 1, "two vhosts, one port, one endpoint: one line"
assert rows[0]["out"]["selectors"] == [
"a.example.test", "www.a.example.test", "b.example.test"], (
"selectors are the union in the order the configs wrote them")
def test_the_same_target_on_two_listen_ports_is_two_connections() -> None:
"""Never merge across `in_port`: `:4443` and `:8443` are two socket spaces.
The real `http.conf` pair — `admin.keycloak` on 8443 and `keycloak` on 4443 both land on
`keycloak:8080` — and the `minecraft` 25565/25566 lesson from the other side. One line for
both would draw one pipe where the operator has two, and the page could not say which inbound
port a connection came in on.
"""
snap = snapshot(nginx=nginx_section([
nginx_server([4443], ["keycloak.example.test"], [proxy_pass(("10.5.5.5", 8080))]),
nginx_server([8443], ["admin.keycloak.example.test"],
[proxy_pass(("10.5.5.5", 8080))])]))
rows = gateway.parse(snap, "gateway")["connections"]
assert [row["in"]["port"] for row in rows] == [4443, 8443]
assert [row["id"] for row in rows] == [
"nginx:4443->10.5.5.5:8080", "nginx:8443->10.5.5.5:8080"]
assert [row["out"]["selectors"] for row in rows] == [
["keycloak.example.test"], ["admin.keycloak.example.test"]]
def test_the_first_name_a_config_writes_for_an_endpoint_is_the_one_kept() -> None:
"""A later claim never displaces a name the config already gave the endpoint.
haproxy claims are built before nginx's, and its `server` line is the only place a name comes
from, so the rule has to be "first name wins" rather than "last write wins": a second route to
the same endpoint that carries no name must not blank the one that does.
"""
snap = snapshot(haproxy=haproxy_section(
[frontend("one", [80], default="be"), frontend("two", [80], default="be")],
[backend("be", [server("", "10.5.5.5", 8080), server("app1", "10.5.5.5", 8080)])]))
rows = gateway.parse(snap, "gateway")["connections"]
assert len(rows) == 1, "both frontends route the same server: one line"
assert rows[0]["out"]["name"] == "app1", "the name the config wrote, not the blank that came first"
def test_the_loopback_hop_counts_once_on_each_side() -> None:
"""A connection to a service on the router itself is **two** `/proc` rows and two lines.
`nginx :4443 → 127.0.0.1:4443` is the fleet's real shape (`stream.conf`'s `default`). One row
has local port `4443` (the side that was accepted, whose client is `127.0.0.1`) and another has
remote `127.0.0.1:4443` (the side the router dialled). Each belongs on its own pipe — inbound
and outbound — and **no single row may be drawn down both**: the inbound arm runs first and
`continue`s, which is the whole enforcement, so the two counts here are two each and not four
on one side. Summed against the table, the four rows are four counted and zero left over.
"""
snap = snapshot(
nginx=nginx_section([nginx_server([4443], ["a.example.test"],
[proxy_pass(("127.0.0.1", 4443))])]),
sockets=sockets_section(
listens=[listen(4443)],
established=[conn(("127.0.0.1", 41000), ("127.0.0.1", 4443)),
conn(("127.0.0.1", 41002), ("127.0.0.1", 4443)),
conn(("127.0.0.1", 4443), ("127.0.0.1", 41000)),
conn(("127.0.0.1", 4443), ("127.0.0.1", 41002))]))
payload = gateway.parse(snap, "gateway")
trunk = listener_at(payload, 4443)
route = routes_to(payload, "127.0.0.1")[0]
assert trunk["connections"] == 2, "the two accepted-side rows, and only those"
assert route["connections"] == 2, "the two dialled-side rows, and only those"
assert route["in"]["port"] == 4443 and route["out"]["port"] == 4443
assert trunk["clients"] == ["127.0.0.1"], "the loopback peer is the client of its own row"
total = trunk["connections"] + endpoint_count(payload, "127.0.0.1", 4443)
assert total == payload["sockets"]["established"] == 4, (
"each row counted at most once; the ≤ identity is an equality here because every row of "
"this table belongs to the one configured route")
def test_an_endpoint_reached_by_one_line_is_not_shared() -> None:
"""`shared` is a fact about the graph, not a flag meaning "I did not decide"."""
snap = snapshot(
nginx=nginx_section([nginx_server([4443], ["a.example.test"],
[proxy_pass(("10.5.5.5", 8080))])]),
sockets=sockets_section(established=[conn(("10.1.1.27", 40000), ("10.5.5.5", 8080))]))
row = routes_to(gateway.parse(snap, "gateway"), "10.5.5.5")[0]
assert row["shared"] is False
assert row["connections"] == 1
# --------------------------------------------------------------------------- #
# phase 32: the resolver section — a name counted through the address the gateway dialled
#
# Everything above this line runs on a snapshot with **no** `resolver` key, which is agent 1.2.0 (or
# 1.3.0 with `--no-resolve`) and therefore phase 30 verbatim. Everything below it spends the
# document `scripts/capture-gateway-config.sh` recorded with `getent` on the router: the resolution
# haproxy and nginx perform **on the box that opens the connection**, which is the only vantage that
# can join `reeseapps-brain-of-reese.reeselink.com` to the `10.1.2.64` `/proc` reports. The
# dashboard asking the same name through its own `resolv.conf` would be a different vantage
# answering a different question — and `test_the_dashboard_still_never_resolves` is the spy that
# keeps it that way.
# --------------------------------------------------------------------------- #
def test_a_name_route_counts_through_the_address_the_gateway_published(
gateway_snapshot: Any) -> None:
"""The whole point of phase 32, measured against the captures.
`stream.conf` routes `brain.reeseapps.com` to `reeseapps-brain-of-reese.reeselink.com:443`,
`/proc` would report a socket to `10.1.2.64:443`, and until this section arrived nothing could
join the two. Now the line carries the config's text and the agent's answer side by side, and
the count comes through the answer.
"""
snap = gateway_snapshot("resolved")
payload = gateway.parse(snap, "gateway")
brain = next(row for row in payload["connections"]
if row["out"]["host"] == "reeseapps-brain-of-reese.reeselink.com")
assert brain["in"] == {"port": 443}
assert brain["out"]["selectors"] == ["brain.reeseapps.com"], "the config's word, untouched"
assert brain["out"]["address"] is False, "the config wrote a name, and still says so"
assert brain["out"]["resolved"] == ["10.1.2.64"], "the agent's answer, verbatim"
assert brain["out"]["via"] == "libc", "the vantage is the agent's own string, copied"
assert brain["out"]["resolved_ts"] == 1_792_886_400.0, "that entry's own timestamp"
assert brain["metric"] == "gateway.out_conns@reeseapps-brain-of-reese.reeselink.com:443", (
"the gauge keeps the config's spelling, so history continues under one name")
assert brain["connections"] == 0, "resolved and idle is `0`, never the dash"
ntfy = next(row for row in payload["connections"]
if row["out"]["host"] == "reeseapps-ntfy.reeselink.com")
live = [row for row in snap["sockets"]["established"]
if (row["remote"]["addr"], row["remote"]["port"]) == ("10.1.1.176", 443)]
assert live, "the recorded table has live ntfy sockets for the join to find"
assert ntfy["connections"] == len(live), "every socket to the address the name became"
assert ntfy["shared"] is False, "one route reaches that endpoint, so nothing to repeat"
def test_an_endpoint_the_config_wrote_as_an_address_claims_no_lookup(
gateway_snapshot: Any) -> None:
"""`resolved` answers *what did this name become*, and an address was never a name.
The agent publishes a `literal` entry for a target that is already an address, and the resolver
block lists it verbatim. Stamping that echo onto the route as `resolved` beside `via: "libc"`
would report a `getaddrinfo` this gateway never ran, so the three provenance fields stay `null`
on an `address: true` row while the count and the gauge keep working exactly as phase 30 left
them. Two facts stay two sentences: `address` is what the config wrote, `resolved` is what the
gateway looked up.
"""
payload = gateway.parse(gateway_snapshot("resolved"), "gateway")
written = [row for row in payload["connections"] if row["out"]["address"] is True]
assert written, "this capture writes two targets as addresses, which is phase 30's joinable pair"
for row in written:
assert row["out"]["host"] in {entry["name"]
for entry in payload["resolver"]["entries"]}, (
"the agent did publish this literal, and the block still lists it")
assert row["out"]["resolved"] is None and row["out"]["via"] is None \
and row["out"]["resolved_ts"] is None, (
f"{row['id']}: an address row must not borrow a lookup claim")
assert row["metric"] == f"gateway.out_conns@{row['out']['host']}:{row['out']['port']}"
assert row["connections"] is not None, "phase 30's address join is untouched by this"
def test_a_socket_lands_on_the_name_route_that_resolved_to_its_address() -> None:
"""The synthetic version, so the claim does not depend on what a capture happened to contain."""
snap = snapshot(
nginx=nginx_section([nginx_server([443], [], [proxy_pass(None, variable="targetBackend")])],
maps=[mapping_block("$targetBackend",
[("brain.example.test",
("brain-backend.example.test", 443))])]),
sockets=sockets_section(
listens=[listen(443)],
established=[conn(("::", 51000), ("10.1.2.64", 443), family="v6"),
conn(("10.1.1.27", 443), ("203.0.113.7", 900))]),
resolver=resolver_section([resolved("brain-backend.example.test", 443, ["10.1.2.64"])]))
payload = gateway.parse(snap, "gateway")
row = routes_to(payload, "brain-backend.example.test")[0]
assert row["out"]["address"] is False and row["out"]["resolved"] == ["10.1.2.64"]
assert row["connections"] == 1
assert row["metric"] == "gateway.out_conns@brain-backend.example.test:443"
assert listener_at(payload, 443)["connections"] == 1, "the inbound half is untouched"
gauge = [g for g in gateway.samples(payload, 5.0)
if g.metric == "gateway.out_conns@brain-backend.example.test:443"]
assert [(g.metric, g.value) for g in gauge] == [
("gateway.out_conns@brain-backend.example.test:443", 1.0)]
def test_the_match_is_an_address_not_a_string() -> None:
"""One address, two spellings, in either direction: the join still lands.
`/proc`'s `inet_ntop` writes an IPv4 peer seen through the IPv6 table as `::ffff:10.1.2.64`
while a resolver answers `10.1.2.64`, and an IPv6 in one table can carry leading zeros and
capitals the other does not. `ipaddress` canonicalises both, so the **match key** is one address
while the **payload** keeps publishing whatever the config and the agent wrote — a page that
printed the normalised form would be quoting a string nobody configured.
A join that compared strings instead would report a resolved, dialled route as idle forever:
the exact failure this phase exists to remove, arriving back through a spelling nobody thought
about. The two directions are asserted separately because each half is a different table.
"""
def build(written: str, socketed: str) -> dict[str, Any]:
snap = snapshot(
haproxy=haproxy_section([frontend("peer", [8001], default="peer-be")],
[backend("peer-be", [server("p", "peer.example.test", 443)])]),
sockets=sockets_section(
established=[conn(("::", 40000), (socketed, 443), family="v6")]),
resolver=resolver_section([resolved("peer.example.test", 443, [written])]))
return gateway.parse(snap, "gateway")
for written, socketed in (("10.1.2.64", "::ffff:10.1.2.64"),
("::ffff:10.1.2.64", "10.1.2.64"),
("2001:db8::1", "2001:0DB8::0001"),
("2001:0DB8::0001", "2001:db8::1")):
payload = build(written, socketed)
row = routes_to(payload, "peer.example.test")[0]
assert row["connections"] == 1, f"{written} vs {socketed} is one address and did not match"
assert row["out"]["resolved"] == [written], "the payload keeps the agent's own spelling"
def test_a_zero_padded_ipv4_is_no_address_to_anyone_and_matches_nothing() -> None:
"""The one spelling this module refuses to fold, and why refusing is the safe half.
`ipaddress` rejects `010.1.2.64` because `inet_aton` reads a leading `0` as **octal**: the
decimal and octal readings are two different addresses, and picking one here would book a
socket on a route by coin flip. So a padded target is a **name** to this module — and the
resolver section, which is libc's own answer, is what turns it into an address the kernel agrees
with. Nothing is guessed in between.
"""
snap = snapshot(
haproxy=haproxy_section([frontend("be", [8080], default="be")],
[backend("be", [server("odd", "010.1.2.64", 443)])]),
sockets=sockets_section(
established=[conn(("10.1.1.27", 40000), ("10.1.2.64", 443)),
conn(("10.1.1.27", 40002), ("8.1.2.64", 443))]))
row = routes_to(gateway.parse(snap, "gateway"), "010.1.2.64")[0]
assert row["out"]["address"] is False, "not an address here, so it is a name"
assert row["connections"] is None and row["metric"] is None, (
"and with no resolver entry it is counted nowhere — neither reading is chosen for it")
# …and the fix is the resolver's, not this module's: libc's own answer places the same route,
# the padded spelling travels in the payload untouched, and neither socket is read as an octet.
placed = snapshot(
haproxy=haproxy_section([frontend("be", [8080], default="be")],
[backend("be", [server("odd", "010.1.2.64", 443)])]),
sockets=sockets_section(
established=[conn(("10.1.1.27", 40000), ("10.1.2.64", 443)),
conn(("10.1.1.27", 40002), ("8.1.2.64", 443))]),
resolver=resolver_section([resolved("010.1.2.64", 443, ["10.1.2.64"])]))
fixed = routes_to(gateway.parse(placed, "gateway"), "010.1.2.64")[0]
assert fixed["out"]["resolved"] == ["10.1.2.64"]
assert fixed["connections"] == 1, (
"the one socket to the address libc returned, not the octal-reading 8.1.2.64 and not both")
def test_two_names_one_endpoint_is_one_count_shared() -> None:
"""Two **names** the gateway placed on one address are one endpoint and one number.
Phase 30 knew this shape only for one address written twice; resolving makes it ordinary, and
the rule did not change: the endpoint's whole count is repeated on both lines with
`shared: true`, never halved and never attributed to one of them.
Two readers then differ, and both are right. **Every total** books the endpoint once
(`distinct_endpoint_counts`, which is what `/api/summary`'s `live_out` reads), while the gauge
is per line-family — one series per name the config wrote, each carrying that endpoint's number,
because a route's own history is asked for by the name on the route. Two names therefore store
two series of one figure; lines that share a **name** (`:4443` and `:8443` reaching one target)
still store exactly one.
"""
snap = snapshot(
nginx=nginx_section([nginx_server([4443], ["a.example.test"],
[proxy_pass(("one.example.test", 8080))]),
nginx_server([8443], ["b.example.test"],
[proxy_pass(("two.example.test", 8080))])]),
sockets=sockets_section(
established=[conn(("10.1.1.27", 40000), ("10.5.5.5", 8080)),
conn(("10.1.1.27", 40002), ("10.5.5.5", 8080))]),
resolver=resolver_section([resolved("one.example.test", 8080, ["10.5.5.5"]),
resolved("two.example.test", 8080, ["10.5.5.5"])]))
payload = gateway.parse(snap, "gateway")
rows = [row for row in payload["connections"] if row["in"]["port"] in (4443, 8443)]
assert [row["connections"] for row in rows] == [2, 2], "the endpoint's total, twice"
assert [row["shared"] for row in rows] == [True, True]
assert gateway.distinct_endpoint_counts(payload["connections"]) == [2], (
"and one figure for anything that adds the outbound side up")
assert out_metrics(payload) == ["gateway.out_conns@one.example.test:8080",
"gateway.out_conns@two.example.test:8080"]
def test_two_spellings_of_one_multi_homed_name_are_one_endpoint() -> None:
"""A name with three addresses is **one** route, not three, and its sockets are one count.
Every key the resolver published points at the same counter row, so a client round-robin across
`10.5.5.5`/`10.5.5.6` reads as three connections on one pipe. The case union-find exists for is
two names whose address sets *overlap* — a multi-homed host and a single-homed alias for one of
its addresses — which would otherwise count the same endpoint twice in one total.
"""
snap = snapshot(
nginx=nginx_section([nginx_server([4443], ["a.example.test"],
[proxy_pass(("multi.example.test", 443))]),
nginx_server([8443], ["b.example.test"],
[proxy_pass(("alias.example.test", 443))])]),
sockets=sockets_section(
established=[conn(("10.1.1.27", 40000), ("10.5.5.5", 443)),
conn(("10.1.1.27", 40002), ("10.5.5.6", 443))]),
resolver=resolver_section([resolved("multi.example.test", 443,
["10.5.5.5", "10.5.5.6"]),
resolved("alias.example.test", 443, ["10.5.5.6"])]))
payload = gateway.parse(snap, "gateway")
rows = [row for row in payload["connections"] if row["in"]["port"] in (4443, 8443)]
assert [row["connections"] for row in rows] == [2, 2], (
"the shared endpoint's two sockets, on both lines — not one socket each")
assert [row["shared"] for row in rows] == [True, True]
assert gateway.distinct_endpoint_counts(payload["connections"]) == [2], (
"overlapping address sets are one endpoint, so a total books them once")
def test_a_name_that_bridges_two_others_merges_all_three_lines_into_one_endpoint() -> None:
"""A→{x}, B→{x,y}, C→{y}: three routes, one endpoint, one figure — transitivity included.
The bridging name is what makes this a *merge* rather than three pairwise comparisons: the first
and third share no address, and yet the same kernel endpoint sits behind both, so counting them
separately would book two endpoints where `/proc` has one. **The config order is load-bearing**
beyond the transitivity: the bridge arrives second and its lowest address (`10.5.5.5`) is lower
than the root the first route already built, so that whole group is re-parented, and the third
route's lookup then has to walk through two links to find it — the path the union-find compresses
so a fan-out of this size stays cheap on the next poll.
"""
snap = snapshot(
nginx=nginx_section([nginx_server([4443], ["a.example.test"],
[proxy_pass(("multi.example.test", 443))]),
nginx_server([8443], ["b.example.test"],
[proxy_pass(("bridge.example.test", 443))]),
nginx_server([8444], ["c.example.test"],
[proxy_pass(("alias.example.test", 443))])]),
sockets=sockets_section(
established=[conn(("10.1.1.27", 40000), ("10.5.5.6", 443)),
conn(("10.1.1.27", 40002), ("10.5.5.7", 443))]),
resolver=resolver_section([resolved("multi.example.test", 443,
["10.5.5.6", "10.5.5.7"]),
resolved("bridge.example.test", 443,
["10.5.5.5", "10.5.5.6"]),
resolved("alias.example.test", 443, ["10.5.5.7"])]))
payload = gateway.parse(snap, "gateway")
rows = [row for row in payload["connections"] if row["in"]["port"] in (4443, 8443, 8444)]
assert len(rows) == 3
assert [row["connections"] for row in rows] == [2, 2, 2], (
"one endpoint's two sockets, repeated on all three lines")
assert [row["shared"] for row in rows] == [True, True, True]
assert gateway.distinct_endpoint_counts(payload["connections"]) == [2], (
"and one figure for any total, however many names bridged their way to it")
def test_a_resolver_entry_that_publishes_nothing_address_like_places_nothing() -> None:
"""A malformed answer is a name with no address, not an address that matches by luck.
The agent publishes strings it got from `getaddrinfo`, and this module has no reason to believe
them: an entry whose `addresses` are not addresses yields no endpoint key, so the route keeps its
dash and its missing gauge exactly as an `nxdomain` does. The alternative — comparing the junk
text against socket addresses — is how a typo in one file could credit a socket to a route.
"""
snap = snapshot(
nginx=nginx_section([nginx_server([4443], ["odd.example.test"],
[proxy_pass(("odd.example.test", 443))])]),
sockets=sockets_section(established=[conn(("10.1.1.27", 40000), ("10.5.5.5", 443))]),
resolver=resolver_section([
resolved("odd.example.test", 443, ["localhost", "10.5.5.5.6", ""])]))
row = routes_to(gateway.parse(snap, "gateway"), "odd.example.test")[0]
assert row["out"]["resolved"] == ["localhost", "10.5.5.5.6"], (
"the agent's bytes verbatim, minus the empty string — the payload is not where this is fixed")
assert row["connections"] is None and row["metric"] is None, (
"and not one of them is an address to match a socket with")
def test_a_line_with_no_endpoint_contributes_nothing_to_a_total() -> None:
"""`distinct_endpoint_counts` skips the unplaced line instead of offering it as a zero.
The caller turns "no figures at all" into `null` (`api._live_total`), which is the arm an
ingress of nothing-but-unresolved-names lands in. A `null` inside the list instead would make
every total on the page `null` because one route has a name nobody could place — and rule 2's
complaint about that shape is precisely that it invents a fault in the healthy routes beside it.
"""
snap = snapshot(
nginx=nginx_section([nginx_server([4443], ["up.example.test"],
[proxy_pass(("up.example.test", 443))]),
nginx_server([8443], ["gone.example.test"],
[proxy_pass(("gone.example.test", 443))])]),
sockets=sockets_section(established=[conn(("10.1.1.27", 40000), ("10.5.5.5", 443))]),
resolver=resolver_section([resolved("up.example.test", 443, ["10.5.5.5"]),
resolved("gone.example.test", 443, [], status="nxdomain")]))
payload = gateway.parse(snap, "gateway")
placed, unplaced = (routes_to(payload, "up.example.test")[0],
routes_to(payload, "gone.example.test")[0])
assert (placed["connections"], unplaced["connections"]) == (1, None)
assert gateway.distinct_endpoint_counts(payload["connections"]) == [1], (
"the placed endpoint alone: no second entry, no None, no zero")
assert gateway.distinct_endpoint_counts([unplaced]) == [], (
"and a payload of nothing-but-dashes is no figures at all, which the API renders as null")
def test_an_unresolved_name_publishes_no_metric_and_no_sample() -> None:
"""`nxdomain`, an error, and no entry at all are one answer here: drawn, never counted.
Three arms in one snapshot — a name the resolver placed, a name it looked up and got nothing
for, and a name it never mentioned — so the difference between "no record" and "nobody asked"
stays visible in the resolver block while the **route rows** behave identically: `connections`
`null`, `metric` `null`, no sample, and never a `0` that reads as an idle route.
"""
snap = snapshot(
nginx=nginx_section([
nginx_server([4443], ["up.example.test"], [proxy_pass(("up.example.test", 443))]),
nginx_server([8443], ["gone.example.test"], [proxy_pass(("gone.example.test", 443))]),
nginx_server([9443], ["silent.example.test"],
[proxy_pass(("silent.example.test", 443))])]),
sockets=sockets_section(
established=[conn(("10.1.1.27", 40000), ("10.5.5.5", 443)),
conn(("10.1.1.27", 40002), ("10.5.5.9", 443))]),
resolver=resolver_section([
resolved("up.example.test", 443, ["10.5.5.5"]),
resolved("gone.example.test", 443, [], status="nxdomain")]))
payload = gateway.parse(snap, "gateway")
assert routes_to(payload, "up.example.test")[0]["connections"] == 1
for name in ("gone.example.test", "silent.example.test"):
row = routes_to(payload, name)[0]
assert row["out"]["resolved"] is None, "never `[]`: unknown is one test, not a length"
assert row["out"]["via"] is None and row["out"]["resolved_ts"] is None
assert row["connections"] is None, "no count, and no 0 standing in for one"
assert row["metric"] is None, "no series for a number that cannot exist"
stored = [g.metric for g in gateway.samples(payload, 5.0)
if g.metric.startswith("gateway.out_conns@")]
assert stored == ["gateway.out_conns@up.example.test:443"]
assert sum(row["connections"] or 0 for row in payload["connections"]) == 1, (
"the socket to the unresolved target is credited to nothing at all")
def test_a_resolver_section_that_did_not_answer_places_nothing(gateway_snapshot: Any) -> None:
"""`available: false` and `available: null` are the agent's answers, and neither joins.
`false` is "there was nothing to resolve" (no proxy config was read) and `null` is "a pass that
could not run"; both arrive with no usable entries, and the payload's `resolver.available` says
which one it was rather than leaving a page to infer it from an empty list.
"""
for available in (False, None):
snap = snapshot(
nginx=nginx_section([nginx_server([4443], ["a.example.test"],
[proxy_pass(("x.example.test", 443))])]),
sockets=sockets_section(
established=[conn(("10.1.1.27", 40000), ("10.5.5.5", 443))]),
resolver=resolver_section([resolved("x.example.test", 443, ["10.5.5.5"])],
available=available, via=None))
payload = gateway.parse(snap, "gateway")
assert payload["resolver"]["available"] is available
assert payload["resolver"]["entries"] == [], "nothing was indexed from a dead pass"
assert routes_to(payload, "x.example.test")[0]["connections"] is None, (
"an entry beside `available: null` is a stale answer, not a joinable address")
def test_the_resolver_block_names_the_entries_the_index_was_built_from(
gateway_snapshot: Any) -> None:
"""The payload carries the document it was drawn from, so provenance is one read.
`entries` is every entry the section published — the index filters **within** it, so a page can
still say *this name has no record* — and each is projected to the five keys the agent owns.
`via` is the section's string. The `stub` arm keeps the same key set with nulls, which is what
the parse/stub parity pin in `tests/test_api.py` compares.
"""
payload = gateway.parse(gateway_snapshot("resolved"), "gateway")
block = payload["resolver"]
assert set(block) == {"available", "via", "entries"}
assert block["available"] is True and block["via"] == "libc"
document = json.loads((FIXTURES / "gateway_resolver.json").read_text())
assert [row["name"] for row in block["entries"]] == [row["name"] for row in document["entries"]]
assert [row["port"] for row in block["entries"]] == [row["port"] for row in document["entries"]]
assert all(set(row) == {"name", "port", "status", "addresses", "ts"}
for row in block["entries"]), "no agent field reaches a page unreviewed"
assert {row["status"] for row in block["entries"]} == {"ok", "literal"}, (
"this capture resolves every name it was asked about, plus the two `127.0.0.1` targets it "
"recognised as addresses without asking at all")
empty = gateway.stub("gateway")
assert empty["resolver"] == {"available": None, "via": None, "entries": []}, (
"a host that has not answered has not said resolution is unavailable")
def test_the_dashboard_still_never_resolves(gateway_snapshot: Any) -> None:
"""The spy pin, restated for phase 32 rather than relaxed by it.
`parse()` may **read** a resolver document and must not produce one, so the spy now runs over
three paths: a synthetic snapshot whose every target is a name, the same with a resolver section
full of answers, and the captured `resolved` ingress — then `samples()` over the result, since
a gauge name is built from a host string and is the other place a resolver could creep in.
Why the split is the design and not a limitation: the agent resolves with `getaddrinfo` on the
host that **opens** these connections (haproxy's `init-addr last,libc,none`), so its answer is
the one that decides what `/proc` then says. The dashboard asking the same name through its own
`resolv.conf` would answer a different question — a different server, a different view, and on
this fleet a name that resolves back through this same router — and would then attribute a
socket to a route on the strength of a lookup the router never made.
"""
calls: list[tuple[Any, ...]] = []
def spy(*args: Any, **kwargs: Any) -> Any: # pragma: no cover - must never run
calls.append((args, kwargs))
raise AssertionError("the gateway join must never resolve a name")
named = snapshot(
haproxy=haproxy_section([frontend("mc", [25565])],
[backend("mc-be", [server("mc1", "minecraft.example.test",
25565)])]),
nginx=nginx_section([nginx_server([443], ["a.example.test"],
[proxy_pass(("app.example.test", 8080))])]),
sockets=sockets_section(listens=[listen(25565), listen(443)],
established=[conn(("10.1.1.27", 443), ("203.0.113.5", 900)),
conn(("10.1.1.27", 1), ("10.2.2.2", 8080))]))
answered = snapshot(
haproxy=named["haproxy"], nginx=named["nginx"], sockets=named["sockets"],
resolver=resolver_section([resolved("minecraft.example.test", 25565, ["10.9.9.9"]),
resolved("app.example.test", 8080, ["10.2.2.2"])]))
original = socket_module.getaddrinfo
socket_module.getaddrinfo = spy
try:
for document in (named, answered, gateway_snapshot("resolved")):
payload = gateway.parse(document, "gateway")
gateway.samples(payload, 5.0)
gateway.warnings(gateway._section(document, "haproxy"),
gateway._section(document, "nginx"))
finally:
socket_module.getaddrinfo = original
assert calls == []
def test_a_snapshot_without_a_resolver_section_is_phase_30_verbatim(
gateway_snapshot: Any) -> None:
"""A 1.2.0 agent, or a 1.3.0 one run with `--no-resolve`, is **not** a changed page.
The topology, the sections, the warnings and the two loopback counts are byte-for-byte what
phase 30 produced; what differs is three `null`s on every `out` and an empty `resolver` block,
plus the routes phase 32 makes countable. `supported` stays `true` — a missing resolver is a
missing measurement, not an unsupported agent, and a fleet running two versions (which this one
did until task 05) must not read half-red.
"""
without = gateway.parse(gateway_snapshot("full"), "gateway")
with_resolver = gateway.parse(gateway_snapshot("resolved"), "gateway")
assert without["supported"] is True and with_resolver["supported"] is True
assert without["resolver"] == {"available": None, "via": None, "entries": []}
assert all(row["out"]["resolved"] is None and row["out"]["via"] is None
and row["out"]["resolved_ts"] is None for row in without["connections"])
def shape(payload: Mapping[str, Any]) -> Any:
"""Everything except the counts, which is the one thing the section may move."""
return {
"routes": [row["id"] for row in payload["connections"]],
"out": [(row["out"]["host"], row["out"]["port"], row["out"]["name"],
row["out"]["kind"], tuple(row["out"]["selectors"]), row["out"]["address"])
for row in payload["connections"]],
"trunks": [(row["port"], row["source"], tuple(row["domains"]))
for row in payload["listeners"]],
"sections": payload["sections"],
"why": payload["why"],
"warnings": payload["warnings"],
"sockets": payload["sockets"],
}
assert shape(without) == shape(with_resolver), (
"the resolver section counts through an address; it moves no route, trunk or warning")
counted = {row["id"] for row in without["connections"] if row["connections"] is not None}
assert counted == {"nginx:443->127.0.0.1:4443", "nginx:8443->127.0.0.1:8081"}, (
"phase 30 counted exactly the two endpoints the config wrote as addresses")
def test_resolution_raises_the_counted_route_count_from_two_to_every_name(
gateway_snapshot: Any) -> None:
"""The change, as one printed number rather than as an implied one.
Phase 30 could count the two routes whose target the configs wrote as an address. The recorded
resolver document places the other thirty-one, so against the captures every route on the
ingress now carries a figure — which is what the operator's "the only animated line is `:443 ->
127.0.0.1:4443`" was really asking for.
"""
before = gateway.parse(gateway_snapshot("full"), "gateway")
after = gateway.parse(gateway_snapshot("resolved"), "gateway")
counted_before = [row for row in before["connections"] if row["connections"] is not None]
counted_after = [row for row in after["connections"] if row["connections"] is not None]
written_as_name = [row for row in after["connections"] if row["out"]["address"] is not True]
unplaced = sorted(row["out"]["host"] for row in written_as_name
if row["connections"] is None)
report = (f"/gateway counted routes: {len(counted_before)} "
f"(phase 30, addresses only) → {len(counted_after)} of {len(after['connections'])}; "
f"the resolver placed {len(written_as_name)} name-written routes")
print(report)
assert len(counted_before) == 2, "phase 30's figure, so the delta is measurable"
assert not unplaced, f"{report} — unplaced: {unplaced}"
assert len(counted_after) == len(after["connections"]), (
f"{report} — a route the document places must carry a count")
def test_two_ports_on_one_host_are_two_gauges_even_when_both_names_resolve() -> None:
"""`minecraft:25565` and `minecraft:25566` stay two series — the port earns its place in the name.
This is why phase 29 put the port in `gateway.out_conns@<host>:<port>` in the first place, and
resolution does not remove the reason: this fleet's two haproxy Minecraft backends name the *same*
domain on two ports, libc answers one address for both, and a gauge keyed on the name alone would
have the second poll overwrite the first in silence. Two endpoints, two counters, two series — and
the fact that they share an address is invisible in the names and visible in the traffic.
"""
snap = snapshot(
haproxy=haproxy_section(
[frontend("mc", [25565], default="mc-a"), frontend("mc2", [25566], default="mc-b")],
[backend("mc-a", [server("mc1", "minecraft.example.test", 25565)]),
backend("mc-b", [server("mc2", "minecraft.example.test", 25566)])]),
sockets=sockets_section(established=[
conn(("10.1.1.27", 40000), ("10.9.9.9", 25565)),
conn(("10.1.1.27", 40002), ("10.9.9.9", 25565)),
conn(("10.1.1.27", 40004), ("10.9.9.9", 25566))]),
resolver=resolver_section([resolved("minecraft.example.test", 25565, ["10.9.9.9"]),
resolved("minecraft.example.test", 25566, ["10.9.9.9"])]))
payload = gateway.parse(snap, "gateway")
by_port = {row["in"]["port"]: row for row in payload["connections"]}
assert by_port[25565]["metric"] == "gateway.out_conns@minecraft.example.test:25565"
assert by_port[25566]["metric"] == "gateway.out_conns@minecraft.example.test:25566"
assert [by_port[25565]["connections"], by_port[25566]["connections"]] == [2, 1], (
"two endpoints, two counts — one address does not merge two ports")
assert by_port[25565]["shared"] is False and by_port[25566]["shared"] is False, (
"sharing an *address* is not sharing an *endpoint*, which is what `shared` means")
stored = gateway.samples(payload)
assert [(s.metric, s.value) for s in stored if "out_conns" in s.metric] == [
("gateway.out_conns@minecraft.example.test:25565", 2),
("gateway.out_conns@minecraft.example.test:25566", 1)], (
"neither gauge overwrites the other on the way to the store")
assert gateway.distinct_endpoint_counts(payload["connections"]) == [2, 1], (
"and the total books both, because the kernel has both")
# --------------------------------------------------------------------------- #
# the map: which maps route, and which are just header words
# --------------------------------------------------------------------------- #
def test_a_map_no_proxy_pass_references_contributes_nothing() -> None:
"""`$connection_upgrade`'s values are `upgrade`/`close` — header words, never a host."""
snap = snapshot(
nginx=nginx_section([nginx_server([443], [], [proxy_pass(None, variable="targetBackend")])],
maps=[mapping_block("connection_upgrade",
[("default", ("upgrade", None)),
("", ("close", None))]),
mapping_block("targetBackend",
[("a.example.test", ("be.example.test", 443))])]),
sockets=sockets_section(listens=[listen(443)]))
payload = gateway.parse(snap, "gateway")
assert [row["key"] for row in payload["nginx"]["map_targets"]] == ["a.example.test"]
assert [row["out"]["host"] for row in payload["connections"]] == ["be.example.test"]
def test_a_map_entry_with_no_address_is_skipped() -> None:
snap = snapshot(
nginx=nginx_section([nginx_server([443], [], [proxy_pass(None, variable="be")])],
maps=[mapping_block("be", [("a.example.test", None),
("b.example.test", ("be2.example.test", 443))])]))
payload = gateway.parse(snap, "gateway")
assert [row["key"] for row in payload["nginx"]["map_targets"]] == ["b.example.test"]
def test_one_endpoint_reached_from_both_configs_is_two_lines_of_one_number() -> None:
"""haproxy routes `:80` to `app.example.test:9000` and an nginx vhost routes `:8443` there too.
Two files, two inbound ports, one endpoint: the graph keeps **both lines** (each is a route a
reader has to see, and merging them would erase which file declares which) and stamps each with
the same endpoint count, `shared`. What the merge rule inside `config_connections` still does is
keep the name a claim carries within one `(source, in_port, host, port)` key — tested here on
the haproxy line, whose `server` name no other claim has.
"""
snap = snapshot(
haproxy=haproxy_section([frontend("app", [80], default="app-be", use=["app-be"])],
[backend("app-be", [server("app1", "app.example.test", 9000)])]),
nginx=nginx_section([nginx_server([8443], ["app.example.test"],
[proxy_pass(("app.example.test", 9000))])]),
sockets=sockets_section(established=[conn(("10.1.1.27", 40000), ("10.5.5.5", 9000))]))
payload = gateway.parse(snap, "gateway")
rows = routes_to(payload, "app.example.test")
assert [row["source"] for row in rows] == ["haproxy", "nginx"]
assert rows[0]["out"]["name"] == "app1", "the haproxy `server` name, kept verbatim"
assert rows[0]["in"]["port"] == 80, "one line despite two routes naming the same backend"
assert rows[1]["out"]["selectors"] == ["app.example.test"]
def test_a_pool_names_its_upstream_servers_and_an_undeclared_pool_draws_nothing() -> None:
"""Derivation 3: `proxy_pass http://<name>` (no port) is an `upstream` block, and its members
are the endpoints. No block of that name is **no connection** — and no invented host either.
The payload cannot say where the traffic lands, so it claims nothing (rule 2). Task 03 decides
whether an undeclared pool is a config-integrity warning; this task draws no line for it, and
says so by pinning an empty graph rather than a line to `backend_pool:80`.
"""
snap = snapshot(nginx=nginx_section(
[nginx_server([4443], ["app.example.test"], [pool_pass("backend_pool")])],
upstreams=[upstream("backend_pool", [("10.0.0.1", 3000), ("10.0.0.2", 3000)])]))
rows = gateway.parse(snap, "gateway")["connections"]
assert [(row["out"]["host"], row["out"]["port"], row["out"]["kind"]) for row in rows] == [
("10.0.0.1", 3000, gateway.KIND_POOL), ("10.0.0.2", 3000, gateway.KIND_POOL)]
assert [row["out"]["selectors"] for row in rows] == [["backend_pool"], ["backend_pool"]], (
"the pool name is the config's own word for why these two endpoints are here")
assert [row["metric"] for row in rows] == [
"gateway.out_conns@10.0.0.1:3000", "gateway.out_conns@10.0.0.2:3000"], (
"a pool member is an address, so both lines are joinable")
orphans = snapshot(nginx=nginx_section(
[nginx_server([4443], ["app.example.test"], [pool_pass("no_such_pool")])]))
payload = gateway.parse(orphans, "gateway")
assert payload["connections"] == [], "an undeclared pool names no endpoint, so it draws nothing"
assert listener_at(payload, 4443), "and the trunk it hangs off still exists"
def test_a_haproxy_connection_carries_the_backend_and_the_server_name(gateway_snapshot: Any) -> None:
"""Both of the config's own names for a haproxy route survive into the graph.
`out.name` is the `server` line's name and `out.selectors` the backend the frontend routes to —
the two words an operator greps for in `haproxy.cfg`, printed rather than paraphrased.
"""
payload = gateway.parse(gateway_snapshot("full"), "gateway")
row = next(row for row in payload["connections"] if row["in"]["port"] == 22)
assert row["out"]["name"] == "gitea"
assert row["out"]["kind"] == gateway.KIND_BACKEND
assert row["out"]["selectors"] == ["gitea-2222"]
assert row["source"] == "haproxy"
def test_a_frontend_that_routes_by_expression_is_a_listener_with_no_connection() -> None:
"""Refusal rule 5, synthetic half: a fetch expression names no backend, so nothing is joined.
The agent reports `use_backend %[req.hdr(host),map_dom(...)]` as **no** name (it interprets no
config), and "no name" and "no routing directive" are one value in this payload. Inventing a
route here would be rule 2, and inventing a *warning* about it would blame the operator for a
thing the payload cannot distinguish.
"""
snap = snapshot(haproxy=haproxy_section([{"name": "expr", "binds": [bind(80)],
"default_backend": None, "backends": []}]))
payload = gateway.parse(snap, "gateway")
assert listener_at(payload, 80), "the trunk exists"
assert payload["connections"] == [], "and no line is drawn out of it"
def test_a_proxy_pass_with_neither_target_nor_variable_draws_nothing() -> None:
"""The other half of refusal rule 5: a unix socket or a bare word names no endpoint."""
snap = snapshot(nginx=nginx_section([nginx_server(
[443], ["a.example.test"],
[{"location": "/", "scheme": None, "target": None, "variable": None}])]))
payload = gateway.parse(snap, "gateway")
assert payload["connections"] == []
assert listener_at(payload, 443), "the listen port is still a trunk the page draws"
def test_a_proxy_pass_naming_no_host_is_no_line() -> None:
"""`target: {host: ""}` is drift, and drift yields no line rather than a line to nowhere."""
snap = snapshot(nginx=nginx_section([nginx_server(
[443], [], [{"location": "/", "scheme": "http", "target": {"host": " ", "port": 80},
"variable": None}])]))
payload = gateway.parse(snap, "gateway")
assert payload["connections"] == []
assert payload["nginx"]["servers"] == [{"port": 443, "server_names": []}]
def test_the_readers_refuse_the_shapes_that_are_not_values() -> None:
"""The three guards every section read goes through, on the inputs that are not data.
Pinned directly (the way `tests/test_config.py` pins `_flag`/`_num`) because each one is the
difference between a pipe and no pipe: an empty string is not an address, a bare string is
not a list of names, and whitespace is not a host.
"""
assert gateway._is_ip_literal("") is False
assert gateway._is_ip_literal(None) is False
assert gateway._is_ip_literal("203.0.113.7") is True
assert gateway._is_ip_literal("[::1]") is False, "/proc spells addresses without brackets"
assert gateway._is_ip_literal("app.example.test") is False
assert gateway._names("not a list") == []
assert gateway._names(None) == []
assert gateway._names([None, "a-be", "a-be", " ", "b-be"]) == ["a-be", "b-be"]
assert gateway._target_of({"host": " ", "port": 80}) is None
assert gateway._target_of(["host", 80]) is None
assert gateway._target_of({"host": "be.example.test", "port": "nope"}) == (
"be.example.test", None)
# --------------------------------------------------------------------------- #
# the degradation arms: three ways to have nothing, none of them red
# --------------------------------------------------------------------------- #
def test_an_agent_without_the_sections_predates_gateway_support(gateway_snapshot: Any) -> None:
"""The 1.1.0 / `--no-proxy` arm: same keys, empty and null, one sentence, no verdict."""
payload = gateway.parse(gateway_snapshot("predates"), "gateway")
assert payload["supported"] is False
assert payload["why"] == gateway.WHY_PREDATES
assert "predates gateway support" in payload["why"]
assert payload["haproxy"] == {"frontends": []}
assert payload["nginx"] == {"servers": [], "map_targets": []}
assert payload["listeners"] == [] and payload["connections"] == []
assert payload["sockets"] == {"listens": None, "established": None}
assert payload["sections"] == {"haproxy": None, "nginx": None, "sockets": None}
assert gateway.samples(payload, 1.0) == [], "an unknown count publishes no gauge"
def test_a_host_running_no_haproxy_says_so_and_keeps_the_nginx_pipes(gateway_snapshot: Any) -> None:
payload = gateway.parse(gateway_snapshot("no_haproxy"), "gateway")
assert payload["supported"] is True, "the agent knows the feature; the host has no haproxy"
assert payload["sections"]["haproxy"] is False
assert "no haproxy" in (payload["why"] or "")
assert "not found" not in (payload["why"] or ""), "the page's sentence, not the agent's log"
assert payload["section_errors"], "the agent's own line still travels, verbatim"
assert any("not found" in line for line in payload["section_errors"])
assert [row["port"] for row in payload["nginx"]["servers"]] == [80, 443, 4443, 8443]
assert [row["port"] for row in payload["listeners"] if row["source"] == "nginx"] == \
[80, 443, 4443, 8443]
assert payload["haproxy"]["frontends"] == []
assert [row for row in payload["connections"] if row["source"] == "haproxy"] == [], (
"no haproxy file, so no haproxy route — and no invented one to fill the gap")
assert [row for row in payload["connections"] if row["source"] == "nginx"], (
"the nginx half of the map is untouched by the other file being absent")
def test_an_unreadable_config_is_a_different_sentence_from_no_config() -> None:
snap = snapshot(haproxy=haproxy_section([], available=None),
nginx=nginx_section([nginx_server([80], ["a.example.test"])]),
sockets=sockets_section(listens=[listen(80)]))
payload = gateway.parse(snap, "gateway")
assert payload["sections"]["haproxy"] is None
assert "could not be read or parsed" in (payload["why"] or "")
assert [row["port"] for row in payload["listeners"]] == [80]
def test_a_socket_table_that_did_not_answer_leaves_every_count_null(
gateway_snapshot: Any) -> None:
"""`available: null` is "no answer", and the only honest counts are `null` (rule 2)."""
payload = gateway.parse(gateway_snapshot("sockets_silent"), "gateway")
assert payload["sections"]["sockets"] is None
assert "sockets:" in (payload["why"] or "")
assert payload["sockets"] == {"listens": None, "established": None}
assert payload["listeners"], "the configured ports are still drawn — the config did answer"
assert all(row["connections"] is None for row in payload["listeners"])
assert payload["connections"], "and so is every route the two files declare"
assert all(row["connections"] is None for row in payload["connections"]), (
"a silent socket table says nothing about any route, not even about the joinable ones")
assert gateway.samples(payload, 1.0) == []
def test_a_partial_snapshot_names_the_missing_section_only() -> None:
"""A hypothetical agent publishing `sockets` but no configs: supported, with a gap named."""
snap = snapshot(sockets=sockets_section(listens=[listen(9999)]))
payload = gateway.parse(snap, "gateway")
assert payload["supported"] is True
assert payload["sections"] == {"haproxy": None, "nginx": None, "sockets": True}
assert payload["listeners"] == [], "9999 is a listener nothing claims, so it is not a pipe"
assert payload["sockets"] == {"listens": 1, "established": 0}, "the kernel's number still travels"
assert "haproxy" in (payload["why"] or "") and "nginx" in (payload["why"] or "")
# --------------------------------------------------------------------------- #
# what the payload must never contain, and the gauges it must
# --------------------------------------------------------------------------- #
def test_no_count_carries_a_verdict_or_a_window(gateway_snapshot: Any) -> None:
"""`net.*` precedent for colour, rule 2 for the `Δ` label: a count is a count.
Both arms are walked, so a `severity` bolted onto a pipe or a `window_seconds` copied from
`sources/host.py` fails here — the first would be a judgement `severity.py` never made, the
second would claim a measurement this source does not take.
"""
for mode in ("full", "predates", "sockets_silent"):
keys = walk_keys(gateway.parse(gateway_snapshot(mode), "gateway"))
assert "severity" not in keys, mode
assert not {key for key in keys if "window" in key or key.startswith("Δ")}, mode
def test_the_payload_is_json_serialisable_end_to_end(gateway_snapshot: Any) -> None:
"""It goes out over `/api/gateway/snapshot` verbatim, so nothing in it may be exotic."""
json.dumps(gateway.parse(gateway_snapshot("full"), "gateway"))
def test_samples_are_one_gauge_per_pipe_under_the_gateway_family(gateway_snapshot: Any) -> None:
snap = gateway_snapshot("full")
payload = gateway.parse(snap, "gateway")
rows = gateway.samples(payload, 1_760_000_000.0)
assert rows, "a live gateway publishes gauges"
assert {row.source for row in rows} == {gateway.SOURCE}
assert {row.host for row in rows} == {"gateway"}
named = {row.metric: row.value for row in rows}
assert named["gateway.in_conns@443"] == float(listener_at(payload, 443)["connections"])
# one gauge per **metric name** that has a number: shared endpoints appear on several lines and
# must be stored once, or the history doubles the traffic for one reading
expected = {row["metric"]: float(row["connections"])
for group in ("listeners", "connections") for row in payload[group]
if row["metric"] is not None and row["connections"] is not None}
assert named == expected, "one gauge per measured endpoint, and none for an unmeasured one"
assert len(rows) == len(named), "no metric is emitted twice by one poll"
for row in payload["connections"]:
if row["metric"] is None:
assert row["connections"] is None, "a line with no gauge has no number either"
def test_parse_gateway_snapshot_stamps_its_samples_from_the_snapshot() -> None:
"""The x-axis is when the agent took the snapshot, the way `sources/disks.py` does it."""
result = gateway.parse_gateway_snapshot(
"gateway", snapshot(haproxy=haproxy_section([frontend("ssh", [22])]),
sockets=sockets_section(listens=[listen(22)])), now=999.0)
assert result.events == [], "a connection count has no transition worth an event line"
assert [sample.ts for sample in result.samples] == [1_760_000_000.0]
assert [sample.metric for sample in result.samples] == ["gateway.in_conns@22"]
def test_a_snapshot_with_no_timestamp_falls_back_to_the_callers_clock() -> None:
doc = snapshot(haproxy=haproxy_section([frontend("ssh", [22])]),
sockets=sockets_section(listens=[listen(22)]))
doc.pop("ts")
result = gateway.parse_gateway_snapshot("gateway", doc, now=42.0)
assert [sample.ts for sample in result.samples] == [42.0]
def test_the_gauges_skip_a_pipe_with_no_number() -> None:
"""No sample for an unknown count — a chart of zeros for a dead socket table is the
failure mode rule 2 names, and `sources/disks.py` already refuses it for unreadable drives."""
payload = gateway.parse(snapshot(haproxy=haproxy_section([frontend("ssh", [22])])), "gateway")
assert payload["listeners"][0]["connections"] is None
assert gateway.samples(payload, 1.0) == []
def test_a_snapshot_whose_sections_are_empty_is_an_honest_empty_page() -> None:
"""A 1.2.0 agent on a box with two empty configs: supported, and nothing drawn."""
snap = snapshot(haproxy=haproxy_section([]), nginx=nginx_section(),
sockets=sockets_section())
payload = gateway.parse(snap, "gateway")
assert payload["supported"] is True and payload["why"] is None
assert payload["listeners"] == [] and payload["connections"] == []
assert payload["sockets"] == {"listens": 0, "established": 0}
def test_junk_in_the_sections_yields_no_pipes_and_no_exception() -> None:
"""The source is a projection of a document it does not control — drift is `n/a`, not a crash."""
snap: dict[str, Any] = {"ts": 1.0, "host": "gateway",
"haproxy": {"available": True, "frontends": [3, None],
"backends": {"nope": 1}},
"nginx": {"available": True, "servers": "not a list", "maps": None},
"sockets": {"available": True, "listens": [None],
"established": [{"local": 5, "remote": None}]}}
payload = gateway.parse(snap, "gateway")
assert payload["listeners"] == [] and payload["connections"] == []
assert payload["sockets"] == {"listens": 0, "established": 1}
def test_a_port_of_zero_or_a_name_of_nothing_is_no_match() -> None:
"""`0` is not a port and an empty target is not a host, so neither becomes a pipe (rule 2)."""
snap = snapshot(
haproxy=haproxy_section([frontend("u", [None, 0]), frontend("ok", [22], default="b")],
[backend("b", [server("?", "", 22), server("z", "z.example.test",
0)])]),
sockets=sockets_section(listens=[listen(0), listen(22)]))
payload = gateway.parse(snap, "gateway")
assert [row["port"] for row in payload["haproxy"]["frontends"]] == [22]
assert [row["out"]["host"] for row in payload["connections"]] == ["z.example.test"]
row = payload["connections"][0]
assert row["out"]["port"] is None, "a 0 is not a port, so the line carries no port"
assert row["metric"] is None and row["connections"] is None, (
"and therefore nothing to join on and nothing to publish")
assert [entry["port"] for entry in payload["listeners"]] == [22], (
"a 0 bind is not a listening port either")
def test_a_use_backend_map_is_reported_as_routes_not_one_chosen_backend() -> None:
"""haproxy picks per request here; the payload names every candidate and chooses none."""
snap = snapshot(haproxy=haproxy_section(
[frontend("fe", [80], default="base-be", use=("blue-be", "green-be"))]))
row = gateway.parse(snap, "gateway")["haproxy"]["frontends"][0]
assert row["backend"] == "base-be"
assert row["routes"] == ["base-be", "blue-be", "green-be"]
# --------------------------------------------------------------------------- #
# config-integrity warnings: seven kinds, each a config that does not connect to itself
#
# Each kind gets a synthetic config written in the shape the agent publishes (the builders above
# are the section document, not a fake tree: `warnings()` is a pure function over sections, and a
# file round-trip would only re-test the agent's parser). The captured configs then pin that the
# seven checks are not a lie about THIS fleet: zero warnings, and a stated reason per kind.
# --------------------------------------------------------------------------- #
def kinds(payload_warnings: Sequence[Mapping[str, str]]) -> list[str]:
"""The `kind` of every warning, in the order `warnings()` produced them."""
return [row["kind"] for row in payload_warnings]
def warning_for(payload_warnings: Sequence[Mapping[str, str]], kind: str) -> Mapping[str, str]:
"""The one warning of `kind`, failing if there is not exactly one of them."""
rows = [row for row in payload_warnings if row["kind"] == kind]
assert len(rows) == 1, f"expected exactly one {kind}, got {rows}"
return rows[0]
def test_a_dangling_use_backend_warns_by_frontend_and_backend() -> None:
"""`haproxy_route_dangling`: a route naming a backend nobody declares.
Synthetic sections (the pure-function route): the check is over the two lists the agent
publishes, and the fault is one line of a file, not a file's layout on disk.
"""
section = haproxy_section(
[frontend("haproxy-22", [22], default="gitea-2222"),
frontend("haproxy-443", [443], use=("live-be", "gone-be"))],
[backend("live-be", [server("s", "10.0.0.1", 80)])])
found = gateway.haproxy_warnings(section)
assert kinds(found) == [gateway.WARN_ROUTE_DANGLING, gateway.WARN_ROUTE_DANGLING]
row = found[0]
assert row["id"] == "haproxy_route_dangling:haproxy-22->gitea-2222"
assert row["kind"] == gateway.WARN_ROUTE_DANGLING
assert row["subject"] == "haproxy-22->gitea-2222"
assert row["message"] == ('frontend haproxy-22 routes to backend "gitea-2222", '
"which no backend block declares")
assert found[1]["id"] == "haproxy_route_dangling:haproxy-443->gone-be", (
"every dangling name is its own warning — collapsing them would hide the second one")
def test_a_backend_no_frontend_routes_to_warns_with_its_server_count() -> None:
"""`haproxy_backend_orphan`: servers with nowhere to come from, counted as the file wrote it."""
section = haproxy_section(
[frontend("haproxy-22", [22], default="used-be")],
[backend("used-be", [server("s", "10.0.0.1", 80)]),
backend("old-tunnel", [server("a", "10.0.0.2", 22), server("b", "10.0.0.3", 22)])])
row = warning_for(gateway.haproxy_warnings(section), gateway.WARN_BACKEND_ORPHAN)
assert row["message"] == ('backend "old-tunnel" has 2 servers and no frontend routes to it')
assert row["id"] == "haproxy_backend_orphan:old-tunnel"
# the singular, because the message prints the config's own count
one = haproxy_section([frontend("f", [22], default="used-be")],
[backend("used-be", [server("s", "10.0.0.1", 80)]),
backend("solo", [server("a", "10.0.0.9", 22)])])
assert "has 1 server and no frontend routes to it" in \
warning_for(gateway.haproxy_warnings(one), gateway.WARN_BACKEND_ORPHAN)["message"]
def test_a_backend_with_no_server_warns_that_it_routes_nowhere() -> None:
"""`haproxy_backend_empty`: the block exists, so a route to it is not dangling — it just ends."""
section = haproxy_section([frontend("haproxy-22", [22], default="gitea-2222")],
[backend("gitea-2222", [])])
found = gateway.haproxy_warnings(section)
assert kinds(found) == [gateway.WARN_BACKEND_EMPTY], (
"an empty backend is not also an orphan: `orphan` needs a server to have anywhere to go")
assert found[0]["message"] == 'backend "gitea-2222" declares no server — it routes nowhere'
def test_a_server_block_that_never_listens_warns() -> None:
"""`nginx_server_unlistened`: the vhost is configured and unreachable, in both its shapes."""
section = nginx_section([
nginx_server([], ["nextcloud.example.com"], [proxy_pass(("10.0.0.5", 11000))]),
nginx_server([], ["named-only.example.com"]),
nginx_server([443], ["fine.example.com"], [proxy_pass(("10.0.0.6", 443))]),
])
found = gateway.nginx_warnings(section)
assert kinds(found) == [gateway.WARN_SERVER_UNLISTENED, gateway.WARN_SERVER_UNLISTENED]
assert found[0]["message"] == ('nginx server "nextcloud.example.com" proxy_passes '
"but never listens")
assert found[1]["message"] == ('nginx server "named-only.example.com" '
"declares a server_name but never listens")
def test_a_proxy_pass_naming_an_undeclared_pool_warns() -> None:
"""`nginx_pool_undeclared`: task 01 drew no connection for it, and this says why."""
section = nginx_section(
[nginx_server([4443], ["app.example.com"], [pool_pass("backend_pool")]),
nginx_server([8443], ["other.example.com"],
[pool_pass("backend_pool")])],
upstreams=[upstream("a_real_pool", [("10.0.0.7", 3000)])])
found = gateway.nginx_warnings(section)
assert kinds(found) == [gateway.WARN_POOL_UNDECLARED, gateway.WARN_POOL_UNDECLARED]
assert found[0]["message"] == ('nginx listen 4443 proxy_passes to pool "backend_pool", '
"which no upstream block declares")
assert found[0]["id"] == "nginx_pool_undeclared:backend_pool@4443"
assert found[1]["id"] == "nginx_pool_undeclared:backend_pool@8443", (
"two listen ports reaching the one missing pool are two broken routes and two subjects")
# and the same block naming two ports is ONE warning, because the repair is one `proxy_pass`
both = nginx_section([nginx_server([4443, 8443], ["app.example.com"],
[pool_pass("backend_pool")])])
row = warning_for(gateway.nginx_warnings(both), gateway.WARN_POOL_UNDECLARED)
assert row["id"] == "nginx_pool_undeclared:backend_pool@4443+8443"
assert row["message"].startswith("nginx listen 4443, 8443 proxy_passes to pool")
def test_a_proxy_pass_variable_no_map_defines_warns() -> None:
"""`nginx_variable_unmapped`: the stream server routes to a variable nothing fills."""
section = nginx_section(
[nginx_server([443], [], [proxy_pass(None, variable="$targetBackend")]),
nginx_server([8443], [], [proxy_pass(None, variable="mapped")])],
maps=[mapping_block("mapped", [("a.example.com", ("10.0.0.8", 443))])])
row = warning_for(gateway.nginx_warnings(section), gateway.WARN_VARIABLE_UNMAPPED)
assert row["id"] == "nginx_variable_unmapped:targetBackend@443"
assert row["message"] == ("nginx listen 443 proxy_passes $targetBackend, "
"which no map block defines"), (
"the `$` is nginx's spelling and the message keeps it, whatever the agent publishes")
def test_a_referenced_map_entry_with_no_address_warns() -> None:
"""`nginx_map_route_missing`: the SNI key is routed to nothing."""
section = nginx_section(
[nginx_server([443], [], [proxy_pass(None, variable="targetBackend")])],
maps=[mapping_block("targetBackend",
[("x.example.com", None),
("ok.example.com", ("10.0.0.9", 443))])])
row = warning_for(gateway.nginx_warnings(section), gateway.WARN_MAP_ROUTE_MISSING)
assert row["id"] == "nginx_map_route_missing:targetBackend:x.example.com"
assert row["message"] == 'map $targetBackend entry "x.example.com" names no address to route to'
# --- the three refusals: each produces NO warning while still producing its other output --- #
def test_a_listen_with_no_proxy_pass_is_not_a_warning() -> None:
"""A static or redirect vhost is legal and looks identical here, so nothing is said about it.
This fleet's own `:80` catch-all is a `return 301`, and the agent publishes it as a `server`
with a `listen`, a `server_name` and no `proxy_pass` — exactly the shape a warning would blame.
Pinned in the direction that matters: the listener still exists, and the warning list stays
empty.
"""
section = nginx_section([nginx_server([80], ["_"])])
assert gateway.nginx_warnings(section) == []
payload = gateway.parse(snapshot(nginx=section, sockets=sockets_section(listens=[listen(80)])),
"gateway")
assert [row["port"] for row in payload["listeners"]] == [80], "the trunk is still drawn"
assert payload["connections"] == [], "and it is drawn alone"
assert payload["warnings"] == []
def test_a_frontend_routing_by_fetch_expression_is_not_a_warning() -> None:
""""no name" and "no routing directive" are one value on this wire, so neither is a fault.
`use_backend %[req.hdr(host),map_dom(...)]` arrives as a frontend with no backend name — the
same document a frontend with no routing at all produces. Warning about it would blame a config
this agent cannot read; the fix is an agent change, not a guess here. The listener survives and
the warning list stays empty.
"""
section = haproxy_section([frontend("dynamic", [443], use=())])
assert gateway.haproxy_warnings(section) == []
payload = gateway.parse(snapshot(haproxy=section,
sockets=sockets_section(listens=[listen(443)])), "gateway")
assert [row["port"] for row in payload["listeners"]] == [443]
assert payload["connections"] == []
assert payload["warnings"] == []
def test_a_proxy_pass_with_neither_target_nor_variable_is_not_a_warning() -> None:
"""A unix socket or a bare word is legitimate, undrawable, and not broken."""
bare = {"location": "/", "scheme": None, "target": None, "variable": None}
section = nginx_section([nginx_server([9000], ["sock.example.com"], [bare])])
assert gateway.nginx_warnings(section) == []
assert gateway.config_connections(None, section) == []
def test_an_unreferenced_map_with_no_address_entries_warns_about_nothing() -> None:
"""The `$connection_upgrade` map's `upgrade`/`close` are header words, not missing routes."""
section = nginx_section(
[nginx_server([80], ["_"], [{"location": "/", "scheme": "http",
"target": {"host": "10.0.0.4", "port": 80},
"variable": None}])],
maps=[mapping_block("connection_upgrade", [("default", None), ("", None)])])
assert gateway.nginx_warnings(section) == [], (
"a map no proxy_pass names is not a routing table, and its null targets are its business")
# --- the arms that must not be judged, and the pins on this fleet's real configs --- #
def test_an_unreadable_or_absent_section_warns_about_nothing(gateway_snapshot: Any) -> None:
"""A host that cannot be read has not been found guilty of anything (invariant 3).
Three arms, driven from the **real** captured tree through the real agent parser, and each one
still reports the half it *can* report: `no_haproxy` keeps its nginx warnings-free record,
`predates` warns about nothing because nothing was read.
"""
for mode in ("no_haproxy", "no_nginx", "predates"):
payload = gateway.parse(gateway_snapshot(mode), "gateway")
assert payload["warnings"] == [], mode
# and the same rule stated on hand-built sections, so it holds for `available: null` beside
# `available: false` (the fixture has no arm for "the file is there and unreadable")
assert gateway.haproxy_warnings(haproxy_section(
[frontend("f", [22], default="nowhere")], available=False)) == []
assert gateway.haproxy_warnings(haproxy_section(
[frontend("f", [22], default="nowhere")], available=None)) == []
assert gateway.haproxy_warnings(None) == []
assert gateway.nginx_warnings(nginx_section(
[nginx_server([], ["x.example.com"])])) != [], "the control: an available section IS judged"
assert gateway.nginx_warnings(nginx_section([nginx_server([], ["x.example.com"])],
available=None)) == []
assert gateway.nginx_warnings(None) == []
def test_the_captured_configs_produce_no_warnings(gateway_snapshot: Any) -> None:
"""This fleet's own ingress is clean, and the reason per kind is part of the pin.
Seven checks that never fire on a real config would be a lie about this fleet and a source of
phantom alerts, so each kind is answered with the fact in the capture that stops it: the six
frontends each name a declared backend, the six backends are each routed and each carry a
`server`, every nginx `server` block listens, no `proxy_pass` names a pool, the one
`proxy_pass $var` has its `map`, and every entry of that referenced map carries an address
(the *unreferenced* `$connection_upgrade` map's null targets are not routes at all).
"""
snap = gateway_snapshot("full")
payload = gateway.parse(snap, "gateway")
assert payload["warnings"] == []
haproxy, nginx = snap["haproxy"], snap["nginx"]
declared = {block["name"] for block in haproxy["backends"]}
routed = {frontend["default_backend"] for frontend in haproxy["frontends"]}
for frontend in haproxy["frontends"]:
routed.update(frontend["backends"])
assert declared == routed == {
"gitea-2222", "sec-gitea-2000", "iperf3-5201", "unifi-external-8080",
"testcraft-25565", "nimcraft-25566"}, "no dangling route and no orphan backend"
assert all(block["servers"] for block in haproxy["backends"]), "no backend routes nowhere"
assert all(server["listen"] for server in nginx["servers"]), "every vhost listens"
assert not [proxy for server in nginx["servers"] for proxy in server["proxy_passes"]
if proxy["target"] is not None and proxy["target"]["port"] is None], (
"no `proxy_pass` names a pool, declared or otherwise, on this gateway")
used = gateway.proxy_variables(nginx)
defined = {block["variable"].lstrip("$") for block in nginx["maps"]}
assert used == {"targetBackend"} and used <= defined, (
"the one variable a `proxy_pass` passes IS defined by a map")
referenced = [block for block in nginx["maps"] if block["variable"].lstrip("$") in used]
assert len(referenced) == 1 and len(referenced[0]["entries"]) == 18
assert all(entry["target"] for entry in referenced[0]["entries"]), (
"all 18 stream routes name an address, `default` included")
unreferenced = [block for block in nginx["maps"] if block not in referenced]
assert any(entry["target"] is None for block in unreferenced for entry in block["entries"]), (
"the capture must still hold null-target entries on the UNREFERENCED map, or this pin "
"proves the refusal rule about nothing")
def test_warnings_are_deterministic_deduplicated_and_json_serialisable() -> None:
"""Same input twice, one output; the same fault twice, one row; every row a plain JSON object.
The page keys its DOM on `id` and `alerts.py` quotes `warnings[0]`, so a list that reordered
between polls would reshuffle the panel and a row that carried a non-JSON value would 500 the
route. Deduplication is exercised the way the real configs do it: one frontend with two `bind`
lines is one broken route, not two, and two frontends with the same broken name are two.
"""
section_h = haproxy_section(
[frontend("twice", [22, 2222], default="gone"),
frontend("also", [33], default="gone")],
[backend("live", [server("s", "10.0.0.1", 80)])])
section_n = nginx_section([nginx_server([], ["x.example.com"], [pool_pass("nope")])])
first = gateway.warnings(section_h, section_n)
assert gateway.warnings(section_h, section_n) == first, "deterministic"
assert gateway.warnings(section_h, section_n) is not first
assert len(first) == len({row["id"] for row in first}), "deduplicated by id"
assert kinds(first) == [gateway.WARN_ROUTE_DANGLING, gateway.WARN_ROUTE_DANGLING,
gateway.WARN_BACKEND_ORPHAN, gateway.WARN_SERVER_UNLISTENED,
gateway.WARN_POOL_UNDECLARED], (
"haproxy kinds first and in config order, then nginx kinds — the order the page lists")
assert first[0]["id"] == "haproxy_route_dangling:twice->gone", (
"one warning per broken line, never one per bind")
assert first[1]["id"] == "haproxy_route_dangling:also->gone"
json.dumps(first)
assert all(set(row) == {"id", "kind", "subject", "message"} for row in first), (
"a fourth key would reach the page and the event feed unnoticed")
assert all(isinstance(value, str) for row in first for value in row.values())
assert not walk_keys(first) & {"severity", "window_seconds"}, (
"a config warning is a structural fact, never a graded number")
def test_a_row_that_names_nothing_contributes_nothing_and_a_repeated_row_is_one() -> None:
"""The two defensive arms of `warnings()`, driven from the shapes a real file can produce.
A `backend` line with no name is not a backend anybody can repair, and a config that declares
the same empty block twice (a copy-paste, an `include` that lands twice on one file) is one
fault with one fix — so the deduplication the docstring promises is measured here rather than
assumed, and the row that cannot be named is dropped instead of warning as `""`.
"""
section = haproxy_section(
[frontend("haproxy-22", [22], default="empty-be")],
[{"servers": [server("s", "10.0.0.1", 22)]}, # a backend with no name at all
backend("empty-be", []), backend("empty-be", [])]) # the same block declared twice
assert len(gateway.haproxy_warnings(section)) == 2, (
"the per-section check reports what the file says — it is `warnings()` that owns the dedup")
found = gateway.warnings(section, None)
assert kinds(found) == [gateway.WARN_BACKEND_EMPTY], found
assert found[0]["id"] == "haproxy_backend_empty:empty-be", (
"one row for the duplicated block — the page keys its DOM on `id` and two identical rows "
"would be one fault counted twice")
def test_warnings_travel_in_the_payload_and_the_stub_share_the_key(gateway_snapshot: Any) -> None:
"""`payload["warnings"]` is there in every arm, and the socket half cannot silence it.
The stub parity matters because the page renders both arms from one renderer; the sockets arm
matters because a config is just as broken when `/proc` will not answer — a warning is a config
fact, not a socket fact, and `[]` in the `predates` arm is the same key holding the honest value
rather than a missing one.
"""
broken = haproxy_section([frontend("haproxy-22", [22], default="gone")])
snap = snapshot(haproxy=broken, nginx=nginx_section(),
sockets=sockets_section(available=None))
payload = gateway.parse(snap, "gateway")
assert kinds(payload["warnings"]) == [gateway.WARN_ROUTE_DANGLING], (
"a silent socket table says nothing about the config, so it changes no warning")
assert payload["sockets"] == {"listens": None, "established": None}
assert gateway.stub("gateway")["warnings"] == []
assert gateway.parse(gateway_snapshot("predates"), "gateway")["warnings"] == []
assert set(gateway.stub("gateway")) - {"unreachable"} <= set(gateway.parse(snap, "gateway"))
Reference in New Issue View Git Blame Copy Permalink