dodil/erp-platform

tests/test_retry.py

"""Which failures are retried, and which are not — the retry is only as good as that line."""
import pytest
from sqlalchemy.exc import OperationalError

from erp_platform import retry


class PgError(Exception):
    def __init__(self, sqlstate, msg="db error"):
        super().__init__(msg)
        self.sqlstate = sqlstate


def wrapped(code, msg="db error"):
    """How psycopg errors arrive: inside SQLAlchemy's OperationalError, the code on .orig."""
    return OperationalError("SELECT 1", {}, PgError(code, msg))


def flaky(failures):
    calls = []

    def fn():
        calls.append(1)
        if len(calls) <= len(failures):
            raise failures[len(calls) - 1]
        return "done"
    return fn, calls


@pytest.mark.parametrize("code", sorted(retry.SAFE))
def test_a_failure_before_commit_is_retried_even_for_a_non_idempotent_command(code):
    fn, calls = flaky([wrapped(code)])
    assert retry.run(fn, idempotent=False, sleep=lambda s: None) == "done" and len(calls) == 2


def test_a_dropped_connection_is_retried_for_a_query_or_idempotent_command():
    fn, calls = flaky([wrapped("08006")])
    assert retry.run(fn, idempotent=True, sleep=lambda s: None) == "done" and len(calls) == 2


def test_a_dropped_connection_is_NOT_retried_for_a_non_idempotent_command():
    """Its COMMIT may have landed: running open_quote again would make a second quote."""
    fn, calls = flaky([wrapped("08006")])
    with pytest.raises(OperationalError):
        retry.run(fn, idempotent=False, sleep=lambda s: None)
    assert len(calls) == 1


@pytest.mark.parametrize("exc", [wrapped("23505", "duplicate key"), wrapped("42P01"), ValueError("a bug"),
                                 KeyError("x")])
def test_anything_else_is_not_retried(exc):
    fn, calls = flaky([exc])
    with pytest.raises(type(exc)):
        retry.run(fn, idempotent=True, sleep=lambda s: None)
    assert len(calls) == 1


def test_codes_and_markers_found_in_text_as_the_cli_and_gateway_render_them():
    assert retry.classify(RuntimeError("ERROR: cannot connect now (SQLSTATE 57P03)")) == "safe"
    assert retry.classify(RuntimeError("tables reader: ConfigurationLimitExceeded")) == "safe"
    assert retry.classify(RuntimeError("the writer could not confirm the WAL is fully uploaded")) == "ambiguous"


def test_it_gives_up_after_the_last_attempt_with_the_real_error():
    fn, calls = flaky([wrapped("57P03")] * 10)
    with pytest.raises(OperationalError):
        retry.run(fn, idempotent=True, sleep=lambda s: None)
    assert len(calls) == retry.ATTEMPTS


def test_the_waits_are_bounded():
    waits = []
    fn, _ = flaky([wrapped("57P03")] * 3)
    retry.run(fn, idempotent=True, sleep=waits.append)
    assert len(waits) == 3 and sum(waits) < sum(retry.BACKOFF) * 1.25