155 lines
6.4 KiB
Python
155 lines
6.4 KiB
Python
"""Tests for the health gate of 'ps --wait'."""
|
|
|
|
import os
|
|
import sys
|
|
|
|
import click
|
|
import pytest
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
import alakazam
|
|
from alakazam import get_pending_containers, wait_for_healthy_apps
|
|
|
|
|
|
def container(service, status, version="1.0.0"):
|
|
return {"service": service, "status": status, "version": version, "chaos": "false", "state": "running", "image": "img"}
|
|
|
|
|
|
class TestPendingContainers:
|
|
def test_healthy_containers_are_not_pending(self):
|
|
containers = [("a.org", container("app", "healthy")), ("a.org", container("db", "running"))]
|
|
assert get_pending_containers(containers) == []
|
|
|
|
@pytest.mark.parametrize("status", ["starting", "unknown", "unhealthy", "starting (health: starting)"])
|
|
def test_unsettled_statuses_are_pending(self, status):
|
|
containers = [("a.org", container("app", status))]
|
|
assert len(get_pending_containers(containers)) == 1
|
|
|
|
def test_init_containers_are_exempt(self):
|
|
"""Init containers run once and exit, they never report a healthy status."""
|
|
containers = [("a.org", container("authentik-init", "unknown")), ("a.org", container("app", "healthy"))]
|
|
assert get_pending_containers(containers) == []
|
|
|
|
def test_an_init_container_does_not_hide_a_real_one(self):
|
|
containers = [("a.org", container("authentik-init", "unknown")), ("a.org", container("app", "starting"))]
|
|
pending = get_pending_containers(containers)
|
|
assert [c["service"] for _, c in pending] == ["app"]
|
|
|
|
|
|
class FakeCluster:
|
|
"""Serves a scripted sequence of container states, one per health check."""
|
|
|
|
def __init__(self, rounds, deployed=True):
|
|
self.rounds = list(rounds)
|
|
self.deployed = deployed
|
|
self.checks = 0
|
|
self.slept = 0
|
|
|
|
def install(self, monkeypatch):
|
|
monkeypatch.setattr(alakazam, "get_apps_by_deployment", self.get_apps_by_deployment)
|
|
monkeypatch.setattr(alakazam, "get_app_containers", self.get_app_containers)
|
|
monkeypatch.setattr(alakazam, "sleep", self.sleep)
|
|
return self
|
|
|
|
def get_apps_by_deployment(self, recipes, deployed=True):
|
|
return {"a.org": [["authentik", "login.a.org", "1.0.0"]]} if self.deployed else {}
|
|
|
|
def get_app_containers(self, instance_apps):
|
|
self.checks += 1
|
|
return self.rounds[min(self.checks - 1, len(self.rounds) - 1)]
|
|
|
|
def sleep(self, seconds):
|
|
self.slept += seconds
|
|
|
|
|
|
HEALTHY = [("login.a.org", container("app", "healthy"))]
|
|
STARTING = [("login.a.org", container("app", "starting"))]
|
|
|
|
|
|
class TestWaitForHealthyApps:
|
|
def test_returns_immediately_when_everything_is_healthy(self, monkeypatch, capsys):
|
|
cluster = FakeCluster([HEALTHY]).install(monkeypatch)
|
|
wait_for_healthy_apps((), timeout=300)
|
|
assert cluster.checks == 1
|
|
assert cluster.slept == 0
|
|
assert "All services are healthy" in capsys.readouterr().out
|
|
|
|
def test_waits_until_the_services_settle(self, monkeypatch):
|
|
cluster = FakeCluster([STARTING, STARTING, HEALTHY]).install(monkeypatch)
|
|
wait_for_healthy_apps((), timeout=300)
|
|
assert cluster.checks == 3
|
|
assert cluster.slept == 2 * alakazam.PS_WAIT_INTERVAL
|
|
|
|
def test_timeout_aborts_and_names_the_services(self, monkeypatch):
|
|
cluster = FakeCluster([STARTING]).install(monkeypatch)
|
|
with pytest.raises(click.ClickException) as excinfo:
|
|
wait_for_healthy_apps((), timeout=0)
|
|
assert "login.a.org app: starting" in excinfo.value.message
|
|
assert excinfo.value.exit_code != 0
|
|
|
|
def test_timeout_checks_at_least_once(self, monkeypatch):
|
|
"""A zero timeout still has to look, otherwise it could not report anything."""
|
|
cluster = FakeCluster([STARTING]).install(monkeypatch)
|
|
with pytest.raises(click.ClickException):
|
|
wait_for_healthy_apps((), timeout=0)
|
|
assert cluster.checks == 1
|
|
|
|
def test_nothing_deployed_is_a_failure(self, monkeypatch):
|
|
"""Reporting the health of an environment that does not exist would turn a broken deploy green."""
|
|
FakeCluster([HEALTHY], deployed=False).install(monkeypatch)
|
|
with pytest.raises(click.ClickException) as excinfo:
|
|
wait_for_healthy_apps((), timeout=300)
|
|
assert "nothing to wait for" in excinfo.value.message
|
|
|
|
def test_an_init_container_alone_does_not_block(self, monkeypatch):
|
|
rounds = [[("login.a.org", container("authentik-init", "unknown")), ("login.a.org", container("app", "healthy"))]]
|
|
cluster = FakeCluster(rounds).install(monkeypatch)
|
|
wait_for_healthy_apps((), timeout=300)
|
|
assert cluster.checks == 1
|
|
|
|
|
|
class TestPsCommand:
|
|
"""Without --wait the command must behave exactly as it did before the flag existed."""
|
|
|
|
@pytest.fixture
|
|
def cluster(self, monkeypatch):
|
|
cluster = FakeCluster([STARTING]).install(monkeypatch)
|
|
monkeypatch.setattr(alakazam, "INSTANCE_CONFIGS", {"a.org": {"authentik": {"app_domain": "login.a.org"}}})
|
|
return cluster
|
|
|
|
def invoke(self, args):
|
|
from click.testing import CliRunner
|
|
return CliRunner().invoke(alakazam.ps, args, standalone_mode=False)
|
|
|
|
def test_plain_ps_checks_once_and_does_not_wait(self, cluster):
|
|
result = self.invoke([])
|
|
assert result.exit_code == 0, result.output
|
|
assert cluster.checks == 1
|
|
assert cluster.slept == 0
|
|
|
|
def test_plain_ps_prints_the_table(self, cluster):
|
|
result = self.invoke([])
|
|
assert "Service" in result.output
|
|
assert "login.a.org" in result.output
|
|
assert "starting" in result.output
|
|
|
|
def test_plain_ps_succeeds_on_unhealthy_services(self, cluster):
|
|
"""Reporting a status is not the same as gating on it."""
|
|
assert self.invoke([]).exit_code == 0
|
|
|
|
def test_plain_ps_without_deployed_apps(self, cluster, monkeypatch):
|
|
cluster.deployed = False
|
|
result = self.invoke([])
|
|
assert result.exit_code == 0
|
|
assert "No apps deployed" in result.output
|
|
|
|
def test_wait_and_watch_are_mutually_exclusive(self, cluster):
|
|
result = self.invoke(["--wait", "--watch"])
|
|
assert isinstance(result.exception, click.UsageError)
|
|
|
|
def test_wait_gates_on_the_status(self, cluster):
|
|
result = self.invoke(["--wait", "--timeout", "0"])
|
|
assert isinstance(result.exception, click.ClickException)
|
|
assert "still not healthy after 0s" in result.exception.message
|