forked from toolshed/alakazam
161 lines
7.0 KiB
Python
161 lines
7.0 KiB
Python
"""Tests for the readiness hooks that gate dependent apps on an app becoming usable."""
|
|
|
|
import os
|
|
import sys
|
|
|
|
import click
|
|
import pytest
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
import alakazam
|
|
import logging
|
|
|
|
from alakazam import abra, build_app_cmd, run_readiness_hook, run_readiness_hooks, run_streamed
|
|
|
|
|
|
class FakeAbra:
|
|
"""Fails a scripted number of times before succeeding, recording every call."""
|
|
|
|
def __init__(self, failures=0, always_fail=False):
|
|
self.failures = failures
|
|
self.always_fail = always_fail
|
|
self.calls = []
|
|
self.slept = []
|
|
|
|
def install(self, monkeypatch):
|
|
monkeypatch.setattr(alakazam, "abra", self)
|
|
monkeypatch.setattr(alakazam, "sleep", self.slept.append)
|
|
return self
|
|
|
|
def __call__(self, *args, **kwargs):
|
|
self.calls.append(args)
|
|
if self.always_fail or len(self.calls) <= self.failures:
|
|
raise RuntimeError("FATA blueprint not applied")
|
|
return "ok"
|
|
|
|
|
|
class TestBuildAppCmd:
|
|
def test_container_command(self):
|
|
assert build_app_cmd("login.a.org", "app check_blueprints") == [
|
|
"app", "cmd", "-T", "", "login.a.org", "app", "--", "check_blueprints",
|
|
]
|
|
|
|
def test_local_command(self):
|
|
assert build_app_cmd("login.a.org", "local enable_sso") == [
|
|
"app", "cmd", "--local", "", "login.a.org", "--", "enable_sso",
|
|
]
|
|
|
|
def test_arguments_are_passed_through(self):
|
|
assert build_app_cmd("login.a.org", "app set_flag on")[-2:] == ["set_flag", "on"]
|
|
|
|
def test_chaos_flag(self):
|
|
assert "-C" in build_app_cmd("login.a.org", "app check_blueprints", "-C")
|
|
|
|
|
|
class TestRunReadinessHook:
|
|
def test_succeeds_on_the_first_attempt(self, monkeypatch, capsys):
|
|
abra = FakeAbra().install(monkeypatch)
|
|
run_readiness_hook("login.a.org", {"cmd": "app check_blueprints"})
|
|
assert len(abra.calls) == 1
|
|
assert abra.slept == []
|
|
assert "is ready" in capsys.readouterr().out
|
|
|
|
def test_retries_until_it_succeeds(self, monkeypatch):
|
|
abra = FakeAbra(failures=2).install(monkeypatch)
|
|
run_readiness_hook("login.a.org", {"cmd": "app check_blueprints", "interval": 5})
|
|
assert len(abra.calls) == 3
|
|
assert abra.slept == [5, 5]
|
|
|
|
def test_gives_up_after_the_configured_retries(self, monkeypatch):
|
|
abra = FakeAbra(always_fail=True).install(monkeypatch)
|
|
with pytest.raises(click.ClickException) as excinfo:
|
|
run_readiness_hook("login.a.org", {"cmd": "app check_blueprints", "retries": 3, "interval": 5})
|
|
assert len(abra.calls) == 3
|
|
assert "failed 3 times" in excinfo.value.message
|
|
assert excinfo.value.exit_code != 0
|
|
|
|
def test_does_not_sleep_after_the_last_attempt(self, monkeypatch):
|
|
abra = FakeAbra(always_fail=True).install(monkeypatch)
|
|
with pytest.raises(click.ClickException):
|
|
run_readiness_hook("login.a.org", {"cmd": "app check_blueprints", "retries": 3, "interval": 5})
|
|
assert abra.slept == [5, 5]
|
|
|
|
def test_initial_delay_precedes_the_first_attempt(self, monkeypatch):
|
|
abra = FakeAbra().install(monkeypatch)
|
|
run_readiness_hook("login.a.org", {"cmd": "app check_blueprints", "initial-delay": 60})
|
|
assert abra.slept == [60]
|
|
|
|
def test_defaults_apply_when_only_a_command_is_given(self, monkeypatch):
|
|
abra = FakeAbra(always_fail=True).install(monkeypatch)
|
|
with pytest.raises(click.ClickException):
|
|
run_readiness_hook("login.a.org", {"cmd": "app check_blueprints"})
|
|
assert len(abra.calls) == alakazam.READINESS_RETRIES
|
|
assert abra.slept == [alakazam.READINESS_INTERVAL] * (alakazam.READINESS_RETRIES - 1)
|
|
|
|
def test_a_hook_without_a_command_is_rejected(self, monkeypatch):
|
|
FakeAbra().install(monkeypatch)
|
|
with pytest.raises(click.ClickException) as excinfo:
|
|
run_readiness_hook("login.a.org", {"retries": 3})
|
|
assert "no 'cmd'" in excinfo.value.message
|
|
|
|
def test_the_last_failure_is_reported(self, monkeypatch):
|
|
FakeAbra(always_fail=True).install(monkeypatch)
|
|
with pytest.raises(click.ClickException) as excinfo:
|
|
run_readiness_hook("login.a.org", {"cmd": "app check_blueprints", "retries": 1})
|
|
assert "blueprint not applied" in excinfo.value.message
|
|
|
|
|
|
class TestVisibility:
|
|
"""A hook can block for half an hour, so silence is indistinguishable from a hang."""
|
|
|
|
def test_the_command_output_is_streamed(self):
|
|
result = run_streamed(["bash", "-c", "echo first; echo second >&2; exit 3"])
|
|
assert result.returncode == 3
|
|
assert result.stdout.decode().splitlines() == ["first", "second"]
|
|
|
|
def test_the_hook_output_is_not_buffered(self, monkeypatch):
|
|
"""abra is asked to stream, otherwise nothing appears until the command has finished."""
|
|
streamed = []
|
|
monkeypatch.setattr(alakazam, "abra", lambda *a, **kw: streamed.append(kw.get("stream")) or "ok")
|
|
monkeypatch.setattr(alakazam, "sleep", lambda seconds: None)
|
|
run_readiness_hook("login.a.org", {"cmd": "app check_blueprints"})
|
|
assert streamed == [True]
|
|
|
|
def test_progress_is_printed(self, monkeypatch, capsys):
|
|
"""Progress is what the run is doing, not a diagnostic, and logging.info needs -l to show."""
|
|
FakeAbra(failures=1).install(monkeypatch)
|
|
run_readiness_hook("login.a.org", {"cmd": "app check_blueprints", "interval": 5})
|
|
out = capsys.readouterr().out
|
|
assert "attempt 1/20" in out
|
|
assert "retry in 5s" in out
|
|
|
|
def test_the_initial_delay_is_announced(self, monkeypatch, capsys):
|
|
FakeAbra().install(monkeypatch)
|
|
run_readiness_hook("login.a.org", {"cmd": "app check_blueprints", "initial-delay": 60})
|
|
assert "Wait 60s" in capsys.readouterr().out
|
|
|
|
def test_streaming_and_machine_output_are_incompatible(self):
|
|
"""Streaming merges abra's log lines into stdout, which would break the JSON parsing."""
|
|
with pytest.raises(ValueError):
|
|
abra("app", "ls", machine_output=True, stream=True)
|
|
|
|
|
|
class TestRunReadinessHooks:
|
|
def test_no_hooks_configured_is_a_no_op(self, monkeypatch):
|
|
abra = FakeAbra().install(monkeypatch)
|
|
run_readiness_hooks("login.a.org", {"version": "1.0.0"})
|
|
assert abra.calls == []
|
|
|
|
def test_hooks_run_in_configuration_order(self, monkeypatch):
|
|
abra = FakeAbra().install(monkeypatch)
|
|
run_readiness_hooks("login.a.org", {"readiness-hooks": [{"cmd": "app first"}, {"cmd": "app second"}]})
|
|
assert [call[-1] for call in abra.calls] == ["first", "second"]
|
|
|
|
def test_a_failing_hook_stops_the_following_ones(self, monkeypatch):
|
|
"""Dependent apps must not proceed on a half ready app."""
|
|
abra = FakeAbra(always_fail=True).install(monkeypatch)
|
|
with pytest.raises(click.ClickException):
|
|
run_readiness_hooks("login.a.org", {"readiness-hooks": [{"cmd": "app first", "retries": 1}, {"cmd": "app second"}]})
|
|
assert all(call[-1] == "first" for call in abra.calls)
|