Introduce a 'runner' test fixture

Instead of defining the CliRunner value in each test, we use a fixture.
The CliRunner is also configured with stdout and stderr separated
because mixing them will pose problem if we use stderr for other
purposes in tests, e.g. to emit log messages from a forth-coming HTTP
server.
This commit is contained in:
Denis Laxalde 2023-09-28 09:41:33 +02:00
parent d34e597e61
commit ea92809cb3
16 changed files with 91 additions and 157 deletions

View file

@ -56,4 +56,3 @@ setup(
},
zip_safe=False,
)

View file

@ -2,6 +2,7 @@ from functools import partial
from typing import Any, Callable
import pytest
from click.testing import CliRunner
from pytest_mock import MockerFixture
from .tools import my_mock
@ -26,3 +27,9 @@ def fake_restapi(
mocker: MockerFixture, use_old_replica_state: bool
) -> Callable[..., Any]:
return partial(my_mock, mocker, use_old_replica_state=use_old_replica_state)
@pytest.fixture
def runner() -> CliRunner:
"""A CliRunner with stdout and stderr not mixed."""
return CliRunner(mix_stderr=False)

View file

@ -3,9 +3,7 @@ from click.testing import CliRunner
from check_patroni.cli import main
def test_api_status_code_200(fake_restapi) -> None:
runner = CliRunner()
def test_api_status_code_200(runner: CliRunner, fake_restapi) -> None:
fake_restapi("node_is_pending_restart_ok")
result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "node_is_pending_restart"]
@ -13,9 +11,7 @@ def test_api_status_code_200(fake_restapi) -> None:
assert result.exit_code == 0
def test_api_status_code_404(fake_restapi) -> None:
runner = CliRunner()
def test_api_status_code_404(runner: CliRunner, fake_restapi) -> None:
fake_restapi("Fake test", status=404)
result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "node_is_pending_restart"]

View file

@ -6,9 +6,9 @@ from click.testing import CliRunner
from check_patroni.cli import main
def test_cluster_config_has_changed_ok_with_hash(fake_restapi) -> None:
runner = CliRunner()
def test_cluster_config_has_changed_ok_with_hash(
runner: CliRunner, fake_restapi
) -> None:
fake_restapi("cluster_config_has_changed")
result = runner.invoke(
main,
@ -28,10 +28,8 @@ def test_cluster_config_has_changed_ok_with_hash(fake_restapi) -> None:
def test_cluster_config_has_changed_ok_with_state_file(
fake_restapi, tmp_path: Path
runner: CliRunner, fake_restapi, tmp_path: Path
) -> None:
runner = CliRunner()
state_file = tmp_path / "cluster_config_has_changed.state_file"
with state_file.open("w") as f:
f.write('{"hash": "96b12d82571473d13e890b893734e731"}')
@ -54,9 +52,9 @@ def test_cluster_config_has_changed_ok_with_state_file(
)
def test_cluster_config_has_changed_ko_with_hash(fake_restapi) -> None:
runner = CliRunner()
def test_cluster_config_has_changed_ko_with_hash(
runner: CliRunner, fake_restapi
) -> None:
fake_restapi("cluster_config_has_changed")
result = runner.invoke(
main,
@ -76,10 +74,8 @@ def test_cluster_config_has_changed_ko_with_hash(fake_restapi) -> None:
def test_cluster_config_has_changed_ko_with_state_file_and_save(
fake_restapi, tmp_path: Path
runner: CliRunner, fake_restapi, tmp_path: Path
) -> None:
runner = CliRunner()
state_file = tmp_path / "cluster_config_has_changed.state_file"
with state_file.open("w") as f:
f.write('{"hash": "96b12d82571473d13e890b8937ffffff"}')
@ -136,10 +132,10 @@ def test_cluster_config_has_changed_ko_with_state_file_and_save(
assert new_config_hash == "96b12d82571473d13e890b893734e731"
def test_cluster_config_has_changed_params(fake_restapi, tmp_path: Path) -> None:
def test_cluster_config_has_changed_params(
runner: CliRunner, fake_restapi, tmp_path: Path
) -> None:
# This one is placed last because it seems like the exceptions are not flushed from stderr for the next tests.
runner = CliRunner()
fake_state_file = tmp_path / "fake_file_name.state_file"
fake_restapi("cluster_config_has_changed")
result = runner.invoke(

View file

@ -3,9 +3,7 @@ from click.testing import CliRunner
from check_patroni.cli import main
def test_cluster_has_leader_ok(fake_restapi) -> None:
runner = CliRunner()
def test_cluster_has_leader_ok(runner: CliRunner, fake_restapi) -> None:
fake_restapi("cluster_has_leader_ok")
result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "cluster_has_leader"]
@ -17,9 +15,7 @@ def test_cluster_has_leader_ok(fake_restapi) -> None:
)
def test_cluster_has_leader_ok_standby_leader(fake_restapi) -> None:
runner = CliRunner()
def test_cluster_has_leader_ok_standby_leader(runner: CliRunner, fake_restapi) -> None:
fake_restapi("cluster_has_leader_ok_standby_leader")
result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "cluster_has_leader"]
@ -31,9 +27,7 @@ def test_cluster_has_leader_ok_standby_leader(fake_restapi) -> None:
)
def test_cluster_has_leader_ko(fake_restapi) -> None:
runner = CliRunner()
def test_cluster_has_leader_ko(runner: CliRunner, fake_restapi) -> None:
fake_restapi("cluster_has_leader_ko")
result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "cluster_has_leader"]

