| # Licensed to the Apache Software Foundation (ASF) under one or more |
| # contributor license agreements. See the NOTICE file distributed with |
| # this work for additional information regarding copyright ownership. |
| # The ASF licenses this file to You under the Apache License, Version 2.0 |
| # (the "License"); you may not use this file except in compliance with |
| # the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| |
| import os |
| import logging |
| import unittest |
| import time |
| from unittest import mock |
| from datetime import datetime, timezone, timedelta |
| from secret_manager import SecretManager, SECRET_MANAGER_LABEL, SecretManagerLoggerAdapter |
| from google.cloud import secretmanager |
| from google.api_core import exceptions |
| |
| class TestSecretManagerLoggerAdapter(unittest.TestCase): |
| """Unit tests for SecretManagerLoggerAdapter class.""" |
| |
| def test_process_adds_prefix(self): |
| """Test that the logger adapter adds the correct prefix.""" |
| logger = logging.getLogger("test") |
| adapter = SecretManagerLoggerAdapter(logger, {}) |
| |
| msg, kwargs = adapter.process("test message", {"key": "value"}) |
| |
| self.assertEqual(msg, "[SecretManager] test message") |
| self.assertEqual(kwargs, {"key": "value"}) |
| |
| class TestSecretManager(unittest.TestCase): |
| """Unit tests for SecretManager class.""" |
| |
| def setUp(self): |
| """Set up test fixtures.""" |
| self.project_id = "test-project" |
| self.logger = logging.getLogger("test") |
| self.logger.setLevel(logging.CRITICAL) # Suppress logging during tests |
| |
| # Mock the SecretManagerServiceClient |
| with mock.patch('secret_manager.secretmanager.SecretManagerServiceClient'): |
| self.manager = SecretManager( |
| self.project_id, |
| self.logger, |
| rotation_interval=30, |
| grace_period=7, |
| max_retries=3 |
| ) |
| |
| self.test_secret_id = "test-secret" |
| self.test_data_id = "test-data" |
| self.test_payload = b"test-payload" |
| |
| def test_init(self): |
| """Test SecretManager initialization.""" |
| with mock.patch('secret_manager.secretmanager.SecretManagerServiceClient'): |
| manager = SecretManager("test-project", self.logger, 15, 3, 5) |
| |
| self.assertEqual(manager.project_id, "test-project") |
| self.assertEqual(manager.rotation_interval, 15) |
| self.assertEqual(manager.grace_period, 3) |
| self.assertEqual(manager.max_retries, 5) |
| self.assertIsInstance(manager.logger, SecretManagerLoggerAdapter) |
| |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_get_secret_ids(self, mock_client): |
| """Test _get_secret_ids method.""" |
| # Mock response with secrets having the correct label |
| mock_secret1 = mock.Mock() |
| mock_secret1.name = "projects/test-project/secrets/secret1" |
| mock_secret1.labels = {"created_by": SECRET_MANAGER_LABEL} |
| |
| mock_secret2 = mock.Mock() |
| mock_secret2.name = "projects/test-project/secrets/secret2" |
| mock_secret2.labels = {"created_by": "other"} |
| |
| mock_secret3 = mock.Mock() |
| mock_secret3.name = "projects/test-project/secrets/secret3" |
| mock_secret3.labels = {"created_by": SECRET_MANAGER_LABEL} |
| |
| mock_client.return_value.list_secrets.return_value = [mock_secret1, mock_secret2, mock_secret3] |
| |
| manager = SecretManager(self.project_id, self.logger) |
| secret_ids = manager._get_secret_ids() |
| |
| self.assertEqual(secret_ids, ["secret1", "secret3"]) |
| mock_client.return_value.list_secrets.assert_called_once() |
| |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_get_secret_ids_exception(self, mock_client): |
| """Test _get_secret_ids method with exception.""" |
| mock_client.return_value.list_secrets.side_effect = Exception("API Error") |
| |
| manager = SecretManager(self.project_id, self.logger) |
| secret_ids = manager._get_secret_ids() |
| |
| self.assertEqual(secret_ids, []) |
| |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_secret_exists_true(self, mock_client): |
| """Test _secret_exists method when secret exists.""" |
| mock_client.return_value.get_secret.return_value = mock.Mock() |
| |
| manager = SecretManager(self.project_id, self.logger) |
| exists = manager._secret_exists(self.test_secret_id) |
| |
| self.assertTrue(exists) |
| mock_client.return_value.get_secret.assert_called_once() |
| |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_secret_exists_false(self, mock_client): |
| """Test _secret_exists method when secret doesn't exist.""" |
| mock_client.return_value.get_secret.side_effect = exceptions.NotFound("Secret not found") |
| |
| manager = SecretManager(self.project_id, self.logger) |
| exists = manager._secret_exists(self.test_secret_id) |
| |
| self.assertFalse(exists) |
| |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_secret_is_managed_true(self, mock_client): |
| """Test _secret_is_managed method when secret is managed.""" |
| mock_secret = mock.Mock() |
| mock_secret.labels = {"created_by": SECRET_MANAGER_LABEL} |
| mock_client.return_value.get_secret.return_value = mock_secret |
| |
| manager = SecretManager(self.project_id, self.logger) |
| is_managed = manager._secret_is_managed(self.test_secret_id) |
| |
| self.assertTrue(is_managed) |
| |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_secret_is_managed_false(self, mock_client): |
| """Test _secret_is_managed method when secret is not managed.""" |
| mock_secret = mock.Mock() |
| mock_secret.labels = {"created_by": "other"} |
| mock_client.return_value.get_secret.return_value = mock_secret |
| |
| manager = SecretManager(self.project_id, self.logger) |
| is_managed = manager._secret_is_managed(self.test_secret_id) |
| |
| self.assertFalse(is_managed) |
| |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_secret_is_managed_not_exists(self, mock_client): |
| """Test _secret_is_managed method when secret doesn't exist.""" |
| mock_client.return_value.get_secret.side_effect = exceptions.NotFound("Secret not found") |
| |
| manager = SecretManager(self.project_id, self.logger) |
| is_managed = manager._secret_is_managed(self.test_secret_id) |
| |
| self.assertFalse(is_managed) |
| |
| @mock.patch('time.sleep') |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_create_secret_success(self, mock_client, mock_sleep): |
| """Test create_secret method success.""" |
| mock_response = mock.Mock() |
| mock_response.name = f"projects/{self.project_id}/secrets/{self.test_secret_id}" |
| mock_client.return_value.create_secret.return_value = mock_response |
| |
| # Mock the sequence of get_secret calls: first raises NotFound, then succeeds |
| call_count = [0] # Use list to make it mutable in nested function |
| |
| def get_secret_side_effect(*args, **kwargs): |
| call_count[0] += 1 |
| if call_count[0] == 1: |
| raise exceptions.NotFound("Not found") # _secret_is_managed returns False |
| else: |
| # For waiting loop - return a mock secret with proper labels |
| mock_secret = mock.Mock() |
| mock_secret.labels = {"created_by": SECRET_MANAGER_LABEL} |
| return mock_secret |
| |
| mock_client.return_value.get_secret.side_effect = get_secret_side_effect |
| |
| manager = SecretManager(self.project_id, self.logger) |
| result = manager.create_secret(self.test_secret_id) |
| |
| self.assertEqual(result, mock_response.name) |
| mock_client.return_value.create_secret.assert_called_once() |
| |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_create_secret_already_managed(self, mock_client): |
| """Test create_secret method when secret already managed.""" |
| mock_secret = mock.Mock() |
| mock_secret.labels = {"created_by": SECRET_MANAGER_LABEL} |
| mock_secret.name = f"projects/{self.project_id}/secrets/{self.test_secret_id}" |
| mock_client.return_value.get_secret.return_value = mock_secret |
| |
| # Mock the secret_path method to return the expected path |
| expected_path = f"projects/{self.project_id}/secrets/{self.test_secret_id}" |
| mock_client.return_value.secret_path.return_value = expected_path |
| |
| manager = SecretManager(self.project_id, self.logger) |
| result = manager.create_secret(self.test_secret_id) |
| |
| self.assertEqual(result, expected_path) |
| mock_client.return_value.create_secret.assert_not_called() |
| |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_get_secret_success(self, mock_client): |
| """Test get_secret method success.""" |
| mock_secret = mock.Mock() |
| mock_secret.labels = {"created_by": SECRET_MANAGER_LABEL} |
| mock_client.return_value.get_secret.return_value = mock_secret |
| |
| manager = SecretManager(self.project_id, self.logger) |
| result = manager.get_secret(self.test_secret_id) |
| |
| self.assertEqual(result, mock_secret) |
| |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_get_secret_not_exists(self, mock_client): |
| """Test get_secret method when secret doesn't exist.""" |
| mock_client.return_value.get_secret.side_effect = exceptions.NotFound("Not found") |
| |
| manager = SecretManager(self.project_id, self.logger) |
| |
| with self.assertRaises(ValueError): |
| manager.get_secret(self.test_secret_id) |
| |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_get_secret_not_managed(self, mock_client): |
| """Test get_secret method when secret is not managed.""" |
| mock_secret = mock.Mock() |
| mock_secret.labels = {"created_by": "other"} |
| mock_client.return_value.get_secret.return_value = mock_secret |
| |
| manager = SecretManager(self.project_id, self.logger) |
| |
| with self.assertRaises(ValueError): |
| manager.get_secret(self.test_secret_id) |
| |
| @mock.patch.object(SecretManager, '_secret_exists') |
| @mock.patch.object(SecretManager, '_secret_is_managed') |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_delete_secret_success(self, mock_client, mock_is_managed, mock_exists): |
| """Test delete_secret method success.""" |
| # Mock that secret is managed |
| mock_is_managed.return_value = True |
| |
| # Mock that secret doesn't exist after deletion |
| mock_exists.return_value = False |
| |
| manager = SecretManager(self.project_id, self.logger) |
| manager.delete_secret(self.test_secret_id) |
| |
| mock_client.return_value.delete_secret.assert_called_once() |
| |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_delete_secret_not_managed(self, mock_client): |
| """Test delete_secret method when secret is not managed.""" |
| mock_secret = mock.Mock() |
| mock_secret.labels = {"created_by": "other"} |
| mock_client.return_value.get_secret.return_value = mock_secret |
| |
| manager = SecretManager(self.project_id, self.logger) |
| |
| # The method should return early without raising exception when secret is not managed |
| manager.delete_secret(self.test_secret_id) |
| |
| # Verify that delete_secret was not called since the secret is not managed |
| mock_client.return_value.delete_secret.assert_not_called() |
| |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_is_different_user_access_same(self, mock_client): |
| """Test is_different_user_access method when access is the same.""" |
| mock_secret = mock.Mock() |
| mock_secret.labels = {"created_by": SECRET_MANAGER_LABEL} |
| mock_client.return_value.get_secret.return_value = mock_secret |
| |
| mock_policy = mock.Mock() |
| mock_binding = mock.Mock() |
| mock_binding.role = "roles/secretmanager.secretAccessor" |
| mock_binding.members = ["user:test@example.com", "user:test2@example.com"] |
| mock_policy.bindings = [mock_binding] |
| mock_client.return_value.get_iam_policy.return_value = mock_policy |
| |
| manager = SecretManager(self.project_id, self.logger) |
| is_different = manager.is_different_user_access( |
| self.test_secret_id, |
| ["test@example.com", "test2@example.com"] |
| ) |
| |
| self.assertFalse(is_different) |
| |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_is_different_user_access_different(self, mock_client): |
| """Test is_different_user_access method when access is different.""" |
| mock_secret = mock.Mock() |
| mock_secret.labels = {"created_by": SECRET_MANAGER_LABEL} |
| mock_client.return_value.get_secret.return_value = mock_secret |
| |
| mock_policy = mock.Mock() |
| mock_binding = mock.Mock() |
| mock_binding.role = "roles/secretmanager.secretAccessor" |
| mock_binding.members = ["user:different@example.com"] |
| mock_policy.bindings = [mock_binding] |
| mock_client.return_value.get_iam_policy.return_value = mock_policy |
| |
| manager = SecretManager(self.project_id, self.logger) |
| is_different = manager.is_different_user_access( |
| self.test_secret_id, |
| ["test@example.com"] |
| ) |
| |
| self.assertTrue(is_different) |
| |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_update_secret_access_success(self, mock_client): |
| """Test update_secret_access method success.""" |
| mock_secret = mock.Mock() |
| mock_secret.labels = {"created_by": SECRET_MANAGER_LABEL} |
| mock_client.return_value.get_secret.return_value = mock_secret |
| |
| mock_policy = mock.Mock() |
| mock_binding = mock.Mock() |
| mock_binding.role = "roles/secretmanager.secretAccessor" |
| mock_binding.members = ["user:old@example.com"] |
| mock_policy.bindings = [mock_binding] |
| mock_client.return_value.get_iam_policy.return_value = mock_policy |
| |
| manager = SecretManager(self.project_id, self.logger) |
| manager.update_secret_access(self.test_secret_id, ["new@example.com"]) |
| |
| mock_client.return_value.set_iam_policy.assert_called_once() |
| self.assertEqual(mock_binding.members, ["user:new@example.com"]) |
| |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_get_secret_versions_success(self, mock_client): |
| """Test _get_secret_versions method success.""" |
| mock_secret = mock.Mock() |
| mock_secret.labels = {"created_by": SECRET_MANAGER_LABEL} |
| mock_client.return_value.get_secret.return_value = mock_secret |
| |
| mock_versions = [mock.Mock(), mock.Mock()] |
| mock_client.return_value.list_secret_versions.return_value = mock_versions |
| |
| manager = SecretManager(self.project_id, self.logger) |
| versions = manager._get_secret_versions(self.test_secret_id) |
| |
| self.assertEqual(versions, mock_versions) |
| |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_secret_version_exists_true(self, mock_client): |
| """Test _secret_version_exists method when version exists.""" |
| mock_secret = mock.Mock() |
| mock_secret.labels = {"created_by": SECRET_MANAGER_LABEL} |
| mock_client.return_value.get_secret.return_value = mock_secret |
| |
| mock_version = mock.Mock() |
| mock_version.name = f"projects/{self.project_id}/secrets/{self.test_secret_id}/versions/1" |
| mock_client.return_value.list_secret_versions.return_value = [mock_version] |
| |
| manager = SecretManager(self.project_id, self.logger) |
| exists = manager._secret_version_exists(self.test_secret_id, "1") |
| |
| self.assertTrue(exists) |
| |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_secret_version_exists_false(self, mock_client): |
| """Test _secret_version_exists method when version doesn't exist.""" |
| mock_secret = mock.Mock() |
| mock_secret.labels = {"created_by": SECRET_MANAGER_LABEL} |
| mock_client.return_value.get_secret.return_value = mock_secret |
| |
| mock_version = mock.Mock() |
| mock_version.name = f"projects/{self.project_id}/secrets/{self.test_secret_id}/versions/2" |
| mock_client.return_value.list_secret_versions.return_value = [mock_version] |
| |
| manager = SecretManager(self.project_id, self.logger) |
| exists = manager._secret_version_exists(self.test_secret_id, "1") |
| |
| self.assertFalse(exists) |
| |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_secret_version_is_enabled_true(self, mock_client): |
| """Test _secret_version_is_enabled method when version is enabled.""" |
| mock_secret = mock.Mock() |
| mock_secret.labels = {"created_by": SECRET_MANAGER_LABEL} |
| mock_client.return_value.get_secret.return_value = mock_secret |
| |
| mock_version = mock.Mock() |
| mock_version.name = f"projects/{self.project_id}/secrets/{self.test_secret_id}/versions/1" |
| mock_version.state = secretmanager.SecretVersion.State.ENABLED |
| mock_client.return_value.list_secret_versions.return_value = [mock_version] |
| |
| manager = SecretManager(self.project_id, self.logger) |
| is_enabled = manager._secret_version_is_enabled(self.test_secret_id, "1") |
| |
| self.assertTrue(is_enabled) |
| |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_secret_version_is_enabled_false(self, mock_client): |
| """Test _secret_version_is_enabled method when version is not enabled.""" |
| mock_secret = mock.Mock() |
| mock_secret.labels = {"created_by": SECRET_MANAGER_LABEL} |
| mock_client.return_value.get_secret.return_value = mock_secret |
| |
| mock_version = mock.Mock() |
| mock_version.name = f"projects/{self.project_id}/secrets/{self.test_secret_id}/versions/1" |
| mock_version.state = secretmanager.SecretVersion.State.DISABLED |
| mock_client.return_value.list_secret_versions.return_value = [mock_version] |
| |
| manager = SecretManager(self.project_id, self.logger) |
| is_enabled = manager._secret_version_is_enabled(self.test_secret_id, "1") |
| |
| self.assertFalse(is_enabled) |
| |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_get_latest_secret_version_id_success(self, mock_client): |
| """Test _get_latest_secret_version_id method success.""" |
| mock_secret = mock.Mock() |
| mock_secret.labels = {"created_by": SECRET_MANAGER_LABEL} |
| mock_client.return_value.get_secret.return_value = mock_secret |
| |
| mock_version1 = mock.Mock() |
| mock_version1.name = f"projects/{self.project_id}/secrets/{self.test_secret_id}/versions/1" |
| mock_version1.state = secretmanager.SecretVersion.State.ENABLED |
| mock_version1.create_time.timestamp.return_value = 1000 |
| |
| mock_version2 = mock.Mock() |
| mock_version2.name = f"projects/{self.project_id}/secrets/{self.test_secret_id}/versions/2" |
| mock_version2.state = secretmanager.SecretVersion.State.ENABLED |
| mock_version2.create_time.timestamp.return_value = 2000 |
| |
| # Return versions in reverse order (latest first) as Google API does |
| mock_client.return_value.list_secret_versions.return_value = [mock_version2, mock_version1] |
| |
| manager = SecretManager(self.project_id, self.logger) |
| latest_id = manager._get_latest_secret_version_id(self.test_secret_id) |
| |
| self.assertEqual(latest_id, "2") |
| |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_get_latest_secret_version_id_no_enabled(self, mock_client): |
| """Test _get_latest_secret_version_id method when no enabled versions.""" |
| mock_secret = mock.Mock() |
| mock_secret.labels = {"created_by": SECRET_MANAGER_LABEL} |
| mock_client.return_value.get_secret.return_value = mock_secret |
| |
| mock_version = mock.Mock() |
| mock_version.name = f"projects/{self.project_id}/secrets/{self.test_secret_id}/versions/1" |
| mock_version.state = secretmanager.SecretVersion.State.DISABLED |
| mock_client.return_value.list_secret_versions.return_value = [mock_version] |
| |
| manager = SecretManager(self.project_id, self.logger) |
| |
| with self.assertRaises(ValueError): |
| manager._get_latest_secret_version_id(self.test_secret_id) |
| |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_is_key_rotation_due_true(self, mock_client): |
| """Test _is_key_rotation_due method when rotation is due.""" |
| past_date = datetime.now(timezone.utc) - timedelta(days=40) |
| mock_secret = mock.Mock() |
| mock_secret.labels = { |
| "created_by": SECRET_MANAGER_LABEL, |
| "last_version_created_at": past_date.strftime("%Y%m%d_%H%M%S") |
| } |
| mock_client.return_value.get_secret.return_value = mock_secret |
| |
| manager = SecretManager(self.project_id, self.logger, rotation_interval=30) |
| is_due = manager._is_key_rotation_due(self.test_secret_id) |
| |
| self.assertTrue(is_due) |
| |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_is_key_rotation_due_false(self, mock_client): |
| """Test _is_key_rotation_due method when rotation is not due.""" |
| recent_date = datetime.now(timezone.utc) - timedelta(days=10) |
| mock_secret = mock.Mock() |
| mock_secret.labels = { |
| "created_by": SECRET_MANAGER_LABEL, |
| "last_version_created_at": recent_date.strftime("%Y%m%d_%H%M%S") |
| } |
| mock_client.return_value.get_secret.return_value = mock_secret |
| |
| manager = SecretManager(self.project_id, self.logger, rotation_interval=30) |
| is_due = manager._is_key_rotation_due(self.test_secret_id) |
| |
| self.assertFalse(is_due) |
| |
| @mock.patch('time.sleep') |
| @mock.patch('google_crc32c.Checksum') |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_add_secret_version_success(self, mock_client, mock_checksum, mock_sleep): |
| """Test add_secret_version method success.""" |
| # Mock checksum |
| mock_checksum_instance = mock.Mock() |
| mock_checksum_instance.hexdigest.return_value = "abcd1234" |
| mock_checksum.return_value = mock_checksum_instance |
| |
| # Mock create_secret behavior - secret already exists |
| mock_secret = mock.Mock() |
| mock_secret.name = f"projects/{self.project_id}/secrets/{self.test_secret_id}" |
| mock_secret.labels = {"created_by": SECRET_MANAGER_LABEL} |
| mock_client.return_value.get_secret.return_value = mock_secret |
| |
| # Mock add_secret_version |
| mock_response = mock.Mock() |
| mock_response.name = f"projects/{self.project_id}/secrets/{self.test_secret_id}/versions/1" |
| mock_client.return_value.add_secret_version.return_value = mock_response |
| |
| # Mock list_secret_versions for waiting and disabling |
| mock_version = mock.Mock() |
| mock_version.name = f"projects/{self.project_id}/secrets/{self.test_secret_id}/versions/1" |
| mock_version.state = secretmanager.SecretVersion.State.ENABLED |
| mock_client.return_value.list_secret_versions.return_value = [mock_version] |
| |
| manager = SecretManager(self.project_id, self.logger) |
| result = manager.add_secret_version(self.test_secret_id, self.test_data_id, self.test_payload) |
| |
| self.assertEqual(result, mock_response.name) |
| mock_client.return_value.add_secret_version.assert_called_once() |
| mock_client.return_value.update_secret.assert_called_once() |
| |
| @mock.patch('google_crc32c.Checksum') |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_get_latest_secret_version_success(self, mock_client, mock_checksum): |
| """Test get_latest_secret_version method success.""" |
| mock_secret = mock.Mock() |
| mock_secret.labels = {"created_by": SECRET_MANAGER_LABEL} |
| mock_client.return_value.get_secret.return_value = mock_secret |
| |
| # Mock latest version |
| mock_version = mock.Mock() |
| mock_version.name = f"projects/{self.project_id}/secrets/{self.test_secret_id}/versions/1" |
| mock_version.state = secretmanager.SecretVersion.State.ENABLED |
| mock_version.create_time.timestamp.return_value = 1000 |
| mock_client.return_value.list_secret_versions.return_value = [mock_version] |
| |
| # Mock access_secret_version |
| mock_response = mock.Mock() |
| mock_response.payload.data = b"test-data:test-payload" |
| mock_response.payload.data_crc32c = int("abcd1234", 16) |
| mock_client.return_value.access_secret_version.return_value = mock_response |
| |
| # Mock checksum |
| mock_checksum_instance = mock.Mock() |
| mock_checksum_instance.hexdigest.return_value = "abcd1234" |
| mock_checksum.return_value = mock_checksum_instance |
| |
| manager = SecretManager(self.project_id, self.logger) |
| data_id, payload = manager.get_latest_secret_version(self.test_secret_id) |
| |
| self.assertEqual(data_id, "test-data") |
| self.assertEqual(payload, b"test-payload") |
| |
| @mock.patch('time.sleep') |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_enable_secret_version_success(self, mock_client, mock_sleep): |
| """Test enable_secret_version method success.""" |
| mock_secret = mock.Mock() |
| mock_secret.labels = {"created_by": SECRET_MANAGER_LABEL} |
| mock_client.return_value.get_secret.return_value = mock_secret |
| |
| # Mock version exists and is not enabled initially |
| mock_disabled_version = mock.Mock() |
| mock_disabled_version.name = f"projects/{self.project_id}/secrets/{self.test_secret_id}/versions/1" |
| mock_disabled_version.state = secretmanager.SecretVersion.State.DISABLED |
| |
| # Mock version becomes enabled after the operation |
| mock_enabled_version = mock.Mock() |
| mock_enabled_version.name = f"projects/{self.project_id}/secrets/{self.test_secret_id}/versions/1" |
| mock_enabled_version.state = secretmanager.SecretVersion.State.ENABLED |
| |
| # First call returns disabled version, second call returns enabled version |
| mock_client.return_value.list_secret_versions.side_effect = [ |
| [mock_disabled_version], # Initial check |
| [mock_enabled_version] # After enabling |
| ] |
| |
| # Mock enable response |
| mock_response = mock.Mock() |
| mock_response.name = mock_disabled_version.name |
| mock_response.state = secretmanager.SecretVersion.State.ENABLED |
| mock_client.return_value.enable_secret_version.return_value = mock_response |
| |
| manager = SecretManager(self.project_id, self.logger) |
| manager.enable_secret_version(self.test_secret_id, "1") |
| |
| mock_client.return_value.enable_secret_version.assert_called_once() |
| |
| @mock.patch('time.sleep') |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_disable_secret_version_success(self, mock_client, mock_sleep): |
| """Test disable_secret_version method success.""" |
| mock_secret = mock.Mock() |
| mock_secret.labels = {"created_by": SECRET_MANAGER_LABEL} |
| mock_client.return_value.get_secret.return_value = mock_secret |
| |
| # Mock version exists and is enabled initially |
| mock_enabled_version = mock.Mock() |
| mock_enabled_version.name = f"projects/{self.project_id}/secrets/{self.test_secret_id}/versions/1" |
| mock_enabled_version.state = secretmanager.SecretVersion.State.ENABLED |
| |
| # Mock version becomes disabled after the operation |
| mock_disabled_version = mock.Mock() |
| mock_disabled_version.name = f"projects/{self.project_id}/secrets/{self.test_secret_id}/versions/1" |
| mock_disabled_version.state = secretmanager.SecretVersion.State.DISABLED |
| |
| # First call returns enabled version, second call returns disabled version |
| mock_client.return_value.list_secret_versions.side_effect = [ |
| [mock_enabled_version], # Initial check |
| [mock_disabled_version] # After disabling |
| ] |
| |
| # Mock disable response |
| mock_response = mock.Mock() |
| mock_response.name = mock_enabled_version.name |
| mock_response.state = secretmanager.SecretVersion.State.DISABLED |
| mock_client.return_value.disable_secret_version.return_value = mock_response |
| |
| manager = SecretManager(self.project_id, self.logger) |
| manager.disable_secret_version(self.test_secret_id, "1") |
| |
| mock_client.return_value.disable_secret_version.assert_called_once() |
| |
| @mock.patch('time.sleep') |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_destroy_secret_version_success(self, mock_client, mock_sleep): |
| """Test destroy_secret_version method success.""" |
| mock_secret = mock.Mock() |
| mock_secret.labels = {"created_by": SECRET_MANAGER_LABEL} |
| mock_client.return_value.get_secret.return_value = mock_secret |
| |
| # Mock version exists and is enabled initially |
| mock_enabled_version = mock.Mock() |
| mock_enabled_version.name = f"projects/{self.project_id}/secrets/{self.test_secret_id}/versions/1" |
| mock_enabled_version.state = secretmanager.SecretVersion.State.ENABLED |
| |
| # Mock version becomes destroyed after the operation |
| mock_destroyed_version = mock.Mock() |
| mock_destroyed_version.name = f"projects/{self.project_id}/secrets/{self.test_secret_id}/versions/1" |
| mock_destroyed_version.state = secretmanager.SecretVersion.State.DESTROYED |
| |
| # Multiple calls to list_secret_versions for different operations |
| mock_client.return_value.list_secret_versions.side_effect = [ |
| [mock_enabled_version], # Initial check in _secret_version_is_enabled |
| [mock_enabled_version], # Check in enable_secret_version before enabling |
| [mock_enabled_version], # After enabling check |
| [mock_destroyed_version] # After destroying check |
| ] |
| |
| # Mock access_secret_version for getting data_id |
| mock_access_response = mock.Mock() |
| mock_access_response.payload.data = b"test-data:test-payload" |
| mock_client.return_value.access_secret_version.return_value = mock_access_response |
| |
| # Mock destroy response |
| mock_destroy_response = mock.Mock() |
| mock_destroy_response.name = mock_enabled_version.name |
| mock_destroy_response.state = secretmanager.SecretVersion.State.DESTROYED |
| mock_client.return_value.destroy_secret_version.return_value = mock_destroy_response |
| |
| # Mock enable response (needed since version is already enabled) |
| mock_enable_response = mock.Mock() |
| mock_enable_response.name = mock_enabled_version.name |
| mock_enable_response.state = secretmanager.SecretVersion.State.ENABLED |
| mock_client.return_value.enable_secret_version.return_value = mock_enable_response |
| |
| manager = SecretManager(self.project_id, self.logger) |
| data_id = manager.destroy_secret_version(self.test_secret_id, "1") |
| |
| self.assertEqual(data_id, "test-data") |
| mock_client.return_value.destroy_secret_version.assert_called_once() |
| |
| @mock.patch.object(SecretManager, 'destroy_secret_version') |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_purge_disabled_secret_versions_success(self, mock_client, mock_destroy): |
| """Test purge_disabled_secret_versions method success.""" |
| mock_secret = mock.Mock() |
| mock_secret.labels = {"created_by": SECRET_MANAGER_LABEL} |
| mock_client.return_value.get_secret.return_value = mock_secret |
| |
| # Mock old disabled version |
| old_time = datetime.now(timezone.utc) - timedelta(days=10) |
| mock_old_version = mock.Mock() |
| mock_old_version.name = f"projects/{self.project_id}/secrets/{self.test_secret_id}/versions/1" |
| mock_old_version.state = secretmanager.SecretVersion.State.DISABLED |
| mock_old_version.create_time.timestamp.return_value = old_time.timestamp() |
| |
| # Mock recent disabled version (within grace period) |
| recent_time = datetime.now(timezone.utc) - timedelta(days=2) |
| mock_recent_version = mock.Mock() |
| mock_recent_version.name = f"projects/{self.project_id}/secrets/{self.test_secret_id}/versions/2" |
| mock_recent_version.state = secretmanager.SecretVersion.State.DISABLED |
| mock_recent_version.create_time.timestamp.return_value = recent_time.timestamp() |
| |
| mock_client.return_value.list_secret_versions.return_value = [mock_old_version, mock_recent_version] |
| |
| # Mock destroy method to return data_id |
| mock_destroy.return_value = "old-data" |
| |
| manager = SecretManager(self.project_id, self.logger, grace_period=7) |
| data_ids = manager.purge_disabled_secret_versions(self.test_secret_id) |
| |
| self.assertEqual(data_ids, ["old-data"]) |
| mock_destroy.assert_called_once_with(self.test_secret_id, "1") |
| |
| @mock.patch.object(SecretManager, 'purge_disabled_secret_versions') |
| @mock.patch('secret_manager.secretmanager.SecretManagerServiceClient') |
| def test_cron_success(self, mock_client, mock_purge): |
| """Test cron method success.""" |
| # Mock _get_secret_ids |
| mock_secret1 = mock.Mock() |
| mock_secret1.name = f"projects/{self.project_id}/secrets/secret1" |
| mock_secret1.labels = {"created_by": SECRET_MANAGER_LABEL} |
| |
| mock_secret2 = mock.Mock() |
| mock_secret2.name = f"projects/{self.project_id}/secrets/secret2" |
| mock_secret2.labels = {"created_by": SECRET_MANAGER_LABEL} |
| |
| mock_client.return_value.list_secrets.return_value = [mock_secret1, mock_secret2] |
| |
| # Mock purge_disabled_secret_versions behavior |
| def mock_purge_side_effect(secret_id): |
| if secret_id == "secret1": |
| return ["purged-data"] |
| else: |
| return [] # secret2 has no versions to purge |
| |
| mock_purge.side_effect = mock_purge_side_effect |
| |
| manager = SecretManager(self.project_id, self.logger, grace_period=7) |
| result = manager.cron() |
| |
| self.assertIn("secret1", result) |
| self.assertEqual(result["secret1"], ["purged-data"]) |
| # secret2 should not be in result since it had no purged versions |
| self.assertNotIn("secret2", result) |
| |
| |
| |
| |
| # Integration tests (skipped unless environment variables are set) |
| @unittest.skipUnless( |
| 'GOOGLE_CLOUD_PROJECT' in os.environ, |
| "Skipping tests because environment variables are not set for Google Cloud project." |
| ) |
| class TestSecretManagerIntegration(unittest.TestCase): |
| """Integration tests for SecretManager with real Google Cloud Secret Manager client.""" |
| |
| def setUp(self): |
| """Set up test fixtures.""" |
| self.project_id = os.environ['GOOGLE_CLOUD_PROJECT'] |
| # Create a logger for integration tests |
| self.logger = logging.getLogger(__name__) |
| self.manager = SecretManager(self.project_id, self.logger, rotation_interval=0, grace_period=0, max_retries=3) |
| self.test_secret_id = f"integration-test-secret-{int(time.time())}" |
| self.test_data_id = f"integration-test-data-{int(time.time())}" |
| self.test_payload = b"integration-test-payload" |
| self.test_allowed_users = ["pabloem@google.com"] |
| |
| def tearDown(self): |
| """Tear down test fixtures.""" |
| # Clean up any secrets created during tests |
| try: |
| if self.test_secret_id in self.manager._get_secret_ids(): |
| self.manager.delete_secret(self.test_secret_id) |
| except Exception as e: |
| self.logger.warning(f"Failed to clean up test secret: {e}") |
| |
| def test_full_secret_lifecycle(self): |
| """Test creating, adding versions, rotating, and deleting a secret.""" |
| # Test creating a secret |
| self.manager.create_secret(self.test_secret_id) |
| self.assertTrue(self.manager._secret_exists(self.test_secret_id)) |
| |
| # Test allowing users to access the secret |
| self.manager.update_secret_access(self.test_secret_id, self.test_allowed_users) |
| self.assertFalse(self.manager.is_different_user_access(self.test_secret_id, self.test_allowed_users)) |
| |
| # Add first version (creates the secret) |
| version1 = self.manager.add_secret_version(self.test_secret_id, self.test_data_id, self.test_payload) |
| self.assertIsNotNone(version1) |
| |
| # Verify secret exists |
| secret = self.manager.get_secret(self.test_secret_id) |
| self.assertEqual(secret.labels["created_by"], SECRET_MANAGER_LABEL) |
| |
| # Add second version |
| version2 = self.manager.add_secret_version(self.test_secret_id, f"{self.test_data_id}-v2", b"second-payload") |
| self.assertIsNotNone(version2) |
| |
| # List versions |
| versions = self.manager._get_secret_versions(self.test_secret_id) |
| self.assertGreaterEqual(len(versions), 2) |
| |
| # Get latest version |
| retrieved_payload = self.manager.get_latest_secret_version(self.test_secret_id) |
| self.assertEqual(retrieved_payload, (f"{self.test_data_id}-v2", b"second-payload")) |
| |
| # Rotate secret |
| latest_version = self.manager.add_secret_version(self.test_secret_id, f"{self.test_data_id}-rotated", b"rotated-payload") |
| |
| # Verify latest version has rotated payload |
| latest_payload = self.manager.get_latest_secret_version(self.test_secret_id) |
| self.assertEqual(latest_payload, (f"{self.test_data_id}-rotated", b"rotated-payload")) |
| |
| # Verify all the other versions are disabled |
| versions = self.manager._get_secret_versions(self.test_secret_id) |
| for version in versions: |
| if version.name != latest_version: |
| self.assertEqual(version.state, secretmanager.SecretVersion.State.DISABLED) |
| |
| # Try cron method (should be no-op since grace period is 0) |
| cron_result = self.manager.cron() |
| self.assertIn(self.test_secret_id, cron_result) |
| self.assertEqual(len(cron_result[self.test_secret_id]), len(versions) - 1) # All but the latest should be purged |
| self.assertNotIn(f"{self.test_data_id}-rotated", cron_result[self.test_secret_id]) # Latest id should not be purged |
| |
| # Try to get the latest version after cron |
| latest_payload_after_cron = self.manager.get_latest_secret_version(self.test_secret_id) |
| self.assertEqual(latest_payload_after_cron, (f"{self.test_data_id}-rotated", b"rotated-payload")) |
| |
| # Delete secret |
| self.manager.delete_secret(self.test_secret_id) |
| |
| # Verify secret is removed from secret_ids |
| self.assertNotIn(self.test_secret_id, self.manager._get_secret_ids()) |
| |
| if __name__ == '__main__': |
| # Configure logging to reduce noise during testing |
| logging.getLogger('google.cloud').setLevel(logging.WARNING) |
| logging.getLogger('google.auth').setLevel(logging.WARNING) |
| |
| # Run the tests |
| unittest.main() |
