diff --git a/lib/charms/vault_k8s/v0/vault_client.py b/lib/charms/vault_k8s/v0/vault_client.py index bf766e88..3c899141 100644 --- a/lib/charms/vault_k8s/v0/vault_client.py +++ b/lib/charms/vault_k8s/v0/vault_client.py @@ -28,7 +28,7 @@ # Increment this PATCH version before using `charmcraft publish-lib` or reset # to 0 if you are raising the major API version -LIBPATCH = 24 +LIBPATCH = 25 RAFT_STATE_ENDPOINT = "v1/sys/storage/raft/autopilot/state" @@ -483,13 +483,16 @@ def create_snapshot(self) -> requests.Response: """Create a snapshot of the Vault data.""" return self._client.sys.take_raft_snapshot() - def restore_snapshot(self, snapshot: IOBase) -> requests.Response: + def restore_snapshot(self, snapshot: IOBase) -> None: """Restore a snapshot of the Vault data. Uses force_restore_raft_snapshot to restore the snapshot even if the unseal key used at backup time is different from the current one. """ - return self._client.sys.force_restore_raft_snapshot(snapshot) + response = self._client.sys.force_restore_raft_snapshot(snapshot) + if not 200 <= response.status_code < 300: + logger.warning("Error while restoring snapshot: %s", response.text) + raise VaultClientError(f"Error while restoring snapshot: {response.text}") def get_raft_cluster_state(self) -> dict: """Get raft cluster state.""" diff --git a/lib/charms/vault_k8s/v0/vault_managers.py b/lib/charms/vault_k8s/v0/vault_managers.py index 04b02c13..cb09195a 100644 --- a/lib/charms/vault_k8s/v0/vault_managers.py +++ b/lib/charms/vault_k8s/v0/vault_managers.py @@ -32,13 +32,14 @@ import os from abc import ABC, abstractmethod from dataclasses import dataclass -from datetime import timedelta +from datetime import datetime, timedelta from enum import Enum, auto from typing import FrozenSet, MutableMapping, TextIO from charms.certificate_transfer_interface.v0.certificate_transfer import ( CertificateTransferProvides, ) +from charms.data_platform_libs.v0.s3 import S3Requirer from charms.tls_certificates_interface.v4.tls_certificates import ( Certificate, CertificateRequestAttributes, @@ -73,6 +74,7 @@ VaultClientError, ) from charms.vault_k8s.v0.vault_kv import VaultKvProvides +from charms.vault_k8s.v0.vault_s3 import S3, S3Error from ops import CharmBase, EventBase, Object, Relation from ops.pebble import PathError @@ -84,7 +86,7 @@ # Increment this PATCH version before using `charmcraft publish-lib` or reset # to 0 if you are raising the major API version -LIBPATCH = 4 +LIBPATCH = 5 SEND_CA_CERT_RELATION_NAME = "send-ca-cert" @@ -599,6 +601,7 @@ class Naming: autounseal_approle_prefix: str = "charm-autounseal-" autounseal_key_prefix: str = "" autounseal_policy_prefix: str = "charm-autounseal-" + backup_s3_key_prefix: str = "vault-backup-" kv_mount_prefix: str = "charm-" kv_secret_prefix: str = "vault-kv-" @@ -617,6 +620,12 @@ def autounseal_approle_name(cls, relation_id: int) -> str: """Return the approle name for the relation.""" return f"{cls.autounseal_approle_prefix}{relation_id}" + @classmethod + def backup_s3_key_name(cls, model_name: str) -> str: + """Return the key name for the S3 backend.""" + timestamp = datetime.now().strftime("%Y-%m-%d-%H-%M-%S") + return f"{cls.backup_s3_key_prefix}{model_name}-{timestamp}" + @classmethod def kv_secret_label(cls, unit_name: str) -> str: """Return the secret label for the KV backend.""" @@ -1276,3 +1285,165 @@ def remove_unit_credentials(juju_facade: JujuFacade, unit_name: str) -> None: unit_name: The name of the unit for which to remove the secret """ juju_facade.remove_secret(Naming.kv_secret_label(unit_name=unit_name)) + + +class BackupManager: + """Encapsulates the business logic for managing backups in Vault from a Charm. + + This class provides the business logic for creating, listing, and restoring + backups of the Vault data. + """ + + REQUIRED_S3_PARAMETERS = ["bucket", "access-key", "secret-key", "endpoint"] + + def __init__( + self, + charm: CharmBase, + s3_requirer: S3Requirer, + relation_name: str, + ): + self._charm = charm + self._juju_facade = JujuFacade(charm) + self._s3_requirer = s3_requirer + self._relation_name = relation_name + + def create_backup(self, vault_client: VaultClient) -> str: + """Create a backup of the Vault data. + + Stores the backup in the S3 bucket provided by the S3 relation. + + Returns: + The S3 key of the backup. + """ + self._validate_s3_prerequisites() + + s3_parameters = self._get_s3_parameters() + + try: + s3 = S3( + access_key=s3_parameters["access-key"], + secret_key=s3_parameters["secret-key"], + endpoint=s3_parameters["endpoint"], + region=s3_parameters.get("region"), + ) + except S3Error as e: + logger.error("Failed to create S3 session. %s", e) + raise ManagerError("Failed to create S3 session") + + if not (s3.create_bucket(bucket_name=s3_parameters["bucket"])): + raise ManagerError("Failed to create S3 bucket") + backup_key = Naming.backup_s3_key_name(self._charm.model.name) + + response = vault_client.create_snapshot() + content_uploaded = s3.upload_content( + content=response.raw, # type: ignore[reportArgumentType] + bucket_name=s3_parameters["bucket"], + key=backup_key, + ) + if not content_uploaded: + raise ManagerError("Failed to upload backup to S3 bucket") + logger.info("Backup uploaded to S3 bucket %s", s3_parameters["bucket"]) + return backup_key + + def list_backups(self) -> list[str]: + """List all the backups available in the S3 bucket. + + Backups are identified by the key prefix from + ``Naming.backup_s3_key_prefix``. + + Returns: + A list of backup keys with the prefix. + """ + self._validate_s3_prerequisites() + + s3_parameters = self._get_s3_parameters() + + try: + s3 = S3( + access_key=s3_parameters["access-key"], + secret_key=s3_parameters["secret-key"], + endpoint=s3_parameters["endpoint"], + region=s3_parameters.get("region"), + ) + except S3Error: + raise ManagerError("Failed to create S3 session") + + try: + backup_ids = s3.get_object_key_list( + bucket_name=s3_parameters["bucket"], prefix=Naming.backup_s3_key_prefix + ) + except S3Error as e: + raise ManagerError(f"Failed to list backups in S3 bucket: {e}") + return backup_ids + + def restore_backup(self, vault_client: VaultClient, backup_key: str) -> None: + """Restore the Vault data from the backup using the ``vault_client`` provided. + + Args: + vault_client: The Vault client to use for restoring the snapshot + backup_key: The S3 key of the backup to restore + """ + self._validate_s3_prerequisites() + + s3_parameters = self._get_s3_parameters() + + try: + s3 = S3( + access_key=s3_parameters["access-key"], + secret_key=s3_parameters["secret-key"], + endpoint=s3_parameters["endpoint"], + region=s3_parameters.get("region"), + ) + except S3Error: + raise ManagerError("Failed to create S3 session") + + try: + snapshot = s3.get_content( + bucket_name=s3_parameters["bucket"], + object_key=backup_key, + ) + except S3Error as e: + raise ManagerError(f"Failed to retrieve snapshot from S3: {e}") + if not snapshot: + raise ManagerError("Snapshot not found in S3 bucket") + + try: + vault_client.restore_snapshot(snapshot=snapshot) + except VaultClientError as e: + raise ManagerError(f"Failed to restore snapshot: {e}") + + def _validate_s3_prerequisites(self) -> str | None: + """Validate the S3 pre-requisites are met. + + Raises: + ManagerError: If any of the pre-requisites are not met. + """ + if not self._juju_facade.is_leader: + raise ManagerError("Only leader unit can perform backup operations") + if not self._juju_facade.relation_exists(self._relation_name): + raise ManagerError("S3 relation not created") + if missing_parameters := self._get_missing_s3_parameters(): + raise ManagerError("S3 parameters missing ({})".format(", ".join(missing_parameters))) + + def _get_missing_s3_parameters(self) -> list[str]: + """Return the list of missing S3 parameters. + + Returns: + List[str]: List of missing required S3 parameters. + """ + s3_parameters = self._s3_requirer.get_s3_connection_info() + return [param for param in self.REQUIRED_S3_PARAMETERS if param not in s3_parameters] + + def _get_s3_parameters(self) -> dict[str, str]: + """Retrieve S3 parameters from the S3 integrator relation. + + Removes leading and trailing whitespaces from the parameters. + + Returns: + Dict[str, str]: Dictionary of the S3 parameters. + """ + s3_parameters = self._s3_requirer.get_s3_connection_info() + for key, value in s3_parameters.items(): + if isinstance(value, str): + s3_parameters[key] = value.strip() + return s3_parameters diff --git a/src/charm.py b/src/charm.py index 96521c68..46f043d7 100755 --- a/src/charm.py +++ b/src/charm.py @@ -10,11 +10,9 @@ import json import logging import socket -from datetime import datetime from typing import Any, Dict, List import hcl -from botocore.response import StreamingBody from charms.data_platform_libs.v0.s3 import S3Requirer from charms.grafana_k8s.v0.grafana_dashboard import GrafanaDashboardProvider from charms.loki_k8s.v1.loki_push_api import LogForwarder @@ -53,13 +51,14 @@ AutounsealConfigurationDetails, AutounsealProviderManager, AutounsealRequirerManager, + BackupManager, File, KVManager, + ManagerError, PKIManager, TLSManager, VaultCertsError, ) -from charms.vault_k8s.v0.vault_s3 import S3, S3Error from jinja2 import Environment, FileSystemLoader from ops import CharmBase, MaintenanceStatus, main from ops.charm import ( @@ -86,7 +85,6 @@ AUTOUNSEAL_MOUNT_PATH = "charm-autounseal" AUTOUNSEAL_PROVIDES_RELATION_NAME = "vault-autounseal-provides" AUTOUNSEAL_REQUIRES_RELATION_NAME = "vault-autounseal-requires" -BACKUP_KEY_PREFIX = "vault-backup" CHARM_POLICY_NAME = "charm-access" CHARM_POLICY_PATH = "src/templates/charm_policy.hcl" CONFIG_TEMPLATE_DIR_PATH = "src/templates/" @@ -94,14 +92,12 @@ CONTAINER_NAME = "vault" CONTAINER_TLS_FILE_DIRECTORY_PATH = "/vault/certs" KV_RELATION_NAME = "vault-kv" -KV_SECRET_PREFIX = "kv-creds-" LOG_FORWARDING_RELATION_NAME = "logging" PEER_RELATION_NAME = "vault-peers" PKI_MOUNT = "charm-pki" PKI_RELATION_NAME = "vault-pki" PKI_ROLE_NAME = "charm" PROMETHEUS_ALERT_RULES_PATH = "./src/prometheus_alert_rules" -REQUIRED_S3_PARAMETERS = ["bucket", "access-key", "secret-key", "endpoint"] S3_RELATION_NAME = "s3-parameters" TLS_CERTIFICATES_PKI_RELATION_NAME = "tls-certificates-pki" VAULT_CHARM_APPROLE_SECRET_LABEL = "vault-approle-auth-details" @@ -283,7 +279,7 @@ def _on_collect_status(self, event: CollectStatusEvent): # noqa: C901 BlockedStatus("Please authorize charm (see `authorize-charm` action)") ) return - if not self._get_active_vault_client(): + if not self._get_authenticated_vault_client(): event.add_status(WaitingStatus("Waiting for vault to finish raft leader election")) event.add_status(ActiveStatus()) @@ -326,7 +322,7 @@ def _configure(self, _: EventBase) -> None: # noqa: C901 return except VaultClientError: return - if not (vault := self._get_active_vault_client()): + if not (vault := self._get_authenticated_vault_client()): return self._configure_pki_secrets_engine(vault) self._sync_vault_autounseal(vault) @@ -531,49 +527,17 @@ def _on_create_backup_action(self, event: ActionEvent) -> None: Args: event: ActionEvent """ - s3_pre_requisites_err = self._check_s3_pre_requisites() - if s3_pre_requisites_err: - event.fail(message=f"S3 pre-requisites not met. {s3_pre_requisites_err}.") - return - - s3_parameters = self._get_s3_parameters() - - try: - s3 = S3( - access_key=s3_parameters["access-key"], - secret_key=s3_parameters["secret-key"], - endpoint=s3_parameters["endpoint"], - region=s3_parameters.get("region"), - ) - except S3Error: - event.fail(message="Failed to create S3 session.") - logger.error("Failed to run create-backup action - Failed to create S3 session.") - return - - if not (s3.create_bucket(bucket_name=s3_parameters["bucket"])): - event.fail(message="Failed to create S3 bucket.") - logger.error("Failed to run create-backup action - Failed to create S3 bucket.") - return - backup_key = self._get_backup_key() - vault = self._get_active_vault_client() - if not vault: + vault_client = self._get_authenticated_vault_client() + if not vault_client: event.fail(message="Failed to initialize Vault client.") - logger.error("Failed to run create-backup action - Failed to initialize Vault client.") return - - response = vault.create_snapshot() - content_uploaded = s3.upload_content( - content=response.raw, # type: ignore[reportArgumentType] - bucket_name=s3_parameters["bucket"], - key=backup_key, - ) - if not content_uploaded: - event.fail(message="Failed to upload backup to S3 bucket.") - logger.error( - "Failed to run create-backup action - Failed to upload backup to S3 bucket." - ) + try: + manager = BackupManager(self, self.s3_requirer, S3_RELATION_NAME) + backup_key = manager.create_backup(vault_client) + except ManagerError as e: + logger.error("Failed to create backup: %s", e) + event.fail(message=f"Failed to create backup: {e}") return - logger.info("Backup uploaded to S3 bucket %s", s3_parameters["bucket"]) event.set_results({"backup-id": backup_key}) def _on_list_backups_action(self, event: ActionEvent) -> None: @@ -584,32 +548,12 @@ def _on_list_backups_action(self, event: ActionEvent) -> None: Args: event: ActionEvent """ - s3_pre_requisites_err = self._check_s3_pre_requisites() - if s3_pre_requisites_err: - event.fail(message=f"S3 pre-requisites not met. {s3_pre_requisites_err}.") - return - - s3_parameters = self._get_s3_parameters() - - try: - s3 = S3( - access_key=s3_parameters["access-key"], - secret_key=s3_parameters["secret-key"], - endpoint=s3_parameters["endpoint"], - region=s3_parameters.get("region"), - ) - except S3Error as e: - event.fail(message="Failed to create S3 session.") - logger.error("Failed to run list-backups action - %s", e) - return - try: - backup_ids = s3.get_object_key_list( - bucket_name=s3_parameters["bucket"], prefix=BACKUP_KEY_PREFIX - ) - except S3Error as e: + manager = BackupManager(self, self.s3_requirer, S3_RELATION_NAME) + backup_ids = manager.list_backups() + except ManagerError as e: logger.error("Failed to list backups: %s", e) - event.fail(message="Failed to run list-backups action - Failed to list backups.") + event.fail(message=f"Failed to list backups: {e}") return event.set_results({"backup-ids": json.dumps(backup_ids)}) @@ -622,96 +566,24 @@ def _on_restore_backup_action(self, event: ActionEvent) -> None: Args: event: ActionEvent """ - s3_pre_requisites_err = self._check_s3_pre_requisites() - if s3_pre_requisites_err: - event.fail(message=f"S3 pre-requisites not met. {s3_pre_requisites_err}.") - return - - s3_parameters = self._get_s3_parameters() + vault_client = self._get_active_vault_client() + if not vault_client: + event.fail(message="Failed to initialize an active Vault client.") + return + key = event.params.get("backup-id") + # This should be enforced by Juju/charmcraft.yaml, but we assert here + # to make the typechecker happy + assert isinstance(key, str) try: - s3 = S3( - access_key=s3_parameters["access-key"], - secret_key=s3_parameters["secret-key"], - endpoint=s3_parameters["endpoint"], - region=s3_parameters.get("region"), - ) - except S3Error as e: - logger.error("Failed to create S3 session: %s", e) - event.fail(message="Failed to create S3 session.") + manager = BackupManager(self, self.s3_requirer, S3_RELATION_NAME) + manager.restore_backup(vault_client, key) + except ManagerError as e: + logger.error("Failed to restore backup: %s", e) + event.fail(message=f"Failed to restore backup: {e}") return - try: - snapshot = s3.get_content( - bucket_name=s3_parameters["bucket"], - object_key=event.params.get("backup-id"), # type: ignore[reportArgumentType] - ) - except S3Error as e: - logger.error("Failed to retrieve snapshot from S3 storage: %s", e) - event.fail(message="Failed to retrieve snapshot from S3 storage.") - return - if not snapshot: - logger.error("Backup %s not found in S3 bucket", event.params.get("backup-id")) - event.fail(message="Backup not found in S3 bucket.") - return - try: - if not (self._restore_vault(snapshot=snapshot)): - logger.error("Failed to restore vault.") - event.fail(message="Failed to restore vault.") - return - except RuntimeError as e: - logger.error("Failed to restore vault: %s", e) - event.fail(message="Failed to restore vault.") - return - try: - if self.juju_facade.secret_exists_with_fields( - fields=("role-id", "secret-id"), - label=VAULT_CHARM_APPROLE_SECRET_LABEL, - ): - approle = self._get_approle_auth_secret() - vault = VaultClient( - url=self._api_address, - ca_cert_path=self.tls.get_tls_file_path_in_charm(File.CA), - ) - if approle and not vault.authenticate(approle): - self.juju_facade.remove_secret(label=VAULT_CHARM_APPROLE_SECRET_LABEL) - except Exception as e: - logger.error("Failed to remove old approle secret: %s", e) - event.fail(message="Failed to remove old approle secret.") event.set_results({"restored": event.params.get("backup-id")}) - def _get_s3_parameters(self) -> Dict[str, str]: - """Retrieve S3 parameters from the S3 integrator relation. - - Removes leading and trailing whitespaces from the parameters. - - Returns: - Dict[str, str]: Dictionary of the S3 parameters. - """ - s3_parameters = self.s3_requirer.get_s3_connection_info() - for key, value in s3_parameters.items(): - if isinstance(value, str): - s3_parameters[key] = value.strip() - return s3_parameters - - def _check_s3_pre_requisites(self) -> str | None: - """Check if the S3 pre-requisites are met.""" - if not self.unit.is_leader(): - return "Only leader unit can perform backup operations" - if not self.juju_facade.relation_exists(S3_RELATION_NAME): - return "S3 relation not created" - if missing_parameters := self._get_missing_s3_parameters(): - return "S3 parameters missing ({})".format(", ".join(missing_parameters)) - return None - - def _get_backup_key(self) -> str: - """Return the backup key. - - Returns: - str: The backup key - """ - timestamp = datetime.now().strftime("%Y-%m-%d-%H-%M-%S") - return f"{BACKUP_KEY_PREFIX}-{self.model.name}-{timestamp}" - def _delete_vault_data(self) -> None: """Delete Vault's data.""" try: @@ -826,15 +698,6 @@ def _get_approle_auth_secret(self) -> AppRole | None: return None return AppRole(role_id, secret_id) if role_id and secret_id else None - def _get_missing_s3_parameters(self) -> List[str]: - """Return the list of missing S3 parameters. - - Returns: - List[str]: List of missing required S3 parameters. - """ - s3_parameters = self.s3_requirer.get_s3_connection_info() - return [param for param in REQUIRED_S3_PARAMETERS if param not in s3_parameters] - def _set_pebble_plan(self) -> None: """Set the pebble plan if different from the currently applied one.""" plan = self._container.get_plan() @@ -844,42 +707,31 @@ def _set_pebble_plan(self) -> None: self._container.replan() logger.info("Pebble layer added") - def _restore_vault(self, snapshot: StreamingBody) -> bool: - """Restore vault using a raft snapshot. - - Args: - snapshot: Snapshot to be restored as a StreamingBody from the S3 storage. + def _get_active_vault_client(self) -> VaultClient | None: + """Return a client for the _active_ vault service. - Returns: - bool: True if the restore was successful, False otherwise. + This may not be the Vault service running on this unit. """ for address in self._get_peer_node_api_addresses(): - vault = VaultClient(address, ca_cert_path=self.tls.get_tls_file_path_in_charm(File.CA)) + try: + vault = VaultClient( + address, ca_cert_path=self.tls.get_tls_file_path_in_charm(File.CA) + ) + except VaultCertsError as e: + logger.warning("Failed to get Vault client: %s", e) + continue if vault.is_active(): - break - else: - logger.error("Failed to find active Vault client, cannot restore snapshot.") - return False - try: - if not (approle := self._get_approle_auth_secret()): - logger.error("Failed to log in to Vault") - return False - vault.authenticate(approle) - # hvac vault client expects bytes or a file-like object to restore the snapshot - # StreamingBody implements the read() method - # so it can be used as a file-like object in this context - response = vault.restore_snapshot(snapshot) - except VaultClientError as e: - logger.error("Failed to restore snapshot: %s", e) - return False - if not 200 <= response.status_code < 300: - logger.error("Failed to restore snapshot: %s", response.json()) - return False - - return True + if not vault.is_api_available(): + return None + if not (approle := self._get_approle_auth_secret()): + return None + if not vault.authenticate(approle): + return None + return vault + return None - def _get_active_vault_client(self) -> VaultClient | None: - """Return an initialized vault client. + def _get_authenticated_vault_client(self) -> VaultClient | None: + """Return an authenticated client for the Vault service on this unit. Returns: Vault: An active Vault client configured with the cluster address diff --git a/tests/unit/fixtures.py b/tests/unit/fixtures.py index 3751b3b3..bd35a88a 100644 --- a/tests/unit/fixtures.py +++ b/tests/unit/fixtures.py @@ -12,11 +12,11 @@ from charms.vault_k8s.v0.vault_managers import ( AutounsealProviderManager, AutounsealRequirerManager, + BackupManager, KVManager, PKIManager, TLSManager, ) -from charms.vault_k8s.v0.vault_s3 import S3 from charm import VaultCharm @@ -33,7 +33,7 @@ class VaultCharmFixtures: patcher_kv_manager = patch("charm.KVManager", autospec=KVManager) patcher_pki_manager = patch("charm.PKIManager", autospec=PKIManager) patcher_s3_requirer = patch("charm.S3Requirer", autospec=S3Requirer) - patcher_s3 = patch("charm.S3", autospec=S3) + patcher_backup_manager = patch("charm.BackupManager", autospec=BackupManager) patcher_socket_fqdn = patch("socket.getfqdn") patcher_pki_requirer_get_assigned_certificate = patch( "charm.TLSCertificatesRequiresV4.get_assigned_certificate" @@ -71,7 +71,7 @@ def setup(self): self.mock_kv_manager = VaultCharmFixtures.patcher_kv_manager.start().return_value self.mock_pki_manager = VaultCharmFixtures.patcher_pki_manager.start().return_value self.mock_s3_requirer = VaultCharmFixtures.patcher_s3_requirer.start().return_value - self.mock_s3 = VaultCharmFixtures.patcher_s3.start() + self.mock_backup_manager = VaultCharmFixtures.patcher_backup_manager.start().return_value self.mock_socket_fqdn = VaultCharmFixtures.patcher_socket_fqdn.start() self.mock_pki_requirer_get_assigned_certificate = ( VaultCharmFixtures.patcher_pki_requirer_get_assigned_certificate.start() diff --git a/tests/unit/lib/charms/vault_k8s/v0/test_vault_managers.py b/tests/unit/lib/charms/vault_k8s/v0/test_vault_managers.py index fd769233..d49b1d2e 100644 --- a/tests/unit/lib/charms/vault_k8s/v0/test_vault_managers.py +++ b/tests/unit/lib/charms/vault_k8s/v0/test_vault_managers.py @@ -2,16 +2,24 @@ from unittest.mock import MagicMock, patch import pytest +from charms.data_platform_libs.v0.s3 import S3Requirer from charms.vault_k8s.v0.juju_facade import NoSuchSecretError, SecretRemovedError from charms.vault_k8s.v0.vault_autounseal import AutounsealDetails -from charms.vault_k8s.v0.vault_client import AuthMethod, SecretsBackend, VaultClient +from charms.vault_k8s.v0.vault_client import ( + AuthMethod, + SecretsBackend, + VaultClient, + VaultClientError, +) from charms.vault_k8s.v0.vault_client import Certificate as VaultClientCertificate from charms.vault_k8s.v0.vault_managers import ( AUTOUNSEAL_POLICY, AutounsealProviderManager, AutounsealRequirerManager, + BackupManager, CertificateRequestAttributes, KVManager, + ManagerError, PKIManager, PrivateKey, ProviderCertificate, @@ -19,6 +27,7 @@ TLSCertificatesRequiresV4, VaultKvProvides, ) +from charms.vault_k8s.v0.vault_s3 import S3Error from charm import AUTOUNSEAL_MOUNT_PATH, VaultCharm from tests.unit.certificates import ( @@ -509,3 +518,208 @@ def test_given_outstanding_requests_when_sync_then_certificates_issued( chain=provider_certificate.chain, ) ) + + +class TestBackupManager: + @pytest.fixture(autouse=True) + @patch("charms.vault_k8s.v0.vault_managers.JujuFacade") + def setup(self, juju_facade_mock: MagicMock, monkeypatch: pytest.MonkeyPatch): + """Configure the test environment for the happy path. + + Individual tests can then mock error scenarios by changing the mocks. + """ + self.juju_facade = juju_facade_mock.return_value + self.juju_facade.is_leader = True + self.juju_facade.relation_exists.return_value = True + self.s3_class = MagicMock() + monkeypatch.setattr("charms.vault_k8s.v0.vault_managers.S3", self.s3_class) + self.s3 = self.s3_class.return_value + self.s3.get_object_key_list.return_value = [ + "vault-backup-my-model-1", + "vault-backup-my-model-2", + ] + self.s3.get_content.return_value = "snapshot content" + + self.charm = MagicMock(spec=VaultCharm) + self.charm.model.name = "my-model" + self.vault_client = MagicMock(spec=VaultClient) + self.s3_requirer = MagicMock(spec=S3Requirer) + self.s3_requirer.get_s3_connection_info.return_value = { + "bucket": "my-bucket", + "access-key": "my-access-key", + "secret-key": "my-secret-key", + "endpoint": "my-endpoint", + "region": "my-region", + } + + self.manager = BackupManager(self.charm, self.s3_requirer, "s3-relation") + + def test_given_non_leader_when_create_backup_then_error_raised(self): + self.juju_facade.is_leader = False + with pytest.raises(ManagerError) as e: + self.manager.create_backup(self.vault_client) + assert str(e.value) == "Only leader unit can perform backup operations" + + def test_given_s3_relation_not_created_when_create_backup_then_error_raised(self): + self.juju_facade.relation_exists.return_value = False + with pytest.raises(ManagerError) as e: + self.manager.create_backup(self.vault_client) + assert str(e.value) == "S3 relation not created" + + @pytest.mark.parametrize( + "missing_key, expected_error_message", + [ + ("bucket", "S3 parameters missing (bucket)"), + ("access-key", "S3 parameters missing (access-key)"), + ("secret-key", "S3 parameters missing (secret-key)"), + ("endpoint", "S3 parameters missing (endpoint)"), + ], + ) + def test_given_missing_s3_parameter_when_create_backup_then_error_raised( + self, missing_key: str, expected_error_message: str + ): + del self.s3_requirer.get_s3_connection_info.return_value[missing_key] + with pytest.raises(ManagerError) as e: + self.manager.create_backup(self.vault_client) + assert str(e.value) == expected_error_message + + def test_given_s3_error_when_create_backup_then_error_raised(self): + self.s3_class.side_effect = S3Error() # throw an error when creating the S3 object + with pytest.raises(ManagerError) as e: + self.manager.create_backup(self.vault_client) + assert str(e.value) == "Failed to create S3 session" + + def test_given_bucket_creation_returns_none_when_create_backup_then_error_raised(self): + self.s3.create_bucket.return_value = None + with pytest.raises(ManagerError) as e: + self.manager.create_backup(self.vault_client) + assert str(e.value) == "Failed to create S3 bucket" + + def test_given_failed_to_upload_backup_when_create_backup_then_error_raised(self): + self.s3.create_bucket.return_value = True + self.s3.upload_content.return_value = False + with pytest.raises(ManagerError) as e: + self.manager.create_backup(self.vault_client) + assert str(e.value) == "Failed to upload backup to S3 bucket" + + def test_given_s3_available_when_create_backup_then_backup_created(self): + key = self.manager.create_backup(self.vault_client) + + self.vault_client.create_snapshot.assert_called_once() + self.s3.upload_content.assert_called_once() + + assert key.startswith("vault-backup-my-model-") + + # List backups + def test_given_non_leader_when_list_backups_then_error_raised(self): + self.juju_facade.is_leader = False + with pytest.raises(ManagerError) as e: + self.manager.list_backups() + assert str(e.value) == "Only leader unit can perform backup operations" + + def test_given_s3_relation_not_created_when_list_backups_then_error_raised(self): + self.juju_facade.is_leader = True + self.juju_facade.relation_exists.return_value = False + with pytest.raises(ManagerError) as e: + self.manager.list_backups() + assert str(e.value) == "S3 relation not created" + + @pytest.mark.parametrize( + "missing_key, expected_error_message", + [ + ("bucket", "S3 parameters missing (bucket)"), + ("access-key", "S3 parameters missing (access-key)"), + ("secret-key", "S3 parameters missing (secret-key)"), + ("endpoint", "S3 parameters missing (endpoint)"), + ], + ) + def test_given_missing_s3_parameter_when_list_backups_then_error_raised( + self, missing_key: str, expected_error_message: str + ): + self.juju_facade.is_leader = True + self.juju_facade.relation_exists.return_value = True + del self.s3_requirer.get_s3_connection_info.return_value[missing_key] + with pytest.raises(ManagerError) as e: + self.manager.list_backups() + assert str(e.value) == expected_error_message + + def test_given_s3_error_when_list_backups_then_error_raised(self): + self.s3_class.side_effect = S3Error() # throw an error when creating the S3 object + with pytest.raises(ManagerError) as e: + self.manager.list_backups() + assert str(e.value) == "Failed to create S3 session" + + def test_given_s3_error_during_get_object_key_when_list_backups_then_error_raised(self): + self.s3.get_object_key_list.side_effect = S3Error("some error message") + with pytest.raises(ManagerError) as e: + self.manager.list_backups() + assert str(e.value) == "Failed to list backups in S3 bucket: some error message" + + def test_given_s3_available_when_list_backups_then_backups_listed(self): + backups = self.manager.list_backups() + assert backups == ["vault-backup-my-model-1", "vault-backup-my-model-2"] + + # Restore backup + def test_given_non_leader_when_restore_backup_then_error_raised(self): + self.juju_facade.is_leader = False + with pytest.raises(ManagerError) as e: + self.manager.restore_backup(self.vault_client, "vault-backup-my-model-1") + assert str(e.value) == "Only leader unit can perform backup operations" + + def test_given_s3_relation_not_created_when_restore_backup_then_error_raised(self): + self.juju_facade.is_leader = True + self.juju_facade.relation_exists.return_value = False + with pytest.raises(ManagerError) as e: + self.manager.restore_backup(self.vault_client, "vault-backup-my-model-1") + assert str(e.value) == "S3 relation not created" + + @pytest.mark.parametrize( + "missing_key, expected_error_message", + [ + ("bucket", "S3 parameters missing (bucket)"), + ("access-key", "S3 parameters missing (access-key)"), + ("secret-key", "S3 parameters missing (secret-key)"), + ("endpoint", "S3 parameters missing (endpoint)"), + ], + ) + def test_given_missing_s3_parameter_when_restore_backup_then_error_raised( + self, missing_key: str, expected_error_message: str + ): + self.juju_facade.is_leader = True + self.juju_facade.relation_exists.return_value = True + del self.s3_requirer.get_s3_connection_info.return_value[missing_key] + with pytest.raises(ManagerError) as e: + self.manager.restore_backup(self.vault_client, "vault-backup-my-model-1") + assert str(e.value) == expected_error_message + + def test_given_s3_error_when_restore_backup_then_error_raised(self): + self.s3_class.side_effect = S3Error() # throw an error when creating the S3 object + with pytest.raises(ManagerError) as e: + self.manager.restore_backup(self.vault_client, "vault-backup-my-model-1") + assert str(e.value) == "Failed to create S3 session" + + def test_given_s3_error_during_download_when_restore_backup_then_error_raised(self): + self.s3.get_content.side_effect = S3Error("some error message") + with pytest.raises(ManagerError) as e: + self.manager.restore_backup(self.vault_client, "vault-backup-my-model-1") + assert str(e.value) == "Failed to retrieve snapshot from S3: some error message" + + def test_given_s3_content_not_found_when_restore_backup_then_error_raised(self): + self.s3.get_content.return_value = None + with pytest.raises(ManagerError) as e: + self.manager.restore_backup(self.vault_client, "vault-backup-my-model-1") + assert str(e.value) == "Snapshot not found in S3 bucket" + + def test_given_vault_client_fails_to_restore_snapshot_when_restore_backup_then_error_raised( + self, + ): + self.vault_client.restore_snapshot.side_effect = VaultClientError("some error message") + with pytest.raises(ManagerError) as e: + self.manager.restore_backup(self.vault_client, "vault-backup-my-model-1") + assert str(e.value) == "Failed to restore snapshot: some error message" + + def test_given_s3_content_and_vault_client_available_when_restore_backup_then_backup_restored( + self, + ): + self.manager.restore_backup(self.vault_client, "vault-backup-my-model-1") + self.vault_client.restore_snapshot.assert_called_once_with(snapshot="snapshot content") diff --git a/tests/unit/test_charm_create_backup_action.py b/tests/unit/test_charm_create_backup_action.py index 3ed5b3b9..4ca38fb7 100644 --- a/tests/unit/test_charm_create_backup_action.py +++ b/tests/unit/test_charm_create_backup_action.py @@ -5,127 +5,12 @@ import ops.testing as testing import pytest -from charms.vault_k8s.v0.vault_s3 import S3Error +from charms.vault_k8s.v0.vault_managers import ManagerError from tests.unit.fixtures import VaultCharmFixtures class TestCharmCreateBackupAction(VaultCharmFixtures): - def test_given_non_leader_when_create_backup_action_then_fails(self): - container = testing.Container( - name="vault", - can_connect=True, - ) - state_in = testing.State( - containers=[container], - leader=False, - ) - - with pytest.raises(testing.ActionFailed) as e: - self.ctx.run(self.ctx.on.action("create-backup"), state_in) - assert ( - e.value.message - == "S3 pre-requisites not met. Only leader unit can perform backup operations." - ) - - def test_given_s3_relation_not_created_when_create_backup_action_then_fails(self): - container = testing.Container( - name="vault", - can_connect=True, - ) - state_in = testing.State( - containers=[container], - leader=True, - ) - with pytest.raises(testing.ActionFailed) as e: - self.ctx.run(self.ctx.on.action("create-backup"), state_in) - assert e.value.message == "S3 pre-requisites not met. S3 relation not created." - - def test_given_missing_s3_parameters_when_create_backup_then_action_fails(self): - container = testing.Container( - name="vault", - can_connect=True, - ) - s3_relation = testing.Relation( - endpoint="s3-parameters", - interface="s3", - ) - state_in = testing.State( - containers=[container], - leader=True, - relations=[s3_relation], - ) - with pytest.raises(testing.ActionFailed) as e: - self.ctx.run(self.ctx.on.action("create-backup"), state_in) - assert ( - e.value.message - == "S3 pre-requisites not met. S3 parameters missing (bucket, access-key, secret-key, endpoint)." - ) - - def test_given_s3_error_when_create_backup_then_action_fails(self): - self.mock_s3_requirer.configure_mock( - **{ - "get_s3_connection_info.return_value": { - "access-key": "my-access-key", - "secret-key": "my-secret-key", - "endpoint": "my-endpoint", - "bucket": "my bucket", - "region": "my-region", - }, - }, - ) - self.mock_s3.side_effect = S3Error() - container = testing.Container( - name="vault", - can_connect=True, - ) - s3_relation = testing.Relation( - endpoint="s3-parameters", - interface="s3", - ) - state_in = testing.State( - containers=[container], - leader=True, - relations=[s3_relation], - ) - with pytest.raises(testing.ActionFailed) as e: - self.ctx.run(self.ctx.on.action("create-backup"), state_in) - assert e.value.message == "Failed to create S3 session." - - def test_given_bucket_creation_returns_none_when_create_backup_then_action_fails(self): - self.mock_s3_requirer.configure_mock( - **{ - "get_s3_connection_info.return_value": { - "access-key": "my-access-key", - "secret-key": "my-secret-key", - "endpoint": "my-endpoint", - "bucket": "my bucket", - "region": "my-region", - }, - }, - ) - self.mock_s3.return_value.configure_mock( - **{ - "create_bucket.return_value": None, - }, - ) - container = testing.Container( - name="vault", - can_connect=True, - ) - s3_relation = testing.Relation( - endpoint="s3-parameters", - interface="s3", - ) - state_in = testing.State( - containers=[container], - leader=True, - relations=[s3_relation], - ) - with pytest.raises(testing.ActionFailed) as e: - self.ctx.run(self.ctx.on.action("create-backup"), state_in) - assert e.value.message == "Failed to create S3 bucket." - def test_given_failed_to_initialize_vault_client_when_create_backup_then_action_fails(self): self.mock_s3_requirer.configure_mock( **{ @@ -155,7 +40,7 @@ def test_given_failed_to_initialize_vault_client_when_create_backup_then_action_ self.ctx.run(self.ctx.on.action("create-backup"), state_in) assert e.value.message == "Failed to initialize Vault client." - def test_given_failed_to_upload_backup_when_create_backup_then_action_fails(self): + def test_given_manager_raises_error_when_create_backup_then_action_fails(self): self.mock_s3_requirer.configure_mock( **{ "get_s3_connection_info.return_value": { @@ -167,11 +52,7 @@ def test_given_failed_to_upload_backup_when_create_backup_then_action_fails(self }, }, ) - self.mock_s3.return_value.configure_mock( - **{ - "upload_content.return_value": False, - }, - ) + self.mock_backup_manager.create_backup.side_effect = ManagerError("some error message") self.mock_vault.configure_mock( **{ "is_api_available.return_value": True, @@ -198,52 +79,4 @@ def test_given_failed_to_upload_backup_when_create_backup_then_action_fails(self ) with pytest.raises(testing.ActionFailed) as e: self.ctx.run(self.ctx.on.action("create-backup"), state_in) - assert e.value.message == "Failed to upload backup to S3 bucket." - - def test_given_s3_available_when_create_backup_then_backup_created(self): - self.mock_s3_requirer.configure_mock( - **{ - "get_s3_connection_info.return_value": { - "access-key": "my-access-key", - "secret-key": "my-secret-key", - "endpoint": "my-endpoint", - "bucket": "my bucket", - "region": "my-region", - }, - }, - ) - self.mock_s3.return_value.configure_mock( - **{ - "upload_content.return_value": True, - }, - ) - self.mock_vault.configure_mock( - **{ - "is_api_available.return_value": True, - "is_active_or_standby.return_value": True, - }, - ) - approle_secret = testing.Secret( - label="vault-approle-auth-details", - tracked_content={"role-id": "role id", "secret-id": "secret id"}, - ) - container = testing.Container( - name="vault", - can_connect=True, - ) - s3_relation = testing.Relation( - endpoint="s3-parameters", - interface="s3", - ) - state_in = testing.State( - containers=[container], - leader=True, - relations=[s3_relation], - secrets=[approle_secret], - ) - self.ctx.run(self.ctx.on.action("create-backup"), state_in) - self.mock_s3.return_value.create_bucket.assert_called_with(bucket_name="my bucket") - self.mock_vault.create_snapshot.assert_called() - self.mock_s3.return_value.upload_content.assert_called() - assert self.ctx.action_results - assert "backup-id" in self.ctx.action_results.keys() + assert e.value.message == "Failed to create backup: some error message" diff --git a/tests/unit/test_charm_list_backups_action.py b/tests/unit/test_charm_list_backups_action.py index 94558ad7..0dd6ddaf 100644 --- a/tests/unit/test_charm_list_backups_action.py +++ b/tests/unit/test_charm_list_backups_action.py @@ -3,68 +3,15 @@ # See LICENSE file for licensing details. -import json - import ops.testing as testing import pytest -from charms.vault_k8s.v0.vault_s3 import S3Error +from charms.vault_k8s.v0.vault_managers import ManagerError from tests.unit.fixtures import VaultCharmFixtures class TestCharmListBackupAction(VaultCharmFixtures): - def test_given_non_leader_when_list_backups_action_then_fails(self): - container = testing.Container( - name="vault", - can_connect=True, - ) - state_in = testing.State( - containers=[container], - leader=False, - ) - with pytest.raises(testing.ActionFailed) as e: - self.ctx.run(self.ctx.on.action("list-backups"), state_in) - assert ( - e.value.message - == "S3 pre-requisites not met. Only leader unit can perform backup operations." - ) - - def test_given_s3_relation_not_created_when_list_backups_action_then_fails(self): - container = testing.Container( - name="vault", - can_connect=True, - ) - state_in = testing.State( - containers=[container], - leader=True, - ) - with pytest.raises(testing.ActionFailed) as e: - self.ctx.run(self.ctx.on.action("list-backups"), state_in) - assert e.value.message == "S3 pre-requisites not met. S3 relation not created." - - def test_given_missing_s3_parameters_when_list_backups_then_action_fails(self): - container = testing.Container( - name="vault", - can_connect=True, - ) - s3_relation = testing.Relation( - endpoint="s3-parameters", - interface="s3", - ) - state_in = testing.State( - containers=[container], - leader=True, - relations=[s3_relation], - ) - - with pytest.raises(testing.ActionFailed) as e: - self.ctx.run(self.ctx.on.action("list-backups"), state_in) - assert ( - e.value.message - == "S3 pre-requisites not met. S3 parameters missing (bucket, access-key, secret-key, endpoint)." - ) - - def test_given_s3_error_during_instantiation_when_list_backups_then_action_fails(self): + def test_given_manager_raises_error_when_list_backups_then_action_fails(self): self.mock_s3_requirer.configure_mock( **{ "get_s3_connection_info.return_value": { @@ -76,80 +23,7 @@ def test_given_s3_error_during_instantiation_when_list_backups_then_action_fails }, }, ) - self.mock_s3.side_effect = S3Error() - container = testing.Container( - name="vault", - can_connect=True, - ) - s3_relation = testing.Relation( - endpoint="s3-parameters", - interface="s3", - ) - state_in = testing.State( - containers=[container], - leader=True, - relations=[s3_relation], - ) - with pytest.raises(testing.ActionFailed) as e: - self.ctx.run(self.ctx.on.action("list-backups"), state_in) - assert e.value.message == "Failed to create S3 session." - - def test_given_s3_error_during_get_object_key_when_list_backups_then_action_fails(self): - self.mock_s3_requirer.configure_mock( - **{ - "get_s3_connection_info.return_value": { - "access-key": "my-access-key", - "secret-key": "my-secret-key", - "endpoint": "my-endpoint", - "bucket": "my bucket", - "region": "my-region", - }, - }, - ) - self.mock_s3.return_value.configure_mock( - **{ - "get_object_key_list.side_effect": S3Error(), - }, - ) - container = testing.Container( - name="vault", - can_connect=True, - ) - s3_relation = testing.Relation( - endpoint="s3-parameters", - interface="s3", - ) - state_in = testing.State( - containers=[container], - leader=True, - relations=[s3_relation], - ) - with pytest.raises(testing.ActionFailed) as e: - self.ctx.run(self.ctx.on.action("list-backups"), state_in) - assert e.value.message == "Failed to run list-backups action - Failed to list backups." - - def test_given_s3_available_when_list_backups_then_backup_listed(self): - self.mock_s3_requirer.configure_mock( - **{ - "get_s3_connection_info.return_value": { - "access-key": "my-access-key", - "secret-key": "my-secret-key", - "endpoint": "my-endpoint", - "bucket": "my bucket", - "region": "my-region", - }, - }, - ) - self.mock_s3.return_value.configure_mock( - **{ - "upload_content.return_value": True, - }, - ) - self.mock_s3.return_value.configure_mock( - **{ - "get_object_key_list.return_value": ["my-backup-id"], - }, - ) + self.mock_backup_manager.list_backups.side_effect = ManagerError("some error message") approle_secret = testing.Secret( label="vault-approle-auth-details", tracked_content={"role-id": "role id", "secret-id": "secret id"}, @@ -169,6 +43,6 @@ def test_given_s3_available_when_list_backups_then_backup_listed(self): secrets=[approle_secret], ) - self.ctx.run(self.ctx.on.action("list-backups"), state_in) - - assert self.ctx.action_results == {"backup-ids": json.dumps(["my-backup-id"])} + with pytest.raises(testing.ActionFailed) as e: + self.ctx.run(self.ctx.on.action("list-backups"), state_in) + assert e.value.message == "Failed to list backups: some error message" diff --git a/tests/unit/test_charm_restore_backup_action.py b/tests/unit/test_charm_restore_backup_action.py index cba957da..62f9012a 100644 --- a/tests/unit/test_charm_restore_backup_action.py +++ b/tests/unit/test_charm_restore_backup_action.py @@ -6,63 +6,13 @@ import ops.testing as testing import pytest import requests -from charms.vault_k8s.v0.vault_s3 import S3Error +from charms.vault_k8s.v0.vault_managers import ManagerError from tests.unit.fixtures import VaultCharmFixtures class TestCharmRestoreBackupAction(VaultCharmFixtures): - def test_given_non_leader_when_restore_backup_action_then_fails(self): - container = testing.Container( - name="vault", - can_connect=True, - ) - state_in = testing.State( - containers=[container], - leader=False, - ) - with pytest.raises(testing.ActionFailed) as e: - self.ctx.run(self.ctx.on.action("restore-backup"), state_in) - assert ( - e.value.message - == "S3 pre-requisites not met. Only leader unit can perform backup operations." - ) - - def test_given_s3_relation_not_created_when_restore_backup_action_then_fails(self): - container = testing.Container( - name="vault", - can_connect=True, - ) - state_in = testing.State( - containers=[container], - leader=True, - ) - with pytest.raises(testing.ActionFailed) as e: - self.ctx.run(self.ctx.on.action("restore-backup"), state_in) - assert e.value.message == "S3 pre-requisites not met. S3 relation not created." - - def test_given_missing_s3_parameters_when_restore_backup_then_action_fails(self): - container = testing.Container( - name="vault", - can_connect=True, - ) - s3_relation = testing.Relation( - endpoint="s3-parameters", - interface="s3", - ) - state_in = testing.State( - containers=[container], - leader=True, - relations=[s3_relation], - ) - with pytest.raises(testing.ActionFailed) as e: - self.ctx.run(self.ctx.on.action("restore-backup"), state_in) - assert ( - e.value.message - == "S3 pre-requisites not met. S3 parameters missing (bucket, access-key, secret-key, endpoint)." - ) - - def test_given_s3_error_during_instantiation_when_restore_backup_then_action_fails(self): + def test_given_manager_raises_error_when_restore_backup_then_action_fails(self): self.mock_s3_requirer.configure_mock( **{ "get_s3_connection_info.return_value": { @@ -74,143 +24,7 @@ def test_given_s3_error_during_instantiation_when_restore_backup_then_action_fai }, }, ) - self.mock_s3.side_effect = S3Error() - container = testing.Container( - name="vault", - can_connect=True, - ) - s3_relation = testing.Relation( - endpoint="s3-parameters", - interface="s3", - ) - state_in = testing.State( - containers=[container], - leader=True, - relations=[s3_relation], - ) - with pytest.raises(testing.ActionFailed) as e: - self.ctx.run(self.ctx.on.action("restore-backup"), state_in) - assert e.value.message == "Failed to create S3 session." - - def test_given_s3_error_during_get_content_when_restore_backup_then_action_fails(self): - self.mock_s3_requirer.configure_mock( - **{ - "get_s3_connection_info.return_value": { - "access-key": "my-access-key", - "secret-key": "my-secret-key", - "endpoint": "my-endpoint", - "bucket": "my bucket", - "region": "my-region", - }, - }, - ) - self.mock_s3.return_value.configure_mock( - **{ - "get_content.side_effect": S3Error(), - }, - ) - container = testing.Container( - name="vault", - can_connect=True, - ) - s3_relation = testing.Relation( - endpoint="s3-parameters", - interface="s3", - ) - state_in = testing.State( - containers=[container], - leader=True, - relations=[s3_relation], - ) - with pytest.raises(testing.ActionFailed) as e: - self.ctx.run(self.ctx.on.action("restore-backup"), state_in) - assert e.value.message == "Failed to retrieve snapshot from S3 storage." - - def test_given_no_snapshot_when_restore_backup_then_action_fails(self): - self.mock_s3_requirer.configure_mock( - **{ - "get_s3_connection_info.return_value": { - "access-key": "my-access-key", - "secret-key": "my-secret-key", - "endpoint": "my-endpoint", - "bucket": "my bucket", - "region": "my-region", - }, - }, - ) - self.mock_s3.return_value.configure_mock( - **{ - "get_content.return_value": None, - }, - ) - container = testing.Container( - name="vault", - can_connect=True, - ) - s3_relation = testing.Relation( - endpoint="s3-parameters", - interface="s3", - ) - state_in = testing.State( - containers=[container], - leader=True, - relations=[s3_relation], - ) - with pytest.raises(testing.ActionFailed) as e: - self.ctx.run(self.ctx.on.action("restore-backup"), state_in) - assert e.value.message == "Backup not found in S3 bucket." - - def test_given_failed_to_restore_vault_when_restore_backup_then_action_fails(self): - self.mock_s3_requirer.configure_mock( - **{ - "get_s3_connection_info.return_value": { - "access-key": "my-access-key", - "secret-key": "my-secret-key", - "endpoint": "my-endpoint", - "bucket": "my bucket", - "region": "my-region", - }, - }, - ) - self.mock_s3.return_value.configure_mock( - **{ - "get_content.return_value": "my snapshot content", - }, - ) - container = testing.Container( - name="vault", - can_connect=True, - ) - s3_relation = testing.Relation( - endpoint="s3-parameters", - interface="s3", - ) - state_in = testing.State( - containers=[container], - leader=True, - relations=[s3_relation], - ) - with pytest.raises(testing.ActionFailed) as e: - self.ctx.run(self.ctx.on.action("restore-backup"), state_in) - assert e.value.message == "Failed to restore vault." - - def test_given_200_response_when_restore_backup_then_action_success(self): - self.mock_s3_requirer.configure_mock( - **{ - "get_s3_connection_info.return_value": { - "access-key": "my-access-key", - "secret-key": "my-secret-key", - "endpoint": "my-endpoint", - "bucket": "my bucket", - "region": "my-region", - }, - }, - ) - self.mock_s3.return_value.configure_mock( - **{ - "get_content.return_value": "my snapshot content", - }, - ) + self.mock_backup_manager.list_backups.side_effect = ManagerError("some error message") response = requests.Response() response.status_code = 200 self.mock_vault.configure_mock( @@ -241,5 +55,6 @@ def test_given_200_response_when_restore_backup_then_action_success(self): state_in, ) - assert self.ctx.action_results == {"restored": "my-backup-id"} - self.mock_vault.restore_snapshot.assert_called_with("my snapshot content") + with pytest.raises(testing.ActionFailed) as e: + self.ctx.run(self.ctx.on.action("list-backups"), state_in) + assert e.value.message == "Failed to list backups: some error message"