View file

@ -4,9 +4,7 @@ from check_patroni.cli import main
# TODO Lag threshold tests
def test_cluster_has_relica_ok(fake_restapi) -> None:
runner = CliRunner()
def test_cluster_has_relica_ok(runner: CliRunner, fake_restapi) -> None:
fake_restapi("cluster_has_replica_ok")
result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "cluster_has_replica"]
@ -18,9 +16,9 @@ def test_cluster_has_relica_ok(fake_restapi) -> None:
)
def test_cluster_has_replica_ok_with_count_thresholds(fake_restapi) -> None:
runner = CliRunner()
def test_cluster_has_replica_ok_with_count_thresholds(
runner: CliRunner, fake_restapi
) -> None:
fake_restapi("cluster_has_replica_ok")
result = runner.invoke(
main,
@ -41,9 +39,9 @@ def test_cluster_has_replica_ok_with_count_thresholds(fake_restapi) -> None:
)
def test_cluster_has_replica_ok_with_sync_count_thresholds(fake_restapi) -> None:
runner = CliRunner()
def test_cluster_has_replica_ok_with_sync_count_thresholds(
runner: CliRunner, fake_restapi
) -> None:
fake_restapi("cluster_has_replica_ok")
result = runner.invoke(
main,
@ -62,9 +60,9 @@ def test_cluster_has_replica_ok_with_sync_count_thresholds(fake_restapi) -> None
)
def test_cluster_has_replica_ok_with_count_thresholds_lag(fake_restapi) -> None:
runner = CliRunner()
def test_cluster_has_replica_ok_with_count_thresholds_lag(
runner: CliRunner, fake_restapi
) -> None:
fake_restapi("cluster_has_replica_ok_lag")
result = runner.invoke(
main,
@ -87,9 +85,9 @@ def test_cluster_has_replica_ok_with_count_thresholds_lag(fake_restapi) -> None:
)
def test_cluster_has_replica_ko_with_count_thresholds(fake_restapi) -> None:
runner = CliRunner()
def test_cluster_has_replica_ko_with_count_thresholds(
runner: CliRunner, fake_restapi
) -> None:
fake_restapi("cluster_has_replica_ko")
result = runner.invoke(
main,
@ -110,9 +108,9 @@ def test_cluster_has_replica_ko_with_count_thresholds(fake_restapi) -> None:
)
def test_cluster_has_replica_ko_with_sync_count_thresholds(fake_restapi) -> None:
runner = CliRunner()
def test_cluster_has_replica_ko_with_sync_count_thresholds(
runner: CliRunner, fake_restapi
) -> None:
fake_restapi("cluster_has_replica_ko")
result = runner.invoke(
main,
@ -133,9 +131,9 @@ def test_cluster_has_replica_ko_with_sync_count_thresholds(fake_restapi) -> None
)
def test_cluster_has_replica_ko_with_count_thresholds_and_lag(fake_restapi) -> None:
runner = CliRunner()
def test_cluster_has_replica_ko_with_count_thresholds_and_lag(
runner: CliRunner, fake_restapi
) -> None:
fake_restapi("cluster_has_replica_ko_lag")
result = runner.invoke(
main,

View file

@ -3,9 +3,7 @@ from click.testing import CliRunner
from check_patroni.cli import main
def test_cluster_has_scheduled_action_ok(fake_restapi) -> None:
runner = CliRunner()
def test_cluster_has_scheduled_action_ok(runner: CliRunner, fake_restapi) -> None:
fake_restapi("cluster_has_scheduled_action_ok")
result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "cluster_has_scheduled_action"]
@ -17,9 +15,9 @@ def test_cluster_has_scheduled_action_ok(fake_restapi) -> None:
)
def test_cluster_has_scheduled_action_ko_switchover(fake_restapi) -> None:
runner = CliRunner()
def test_cluster_has_scheduled_action_ko_switchover(
runner: CliRunner, fake_restapi
) -> None:
fake_restapi("cluster_has_scheduled_action_ko_switchover")
result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "cluster_has_scheduled_action"]
@ -31,9 +29,9 @@ def test_cluster_has_scheduled_action_ko_switchover(fake_restapi) -> None:
)
def test_cluster_has_scheduled_action_ko_restart(fake_restapi) -> None:
runner = CliRunner()
def test_cluster_has_scheduled_action_ko_restart(
runner: CliRunner, fake_restapi
) -> None:
fake_restapi("cluster_has_scheduled_action_ko_restart")
result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "cluster_has_scheduled_action"]

