dprevot/check-patroni
1
0
Fork 0
Code Issues Pull requests Packages Projects Releases Wiki Activity

Introduce a fake_restapi test fixture

Browse source 
This fixture itself uses the 'use_old_replica_state' fixture, so that
it's no longer needed to use it explicitly in test functions.
This commit is contained in:
Denis Laxalde 2023-09-28 10:57:24 +02:00
parent c3cdb8cdd4
commit bc2d2917c3
15 changed files with 124 additions and 274 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
tests/conftest.py
View file

@ -1,4 +1,10 @@
from typing import Any from functools import partial
from typing import Any, Callable
import pytest
from pytest_mock import MockerFixture
from .tools import my_mock
def pytest_addoption(parser: Any) -> None: def pytest_addoption(parser: Any) -> None:
@ -13,3 +19,10 @@ def pytest_generate_tests(metafunc: Any) -> None:
metafunc.parametrize( metafunc.parametrize(
"use_old_replica_state", [metafunc.config.getoption("use_old_replica_state")] "use_old_replica_state", [metafunc.config.getoption("use_old_replica_state")]
) )
@pytest.fixture
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)

15
tests/test_api.py
View file

@ -1,29 +1,22 @@
from click.testing import CliRunner from click.testing import CliRunner
from pytest_mock import MockerFixture
from check_patroni.cli import main from check_patroni.cli import main
from .tools import my_mock
def test_api_status_code_200(fake_restapi) -> None:
def test_api_status_code_200(
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "node_is_pending_restart_ok") fake_restapi("node_is_pending_restart_ok")
result = runner.invoke( result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "node_is_pending_restart"] main, ["-e", "https://10.20.199.3:8008", "node_is_pending_restart"]
) )
assert result.exit_code == 0 assert result.exit_code == 0
def test_api_status_code_404( def test_api_status_code_404(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "Fake test", status=404) fake_restapi("Fake test", status=404)
result = runner.invoke( result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "node_is_pending_restart"] main, ["-e", "https://10.20.199.3:8008", "node_is_pending_restart"]
) )

34
tests/test_cluster_config_has_changed.py
View file

