forked from toolshed/alakazam
104 lines
4.0 KiB
Python
104 lines
4.0 KiB
Python
"""Tests for which abra failures may be retried, and which configuration keys are ignored."""
|
|
|
|
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 abra, is_connection_error, update_configs
|
|
|
|
|
|
class FakeProcess:
|
|
def __init__(self, returncode, stderr=b""):
|
|
self.returncode = returncode
|
|
self.stdout = b""
|
|
self.stderr = stderr
|
|
|
|
|
|
UNREACHABLE = b"FATA error during connect: kex_exchange_identification: Connection closed by remote host"
|
|
REJECTED = b"FATA login.a.org is still deployed"
|
|
|
|
|
|
class TestRetryLoop:
|
|
"""A server that drops the ssh handshake is usually throttling, so the retries have to spread out."""
|
|
|
|
@pytest.fixture
|
|
def runs(self, monkeypatch):
|
|
calls = {"results": [], "slept": []}
|
|
monkeypatch.setattr(alakazam, "sleep", calls["slept"].append)
|
|
monkeypatch.setattr(alakazam.subprocess, "run",
|
|
lambda cmd, capture_output: calls["results"].pop(0))
|
|
return calls
|
|
|
|
def test_a_successful_command_does_not_retry(self, runs):
|
|
runs["results"] = [FakeProcess(0)]
|
|
abra("app", "ls")
|
|
assert runs["slept"] == []
|
|
|
|
def test_the_delay_doubles_between_attempts(self, runs):
|
|
runs["results"] = [FakeProcess(1, UNREACHABLE)] * alakazam.ABRA_RETRIES
|
|
with pytest.raises(RuntimeError):
|
|
abra("app", "ls")
|
|
expected = [alakazam.ABRA_RETRY_DELAY * 2 ** n for n in range(alakazam.ABRA_RETRIES - 1)]
|
|
assert runs["slept"] == expected
|
|
|
|
def test_it_stops_retrying_once_the_command_succeeds(self, runs):
|
|
runs["results"] = [FakeProcess(1, UNREACHABLE), FakeProcess(1, UNREACHABLE), FakeProcess(0)]
|
|
abra("app", "ls")
|
|
assert runs["slept"] == [alakazam.ABRA_RETRY_DELAY, alakazam.ABRA_RETRY_DELAY * 2]
|
|
|
|
def test_a_rejected_operation_is_not_repeated(self, runs):
|
|
"""Repeating a command the server answered would only waste the whole backoff."""
|
|
runs["results"] = [FakeProcess(1, REJECTED)]
|
|
with pytest.raises(RuntimeError):
|
|
abra("app", "ls")
|
|
assert runs["slept"] == []
|
|
|
|
|
|
class TestConnectionErrors:
|
|
"""Only unreachable servers may be retried, a rejected operation must not be repeated."""
|
|
|
|
@pytest.mark.parametrize("output", [
|
|
'error during connect: Get "http://docker.example.com/v1.51/info"',
|
|
"kex_exchange_identification: Connection closed by remote host",
|
|
"Connection closed by 152.53.133.16 port 22",
|
|
"connection reset by peer",
|
|
"ssh: connect to host example.com port 22: Connection refused",
|
|
"Connection timed out",
|
|
"No route to host",
|
|
"write: broken pipe",
|
|
])
|
|
def test_unreachable_server(self, output):
|
|
assert is_connection_error(output)
|
|
|
|
@pytest.mark.parametrize("output", [
|
|
"FATA email_pass doesn't exist on server?",
|
|
"FATA login.a.org is still deployed",
|
|
"FATA secret is in use",
|
|
"",
|
|
])
|
|
def test_operational_failures_are_not_retried(self, output):
|
|
assert not is_connection_error(output)
|
|
|
|
|
|
class TestUnknownConfigKeys:
|
|
def test_an_unknown_key_never_reaches_the_env_file(self, tmp_path):
|
|
"""A promotion mechanism sets 'hold: <reason>' on a recipe, which must pass through untouched."""
|
|
env = tmp_path / "app.env"
|
|
env.write_text("#SMTP_HOST=mail.example.com\n")
|
|
update_configs(env, {"hold": "waiting for the upstream fix", "env": {"SMTP_HOST": "mail.a.org"}})
|
|
content = env.read_text()
|
|
assert "SMTP_HOST=mail.a.org" in content
|
|
assert "hold" not in content
|
|
assert "waiting for the upstream fix" not in content
|
|
|
|
def test_an_unknown_key_alone_changes_nothing(self, tmp_path):
|
|
env = tmp_path / "app.env"
|
|
env.write_text("#SMTP_HOST=mail.example.com\n")
|
|
update_configs(env, {"hold": "waiting for the upstream fix"})
|
|
assert env.read_text() == "#SMTP_HOST=mail.example.com\n"
|