View file

@ -3,9 +3,7 @@ from click.testing import CliRunner
from check_patroni.cli import main
def test_cluster_is_in_maintenance_ok(fake_restapi) -> None:
runner = CliRunner()
def test_cluster_is_in_maintenance_ok(runner: CliRunner, fake_restapi) -> None:
fake_restapi("cluster_is_in_maintenance_ok")
result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "cluster_is_in_maintenance"]
@ -17,9 +15,7 @@ def test_cluster_is_in_maintenance_ok(fake_restapi) -> None:
)
def test_cluster_is_in_maintenance_ko(fake_restapi) -> None:
runner = CliRunner()
def test_cluster_is_in_maintenance_ko(runner: CliRunner, fake_restapi) -> None:
fake_restapi("cluster_is_in_maintenance_ko")
result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "cluster_is_in_maintenance"]
@ -31,9 +27,9 @@ def test_cluster_is_in_maintenance_ko(fake_restapi) -> None:
)
def test_cluster_is_in_maintenance_ok_pause_false(fake_restapi) -> None:
runner = CliRunner()
def test_cluster_is_in_maintenance_ok_pause_false(
runner: CliRunner, fake_restapi
) -> None:
fake_restapi("cluster_is_in_maintenance_ok_pause_false")
result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "cluster_is_in_maintenance"]

View file

