diff --git a/lib/charms/tempo_k8s/v2/tracing.py b/lib/charms/tempo_k8s/v2/tracing.py index dfb23365fb..81bf1f1f5f 100644 --- a/lib/charms/tempo_k8s/v2/tracing.py +++ b/lib/charms/tempo_k8s/v2/tracing.py @@ -97,7 +97,7 @@ def __init__(self, *args): ) from ops.framework import EventSource, Object from ops.model import ModelError, Relation -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, Field # The unique Charmhub library identifier, never change it LIBID = "12977e9aa0b34367903d8afeb8c3d85d" @@ -107,7 +107,7 @@ def __init__(self, *args): # Increment this PATCH version before using `charmcraft publish-lib` or reset # to 0 if you are raising the major API version -LIBPATCH = 8 +LIBPATCH = 10 PYDEPS = ["pydantic"] @@ -338,7 +338,7 @@ class Config: class ProtocolType(BaseModel): """Protocol Type.""" - model_config = ConfigDict( + model_config = ConfigDict( # type: ignore # Allow serializing enum values. use_enum_values=True ) @@ -902,7 +902,16 @@ def _get_endpoint( def get_endpoint( self, protocol: ReceiverProtocol, relation: Optional[Relation] = None ) -> Optional[str]: - """Receiver endpoint for the given protocol.""" + """Receiver endpoint for the given protocol. + + It could happen that this function gets called before the provider publishes the endpoints. + In such a scenario, if a non-leader unit calls this function, a permission denied exception will be raised due to + restricted access. To prevent this, this function needs to be guarded by the `is_ready` check. + + Raises: + ProtocolNotRequestedError: + If the charm unit is the leader unit and attempts to obtain an endpoint for a protocol it did not request. + """ endpoint = self._get_endpoint(relation or self._relation, protocol=protocol) if not endpoint: requested_protocols = set() @@ -925,7 +934,7 @@ def get_endpoint( def charm_tracing_config( endpoint_requirer: TracingEndpointRequirer, cert_path: Optional[Union[Path, str]] ) -> Tuple[Optional[str], Optional[str]]: - """Utility function to determine the charm_tracing config you will likely want. + """Return the charm_tracing config you likely want. If no endpoint is provided: disable charm tracing. diff --git a/tests/unit/test_backups.py b/tests/unit/test_backups.py index c46a4e272d..4769e9251f 100644 --- a/tests/unit/test_backups.py +++ b/tests/unit/test_backups.py @@ -3,7 +3,6 @@ from pathlib import PosixPath from subprocess import PIPE, CompletedProcess, TimeoutExpired from typing import OrderedDict -from unittest import TestCase from unittest.mock import ANY, MagicMock, PropertyMock, call, mock_open, patch import botocore as botocore @@ -26,9 +25,6 @@ FAILED_TO_INITIALIZE_STANZA_ERROR_MESSAGE = "failed to initialize stanza, check your S3 settings" S3_PARAMETERS_RELATION = "s3-parameters" -# used for assert functions -tc = TestCase() - @pytest.fixture(autouse=True) def harness(): @@ -43,18 +39,15 @@ def harness(): def test_stanza_name(harness): - tc.assertEqual( - harness.charm.backup.stanza_name, - f"{harness.charm.model.name}.{harness.charm.cluster_name}", + assert ( + harness.charm.backup.stanza_name + == f"{harness.charm.model.name}.{harness.charm.cluster_name}" ) def test_tls_ca_chain_filename(harness): # Test when the TLS CA chain is not available. - tc.assertEqual( - harness.charm.backup._tls_ca_chain_filename, - "", - ) + assert harness.charm.backup._tls_ca_chain_filename == "" # Test when the TLS CA chain is available. with harness.hooks_disabled(): @@ -70,33 +63,30 @@ def test_tls_ca_chain_filename(harness): "tls-ca-chain": '["fake-tls-ca-chain"]', }, ) - tc.assertEqual( - harness.charm.backup._tls_ca_chain_filename, - "/var/snap/charmed-postgresql/common/pgbackrest-tls-ca-chain.crt", + assert ( + harness.charm.backup._tls_ca_chain_filename + == "/var/snap/charmed-postgresql/common/pgbackrest-tls-ca-chain.crt" ) def test_are_backup_settings_ok(harness): # Test without S3 relation. - tc.assertEqual( - harness.charm.backup._are_backup_settings_ok(), - (False, "Relation with s3-integrator charm missing, cannot create/restore backup."), + assert harness.charm.backup._are_backup_settings_ok() == ( + False, + "Relation with s3-integrator charm missing, cannot create/restore backup.", ) # Test when there are missing S3 parameters. harness.add_relation(S3_PARAMETERS_RELATION, "s3-integrator") - tc.assertEqual( - harness.charm.backup._are_backup_settings_ok(), - (False, "Missing S3 parameters: ['bucket', 'access-key', 'secret-key']"), + assert harness.charm.backup._are_backup_settings_ok() == ( + False, + "Missing S3 parameters: ['bucket', 'access-key', 'secret-key']", ) # Test when all required parameters are provided. with patch("charm.PostgreSQLBackups._retrieve_s3_parameters") as _retrieve_s3_parameters: _retrieve_s3_parameters.return_value = ["bucket", "access-key", "secret-key"], [] - tc.assertEqual( - harness.charm.backup._are_backup_settings_ok(), - (True, None), - ) + assert harness.charm.backup._are_backup_settings_ok() == (True, None) def test_can_initialise_stanza(harness): @@ -104,10 +94,7 @@ def test_can_initialise_stanza(harness): # Test when Patroni or PostgreSQL hasn't started yet # and the unit hasn't joined the peer relation yet. _member_started.return_value = False - tc.assertEqual( - harness.charm.backup._can_initialise_stanza, - False, - ) + assert not harness.charm.backup._can_initialise_stanza # Test when the unit hasn't configured TLS yet while other unit already has TLS enabled. harness.add_relation_unit( @@ -119,17 +106,11 @@ def test_can_initialise_stanza(harness): f"{harness.charm.app.name}/1", {"tls": "enabled"}, ) - tc.assertEqual( - harness.charm.backup._can_initialise_stanza, - False, - ) + assert not harness.charm.backup._can_initialise_stanza # Test when everything is ok to initialise the stanza. _member_started.return_value = True - tc.assertEqual( - harness.charm.backup._can_initialise_stanza, - True, - ) + assert harness.charm.backup._can_initialise_stanza def test_can_unit_perform_backup(harness): @@ -144,18 +125,18 @@ def test_can_unit_perform_backup(harness): peer_rel_id = harness.model.get_relation(PEER).id # Test when the charm fails to retrieve the primary. _is_primary.side_effect = RetryError(last_attempt=1) - tc.assertEqual( - harness.charm.backup._can_unit_perform_backup(), - (False, "Unit cannot perform backups as the database seems to be offline"), + assert harness.charm.backup._can_unit_perform_backup() == ( + False, + "Unit cannot perform backups as the database seems to be offline", ) # Test when the unit is in a blocked state. _is_primary.side_effect = None _is_primary.return_value = True harness.charm.unit.status = BlockedStatus("fake blocked state") - tc.assertEqual( - harness.charm.backup._can_unit_perform_backup(), - (False, "Unit is in a blocking state"), + assert harness.charm.backup._can_unit_perform_backup() == ( + False, + "Unit is in a blocking state", ) # Test when running the check in the primary, there are replicas and TLS is enabled. @@ -167,9 +148,9 @@ def test_can_unit_perform_backup(harness): harness.charm.unit.name, {"tls": "True"}, ) - tc.assertEqual( - harness.charm.backup._can_unit_perform_backup(), - (False, "Unit cannot perform backups as it is the cluster primary"), + assert harness.charm.backup._can_unit_perform_backup() == ( + False, + "Unit cannot perform backups as it is the cluster primary", ) # Test when running the check in a replica and TLS is disabled. @@ -180,24 +161,24 @@ def test_can_unit_perform_backup(harness): harness.charm.unit.name, {"tls": ""}, ) - tc.assertEqual( - harness.charm.backup._can_unit_perform_backup(), - (False, "Unit cannot perform backups as TLS is not enabled"), + assert harness.charm.backup._can_unit_perform_backup() == ( + False, + "Unit cannot perform backups as TLS is not enabled", ) # Test when Patroni or PostgreSQL hasn't started yet. _is_primary.return_value = True _member_started.return_value = False - tc.assertEqual( - harness.charm.backup._can_unit_perform_backup(), - (False, "Unit cannot perform backups as it's not in running state"), + assert harness.charm.backup._can_unit_perform_backup() == ( + False, + "Unit cannot perform backups as it's not in running state", ) # Test when the stanza was not initialised yet. _member_started.return_value = True - tc.assertEqual( - harness.charm.backup._can_unit_perform_backup(), - (False, "Stanza was not initialised"), + assert harness.charm.backup._can_unit_perform_backup() == ( + False, + "Stanza was not initialised", ) # Test when S3 parameters are not ok. @@ -208,17 +189,11 @@ def test_can_unit_perform_backup(harness): {"stanza": harness.charm.backup.stanza_name}, ) _are_backup_settings_ok.return_value = (False, "fake error message") - tc.assertEqual( - harness.charm.backup._can_unit_perform_backup(), - (False, "fake error message"), - ) + assert harness.charm.backup._can_unit_perform_backup() == (False, "fake error message") # Test when everything is ok to run a backup. _are_backup_settings_ok.return_value = (True, None) - tc.assertEqual( - harness.charm.backup._can_unit_perform_backup(), - (True, None), - ) + assert harness.charm.backup._can_unit_perform_backup() == (True, None) def test_can_use_s3_repository(harness): @@ -246,14 +221,15 @@ def test_can_use_s3_repository(harness): # Test when nothing is returned from the pgBackRest info command. _execute_command.side_effect = TimeoutExpired(cmd="fake command", timeout=30) - with tc.assertRaises(TimeoutError): + with pytest.raises(TimeoutError): harness.charm.backup.can_use_s3_repository() + assert False _execute_command.side_effect = None _execute_command.return_value = (1, "", "") - tc.assertEqual( - harness.charm.backup.can_use_s3_repository(), - (False, FAILED_TO_INITIALIZE_STANZA_ERROR_MESSAGE), + assert harness.charm.backup.can_use_s3_repository() == ( + False, + FAILED_TO_INITIALIZE_STANZA_ERROR_MESSAGE, ) # Test when the unit is a replica. @@ -263,16 +239,12 @@ def test_can_use_s3_repository(harness): "", ) _execute_command.return_value = pgbackrest_info_same_cluster_backup_output - tc.assertEqual( - harness.charm.backup.can_use_s3_repository(), - (True, None), - ) + assert harness.charm.backup.can_use_s3_repository() == (True, None) # Assert that the stanza name is still in the unit relation data. - tc.assertEqual( - harness.get_relation_data(peer_rel_id, harness.charm.app), - {"stanza": harness.charm.backup.stanza_name}, - ) + assert harness.get_relation_data(peer_rel_id, harness.charm.app) == { + "stanza": harness.charm.backup.stanza_name + } # Test when the unit is the leader and the workload is running, # but an exception happens when retrieving the cluster system id. @@ -283,11 +255,9 @@ def test_can_use_s3_repository(harness): ] with harness.hooks_disabled(): harness.set_leader() - with tc.assertRaises(Exception): - tc.assertEqual( - harness.charm.backup.can_use_s3_repository(), - (False, ANOTHER_CLUSTER_REPOSITORY_ERROR_MESSAGE), - ) + with pytest.raises(Exception, match="fake error"): + harness.charm.backup.can_use_s3_repository() + assert False _update_config.assert_not_called() _member_started.assert_not_called() _reload_patroni_configuration.assert_not_called() @@ -313,16 +283,16 @@ def test_can_use_s3_repository(harness): harness.charm.app.name, {"stanza": harness.charm.backup.stanza_name}, ) - tc.assertEqual( - harness.charm.backup.can_use_s3_repository(), - (False, ANOTHER_CLUSTER_REPOSITORY_ERROR_MESSAGE), + assert harness.charm.backup.can_use_s3_repository() == ( + False, + ANOTHER_CLUSTER_REPOSITORY_ERROR_MESSAGE, ) _update_config.assert_called_once() _member_started.assert_called_once() _reload_patroni_configuration.assert_called_once() # Assert that the stanza name is not present in the unit relation data anymore. - tc.assertEqual(harness.get_relation_data(peer_rel_id, harness.charm.app), {}) + assert harness.get_relation_data(peer_rel_id, harness.charm.app) == {} # Test when the cluster system id can be retrieved, but it's different from the stanza system id. _update_config.reset_mock() @@ -348,16 +318,16 @@ def test_can_use_s3_repository(harness): harness.charm.app.name, {"stanza": harness.charm.backup.stanza_name}, ) - tc.assertEqual( - harness.charm.backup.can_use_s3_repository(), - (False, ANOTHER_CLUSTER_REPOSITORY_ERROR_MESSAGE), + assert harness.charm.backup.can_use_s3_repository() == ( + False, + ANOTHER_CLUSTER_REPOSITORY_ERROR_MESSAGE, ) _update_config.assert_called_once() _member_started.assert_called_once() _reload_patroni_configuration.assert_called_once() # Assert that the stanza name is not present in the unit relation data anymore. - tc.assertEqual(harness.get_relation_data(peer_rel_id, harness.charm.app), {}) + assert harness.get_relation_data(peer_rel_id, harness.charm.app) == {} # Test when the workload is not running. _update_config.reset_mock() @@ -374,16 +344,16 @@ def test_can_use_s3_repository(harness): pgbackrest_info_same_cluster_backup_output, other_instance_system_identifier_output, ] - tc.assertEqual( - harness.charm.backup.can_use_s3_repository(), - (False, ANOTHER_CLUSTER_REPOSITORY_ERROR_MESSAGE), + assert harness.charm.backup.can_use_s3_repository() == ( + False, + ANOTHER_CLUSTER_REPOSITORY_ERROR_MESSAGE, ) _update_config.assert_called_once() _member_started.assert_called_once() _reload_patroni_configuration.assert_not_called() # Assert that the stanza name is not present in the unit relation data anymore. - tc.assertEqual(harness.get_relation_data(peer_rel_id, harness.charm.app), {}) + assert harness.get_relation_data(peer_rel_id, harness.charm.app) == {} # Test when there is no backup from another cluster in the S3 repository. with harness.hooks_disabled(): @@ -396,36 +366,30 @@ def test_can_use_s3_repository(harness): pgbackrest_info_same_cluster_backup_output, same_instance_system_identifier_output, ] - tc.assertEqual( - harness.charm.backup.can_use_s3_repository(), - (True, None), - ) + assert harness.charm.backup.can_use_s3_repository() == (True, None) # Assert that the stanza name is still in the unit relation data. - tc.assertEqual( - harness.get_relation_data(peer_rel_id, harness.charm.app), - {"stanza": harness.charm.backup.stanza_name}, - ) + assert harness.get_relation_data(peer_rel_id, harness.charm.app) == { + "stanza": harness.charm.backup.stanza_name + } def test_construct_endpoint(harness): # Test with an AWS endpoint without region. s3_parameters = {"endpoint": "https://s3.amazonaws.com", "region": ""} - tc.assertEqual( - harness.charm.backup._construct_endpoint(s3_parameters), "https://s3.amazonaws.com" - ) + assert harness.charm.backup._construct_endpoint(s3_parameters) == "https://s3.amazonaws.com" # Test with an AWS endpoint with region. s3_parameters["region"] = "us-east-1" - tc.assertEqual( - harness.charm.backup._construct_endpoint(s3_parameters), - "https://s3.us-east-1.amazonaws.com", + assert ( + harness.charm.backup._construct_endpoint(s3_parameters) + == "https://s3.us-east-1.amazonaws.com" ) # Test with another cloud endpoint. s3_parameters["endpoint"] = "https://storage.googleapis.com" - tc.assertEqual( - harness.charm.backup._construct_endpoint(s3_parameters), "https://storage.googleapis.com" + assert ( + harness.charm.backup._construct_endpoint(s3_parameters) == "https://storage.googleapis.com" ) @@ -459,8 +423,9 @@ def test_create_bucket_if_not_exists(harness, tls_ca_chain_filename): [], ) _resource.side_effect = ValueError - with tc.assertRaises(ValueError): + with pytest.raises(ValueError): harness.charm.backup._create_bucket_if_not_exists() + assert False # Test when the bucket already exists. _resource.reset_mock() @@ -495,8 +460,9 @@ def test_create_bucket_if_not_exists(harness, tls_ca_chain_filename): error_response={"Error": {"Code": 1, "message": "fake error"}}, operation_name="fake operation name", ) - with tc.assertRaises(ClientError): + with pytest.raises(ClientError): harness.charm.backup._create_bucket_if_not_exists() + assert False head_bucket.assert_called_once() create.assert_called_once() wait_until_exists.assert_not_called() @@ -507,8 +473,9 @@ def test_create_bucket_if_not_exists(harness, tls_ca_chain_filename): head_bucket.side_effect = botocore.exceptions.ConnectTimeoutError( endpoint_url="fake endpoint URL" ) - with tc.assertRaises(botocore.exceptions.ConnectTimeoutError): + with pytest.raises(botocore.exceptions.ConnectTimeoutError): harness.charm.backup._create_bucket_if_not_exists() + assert False head_bucket.assert_called_once() create.assert_not_called() wait_until_exists.assert_not_called() @@ -522,7 +489,7 @@ def test_empty_data_files(harness): ): # Test when the data directory doesn't exist. _exists.return_value = False - tc.assertTrue(harness.charm.backup._empty_data_files()) + assert harness.charm.backup._empty_data_files() _rmtree.assert_not_called() # Test when the removal of the data files fails. @@ -530,13 +497,13 @@ def test_empty_data_files(harness): _exists.return_value = True _is_dir.return_value = True _rmtree.side_effect = OSError - tc.assertFalse(harness.charm.backup._empty_data_files()) + assert not harness.charm.backup._empty_data_files() _rmtree.assert_called_once_with(path) # Test when data files are successfully removed. _rmtree.reset_mock() _rmtree.side_effect = None - tc.assertTrue(harness.charm.backup._empty_data_files()) + assert harness.charm.backup._empty_data_files() _rmtree.assert_called_once_with(path) @@ -553,19 +520,15 @@ def test_change_connectivity_to_database(harness): # Test when connectivity should be turned on. harness.charm.backup._change_connectivity_to_database(True) - tc.assertEqual( - harness.get_relation_data(peer_rel_id, harness.charm.unit), - {"connectivity": "on"}, - ) + assert harness.get_relation_data(peer_rel_id, harness.charm.unit) == {"connectivity": "on"} _update_config.assert_called_once() # Test when connectivity should be turned off. _update_config.reset_mock() harness.charm.backup._change_connectivity_to_database(False) - tc.assertEqual( - harness.get_relation_data(peer_rel_id, harness.charm.unit), - {"connectivity": "off"}, - ) + assert harness.get_relation_data(peer_rel_id, harness.charm.unit) == { + "connectivity": "off" + } _update_config.assert_called_once() @@ -577,7 +540,7 @@ def test_execute_command(harness): # Test when the command fails. command = "rm -r /var/lib/postgresql/data/pgdata".split() _run.return_value = CompletedProcess(command, 1, b"", b"fake stderr") - tc.assertEqual(harness.charm.backup._execute_command(command), (1, "", "fake stderr")) + assert harness.charm.backup._execute_command(command) == (1, "", "fake stderr") _run.assert_called_once_with( command, input=None, stdout=PIPE, stderr=PIPE, preexec_fn=ANY, timeout=None ) @@ -588,10 +551,9 @@ def test_execute_command(harness): _getpwnam.reset_mock() _run.side_effect = None _run.return_value = CompletedProcess(command, 0, b"fake stdout", b"") - tc.assertEqual( - harness.charm.backup._execute_command(command, command_input=b"fake input", timeout=5), - (0, "fake stdout", ""), - ) + assert harness.charm.backup._execute_command( + command, command_input=b"fake input", timeout=5 + ) == (0, "fake stdout", "") _run.assert_called_once_with( command, input=b"fake input", stdout=PIPE, stderr=PIPE, preexec_fn=ANY, timeout=5 ) @@ -698,16 +660,13 @@ def test_list_backups(harness): with patch("charm.PostgreSQLBackups._execute_command") as _execute_command: # Test when the command that list the backups fails. _execute_command.return_value = (1, "", "fake stderr") - with tc.assertRaises(ListBackupsError): - tc.assertEqual( - harness.charm.backup._list_backups(show_failed=True), OrderedDict[str, str]() - ) + with pytest.raises(ListBackupsError): + harness.charm.backup._list_backups(show_failed=True) + assert False # Test when no backups are available. _execute_command.return_value = (0, "[]", "") - tc.assertEqual( - harness.charm.backup._list_backups(show_failed=True), OrderedDict[str, str]() - ) + assert harness.charm.backup._list_backups(show_failed=True) == OrderedDict[str, str]() # Test when some backups are available. _execute_command.return_value = ( @@ -715,19 +674,15 @@ def test_list_backups(harness): '[{"backup":[{"label":"20230101-090000F","error":"fake error"},{"label":"20230101-100000F","error":null}],"name":"test-stanza"}]', "", ) - tc.assertEqual( - harness.charm.backup._list_backups(show_failed=True), - OrderedDict[str, str]([ - ("2023-01-01T09:00:00Z", "test-stanza"), - ("2023-01-01T10:00:00Z", "test-stanza"), - ]), - ) + assert harness.charm.backup._list_backups(show_failed=True) == OrderedDict[str, str]([ + ("2023-01-01T09:00:00Z", "test-stanza"), + ("2023-01-01T10:00:00Z", "test-stanza"), + ]) # Test when some backups are available, but it's not desired to list failed backups. - tc.assertEqual( - harness.charm.backup._list_backups(show_failed=False), - OrderedDict[str, str]([("2023-01-01T10:00:00Z", "test-stanza")]), - ) + assert harness.charm.backup._list_backups(show_failed=False) == OrderedDict[str, str]([ + ("2023-01-01T10:00:00Z", "test-stanza") + ]) def test_initialise_stanza(harness): @@ -771,19 +726,18 @@ def test_initialise_stanza(harness): harness.charm.unit.status = BlockedStatus(blocked_state) harness.charm.backup._initialise_stanza() _execute_command.assert_called_once_with(stanza_creation_command) - tc.assertIsInstance(harness.charm.unit.status, BlockedStatus) - tc.assertEqual( - harness.charm.unit.status.message, FAILED_TO_INITIALIZE_STANZA_ERROR_MESSAGE - ) + assert isinstance(harness.charm.unit.status, BlockedStatus) + assert harness.charm.unit.status.message == FAILED_TO_INITIALIZE_STANZA_ERROR_MESSAGE # Assert there is no stanza name in the application relation databag. - tc.assertEqual(harness.get_relation_data(peer_rel_id, harness.charm.app), {}) + assert harness.get_relation_data(peer_rel_id, harness.charm.app) == {} # Test when the failure in the stanza creation is due to a timeout. _execute_command.reset_mock() _execute_command.return_value = (49, "", "fake stderr") - with tc.assertRaises(TimeoutError): + with pytest.raises(TimeoutError): harness.charm.backup._initialise_stanza() + assert False # Test when the archiving is working correctly (pgBackRest check command succeeds) # and the unit is not the leader. @@ -791,15 +745,12 @@ def test_initialise_stanza(harness): _execute_command.return_value = (0, "fake stdout", "") _member_started.return_value = True harness.charm.backup._initialise_stanza() - tc.assertEqual(harness.get_relation_data(peer_rel_id, harness.charm.app), {}) - tc.assertEqual( - harness.get_relation_data(peer_rel_id, harness.charm.unit), - { - "stanza": f"{harness.charm.model.name}.postgresql", - "init-pgbackrest": "True", - }, - ) - tc.assertIsInstance(harness.charm.unit.status, MaintenanceStatus) + assert harness.get_relation_data(peer_rel_id, harness.charm.app) == {} + assert harness.get_relation_data(peer_rel_id, harness.charm.unit) == { + "stanza": f"{harness.charm.model.name}.postgresql", + "init-pgbackrest": "True", + } + assert isinstance(harness.charm.unit.status, MaintenanceStatus) # Test when the unit is the leader. with harness.hooks_disabled(): @@ -809,13 +760,13 @@ def test_initialise_stanza(harness): ) harness.charm.backup._initialise_stanza() _update_config.assert_not_called() - tc.assertEqual( - harness.get_relation_data(peer_rel_id, harness.charm.app), - {"stanza": "None.postgresql", "init-pgbackrest": "True"}, - ) + assert harness.get_relation_data(peer_rel_id, harness.charm.app) == { + "stanza": "None.postgresql", + "init-pgbackrest": "True", + } _member_started.assert_not_called() _reload_patroni_configuration.assert_not_called() - tc.assertIsInstance(harness.charm.unit.status, MaintenanceStatus) + assert isinstance(harness.charm.unit.status, MaintenanceStatus) def test_check_stanza(harness): @@ -855,16 +806,14 @@ def test_check_stanza(harness): _execute_command.return_value = (49, "", "fake stderr") _member_started.return_value = True harness.charm.backup.check_stanza() - tc.assertEqual(_update_config.call_count, 2) - tc.assertEqual(harness.get_relation_data(peer_rel_id, harness.charm.app), {}) - tc.assertEqual(_member_started.call_count, 5) - tc.assertEqual(_reload_patroni_configuration.call_count, 5) - tc.assertEqual(harness.get_relation_data(peer_rel_id, harness.charm.app), {}) - tc.assertEqual(harness.get_relation_data(peer_rel_id, harness.charm.unit), {}) - tc.assertIsInstance(harness.charm.unit.status, BlockedStatus) - tc.assertEqual( - harness.charm.unit.status.message, FAILED_TO_INITIALIZE_STANZA_ERROR_MESSAGE - ) + assert _update_config.call_count == 2 + assert harness.get_relation_data(peer_rel_id, harness.charm.app) == {} + assert _member_started.call_count == 5 + assert _reload_patroni_configuration.call_count == 5 + assert harness.get_relation_data(peer_rel_id, harness.charm.app) == {} + assert harness.get_relation_data(peer_rel_id, harness.charm.unit) == {} + assert isinstance(harness.charm.unit.status, BlockedStatus) + assert harness.charm.unit.status.message == FAILED_TO_INITIALIZE_STANZA_ERROR_MESSAGE # Test when the archiving is working correctly (pgBackRest check command succeeds) # and the unit is not the leader. @@ -889,15 +838,14 @@ def test_check_stanza(harness): _update_config.assert_called_once() _member_started.assert_called_once() _reload_patroni_configuration.assert_called_once() - tc.assertEqual( - harness.get_relation_data(peer_rel_id, harness.charm.app), - {"stanza": "test-stanza", "init-pgbackrest": "True"}, - ) - tc.assertEqual( - harness.get_relation_data(peer_rel_id, harness.charm.unit), - {"stanza": "test-stanza"}, - ) - tc.assertIsInstance(harness.charm.unit.status, ActiveStatus) + assert harness.get_relation_data(peer_rel_id, harness.charm.app) == { + "stanza": "test-stanza", + "init-pgbackrest": "True", + } + assert harness.get_relation_data(peer_rel_id, harness.charm.unit) == { + "stanza": "test-stanza" + } + assert isinstance(harness.charm.unit.status, ActiveStatus) # Test when the unit is the leader. harness.charm.unit.status = BlockedStatus("fake blocked state") @@ -920,15 +868,13 @@ def test_check_stanza(harness): _update_config.assert_called_once() _member_started.assert_called_once() _reload_patroni_configuration.assert_called_once() - tc.assertEqual( - harness.get_relation_data(peer_rel_id, harness.charm.app), - {"stanza": "test-stanza"}, - ) - tc.assertEqual( - harness.get_relation_data(peer_rel_id, harness.charm.unit), - {"stanza": "test-stanza"}, - ) - tc.assertIsInstance(harness.charm.unit.status, ActiveStatus) + assert harness.get_relation_data(peer_rel_id, harness.charm.app) == { + "stanza": "test-stanza" + } + assert harness.get_relation_data(peer_rel_id, harness.charm.unit) == { + "stanza": "test-stanza" + } + assert isinstance(harness.charm.unit.status, ActiveStatus) def test_coordinate_stanza_fields(harness): @@ -940,9 +886,9 @@ def test_coordinate_stanza_fields(harness): # Test when the stanza name is neither in the application relation databag nor in the unit relation databag. harness.charm.backup.coordinate_stanza_fields() - tc.assertEqual(harness.get_relation_data(peer_rel_id, harness.charm.app), {}) - tc.assertEqual(harness.get_relation_data(peer_rel_id, harness.charm.unit), {}) - tc.assertEqual(harness.get_relation_data(peer_rel_id, new_unit), {}) + assert harness.get_relation_data(peer_rel_id, harness.charm.app) == {} + assert harness.get_relation_data(peer_rel_id, harness.charm.unit) == {} + assert harness.get_relation_data(peer_rel_id, new_unit) == {} # Test when the stanza name is in the unit relation databag but the unit is not the leader. stanza_name = f"{harness.charm.model.name}.{harness.charm.app.name}" @@ -951,61 +897,52 @@ def test_coordinate_stanza_fields(harness): peer_rel_id, new_unit_name, {"stanza": stanza_name, "init-pgbackrest": "True"} ) harness.charm.backup.coordinate_stanza_fields() - tc.assertEqual(harness.get_relation_data(peer_rel_id, harness.charm.app), {}) - tc.assertEqual(harness.get_relation_data(peer_rel_id, harness.charm.unit), {}) - tc.assertEqual( - harness.get_relation_data(peer_rel_id, new_unit), - {"stanza": stanza_name, "init-pgbackrest": "True"}, - ) + assert harness.get_relation_data(peer_rel_id, harness.charm.app) == {} + assert harness.get_relation_data(peer_rel_id, harness.charm.unit) == {} + assert harness.get_relation_data(peer_rel_id, new_unit) == { + "stanza": stanza_name, + "init-pgbackrest": "True", + } # Test when the unit is the leader. with harness.hooks_disabled(): harness.set_leader() harness.charm.backup.coordinate_stanza_fields() - tc.assertEqual( - harness.get_relation_data(peer_rel_id, harness.charm.app), - {"stanza": stanza_name, "init-pgbackrest": "True"}, - ) - tc.assertEqual(harness.get_relation_data(peer_rel_id, harness.charm.unit), {}) - tc.assertEqual( - harness.get_relation_data(peer_rel_id, new_unit), - {"stanza": stanza_name, "init-pgbackrest": "True"}, - ) + assert harness.get_relation_data(peer_rel_id, harness.charm.app) == { + "stanza": stanza_name, + "init-pgbackrest": "True", + } + assert harness.get_relation_data(peer_rel_id, harness.charm.unit) == {} + assert harness.get_relation_data(peer_rel_id, new_unit) == { + "stanza": stanza_name, + "init-pgbackrest": "True", + } # Test when the stanza was already checked in the primary non-leader unit. with harness.hooks_disabled(): harness.update_relation_data(peer_rel_id, new_unit_name, {"init-pgbackrest": ""}) harness.charm.backup.coordinate_stanza_fields() - tc.assertEqual( - harness.get_relation_data(peer_rel_id, harness.charm.app), - {"stanza": stanza_name}, - ) - tc.assertEqual(harness.get_relation_data(peer_rel_id, harness.charm.unit), {}) - tc.assertEqual(harness.get_relation_data(peer_rel_id, new_unit), {"stanza": stanza_name}) + assert harness.get_relation_data(peer_rel_id, harness.charm.app) == {"stanza": stanza_name} + assert harness.get_relation_data(peer_rel_id, harness.charm.unit) == {} + assert harness.get_relation_data(peer_rel_id, new_unit) == {"stanza": stanza_name} # Test when the "init-pgbackrest" flag was removed from the application relation databag # and this is the unit that has the stanza name in the unit relation databag. with harness.hooks_disabled(): harness.update_relation_data(peer_rel_id, harness.charm.unit.name, {"stanza": stanza_name}) harness.charm.backup.coordinate_stanza_fields() - tc.assertEqual( - harness.get_relation_data(peer_rel_id, harness.charm.app), - {"stanza": stanza_name}, - ) - tc.assertEqual(harness.get_relation_data(peer_rel_id, harness.charm.unit), {}) - tc.assertEqual(harness.get_relation_data(peer_rel_id, new_unit), {"stanza": stanza_name}) + assert harness.get_relation_data(peer_rel_id, harness.charm.app) == {"stanza": stanza_name} + assert harness.get_relation_data(peer_rel_id, harness.charm.unit) == {} + assert harness.get_relation_data(peer_rel_id, new_unit) == {"stanza": stanza_name} # Test when the unit is not the leader. with harness.hooks_disabled(): harness.set_leader(False) harness.update_relation_data(peer_rel_id, harness.charm.unit.name, {"stanza": stanza_name}) harness.charm.backup.coordinate_stanza_fields() - tc.assertEqual( - harness.get_relation_data(peer_rel_id, harness.charm.app), - {"stanza": stanza_name}, - ) - tc.assertEqual(harness.get_relation_data(peer_rel_id, harness.charm.unit), {}) - tc.assertEqual(harness.get_relation_data(peer_rel_id, new_unit), {"stanza": stanza_name}) + assert harness.get_relation_data(peer_rel_id, harness.charm.app) == {"stanza": stanza_name} + assert harness.get_relation_data(peer_rel_id, harness.charm.unit) == {} + assert harness.get_relation_data(peer_rel_id, new_unit) == {"stanza": stanza_name} def test_is_primary_pgbackrest_service_running(harness): @@ -1019,20 +956,20 @@ def test_is_primary_pgbackrest_service_running(harness): # Test when the pgBackRest fails to contact the primary server. _get_primary.side_effect = None _execute_command.return_value = (1, "", "fake stderr") - tc.assertFalse(harness.charm.backup._is_primary_pgbackrest_service_running) + assert not harness.charm.backup._is_primary_pgbackrest_service_running _execute_command.assert_called_once() # Test when the endpoint is not generated. _execute_command.reset_mock() _primary_endpoint.return_value = None - tc.assertFalse(harness.charm.backup._is_primary_pgbackrest_service_running) + assert not harness.charm.backup._is_primary_pgbackrest_service_running _execute_command.assert_not_called() # Test when the pgBackRest succeeds on contacting the primary server. _execute_command.reset_mock() _execute_command.return_value = (0, "fake stdout", "") _primary_endpoint.return_value = "fake_endpoint" - tc.assertTrue(harness.charm.backup._is_primary_pgbackrest_service_running) + assert harness.charm.backup._is_primary_pgbackrest_service_running _execute_command.assert_called_once() @@ -1119,13 +1056,12 @@ def test_on_s3_credential_changed(harness): relation=harness.model.get_relation(S3_PARAMETERS_RELATION, s3_rel_id) ) _render_pgbackrest_conf_file.assert_called_once() - tc.assertNotIn( - "require-change-bucket-after-restore", - harness.get_relation_data(peer_rel_id, harness.charm.app), + assert "require-change-bucket-after-restore" not in harness.get_relation_data( + peer_rel_id, harness.charm.app ) _is_primary.assert_called_once() _create_bucket_if_not_exists.assert_not_called() - tc.assertIsInstance(harness.charm.unit.status, ActiveStatus) + assert isinstance(harness.charm.unit.status, ActiveStatus) _can_use_s3_repository.assert_not_called() _initialise_stanza.assert_not_called() @@ -1155,16 +1091,16 @@ def test_on_s3_credential_changed(harness): relation=harness.model.get_relation(S3_PARAMETERS_RELATION, s3_rel_id) ) _render_pgbackrest_conf_file.assert_called_once() - tc.assertEqual( + assert ( harness.get_relation_data(peer_rel_id, harness.charm.app)[ "require-change-bucket-after-restore" - ], - "True", + ] + == "True" ) _create_bucket_if_not_exists.assert_called_once() - tc.assertIsInstance(harness.charm.unit.status, BlockedStatus) - tc.assertEqual( - harness.charm.unit.status.message, FAILED_TO_ACCESS_CREATE_BUCKET_ERROR_MESSAGE + assert isinstance(harness.charm.unit.status, BlockedStatus) + assert ( + harness.charm.unit.status.message == FAILED_TO_ACCESS_CREATE_BUCKET_ERROR_MESSAGE ) _can_use_s3_repository.assert_not_called() _initialise_stanza.assert_not_called() @@ -1176,8 +1112,8 @@ def test_on_s3_credential_changed(harness): harness.charm.backup.s3_client.on.credentials_changed.emit( relation=harness.model.get_relation(S3_PARAMETERS_RELATION, s3_rel_id) ) - tc.assertIsInstance(harness.charm.unit.status, BlockedStatus) - tc.assertEqual(harness.charm.unit.status.message, "fake validation message") + assert isinstance(harness.charm.unit.status, BlockedStatus) + assert harness.charm.unit.status.message == "fake validation message" _create_bucket_if_not_exists.assert_called_once() _can_use_s3_repository.assert_called_once() _initialise_stanza.assert_not_called() @@ -1197,12 +1133,12 @@ def test_on_s3_credential_gone(harness): # Test that unrelated blocks will remain harness.charm.unit.status = BlockedStatus("test block") harness.charm.backup._on_s3_credential_gone(None) - tc.assertIsInstance(harness.charm.unit.status, BlockedStatus) + assert isinstance(harness.charm.unit.status, BlockedStatus) # Test that s3 related blocks will be cleared harness.charm.unit.status = BlockedStatus(ANOTHER_CLUSTER_REPOSITORY_ERROR_MESSAGE) harness.charm.backup._on_s3_credential_gone(None) - tc.assertIsInstance(harness.charm.unit.status, ActiveStatus) + assert isinstance(harness.charm.unit.status, ActiveStatus) # Test removal of relation data when the unit is not the leader. with harness.hooks_disabled(): @@ -1217,11 +1153,11 @@ def test_on_s3_credential_gone(harness): {"stanza": "test-stanza", "init-pgbackrest": "True"}, ) harness.charm.backup._on_s3_credential_gone(None) - tc.assertEqual( - harness.get_relation_data(peer_rel_id, harness.charm.app), - {"stanza": "test-stanza", "init-pgbackrest": "True"}, - ) - tc.assertEqual(harness.get_relation_data(peer_rel_id, harness.charm.unit), {}) + assert harness.get_relation_data(peer_rel_id, harness.charm.app) == { + "stanza": "test-stanza", + "init-pgbackrest": "True", + } + assert harness.get_relation_data(peer_rel_id, harness.charm.unit) == {} # Test removal of relation data when the unit is the leader. with harness.hooks_disabled(): @@ -1232,8 +1168,8 @@ def test_on_s3_credential_gone(harness): {"stanza": "test-stanza", "init-pgbackrest": "True"}, ) harness.charm.backup._on_s3_credential_gone(None) - tc.assertEqual(harness.get_relation_data(peer_rel_id, harness.charm.app), {}) - tc.assertEqual(harness.get_relation_data(peer_rel_id, harness.charm.unit), {}) + assert harness.get_relation_data(peer_rel_id, harness.charm.app) == {} + assert harness.get_relation_data(peer_rel_id, harness.charm.unit) == {} def test_on_create_backup_action(harness): @@ -1391,7 +1327,7 @@ def test_on_create_backup_action(harness): mock_s3_parameters, ), ]) - tc.assertEqual(_change_connectivity_to_database.call_count, 2) + assert _change_connectivity_to_database.call_count == 2 mock_event.fail.assert_not_called() mock_event.set_results.assert_called_once_with({"backup-status": "backup created"}) @@ -1473,7 +1409,7 @@ def test_on_restore_action(harness): _start_patroni.assert_not_called() mock_event.fail.assert_not_called() mock_event.set_results.assert_not_called() - tc.assertNotIsInstance(harness.charm.unit.status, MaintenanceStatus) + assert not isinstance(harness.charm.unit.status, MaintenanceStatus) # Test when the user provides an invalid backup id. mock_event.params = {"backup-id": "2023-01-01T10:00:00Z"} @@ -1491,7 +1427,7 @@ def test_on_restore_action(harness): _update_config.assert_not_called() _start_patroni.assert_not_called() mock_event.set_results.assert_not_called() - tc.assertNotIsInstance(harness.charm.unit.status, MaintenanceStatus) + assert not isinstance(harness.charm.unit.status, MaintenanceStatus) # Test when the charm fails to stop the workload. mock_event.reset_mock() @@ -1526,16 +1462,13 @@ def test_on_restore_action(harness): _empty_data_files.return_value = True _execute_command.return_value = (1, "", "fake stderr") _fetch_backup_from_id.return_value = "20230101-090000F" - tc.assertEqual(harness.get_relation_data(peer_rel_id, harness.charm.app), {}) + assert harness.get_relation_data(peer_rel_id, harness.charm.app) == {} harness.charm.backup._on_restore_action(mock_event) - tc.assertEqual( - harness.get_relation_data(peer_rel_id, harness.charm.app), - { - "restoring-backup": "20230101-090000F", - "restore-stanza": f"{harness.charm.model.name}.{harness.charm.cluster_name}", - "require-change-bucket-after-restore": "True", - }, - ) + assert harness.get_relation_data(peer_rel_id, harness.charm.app) == { + "restoring-backup": "20230101-090000F", + "restore-stanza": f"{harness.charm.model.name}.{harness.charm.cluster_name}", + "require-change-bucket-after-restore": "True", + } _execute_command.assert_called_once_with( [ "charmed-postgresql.patronictl", @@ -1572,13 +1505,13 @@ def test_pre_restore_checks(harness): # Test when S3 parameters are not ok. mock_event = MagicMock(params={}) _are_backup_settings_ok.return_value = (False, "fake error message") - tc.assertEqual(harness.charm.backup._pre_restore_checks(mock_event), False) + assert not harness.charm.backup._pre_restore_checks(mock_event) mock_event.fail.assert_called_once() # Test when no backup id is provided. mock_event.reset_mock() _are_backup_settings_ok.return_value = (True, None) - tc.assertEqual(harness.charm.backup._pre_restore_checks(mock_event), False) + assert not harness.charm.backup._pre_restore_checks(mock_event) mock_event.fail.assert_called_once() # Test when the unit is in a blocked state that is not recoverable by changing @@ -1586,7 +1519,7 @@ def test_pre_restore_checks(harness): mock_event.reset_mock() mock_event.params = {"backup-id": "2023-01-01T09:00:00Z"} harness.charm.unit.status = BlockedStatus("fake blocked state") - tc.assertEqual(harness.charm.backup._pre_restore_checks(mock_event), False) + assert not harness.charm.backup._pre_restore_checks(mock_event) mock_event.fail.assert_called_once() # Test when the unit is in a blocked state that is recoverable by changing S3 parameters, @@ -1594,20 +1527,20 @@ def test_pre_restore_checks(harness): mock_event.reset_mock() harness.charm.unit.status = BlockedStatus(ANOTHER_CLUSTER_REPOSITORY_ERROR_MESSAGE) _planned_units.return_value = 2 - tc.assertEqual(harness.charm.backup._pre_restore_checks(mock_event), False) + assert not harness.charm.backup._pre_restore_checks(mock_event) mock_event.fail.assert_called_once() # Test when the cluster has only one unit, but it's not the leader yet. mock_event.reset_mock() _planned_units.return_value = 1 - tc.assertEqual(harness.charm.backup._pre_restore_checks(mock_event), False) + assert not harness.charm.backup._pre_restore_checks(mock_event) mock_event.fail.assert_called_once() # Test when everything is ok to run a restore. mock_event.reset_mock() with harness.hooks_disabled(): harness.set_leader() - tc.assertEqual(harness.charm.backup._pre_restore_checks(mock_event), True) + assert harness.charm.backup._pre_restore_checks(mock_event) mock_event.fail.assert_not_called() @@ -1684,7 +1617,7 @@ def test_render_pgbackrest_conf_file(harness, tls_ca_chain_filename): harness.charm.backup._render_pgbackrest_conf_file() # Check the template is opened read-only in the call to open. - tc.assertEqual(mock.call_args_list[0][0], ("templates/pgbackrest.conf.j2", "r")) + assert mock.call_args_list[0][0] == ("templates/pgbackrest.conf.j2", "r") # Ensure the correct rendered template is sent to _render_file method. calls = [ @@ -1714,7 +1647,7 @@ def test_restart_database(harness): harness.charm.backup._restart_database() # Assert that the backup id is not in the application relation databag anymore. - tc.assertEqual(harness.get_relation_data(peer_rel_id, harness.charm.app), {}) + assert harness.get_relation_data(peer_rel_id, harness.charm.app) == {} _update_config.assert_called_once() _start_patroni.assert_called_once() @@ -1726,9 +1659,9 @@ def test_retrieve_s3_parameters(harness): ) as _get_s3_connection_info: # Test when there are missing S3 parameters. _get_s3_connection_info.return_value = {} - tc.assertEqual( - harness.charm.backup._retrieve_s3_parameters(), - ({}, ["bucket", "access-key", "secret-key"]), + assert harness.charm.backup._retrieve_s3_parameters() == ( + {}, + ["bucket", "access-key", "secret-key"], ) # Test when only the required parameters are provided. @@ -1737,21 +1670,18 @@ def test_retrieve_s3_parameters(harness): "access-key": "test-access-key", "secret-key": "test-secret-key", } - tc.assertEqual( - harness.charm.backup._retrieve_s3_parameters(), - ( - { - "access-key": "test-access-key", - "bucket": "test-bucket", - "delete-older-than-days": "9999999", - "endpoint": "https://s3.amazonaws.com", - "path": "/", - "region": None, - "s3-uri-style": "host", - "secret-key": "test-secret-key", - }, - [], - ), + assert harness.charm.backup._retrieve_s3_parameters() == ( + { + "access-key": "test-access-key", + "bucket": "test-bucket", + "delete-older-than-days": "9999999", + "endpoint": "https://s3.amazonaws.com", + "path": "/", + "region": None, + "s3-uri-style": "host", + "secret-key": "test-secret-key", + }, + [], ) # Test when all parameters are provided. @@ -1765,21 +1695,18 @@ def test_retrieve_s3_parameters(harness): "s3-uri-style": " path ", "delete-older-than-days": "30", } - tc.assertEqual( - harness.charm.backup._retrieve_s3_parameters(), - ( - { - "access-key": "test-access-key", - "bucket": "test-bucket", - "endpoint": "https://storage.googleapis.com", - "path": "/test-path", - "region": "us-east-1", - "s3-uri-style": "path", - "secret-key": "test-secret-key", - "delete-older-than-days": "30", - }, - [], - ), + assert harness.charm.backup._retrieve_s3_parameters() == ( + { + "access-key": "test-access-key", + "bucket": "test-bucket", + "endpoint": "https://storage.googleapis.com", + "path": "/test-path", + "region": "us-east-1", + "s3-uri-style": "path", + "secret-key": "test-secret-key", + "delete-older-than-days": "30", + }, + [], ) @@ -1809,30 +1736,21 @@ def test_start_stop_pgbackrest_service(harness): restart = MagicMock() stop = MagicMock() _snap_cache.return_value = {"charmed-postgresql": MagicMock(restart=restart, stop=stop)} - tc.assertEqual( - harness.charm.backup.start_stop_pgbackrest_service(), - True, - ) + assert harness.charm.backup.start_stop_pgbackrest_service() stop.assert_not_called() restart.assert_not_called() # Test when it was not possible to render the pgBackRest configuration file. _are_backup_settings_ok.return_value = (True, None) _render_pgbackrest_conf_file.return_value = False - tc.assertEqual( - harness.charm.backup.start_stop_pgbackrest_service(), - False, - ) + assert not harness.charm.backup.start_stop_pgbackrest_service() stop.assert_not_called() restart.assert_not_called() # Test when TLS is not enabled (should stop the service). _render_pgbackrest_conf_file.return_value = True _is_tls_enabled.return_value = False - tc.assertEqual( - harness.charm.backup.start_stop_pgbackrest_service(), - True, - ) + assert harness.charm.backup.start_stop_pgbackrest_service() stop.assert_called_once() restart.assert_not_called() @@ -1840,10 +1758,7 @@ def test_start_stop_pgbackrest_service(harness): stop.reset_mock() _is_tls_enabled.return_value = True _peer_members_ips.return_value = [] - tc.assertEqual( - harness.charm.backup.start_stop_pgbackrest_service(), - True, - ) + assert harness.charm.backup.start_stop_pgbackrest_service() stop.assert_called_once() restart.assert_not_called() @@ -1852,19 +1767,13 @@ def test_start_stop_pgbackrest_service(harness): _peer_members_ips.return_value = ["1.1.1.1"] _is_primary.return_value = False _is_primary_pgbackrest_service_running.return_value = False - tc.assertEqual( - harness.charm.backup.start_stop_pgbackrest_service(), - False, - ) + assert not harness.charm.backup.start_stop_pgbackrest_service() stop.assert_not_called() restart.assert_not_called() # Test when the service has already started in the primary. _is_primary_pgbackrest_service_running.return_value = True - tc.assertEqual( - harness.charm.backup.start_stop_pgbackrest_service(), - True, - ) + assert harness.charm.backup.start_stop_pgbackrest_service() stop.assert_not_called() restart.assert_called_once() @@ -1872,10 +1781,7 @@ def test_start_stop_pgbackrest_service(harness): restart.reset_mock() _is_primary.return_value = True _is_primary_pgbackrest_service_running.return_value = False - tc.assertEqual( - harness.charm.backup.start_stop_pgbackrest_service(), - True, - ) + assert harness.charm.backup.start_stop_pgbackrest_service() stop.assert_not_called() restart.assert_called_once() @@ -1911,10 +1817,7 @@ def test_upload_content_to_s3(harness, tls_ca_chain_filename): _resource.side_effect = ValueError _construct_endpoint.return_value = "https://s3.us-east-1.amazonaws.com" _named_temporary_file.return_value.__enter__.return_value.name = "/tmp/test-file" - tc.assertEqual( - harness.charm.backup._upload_content_to_s3(content, s3_path, s3_parameters), - False, - ) + assert not harness.charm.backup._upload_content_to_s3(content, s3_path, s3_parameters) _resource.assert_called_once_with( "s3", endpoint_url="https://s3.us-east-1.amazonaws.com", @@ -1926,10 +1829,7 @@ def test_upload_content_to_s3(harness, tls_ca_chain_filename): _resource.reset_mock() _resource.side_effect = None upload_file.side_effect = S3UploadFailedError - tc.assertEqual( - harness.charm.backup._upload_content_to_s3(content, s3_path, s3_parameters), - False, - ) + assert not harness.charm.backup._upload_content_to_s3(content, s3_path, s3_parameters) _resource.assert_called_once_with( "s3", endpoint_url="https://s3.us-east-1.amazonaws.com", @@ -1943,10 +1843,7 @@ def test_upload_content_to_s3(harness, tls_ca_chain_filename): _named_temporary_file.reset_mock() upload_file.reset_mock() upload_file.side_effect = None - tc.assertEqual( - harness.charm.backup._upload_content_to_s3(content, s3_path, s3_parameters), - True, - ) + assert harness.charm.backup._upload_content_to_s3(content, s3_path, s3_parameters) _resource.assert_called_once_with( "s3", endpoint_url="https://s3.us-east-1.amazonaws.com", diff --git a/tests/unit/test_charm.py b/tests/unit/test_charm.py index 5cb0709891..9f4dd50535 100644 --- a/tests/unit/test_charm.py +++ b/tests/unit/test_charm.py @@ -5,7 +5,6 @@ import logging import platform import subprocess -from unittest import TestCase from unittest.mock import MagicMock, Mock, PropertyMock, call, mock_open, patch, sentinel import psycopg2 @@ -42,7 +41,6 @@ CREATE_CLUSTER_CONF_PATH = "/etc/postgresql-common/createcluster.d/pgcharm.conf" # used for assert functions -tc = TestCase() @pytest.fixture(autouse=True) @@ -2398,7 +2396,7 @@ def test_set_primary_status_message(harness, is_leader): if is_leader else databag_containing_restore_data ) - tc.assertIsInstance(harness.charm.unit.status, BlockedStatus) + assert isinstance(harness.charm.unit.status, BlockedStatus) # Test other scenarios. with harness.hooks_disabled(): @@ -2431,28 +2429,28 @@ def test_set_primary_status_message(harness, is_leader): _is_standby_leader.side_effect = values[1] _is_standby_leader.return_value = None harness.charm._set_primary_status_message() - tc.assertIsInstance(harness.charm.unit.status, MaintenanceStatus) + assert isinstance(harness.charm.unit.status, MaintenanceStatus) else: _is_standby_leader.side_effect = None _is_standby_leader.return_value = values[1] harness.charm._set_primary_status_message() - tc.assertIsInstance( + assert isinstance( harness.charm.unit.status, ActiveStatus if values[0] == harness.charm.unit.name or values[1] or values[2] else MaintenanceStatus, ) - tc.assertEqual( - harness.charm.unit.status.message, + status = ( "Primary" if values[0] == harness.charm.unit.name - else ("Standby" if values[1] else ("" if values[2] else "fake status")), + else ("Standby" if values[1] else "" if values[2] else "fake status") ) + assert harness.charm.unit.status.message == status else: _get_primary.side_effect = values[0] _get_primary.return_value = None harness.charm._set_primary_status_message() - tc.assertIsInstance(harness.charm.unit.status, MaintenanceStatus) + assert isinstance(harness.charm.unit.status, MaintenanceStatus) def test_override_patroni_restart_condition(harness): diff --git a/tests/unit/test_cluster.py b/tests/unit/test_cluster.py index 370f47c147..aa95b3bc8c 100644 --- a/tests/unit/test_cluster.py +++ b/tests/unit/test_cluster.py @@ -1,7 +1,6 @@ # Copyright 2021 Canonical Ltd. # See LICENSE file for licensing details. -from unittest import TestCase from unittest.mock import MagicMock, Mock, PropertyMock, mock_open, patch, sentinel import pytest @@ -25,9 +24,6 @@ PATRONI_SERVICE = "patroni" CREATE_CLUSTER_CONF_PATH = "/var/snap/charmed-postgresql/current/etc/postgresql/postgresql.conf" -# used for assert functions -tc = TestCase() - # This method will be used by the mock to replace requests.get def mocked_requests_get(*args, **kwargs): @@ -92,7 +88,7 @@ def test_get_alternative_patroni_url(peers_ips, patroni): # Test the first URL that is returned (it should have the current unit IP). url = patroni._get_alternative_patroni_url(attempt) - tc.assertEqual(url, f"http://{patroni.unit_ip}:8008") + assert url == f"http://{patroni.unit_ip}:8008" # Test returning the other servers URLs. for attempt_number in range(attempt.retry_state.attempt_number + 1, len(peers_ips) + 2): @@ -108,8 +104,9 @@ def test_get_member_ip(peers_ips, patroni): ): # Test error on trying to get the member IP. _get_alternative_patroni_url.side_effect = "http://server2" - with tc.assertRaises(tenacity.RetryError): + with pytest.raises(tenacity.RetryError): patroni.get_member_ip(patroni.member_name) + assert False # Test using an alternative Patroni URL. _get_alternative_patroni_url.side_effect = [ @@ -118,17 +115,17 @@ def test_get_member_ip(peers_ips, patroni): "http://server1", ] ip = patroni.get_member_ip(patroni.member_name) - tc.assertEqual(ip, "1.1.1.1") + assert ip == "1.1.1.1" # Test using the current Patroni URL. _get_alternative_patroni_url.side_effect = ["http://server1"] ip = patroni.get_member_ip(patroni.member_name) - tc.assertEqual(ip, "1.1.1.1") + assert ip == "1.1.1.1" # Test when not having that specific member in the cluster. _get_alternative_patroni_url.side_effect = ["http://server1"] ip = patroni.get_member_ip("other-member-name") - tc.assertIsNone(ip) + assert ip is None def test_get_patroni_health(peers_ips, patroni): @@ -145,12 +142,13 @@ def test_get_patroni_health(peers_ips, patroni): _stop_after_delay.assert_called_once_with(60) _wait_fixed.assert_called_once_with(7) - tc.assertEqual(health, {"state": "running"}) + assert health == {"state": "running"} # Test when the Patroni API is not reachable. _patroni_url.return_value = "http://server2" - with tc.assertRaises(tenacity.RetryError): + with pytest.raises(tenacity.RetryError): patroni.get_patroni_health() + assert False def test_get_postgresql_version(peers_ips, patroni): @@ -163,7 +161,7 @@ def test_get_postgresql_version(peers_ips, patroni): ] version = patroni.get_postgresql_version() - tc.assertEqual(version, "14.0") + assert version == "14.0" _snap_client.assert_called_once_with() _get_installed_snaps.assert_called_once_with() @@ -175,8 +173,9 @@ def test_get_primary(peers_ips, patroni): ): # Test error on trying to get the member IP. _get_alternative_patroni_url.side_effect = "http://server2" - with tc.assertRaises(tenacity.RetryError): + with pytest.raises(tenacity.RetryError): patroni.get_primary(patroni.member_name) + assert False # Test using an alternative Patroni URL. _get_alternative_patroni_url.side_effect = [ @@ -185,17 +184,17 @@ def test_get_primary(peers_ips, patroni): "http://server1", ] primary = patroni.get_primary() - tc.assertEqual(primary, "postgresql-0") + assert primary == "postgresql-0" # Test using the current Patroni URL. _get_alternative_patroni_url.side_effect = ["http://server1"] primary = patroni.get_primary() - tc.assertEqual(primary, "postgresql-0") + assert primary == "postgresql-0" # Test requesting the primary in the unit name pattern. _get_alternative_patroni_url.side_effect = ["http://server1"] primary = patroni.get_primary(unit_name_pattern=True) - tc.assertEqual(primary, "postgresql/0") + assert primary == "postgresql/0" def test_is_creating_backup(peers_ips, patroni): @@ -208,13 +207,13 @@ def test_is_creating_backup(peers_ips, patroni): {"name": "postgresql-1", "tags": {"is_creating_backup": True}}, ] } - tc.assertTrue(patroni.is_creating_backup) + assert patroni.is_creating_backup # Test when no member is creating a backup. response.json.return_value = { "members": [{"name": "postgresql-0"}, {"name": "postgresql-1"}] } - tc.assertFalse(patroni.is_creating_backup) + assert not patroni.is_creating_backup def test_is_replication_healthy(peers_ips, patroni): @@ -245,15 +244,15 @@ def test_is_member_isolated(peers_ips, patroni): ): # Test when it wasn't possible to connect to the Patroni API. _patroni_url.return_value = "http://server3" - tc.assertFalse(patroni.is_member_isolated) + assert not patroni.is_member_isolated # Test when the member isn't isolated from the cluster. _patroni_url.return_value = "http://server1" - tc.assertFalse(patroni.is_member_isolated) + assert not patroni.is_member_isolated # Test when the member is isolated from the cluster. _patroni_url.return_value = "http://server4" - tc.assertTrue(patroni.is_member_isolated) + assert patroni.is_member_isolated def test_render_file(peers_ips, patroni): @@ -277,7 +276,7 @@ def test_render_file(peers_ips, patroni): patroni.render_file(filename, "rendered-content", 0o640) # Check the rendered file is opened with "w+" mode. - tc.assertEqual(mock.call_args_list[0][0], (filename, "w+")) + assert mock.call_args_list[0][0] == (filename, "w+") # Ensure that the correct user is lookup up. _pwnam.assert_called_with("snap_daemon") # Ensure the file is chmod'd correctly. @@ -342,7 +341,7 @@ def test_render_patroni_yml_file(peers_ips, patroni): patroni.render_patroni_yml_file() # Check the template is opened read-only in the call to open. - tc.assertEqual(mock.call_args_list[0][0], ("templates/patroni.yml.j2", "r")) + assert mock.call_args_list[0][0] == ("templates/patroni.yml.j2", "r") # Ensure the correct rendered template is sent to _render_file method. _render_file.assert_called_once_with( "/var/snap/charmed-postgresql/current/etc/patroni/patroni.yaml", @@ -457,8 +456,9 @@ def test_update_synchronous_node_count(peers_ips, patroni): # Test when the request fails. response.status_code = 500 - with tc.assertRaises(RetryError): + with pytest.raises(RetryError): patroni.update_synchronous_node_count() + assert False def test_configure_patroni_on_unit(peers_ips, patroni): @@ -626,8 +626,9 @@ def test_get_patroni_restart_condition(patroni): # Test when there is no restart condition set. _open.return_value.__enter__.return_value.read.return_value = "" - with tc.assertRaises(RuntimeError): + with pytest.raises(RuntimeError): patroni.get_patroni_restart_condition() + assert False @pytest.mark.parametrize("new_restart_condition", ["on-success", "on-failure"])