@ -1,18 +1,15 @@
import nagiosplugin import nagiosplugin
from click.testing import CliRunner from click.testing import CliRunner
from pytest_mock import MockerFixture
from check_patroni.cli import main from check_patroni.cli import main
from .tools import here, my_mock from .tools import here
def test_cluster_config_has_changed_ok_with_hash( def test_cluster_config_has_changed_ok_with_hash(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "cluster_config_has_changed") fake_restapi("cluster_config_has_changed")
result = runner.invoke( result = runner.invoke(
main, main,
[ [
@ -30,15 +27,13 @@ def test_cluster_config_has_changed_ok_with_hash(
) )
def test_cluster_config_has_changed_ok_with_state_file( def test_cluster_config_has_changed_ok_with_state_file(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
with open(here / "cluster_config_has_changed.state_file", "w") as f: with open(here / "cluster_config_has_changed.state_file", "w") as f:
f.write('{"hash": "96b12d82571473d13e890b893734e731"}') f.write('{"hash": "96b12d82571473d13e890b893734e731"}')
my_mock(mocker, "cluster_config_has_changed") fake_restapi("cluster_config_has_changed")
result = runner.invoke( result = runner.invoke(
main, main,
[ [
@ -56,12 +51,10 @@ def test_cluster_config_has_changed_ok_with_state_file(
) )
def test_cluster_config_has_changed_ko_with_hash( def test_cluster_config_has_changed_ko_with_hash(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "cluster_config_has_changed") fake_restapi("cluster_config_has_changed")
result = runner.invoke( result = runner.invoke(
main, main,
[ [
@ -79,16 +72,13 @@ def test_cluster_config_has_changed_ko_with_hash(
) )
def test_cluster_config_has_changed_ko_with_state_file_and_save( def test_cluster_config_has_changed_ko_with_state_file_and_save(fake_restapi) -> None:
mocker: MockerFixture,
use_old_replica_state: bool,
) -> None:
runner = CliRunner() runner = CliRunner()
with open(here / "cluster_config_has_changed.state_file", "w") as f: with open(here / "cluster_config_has_changed.state_file", "w") as f:
f.write('{"hash": "96b12d82571473d13e890b8937ffffff"}') f.write('{"hash": "96b12d82571473d13e890b8937ffffff"}')
my_mock(mocker, "cluster_config_has_changed") fake_restapi("cluster_config_has_changed")
# test without saving the new hash # test without saving the new hash
result = runner.invoke( result = runner.invoke(
main, main,
@ -139,13 +129,11 @@ def test_cluster_config_has_changed_ko_with_state_file_and_save(
assert new_config_hash == "96b12d82571473d13e890b893734e731" assert new_config_hash == "96b12d82571473d13e890b893734e731"
def test_cluster_config_has_changed_params( def test_cluster_config_has_changed_params(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
# This one is placed last because it seems like the exceptions are not flushed from stderr for the next tests. # This one is placed last because it seems like the exceptions are not flushed from stderr for the next tests.
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "cluster_config_has_changed") fake_restapi("cluster_config_has_changed")
result = runner.invoke( result = runner.invoke(
main, main,
[ [

21
tests/test_cluster_has_leader.py
View file

@ -1,17 +1,12 @@
from click.testing import CliRunner from click.testing import CliRunner
from pytest_mock import MockerFixture
from check_patroni.cli import main from check_patroni.cli import main
from .tools import my_mock
def test_cluster_has_leader_ok(fake_restapi) -> None:
def test_cluster_has_leader_ok(
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "cluster_has_leader_ok") fake_restapi("cluster_has_leader_ok")
result = runner.invoke( result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "cluster_has_leader"] main, ["-e", "https://10.20.199.3:8008", "cluster_has_leader"]
) )
@ -22,12 +17,10 @@ def test_cluster_has_leader_ok(
) )
def test_cluster_has_leader_ok_standby_leader( def test_cluster_has_leader_ok_standby_leader(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "cluster_has_leader_ok_standby_leader") fake_restapi("cluster_has_leader_ok_standby_leader")
result = runner.invoke( result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "cluster_has_leader"] main, ["-e", "https://10.20.199.3:8008", "cluster_has_leader"]
) )
@ -38,12 +31,10 @@ def test_cluster_has_leader_ok_standby_leader(
) )
def test_cluster_has_leader_ko( def test_cluster_has_leader_ko(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "cluster_has_leader_ko") fake_restapi("cluster_has_leader_ko")
result = runner.invoke( result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "cluster_has_leader"] main, ["-e", "https://10.20.199.3:8008", "cluster_has_leader"]
) )

65
tests/test_cluster_has_replica.py
View file

@ -1,20 +1,13 @@
from click.testing import CliRunner from click.testing import CliRunner
from pytest_mock import MockerFixture
from check_patroni.cli import main from check_patroni.cli import main
from .tools import my_mock
# TODO Lag threshold tests # TODO Lag threshold tests
def test_cluster_has_relica_ok( def test_cluster_has_relica_ok(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock( fake_restapi("cluster_has_replica_ok")
mocker, "cluster_has_replica_ok", use_old_replica_state=use_old_replica_state
)
result = runner.invoke( result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "cluster_has_replica"] main, ["-e", "https://10.20.199.3:8008", "cluster_has_replica"]
) )
@ -25,14 +18,10 @@ def test_cluster_has_relica_ok(
) )
def test_cluster_has_replica_ok_with_count_thresholds( def test_cluster_has_replica_ok_with_count_thresholds(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock( fake_restapi("cluster_has_replica_ok")
mocker, "cluster_has_replica_ok", use_old_replica_state=use_old_replica_state
)
result = runner.invoke( result = runner.invoke(
main, main,
[ [
@ -52,14 +41,10 @@ def test_cluster_has_replica_ok_with_count_thresholds(
) )
def test_cluster_has_replica_ok_with_sync_count_thresholds( def test_cluster_has_replica_ok_with_sync_count_thresholds(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock( fake_restapi("cluster_has_replica_ok")
mocker, "cluster_has_replica_ok", use_old_replica_state=use_old_replica_state
)
result = runner.invoke( result = runner.invoke(
main, main,
[ [
@ -77,17 +62,10 @@ def test_cluster_has_replica_ok_with_sync_count_thresholds(
) )
def test_cluster_has_replica_ok_with_count_thresholds_lag( def test_cluster_has_replica_ok_with_count_thresholds_lag(fake_restapi) -> None:
mocker: MockerFixture,
use_old_replica_state: bool,
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock( fake_restapi("cluster_has_replica_ok_lag")
mocker,
"cluster_has_replica_ok_lag",
use_old_replica_state=use_old_replica_state,
)
result = runner.invoke( result = runner.invoke(
main, main,
[ [
@ -109,14 +87,10 @@ def test_cluster_has_replica_ok_with_count_thresholds_lag(
) )
def test_cluster_has_replica_ko_with_count_thresholds( def test_cluster_has_replica_ko_with_count_thresholds(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock( fake_restapi("cluster_has_replica_ko")
mocker, "cluster_has_replica_ko", use_old_replica_state=use_old_replica_state
)
result = runner.invoke( result = runner.invoke(
main, main,
[ [
@ -136,14 +110,10 @@ def test_cluster_has_replica_ko_with_count_thresholds(
) )
def test_cluster_has_replica_ko_with_sync_count_thresholds( def test_cluster_has_replica_ko_with_sync_count_thresholds(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock( fake_restapi("cluster_has_replica_ko")
mocker, "cluster_has_replica_ko", use_old_replica_state=use_old_replica_state
)
result = runner.invoke( result = runner.invoke(
main, main,
[ [
@ -163,17 +133,10 @@ def test_cluster_has_replica_ko_with_sync_count_thresholds(
) )
def test_cluster_has_replica_ko_with_count_thresholds_and_lag( def test_cluster_has_replica_ko_with_count_thresholds_and_lag(fake_restapi) -> None:
mocker: MockerFixture,
use_old_replica_state: bool,
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock( fake_restapi("cluster_has_replica_ko_lag")
mocker,
"cluster_has_replica_ko_lag",
use_old_replica_state=use_old_replica_state,
)
result = runner.invoke( result = runner.invoke(
main, main,
[ [

21
tests/test_cluster_has_scheduled_action.py
View file

@ -1,17 +1,12 @@
from click.testing import CliRunner from click.testing import CliRunner
from pytest_mock import MockerFixture
from check_patroni.cli import main from check_patroni.cli import main
from .tools import my_mock
def test_cluster_has_scheduled_action_ok(fake_restapi) -> None:
def test_cluster_has_scheduled_action_ok(
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "cluster_has_scheduled_action_ok") fake_restapi("cluster_has_scheduled_action_ok")
result = runner.invoke( result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "cluster_has_scheduled_action"] main, ["-e", "https://10.20.199.3:8008", "cluster_has_scheduled_action"]
) )
@ -22,12 +17,10 @@ def test_cluster_has_scheduled_action_ok(
) )
def test_cluster_has_scheduled_action_ko_switchover( def test_cluster_has_scheduled_action_ko_switchover(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "cluster_has_scheduled_action_ko_switchover") fake_restapi("cluster_has_scheduled_action_ko_switchover")
result = runner.invoke( result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "cluster_has_scheduled_action"] main, ["-e", "https://10.20.199.3:8008", "cluster_has_scheduled_action"]
) )
@ -38,12 +31,10 @@ def test_cluster_has_scheduled_action_ko_switchover(
) )
def test_cluster_has_scheduled_action_ko_restart( def test_cluster_has_scheduled_action_ko_restart(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "cluster_has_scheduled_action_ko_restart") fake_restapi("cluster_has_scheduled_action_ko_restart")
result = runner.invoke( result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "cluster_has_scheduled_action"] main, ["-e", "https://10.20.199.3:8008", "cluster_has_scheduled_action"]
) )

21
tests/test_cluster_is_in_maintenance.py
View file

@ -1,17 +1,12 @@
from click.testing import CliRunner from click.testing import CliRunner
from pytest_mock import MockerFixture
from check_patroni.cli import main from check_patroni.cli import main
from .tools import my_mock
def test_cluster_is_in_maintenance_ok(fake_restapi) -> None:
def test_cluster_is_in_maintenance_ok(
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "cluster_is_in_maintenance_ok") fake_restapi("cluster_is_in_maintenance_ok")
result = runner.invoke( result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "cluster_is_in_maintenance"] main, ["-e", "https://10.20.199.3:8008", "cluster_is_in_maintenance"]
) )
@ -22,12 +17,10 @@ def test_cluster_is_in_maintenance_ok(
) )
def test_cluster_is_in_maintenance_ko( def test_cluster_is_in_maintenance_ko(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "cluster_is_in_maintenance_ko") fake_restapi("cluster_is_in_maintenance_ko")
result = runner.invoke( result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "cluster_is_in_maintenance"] main, ["-e", "https://10.20.199.3:8008", "cluster_is_in_maintenance"]
) )
@ -38,12 +31,10 @@ def test_cluster_is_in_maintenance_ko(
) )
def test_cluster_is_in_maintenance_ok_pause_false( def test_cluster_is_in_maintenance_ok_pause_false(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "cluster_is_in_maintenance_ok_pause_false") fake_restapi("cluster_is_in_maintenance_ok_pause_false")
result = runner.invoke( result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "cluster_is_in_maintenance"] main, ["-e", "https://10.20.199.3:8008", "cluster_is_in_maintenance"]
) )

55
tests/test_cluster_node_count.py
View file

@ -1,19 +1,12 @@
from click.testing import CliRunner from click.testing import CliRunner
from pytest_mock import MockerFixture
from check_patroni.cli import main from check_patroni.cli import main
from .tools import my_mock
def test_cluster_node_count_ok(fake_restapi, use_old_replica_state: bool) -> None:
def test_cluster_node_count_ok(
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock( fake_restapi("cluster_node_count_ok")
mocker, "cluster_node_count_ok", use_old_replica_state=use_old_replica_state
)
result = runner.invoke( result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "cluster_node_count"] main, ["-e", "https://10.20.199.3:8008", "cluster_node_count"]
) )
@ -31,13 +24,11 @@ def test_cluster_node_count_ok(
def test_cluster_node_count_ok_with_thresholds( def test_cluster_node_count_ok_with_thresholds(
mocker: MockerFixture, use_old_replica_state: bool fake_restapi, use_old_replica_state: bool
) -> None: ) -> None:
runner = CliRunner() runner = CliRunner()
my_mock( fake_restapi("cluster_node_count_ok")
mocker, "cluster_node_count_ok", use_old_replica_state=use_old_replica_state
)
result = runner.invoke( result = runner.invoke(
main, main,
[ [
@ -68,15 +59,11 @@ def test_cluster_node_count_ok_with_thresholds(
def test_cluster_node_count_healthy_warning( def test_cluster_node_count_healthy_warning(
mocker: MockerFixture, use_old_replica_state: bool fake_restapi, use_old_replica_state: bool
) -> None: ) -> None:
runner = CliRunner() runner = CliRunner()
my_mock( fake_restapi("cluster_node_count_healthy_warning")
mocker,
"cluster_node_count_healthy_warning",
use_old_replica_state=use_old_replica_state,
)
result = runner.invoke( result = runner.invoke(
main, main,
[ [
@ -102,16 +89,10 @@ def test_cluster_node_count_healthy_warning(
) )
def test_cluster_node_count_healthy_critical( def test_cluster_node_count_healthy_critical(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock( fake_restapi("cluster_node_count_healthy_critical")
mocker,
"cluster_node_count_healthy_critical",
use_old_replica_state=use_old_replica_state,
)
result = runner.invoke( result = runner.invoke(
main, main,
[ [
@ -131,16 +112,10 @@ def test_cluster_node_count_healthy_critical(
) )
def test_cluster_node_count_warning( def test_cluster_node_count_warning(fake_restapi, use_old_replica_state: bool) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock( fake_restapi("cluster_node_count_warning")
mocker,
"cluster_node_count_warning",
use_old_replica_state=use_old_replica_state,
)
result = runner.invoke( result = runner.invoke(
main, main,
[ [
@ -166,16 +141,10 @@ def test_cluster_node_count_warning(
) )
def test_cluster_node_count_critical( def test_cluster_node_count_critical(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock( fake_restapi("cluster_node_count_critical")
mocker,
"cluster_node_count_critical",
use_old_replica_state=use_old_replica_state,
)
result = runner.invoke( result = runner.invoke(
main, main,
[ [

11
tests/test_node_is_alive.py
View file

@ -1,15 +1,12 @@
from click.testing import CliRunner from click.testing import CliRunner
from pytest_mock import MockerFixture
from check_patroni.cli import main from check_patroni.cli import main
from .tools import my_mock
def test_node_is_alive_ok(fake_restapi) -> None:
def test_node_is_alive_ok(mocker: MockerFixture, use_old_replica_state: bool) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, None) fake_restapi(None)
result = runner.invoke(main, ["-e", "https://10.20.199.3:8008", "node_is_alive"]) result = runner.invoke(main, ["-e", "https://10.20.199.3:8008", "node_is_alive"])
assert result.exit_code == 0 assert result.exit_code == 0
assert ( assert (
@ -18,10 +15,10 @@ def test_node_is_alive_ok(mocker: MockerFixture, use_old_replica_state: bool) ->
) )
def test_node_is_alive_ko(mocker: MockerFixture, use_old_replica_state: bool) -> None: def test_node_is_alive_ko(fake_restapi) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, None, status=404) fake_restapi(None, status=404)
result = runner.invoke(main, ["-e", "https://10.20.199.3:8008", "node_is_alive"]) result = runner.invoke(main, ["-e", "https://10.20.199.3:8008", "node_is_alive"])
assert result.exit_code == 2 assert result.exit_code == 2
assert ( assert (

15
tests/test_node_is_leader.py
View file

@ -1,15 +1,12 @@
from click.testing import CliRunner from click.testing import CliRunner
from pytest_mock import MockerFixture
from check_patroni.cli import main from check_patroni.cli import main
from .tools import my_mock
def test_node_is_leader_ok(fake_restapi) -> None:
def test_node_is_leader_ok(mocker: MockerFixture, use_old_replica_state: bool) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "node_is_leader_ok") fake_restapi("node_is_leader_ok")
result = runner.invoke(main, ["-e", "https://10.20.199.3:8008", "node_is_leader"]) result = runner.invoke(main, ["-e", "https://10.20.199.3:8008", "node_is_leader"])
assert result.exit_code == 0 assert result.exit_code == 0
assert ( assert (
@ -17,7 +14,7 @@ def test_node_is_leader_ok(mocker: MockerFixture, use_old_replica_state: bool) -
== "NODEISLEADER OK - This node is a leader node. | is_leader=1;;@0\n" == "NODEISLEADER OK - This node is a leader node. | is_leader=1;;@0\n"
) )
my_mock(mocker, "node_is_leader_ok_standby_leader") fake_restapi("node_is_leader_ok_standby_leader")
result = runner.invoke( result = runner.invoke(
main, main,
["-e", "https://10.20.199.3:8008", "node_is_leader", "--is-standby-leader"], ["-e", "https://10.20.199.3:8008", "node_is_leader", "--is-standby-leader"],
@ -30,10 +27,10 @@ def test_node_is_leader_ok(mocker: MockerFixture, use_old_replica_state: bool) -
) )
def test_node_is_leader_ko(mocker: MockerFixture, use_old_replica_state: bool) -> None: def test_node_is_leader_ko(fake_restapi) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "node_is_leader_ko", status=503) fake_restapi("node_is_leader_ko", status=503)
result = runner.invoke(main, ["-e", "https://10.20.199.3:8008", "node_is_leader"]) result = runner.invoke(main, ["-e", "https://10.20.199.3:8008", "node_is_leader"])
assert result.exit_code == 2 assert result.exit_code == 2
assert ( assert (
@ -41,7 +38,7 @@ def test_node_is_leader_ko(mocker: MockerFixture, use_old_replica_state: bool) -
== "NODEISLEADER CRITICAL - This node is not a leader node. | is_leader=0;;@0\n" == "NODEISLEADER CRITICAL - This node is not a leader node. | is_leader=0;;@0\n"
) )
my_mock(mocker, "node_is_leader_ko_standby_leader", status=503) fake_restapi("node_is_leader_ko_standby_leader", status=503)
result = runner.invoke( result = runner.invoke(
main, main,
["-e", "https://10.20.199.3:8008", "node_is_leader", "--is-standby-leader"], ["-e", "https://10.20.199.3:8008", "node_is_leader", "--is-standby-leader"],

15
tests/test_node_is_pending_restart.py
View file

@ -1,17 +1,12 @@
from click.testing import CliRunner from click.testing import CliRunner
from pytest_mock import MockerFixture
from check_patroni.cli import main from check_patroni.cli import main
from .tools import my_mock
def test_node_is_pending_restart_ok(fake_restapi) -> None:
def test_node_is_pending_restart_ok(
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "node_is_pending_restart_ok") fake_restapi("node_is_pending_restart_ok")
result = runner.invoke( result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "node_is_pending_restart"] main, ["-e", "https://10.20.199.3:8008", "node_is_pending_restart"]
) )
@ -22,12 +17,10 @@ def test_node_is_pending_restart_ok(
) )
def test_node_is_pending_restart_ko( def test_node_is_pending_restart_ko(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "node_is_pending_restart_ko") fake_restapi("node_is_pending_restart_ko")
result = runner.invoke( result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "node_is_pending_restart"] main, ["-e", "https://10.20.199.3:8008", "node_is_pending_restart"]
) )

11
tests/test_node_is_primary.py
View file

@ -1,15 +1,12 @@
from click.testing import CliRunner from click.testing import CliRunner
from pytest_mock import MockerFixture
from check_patroni.cli import main from check_patroni.cli import main
from .tools import my_mock
def test_node_is_primary_ok(fake_restapi) -> None:
def test_node_is_primary_ok(mocker: MockerFixture, use_old_replica_state: bool) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "node_is_primary_ok") fake_restapi("node_is_primary_ok")
result = runner.invoke(main, ["-e", "https://10.20.199.3:8008", "node_is_primary"]) result = runner.invoke(main, ["-e", "https://10.20.199.3:8008", "node_is_primary"])
assert result.exit_code == 0 assert result.exit_code == 0
assert ( assert (
@ -18,10 +15,10 @@ def test_node_is_primary_ok(mocker: MockerFixture, use_old_replica_state: bool)
) )
def test_node_is_primary_ko(mocker: MockerFixture, use_old_replica_state: bool) -> None: def test_node_is_primary_ko(fake_restapi) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "node_is_primary_ko", status=503) fake_restapi("node_is_primary_ko", status=503)
result = runner.invoke(main, ["-e", "https://10.20.199.3:8008", "node_is_primary"]) result = runner.invoke(main, ["-e", "https://10.20.199.3:8008", "node_is_primary"])
assert result.exit_code == 2 assert result.exit_code == 2
assert ( assert (

51
tests/test_node_is_replica.py
View file

@ -1,15 +1,12 @@
from click.testing import CliRunner from click.testing import CliRunner
from pytest_mock import MockerFixture
from check_patroni.cli import main from check_patroni.cli import main
from .tools import my_mock
def test_node_is_replica_ok(fake_restapi) -> None:
def test_node_is_replica_ok(mocker: MockerFixture, use_old_replica_state: bool) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "node_is_replica_ok") fake_restapi("node_is_replica_ok")
result = runner.invoke(main, ["-e", "https://10.20.199.3:8008", "node_is_replica"]) result = runner.invoke(main, ["-e", "https://10.20.199.3:8008", "node_is_replica"])
assert result.exit_code == 0 assert result.exit_code == 0
assert ( assert (
@ -18,10 +15,10 @@ def test_node_is_replica_ok(mocker: MockerFixture, use_old_replica_state: bool)
) )
def test_node_is_replica_ko(mocker: MockerFixture, use_old_replica_state: bool) -> None: def test_node_is_replica_ko(fake_restapi) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "node_is_replica_ko", status=503) fake_restapi("node_is_replica_ko", status=503)
result = runner.invoke(main, ["-e", "https://10.20.199.3:8008", "node_is_replica"]) result = runner.invoke(main, ["-e", "https://10.20.199.3:8008", "node_is_replica"])
assert result.exit_code == 2 assert result.exit_code == 2
assert ( assert (
@ -30,13 +27,11 @@ def test_node_is_replica_ko(mocker: MockerFixture, use_old_replica_state: bool)
) )
def test_node_is_replica_ko_lag( def test_node_is_replica_ko_lag(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
# We don't do the check ourselves, patroni does it and changes the return code # We don't do the check ourselves, patroni does it and changes the return code
my_mock(mocker, "node_is_replica_ok", status=503) fake_restapi("node_is_replica_ok", status=503)
result = runner.invoke( result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "node_is_replica", "--max-lag", "100"] main, ["-e", "https://10.20.199.3:8008", "node_is_replica", "--max-lag", "100"]
) )
@ -46,7 +41,7 @@ def test_node_is_replica_ko_lag(
== "NODEISREPLICA CRITICAL - This node is not a running replica with no noloadbalance tag and a lag under 100. | is_replica=0;;@0\n" == "NODEISREPLICA CRITICAL - This node is not a running replica with no noloadbalance tag and a lag under 100. | is_replica=0;;@0\n"
) )
my_mock(mocker, "node_is_replica_ok", status=503) fake_restapi("node_is_replica_ok", status=503)
result = runner.invoke( result = runner.invoke(
main, main,
[ [
@ -65,13 +60,11 @@ def test_node_is_replica_ko_lag(
) )
def test_node_is_replica_sync_ok( def test_node_is_replica_sync_ok(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
# We don't do the check ourselves, patroni does it and changes the return code # We don't do the check ourselves, patroni does it and changes the return code
my_mock(mocker, "node_is_replica_ok") fake_restapi("node_is_replica_ok")
result = runner.invoke( result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "node_is_replica", "--is-sync"] main, ["-e", "https://10.20.199.3:8008", "node_is_replica", "--is-sync"]
) )
@ -82,13 +75,11 @@ def test_node_is_replica_sync_ok(
) )
def test_node_is_replica_sync_ko( def test_node_is_replica_sync_ko(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
# We don't do the check ourselves, patroni does it and changes the return code # We don't do the check ourselves, patroni does it and changes the return code
my_mock(mocker, "node_is_replica_ok", status=503) fake_restapi("node_is_replica_ok", status=503)
result = runner.invoke( result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "node_is_replica", "--is-sync"] main, ["-e", "https://10.20.199.3:8008", "node_is_replica", "--is-sync"]
) )
@ -99,13 +90,11 @@ def test_node_is_replica_sync_ko(
) )
def test_node_is_replica_async_ok( def test_node_is_replica_async_ok(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
# We don't do the check ourselves, patroni does it and changes the return code # We don't do the check ourselves, patroni does it and changes the return code
my_mock(mocker, "node_is_replica_ok") fake_restapi("node_is_replica_ok")
result = runner.invoke( result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "node_is_replica", "--is-async"] main, ["-e", "https://10.20.199.3:8008", "node_is_replica", "--is-async"]
) )
@ -116,13 +105,11 @@ def test_node_is_replica_async_ok(
) )
def test_node_is_replica_async_ko( def test_node_is_replica_async_ko(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
# We don't do the check ourselves, patroni does it and changes the return code # We don't do the check ourselves, patroni does it and changes the return code
my_mock(mocker, "node_is_replica_ok", status=503) fake_restapi("node_is_replica_ok", status=503)
result = runner.invoke( result = runner.invoke(
main, ["-e", "https://10.20.199.3:8008", "node_is_replica", "--is-async"] main, ["-e", "https://10.20.199.3:8008", "node_is_replica", "--is-async"]
) )
@ -133,13 +120,11 @@ def test_node_is_replica_async_ko(
) )
def test_node_is_replica_params( def test_node_is_replica_params(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
# We don't do the check ourselves, patroni does it and changes the return code # We don't do the check ourselves, patroni does it and changes the return code
my_mock(mocker, "node_is_replica_ok") fake_restapi("node_is_replica_ok")
result = runner.invoke( result = runner.invoke(
main, main,
[ [
@ -157,7 +142,7 @@ def test_node_is_replica_params(
) )
# We don't do the check ourselves, patroni does it and changes the return code # We don't do the check ourselves, patroni does it and changes the return code
my_mock(mocker, "node_is_replica_ok") fake_restapi("node_is_replica_ok")
result = runner.invoke( result = runner.invoke(
main, main,
[ [

15
tests/test_node_patroni_version.py
View file

@ -1,17 +1,12 @@
from click.testing import CliRunner from click.testing import CliRunner
from pytest_mock import MockerFixture
from check_patroni.cli import main from check_patroni.cli import main
from .tools import my_mock
def test_node_patroni_version_ok(fake_restapi) -> None:
def test_node_patroni_version_ok(
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "node_patroni_version") fake_restapi("node_patroni_version")
result = runner.invoke( result = runner.invoke(
main, main,
[ [
@ -29,12 +24,10 @@ def test_node_patroni_version_ok(
) )
def test_node_patroni_version_ko( def test_node_patroni_version_ko(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "node_patroni_version") fake_restapi("node_patroni_version")
result = runner.invoke( result = runner.invoke(
main, main,
[ [

33
tests/test_node_tl_has_changed.py
View file

@ -1,18 +1,15 @@
import nagiosplugin import nagiosplugin
from click.testing import CliRunner from click.testing import CliRunner
from pytest_mock import MockerFixture
from check_patroni.cli import main from check_patroni.cli import main
from .tools import here, my_mock from .tools import here
def test_node_tl_has_changed_ok_with_timeline( def test_node_tl_has_changed_ok_with_timeline(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "node_tl_has_changed") fake_restapi("node_tl_has_changed")
result = runner.invoke( result = runner.invoke(
main, main,
[ [
@ -30,15 +27,13 @@ def test_node_tl_has_changed_ok_with_timeline(
) )
def test_node_tl_has_changed_ok_with_state_file( def test_node_tl_has_changed_ok_with_state_file(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
with open(here / "node_tl_has_changed.state_file", "w") as f: with open(here / "node_tl_has_changed.state_file", "w") as f:
f.write('{"timeline": 58}') f.write('{"timeline": 58}')
my_mock(mocker, "node_tl_has_changed") fake_restapi("node_tl_has_changed")
result = runner.invoke( result = runner.invoke(
main, main,
[ [
@ -56,12 +51,10 @@ def test_node_tl_has_changed_ok_with_state_file(
) )
def test_node_tl_has_changed_ko_with_timeline( def test_node_tl_has_changed_ko_with_timeline(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "node_tl_has_changed") fake_restapi("node_tl_has_changed")
result = runner.invoke( result = runner.invoke(
main, main,
[ [
@ -79,15 +72,13 @@ def test_node_tl_has_changed_ko_with_timeline(
) )
def test_node_tl_has_changed_ko_with_state_file_and_save( def test_node_tl_has_changed_ko_with_state_file_and_save(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
runner = CliRunner() runner = CliRunner()
with open(here / "node_tl_has_changed.state_file", "w") as f: with open(here / "node_tl_has_changed.state_file", "w") as f:
f.write('{"timeline": 700}') f.write('{"timeline": 700}')
my_mock(mocker, "node_tl_has_changed") fake_restapi("node_tl_has_changed")
# test without saving the new tl # test without saving the new tl
result = runner.invoke( result = runner.invoke(
main, main,
@ -138,13 +129,11 @@ def test_node_tl_has_changed_ko_with_state_file_and_save(
assert new_tl == 58 assert new_tl == 58
def test_node_tl_has_changed_params( def test_node_tl_has_changed_params(fake_restapi) -> None:
mocker: MockerFixture, use_old_replica_state: bool
) -> None:
# This one is placed last because it seems like the exceptions are not flushed from stderr for the next tests. # This one is placed last because it seems like the exceptions are not flushed from stderr for the next tests.
runner = CliRunner() runner = CliRunner()
my_mock(mocker, "node_tl_has_changed") fake_restapi("node_tl_has_changed")
result = runner.invoke( result = runner.invoke(
main, main,
[ [