@ -3,9 +3,9 @@ from click.testing import CliRunner
from check_patroni.cli import main
def test_cluster_node_count_ok(fake_restapi, use_old_replica_state: bool) -> None:
runner = CliRunner()
def test_cluster_node_count_ok(
runner: CliRunner, fake_restapi, use_old_replica_state: bool
) -> None:
fake_restapi("cluster_node_count_ok")
result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "cluster_node_count"]
@ -24,10 +24,8 @@ def test_cluster_node_count_ok(fake_restapi, use_old_replica_state: bool) -> Non
def test_cluster_node_count_ok_with_thresholds(
fake_restapi, use_old_replica_state: bool
runner: CliRunner, fake_restapi, use_old_replica_state: bool
) -> None:
runner = CliRunner()
fake_restapi("cluster_node_count_ok")
result = runner.invoke(
main,
@ -59,10 +57,8 @@ def test_cluster_node_count_ok_with_thresholds(
def test_cluster_node_count_healthy_warning(
fake_restapi, use_old_replica_state: bool
runner: CliRunner, fake_restapi, use_old_replica_state: bool
) -> None:
runner = CliRunner()
fake_restapi("cluster_node_count_healthy_warning")
result = runner.invoke(
main,
@ -89,9 +85,7 @@ def test_cluster_node_count_healthy_warning(
)
def test_cluster_node_count_healthy_critical(fake_restapi) -> None:
runner = CliRunner()
def test_cluster_node_count_healthy_critical(runner: CliRunner, fake_restapi) -> None:
fake_restapi("cluster_node_count_healthy_critical")
result = runner.invoke(
main,
@ -112,9 +106,9 @@ def test_cluster_node_count_healthy_critical(fake_restapi) -> None:
)
def test_cluster_node_count_warning(fake_restapi, use_old_replica_state: bool) -> None:
runner = CliRunner()
def test_cluster_node_count_warning(
runner: CliRunner, fake_restapi, use_old_replica_state: bool
) -> None:
fake_restapi("cluster_node_count_warning")
result = runner.invoke(
main,
@ -141,9 +135,7 @@ def test_cluster_node_count_warning(fake_restapi, use_old_replica_state: bool) -
)
def test_cluster_node_count_critical(fake_restapi) -> None:
runner = CliRunner()
def test_cluster_node_count_critical(runner: CliRunner, fake_restapi) -> None:
fake_restapi("cluster_node_count_critical")
result = runner.invoke(
main,

View file

@ -3,9 +3,7 @@ from click.testing import CliRunner
from check_patroni.cli import main
def test_node_is_alive_ok(fake_restapi) -> None:
runner = CliRunner()
def test_node_is_alive_ok(runner: CliRunner, fake_restapi) -> None:
fake_restapi(None)
result = runner.invoke(main, ["-e", "https://10.20.199.3:8008", "node_is_alive"])
assert result.exit_code == 0
@ -15,9 +13,7 @@ def test_node_is_alive_ok(fake_restapi) -> None:
)
def test_node_is_alive_ko(fake_restapi) -> None:
runner = CliRunner()
def test_node_is_alive_ko(runner: CliRunner, fake_restapi) -> None:
fake_restapi(None, status=404)
result = runner.invoke(main, ["-e", "https://10.20.199.3:8008", "node_is_alive"])
assert result.exit_code == 2

View file

@ -3,9 +3,7 @@ from click.testing import CliRunner
from check_patroni.cli import main
def test_node_is_leader_ok(fake_restapi) -> None:
runner = CliRunner()
def test_node_is_leader_ok(runner: CliRunner, fake_restapi) -> None:
fake_restapi("node_is_leader_ok")
result = runner.invoke(main, ["-e", "https://10.20.199.3:8008", "node_is_leader"])
assert result.exit_code == 0
@ -27,9 +25,7 @@ def test_node_is_leader_ok(fake_restapi) -> None:
)
def test_node_is_leader_ko(fake_restapi) -> None:
runner = CliRunner()
def test_node_is_leader_ko(runner: CliRunner, fake_restapi) -> None:
fake_restapi("node_is_leader_ko", status=503)
result = runner.invoke(main, ["-e", "https://10.20.199.3:8008", "node_is_leader"])
assert result.exit_code == 2

View file

@ -3,9 +3,7 @@ from click.testing import CliRunner
from check_patroni.cli import main
def test_node_is_pending_restart_ok(fake_restapi) -> None:
runner = CliRunner()
def test_node_is_pending_restart_ok(runner: CliRunner, fake_restapi) -> None:
fake_restapi("node_is_pending_restart_ok")
result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "node_is_pending_restart"]
@ -17,9 +15,7 @@ def test_node_is_pending_restart_ok(fake_restapi) -> None:
)
def test_node_is_pending_restart_ko(fake_restapi) -> None:
runner = CliRunner()
def test_node_is_pending_restart_ko(runner: CliRunner, fake_restapi) -> None:
fake_restapi("node_is_pending_restart_ko")
result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "node_is_pending_restart"]

