2023-11-08 17:50:32 +01:00
from pathlib import Path
from typing import Iterator , Union
import pytest
2022-02-07 15:11:05 +01:00
from click . testing import CliRunner
2021-08-11 19:09:14 +02:00
2022-02-07 15:11:05 +01:00
from check_patroni . cli import main
2023-03-20 11:44:19 +01:00
2023-11-08 17:50:32 +01:00
from . import PatroniAPI , cluster_api_set_replica_running
2021-08-11 19:09:14 +02:00
2023-11-08 17:50:32 +01:00
@pytest.fixture
def cluster_has_leader_ok (
patroni_api : PatroniAPI , old_replica_state : bool , datadir : Path , tmp_path : Path
) - > Iterator [ None ] :
cluster_path : Union [ str , Path ] = " cluster_has_leader_ok.json "
patroni_path = " cluster_has_replica_patroni_verion_3.1.0.json "
if old_replica_state :
cluster_path = cluster_api_set_replica_running ( datadir / cluster_path , tmp_path )
patroni_path = " cluster_has_replica_patroni_verion_3.0.0.json "
with patroni_api . routes ( { " cluster " : cluster_path , " patroni " : patroni_path } ) :
yield None
2023-09-28 09:41:33 +02:00
2023-11-08 17:50:32 +01:00
@pytest.mark.usefixtures ( " cluster_has_leader_ok " )
2023-09-28 09:41:33 +02:00
def test_cluster_has_leader_ok ( runner : CliRunner , patroni_api : PatroniAPI ) - > None :
2023-11-08 17:50:32 +01:00
result = runner . invoke ( main , [ " -e " , patroni_api . endpoint , " cluster_has_leader " ] )
2023-08-22 11:08:38 +02:00
assert (
result . stdout
2023-11-08 17:50:32 +01:00
== " CLUSTERHASLEADER OK - The cluster has a running leader. | has_leader=1;;@0 is_leader=1 is_standby_leader=0 is_standby_leader_in_arc_rec=0;@1:1 \n "
2023-08-22 11:08:38 +02:00
)
2023-11-08 17:50:32 +01:00
assert result . exit_code == 0
2021-08-11 19:09:14 +02:00
2023-11-08 17:50:32 +01:00
@pytest.fixture
def cluster_has_leader_ok_standby_leader (
patroni_api : PatroniAPI , old_replica_state : bool , datadir : Path , tmp_path : Path
) - > Iterator [ None ] :
cluster_path : Union [ str , Path ] = " cluster_has_leader_ok_standby_leader.json "
patroni_path = " cluster_has_replica_patroni_verion_3.1.0.json "
if old_replica_state :
cluster_path = cluster_api_set_replica_running ( datadir / cluster_path , tmp_path )
patroni_path = " cluster_has_replica_patroni_verion_3.0.0.json "
with patroni_api . routes ( { " cluster " : cluster_path , " patroni " : patroni_path } ) :
yield None
2021-08-11 19:09:14 +02:00
2023-11-08 17:50:32 +01:00
@pytest.mark.usefixtures ( " cluster_has_leader_ok_standby_leader " )
2023-09-28 09:41:33 +02:00
def test_cluster_has_leader_ok_standby_leader (
runner : CliRunner , patroni_api : PatroniAPI
) - > None :
2023-11-08 17:50:32 +01:00
result = runner . invoke ( main , [ " -e " , patroni_api . endpoint , " cluster_has_leader " ] )
2023-08-23 16:58:08 +02:00
assert (
result . stdout
2023-11-08 17:50:32 +01:00
== " CLUSTERHASLEADER OK - The cluster has a running leader. | has_leader=1;;@0 is_leader=0 is_standby_leader=1 is_standby_leader_in_arc_rec=0;@1:1 \n "
2023-08-23 16:58:08 +02:00
)
2023-11-08 17:50:32 +01:00
assert result . exit_code == 0
2023-08-23 16:58:08 +02:00
2023-11-08 17:50:32 +01:00
@pytest.fixture
def cluster_has_leader_ko (
patroni_api : PatroniAPI , old_replica_state : bool , datadir : Path , tmp_path : Path
) - > Iterator [ None ] :
cluster_path : Union [ str , Path ] = " cluster_has_leader_ko.json "
patroni_path = " cluster_has_replica_patroni_verion_3.1.0.json "
if old_replica_state :
cluster_path = cluster_api_set_replica_running ( datadir / cluster_path , tmp_path )
patroni_path = " cluster_has_replica_patroni_verion_3.0.0.json "
with patroni_api . routes ( { " cluster " : cluster_path , " patroni " : patroni_path } ) :
yield None
2023-08-23 16:58:08 +02:00
2023-11-08 17:50:32 +01:00
@pytest.mark.usefixtures ( " cluster_has_leader_ko " )
2023-09-28 09:41:33 +02:00
def test_cluster_has_leader_ko ( runner : CliRunner , patroni_api : PatroniAPI ) - > None :
2023-11-08 17:50:32 +01:00
result = runner . invoke ( main , [ " -e " , patroni_api . endpoint , " cluster_has_leader " ] )
assert (
result . stdout
== " CLUSTERHASLEADER CRITICAL - The cluster has no running leader or the standby leader is in archive recovery. | has_leader=0;;@0 is_leader=0 is_standby_leader=0 is_standby_leader_in_arc_rec=0;@1:1 \n "
)
assert result . exit_code == 2
@pytest.fixture
def cluster_has_leader_ko_standby_leader (
patroni_api : PatroniAPI , old_replica_state : bool , datadir : Path , tmp_path : Path
) - > Iterator [ None ] :
cluster_path : Union [ str , Path ] = " cluster_has_leader_ko_standby_leader.json "
patroni_path = " cluster_has_replica_patroni_verion_3.1.0.json "
if old_replica_state :
cluster_path = cluster_api_set_replica_running ( datadir / cluster_path , tmp_path )
patroni_path = " cluster_has_replica_patroni_verion_3.0.0.json "
with patroni_api . routes ( { " cluster " : cluster_path , " patroni " : patroni_path } ) :
yield None
@pytest.mark.usefixtures ( " cluster_has_leader_ko_standby_leader " )
def test_cluster_has_leader_ko_standby_leader (
runner : CliRunner , patroni_api : PatroniAPI
) - > None :
result = runner . invoke ( main , [ " -e " , patroni_api . endpoint , " cluster_has_leader " ] )
assert (
result . stdout
== " CLUSTERHASLEADER CRITICAL - The cluster has no running leader or the standby leader is in archive recovery. | has_leader=0;;@0 is_leader=0 is_standby_leader=0 is_standby_leader_in_arc_rec=0;@1:1 \n "
)
2021-08-11 19:09:14 +02:00
assert result . exit_code == 2
2023-11-08 17:50:32 +01:00
@pytest.fixture
def cluster_has_leader_ko_standby_leader_archiving (
patroni_api : PatroniAPI , old_replica_state : bool , datadir : Path , tmp_path : Path
) - > Iterator [ None ] :
2024-02-27 11:25:24 +01:00
cluster_path : Union [ str , Path ] = (
" cluster_has_leader_ko_standby_leader_archiving.json "
)
2023-11-08 17:50:32 +01:00
patroni_path = " cluster_has_replica_patroni_verion_3.1.0.json "
if old_replica_state :
cluster_path = cluster_api_set_replica_running ( datadir / cluster_path , tmp_path )
patroni_path = " cluster_has_replica_patroni_verion_3.0.0.json "
with patroni_api . routes ( { " cluster " : cluster_path , " patroni " : patroni_path } ) :
yield None
@pytest.mark.usefixtures ( " cluster_has_leader_ko_standby_leader_archiving " )
def test_cluster_has_leader_ko_standby_leader_archiving (
2023-12-19 16:56:52 +01:00
runner : CliRunner , patroni_api : PatroniAPI , old_replica_state : bool
2023-11-08 17:50:32 +01:00
) - > None :
result = runner . invoke ( main , [ " -e " , patroni_api . endpoint , " cluster_has_leader " ] )
2023-12-19 16:56:52 +01:00
if old_replica_state :
assert (
result . stdout
== " CLUSTERHASLEADER OK - The cluster has a running leader. | has_leader=1;;@0 is_leader=0 is_standby_leader=1 is_standby_leader_in_arc_rec=0;@1:1 \n "
)
assert result . exit_code == 0
else :
assert (
result . stdout
== " CLUSTERHASLEADER WARNING - The cluster has no running leader or the standby leader is in archive recovery. | has_leader=1;;@0 is_leader=0 is_standby_leader=1 is_standby_leader_in_arc_rec=1;@1:1 \n "
)
assert result . exit_code == 1