View file

@ -3,9 +3,7 @@ from click.testing import CliRunner
from check_patroni.cli import main
def test_node_is_primary_ok(fake_restapi) -> None:
runner = CliRunner()
def test_node_is_primary_ok(runner: CliRunner, fake_restapi) -> None:
fake_restapi("node_is_primary_ok")
result = runner.invoke(main, ["-e", "https://10.20.199.3:8008", "node_is_primary"])
assert result.exit_code == 0
@ -15,9 +13,7 @@ def test_node_is_primary_ok(fake_restapi) -> None:
)
def test_node_is_primary_ko(fake_restapi) -> None:
runner = CliRunner()
def test_node_is_primary_ko(runner: CliRunner, fake_restapi) -> None:
fake_restapi("node_is_primary_ko", status=503)
result = runner.invoke(main, ["-e", "https://10.20.199.3:8008", "node_is_primary"])
assert result.exit_code == 2

View file

@ -3,9 +3,7 @@ from click.testing import CliRunner
from check_patroni.cli import main
def test_node_is_replica_ok(fake_restapi) -> None:
runner = CliRunner()
def test_node_is_replica_ok(runner: CliRunner, fake_restapi) -> None:
fake_restapi("node_is_replica_ok")
result = runner.invoke(main, ["-e", "https://10.20.199.3:8008", "node_is_replica"])
assert result.exit_code == 0
@ -15,9 +13,7 @@ def test_node_is_replica_ok(fake_restapi) -> None:
)
def test_node_is_replica_ko(fake_restapi) -> None:
runner = CliRunner()
def test_node_is_replica_ko(runner: CliRunner, fake_restapi) -> None:
fake_restapi("node_is_replica_ko", status=503)
result = runner.invoke(main, ["-e", "https://10.20.199.3:8008", "node_is_replica"])
assert result.exit_code == 2
@ -27,9 +23,7 @@ def test_node_is_replica_ko(fake_restapi) -> None:
)
def test_node_is_replica_ko_lag(fake_restapi) -> None:
runner = CliRunner()
def test_node_is_replica_ko_lag(runner: CliRunner, fake_restapi) -> None:
# We don't do the check ourselves, patroni does it and changes the return code
fake_restapi("node_is_replica_ok", status=503)
result = runner.invoke(
@ -60,9 +54,7 @@ def test_node_is_replica_ko_lag(fake_restapi) -> None:
)
def test_node_is_replica_sync_ok(fake_restapi) -> None:
runner = CliRunner()
def test_node_is_replica_sync_ok(runner: CliRunner, fake_restapi) -> None:
# We don't do the check ourselves, patroni does it and changes the return code
fake_restapi("node_is_replica_ok")
result = runner.invoke(
@ -75,9 +67,7 @@ def test_node_is_replica_sync_ok(fake_restapi) -> None:
)
def test_node_is_replica_sync_ko(fake_restapi) -> None:
runner = CliRunner()
def test_node_is_replica_sync_ko(runner: CliRunner, fake_restapi) -> None:
# We don't do the check ourselves, patroni does it and changes the return code
fake_restapi("node_is_replica_ok", status=503)
result = runner.invoke(
@ -90,9 +80,7 @@ def test_node_is_replica_sync_ko(fake_restapi) -> None:
)
def test_node_is_replica_async_ok(fake_restapi) -> None:
runner = CliRunner()
def test_node_is_replica_async_ok(runner: CliRunner, fake_restapi) -> None:
# We don't do the check ourselves, patroni does it and changes the return code
fake_restapi("node_is_replica_ok")
result = runner.invoke(
@ -105,9 +93,7 @@ def test_node_is_replica_async_ok(fake_restapi) -> None:
)
def test_node_is_replica_async_ko(fake_restapi) -> None:
runner = CliRunner()
def test_node_is_replica_async_ko(runner: CliRunner, fake_restapi) -> None:
# We don't do the check ourselves, patroni does it and changes the return code
fake_restapi("node_is_replica_ok", status=503)
result = runner.invoke(
@ -120,9 +106,7 @@ def test_node_is_replica_async_ko(fake_restapi) -> None:
)
def test_node_is_replica_params(fake_restapi) -> None:
runner = CliRunner()
def test_node_is_replica_params(runner: CliRunner, fake_restapi) -> None:
# We don't do the check ourselves, patroni does it and changes the return code
fake_restapi("node_is_replica_ok")
result = runner.invoke(

View file

@ -3,9 +3,7 @@ from click.testing import CliRunner
from check_patroni.cli import main
def test_node_patroni_version_ok(fake_restapi) -> None:
runner = CliRunner()
def test_node_patroni_version_ok(runner: CliRunner, fake_restapi) -> None:
fake_restapi("node_patroni_version")
result = runner.invoke(
main,
@ -24,9 +22,7 @@ def test_node_patroni_version_ok(fake_restapi) -> None:
)
def test_node_patroni_version_ko(fake_restapi) -> None:
runner = CliRunner()
def test_node_patroni_version_ko(runner: CliRunner, fake_restapi) -> None:
fake_restapi("node_patroni_version")
result = runner.invoke(
main,

View file

@ -6,9 +6,7 @@ from click.testing import CliRunner
from check_patroni.cli import main
def test_node_tl_has_changed_ok_with_timeline(fake_restapi) -> None:
runner = CliRunner()
def test_node_tl_has_changed_ok_with_timeline(runner: CliRunner, fake_restapi) -> None:
fake_restapi("node_tl_has_changed")
result = runner.invoke(
main,
@ -27,9 +25,9 @@ def test_node_tl_has_changed_ok_with_timeline(fake_restapi) -> None:
)
def test_node_tl_has_changed_ok_with_state_file(fake_restapi, tmp_path: Path) -> None:
runner = CliRunner()
def test_node_tl_has_changed_ok_with_state_file(
runner: CliRunner, fake_restapi, tmp_path: Path
) -> None:
state_file = tmp_path / "node_tl_has_changed.state_file"
with state_file.open("w") as f:
f.write('{"timeline": 58}')
@ -52,9 +50,7 @@ def test_node_tl_has_changed_ok_with_state_file(fake_restapi, tmp_path: Path) ->
)
def test_node_tl_has_changed_ko_with_timeline(fake_restapi) -> None:
runner = CliRunner()
def test_node_tl_has_changed_ko_with_timeline(runner: CliRunner, fake_restapi) -> None:
fake_restapi("node_tl_has_changed")
result = runner.invoke(
main,
@ -74,10 +70,8 @@ def test_node_tl_has_changed_ko_with_timeline(fake_restapi) -> None:
def test_node_tl_has_changed_ko_with_state_file_and_save(
fake_restapi, tmp_path: Path
runner: CliRunner, fake_restapi, tmp_path: Path
) -> None:
runner = CliRunner()
state_file = tmp_path / "node_tl_has_changed.state_file"
with state_file.open("w") as f:
f.write('{"timeline": 700}')
@ -133,10 +127,10 @@ def test_node_tl_has_changed_ko_with_state_file_and_save(
assert new_tl == 58
def test_node_tl_has_changed_params(fake_restapi, tmp_path: Path) -> None:
def test_node_tl_has_changed_params(
runner: CliRunner, fake_restapi, tmp_path: Path
) -> None:
# This one is placed last because it seems like the exceptions are not flushed from stderr for the next tests.
runner = CliRunner()
fake_state_file = tmp_path / "fake_file_name.state_file"
fake_restapi("node_tl_has_changed")
result = runner.invoke(