Merge pull request #3 from narenaryan/test/add-unit-tests

tests: add unit tests for vaults

.github/workflows/test.yml

@@ -0,0 +1,36 @@
+name: Run unit tests
+
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+    branches:
+      - "*"
+env:
+  PYTHONPATH: ./src # Needed for tests to discover whispr package
+jobs:
+  test:
+    runs-on: ubuntu-22.04
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v3
+
+      - name: Set up Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: "3.10"
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements_test.txt
+          pip install coveralls
+
+      - name: Run Test
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        run: |
+          pytest --cov=whispr tests
+          coveralls

.gitignore

@@ -28,3 +28,4 @@ lib64/
 # Secrets
 .env
 *.creds
+.coverage*

README.md

@@ -1,10 +1,11 @@
 [![Downloads](https://static.pepy.tech/badge/whispr/month)](https://pepy.tech/project/whispr)
+[![Coverage Status](https://coveralls.io/repos/github/narenaryan/whispr/badge.svg)](https://coveralls.io/github/narenaryan/whispr)
 
 # Whispr
 
 ![Logo](./logo.png)
 
-Whispr (Pronounced as whisp-r) is a CLI tool to safely inject secrets from your favorite secret vault (Ex: AWS Secrets Manager, Azure Key Vault etc.) into your app's environment. This is very useful for enabling secure local software development.
+Whispr (Pronounced as whisper) is a CLI tool to safely inject secrets from your favorite secret vault (Ex: AWS Secrets Manager, Azure Key Vault etc.) into your app's environment. This is very useful for enabling secure local software development.
 
 Whispr uses keys (with empty values) specified in a `.env` file and fetches respective secrets from a vault, and sets them as environment variables before launching an application.
 
@@ -45,7 +46,7 @@ pip install whispr
 
 Run `whispr init <vault_type>` in your terminal to create a `whispr.yaml` file in your project root. This file will store your configuration settings.
 
-The available vault types are: `aws`, `azure`, and `gcp`.  
+The available vault types are: `aws`, `azure`, and `gcp`.
 
 **Example whispr.yaml contents (For: AWS):**
 ```yaml
@@ -71,7 +72,7 @@ POSTGRES_PASSWORD=
 
 *   Authenticate to AWS using Short-term credentials.
 *   Alternatively, set temporary AWS credentials using a config file or environment variables.
-  
+
 **Note**: Use respective authentication methods for other vaults.
 
 ## Launch any Application using Whispr

pyproject.toml

@@ -40,7 +40,7 @@ dependencies = [
     "structlog==24.4.0",
     "azure-keyvault==4.2.0",
     "azure-identity==1.19.0",
-    "hvac==2.3.0"
+    "hvac==2.3.0",
 ]
 [project.urls]
 Documentation = "https://github.com/narenaryan/whispr/blob/main/README.md"
@@ -60,9 +60,10 @@ python = ["3.8", "3.9", "3.10", "3.11", "3.12"]
 dependencies = ["mypy>=1.0.0"]
 
 [tool.coverage.run]
-source_pkgs = ["whispr", "tests"]
+source_pkgs = ["whispr"]
 branch = true
 parallel = true
+relative_files = true
 omit = ["src/whispr/__about__.py"]
 
 [tool.coverage.paths]

requirements_test.txt

@@ -0,0 +1,4 @@
+-r requirements.txt
+coverage==7.6.4
+pytest==8.3.3
+pytest-cov==5.0.0

tests/test_aws.py

@@ -0,0 +1,79 @@
+"""Tests for AWS module"""
+
+import unittest
+from unittest.mock import Mock, MagicMock, patch
+
+import botocore.exceptions
+import structlog
+
+from whispr.vault import SimpleVault
+from whispr.aws import AWSVault
+
+
+class AWSVaultTestCase(unittest.TestCase):
+    """Unit tests for AWSVault class, which fetches secrets from AWS Secrets Manager."""
+
+    def setUp(self):
+        """Set up mocks for logger and AWS client before each test."""
+        self.mock_logger = MagicMock()
+        self.mock_client = MagicMock()
+        self.vault = AWSVault(logger=self.mock_logger, client=self.mock_client)
+
+    def test_initialization(self):
+        """Test that AWSVault initializes with logger and client correctly."""
+        self.assertEqual(self.vault.logger, self.mock_logger)
+        self.assertEqual(self.vault.client, self.mock_client)
+
+    def test_fetch_secrets_success(self):
+        """Test successful fetch of secrets from AWS Secrets Manager."""
+        self.mock_client.get_secret_value.return_value = {
+            "SecretString": '{"key": "value"}'
+        }
+        result = self.vault.fetch_secrets("test_secret")
+        self.assertEqual(result, '{"key": "value"}')
+        self.mock_client.get_secret_value.assert_called_with(SecretId="test_secret")
+
+    def test_fetch_secrets_resource_not_found(self):
+        """Test fetch_secrets handles ResourceNotFoundException gracefully."""
+        # Set up the client to raise ResourceNotFoundException
+        self.mock_client.get_secret_value.side_effect = botocore.exceptions.ClientError(
+            {"Error": {"Code": "ResourceNotFoundException"}}, "get_secret_value"
+        )
+
+        result = self.vault.fetch_secrets("non_existent_secret")
+        self.assertEqual(result, "")
+        self.mock_logger.error.assert_called_with(
+            "The secret is not found on AWS. Did you set the right AWS_DEFAULT_REGION ?",
+            secret_name="non_existent_secret",
+        )
+
+    @patch("whispr.aws.AWSVault.fetch_secrets")
+    def test_fetch_secrets_unrecognized_client_exception(self, mock_fetch_secrets):
+        """Test fetch_secrets handles UnrecognizedClientException gracefully."""
+        mock_fetch_secrets.side_effect = botocore.exceptions.ClientError(
+            {"Error": {"Code": "UnrecognizedClientException"}}, "get_secret_value"
+        )
+
+        with self.assertRaises(botocore.exceptions.ClientError):
+            result = self.vault.fetch_secrets("incorrect_credentials_secret")
+            self.assertEqual(result, "")
+            self.mock_logger.error.assert_called_with(
+                "Incorrect AWS credentials set for operation. Please verify them and retry."
+            )
+
+    def test_fetch_secrets_generic_exception(self):
+        """Test fetch_secrets raises exception and logs an error for generic exceptions."""
+        # Set up the client to raise a generic exception
+        exception_message = "Some generic error"
+        self.mock_client.get_secret_value.side_effect = Exception(exception_message)
+
+        with self.assertRaises(Exception) as context:
+            self.vault.fetch_secrets("generic_error_secret")
+        self.assertEqual(str(context.exception), exception_message)
+
+        # Extract the actual call to the logger and check its arguments
+        self.assertTrue(self.mock_logger.error.called)
+        error_call = self.mock_logger.error.call_args
+        self.assertEqual(error_call[0][0], "Error fetching secret")
+        self.assertIsInstance(error_call[1]["error"], Exception)
+        self.assertEqual(str(error_call[1]["error"]), exception_message)

tests/test_azure.py

@@ -0,0 +1,74 @@
+"""Tests for Azure module"""
+
+import unittest
+from unittest.mock import Mock, MagicMock
+
+import structlog
+from azure.core.exceptions import ResourceNotFoundError
+
+from whispr.vault import SimpleVault
+from whispr.azure import AzureVault
+
+
+class AzureVaultTestCase(unittest.TestCase):
+    """Unit tests for AzureVault class, which fetches secrets from Azure Key Vault."""
+
+    def setUp(self):
+        """Set up mocks for logger, Azure client, and vault URL before each test."""
+        self.mock_logger = MagicMock()
+        self.mock_client = MagicMock()
+        self.vault_url = "https://example-vault.vault.azure.net/"
+        self.vault = AzureVault(
+            logger=self.mock_logger, client=self.mock_client, vault_url=self.vault_url
+        )
+
+    def test_initialization(self):
+        """Test that AzureVault initializes with logger, client, and vault_url correctly."""
+        self.assertEqual(self.vault.logger, self.mock_logger)
+        self.assertEqual(self.vault.client, self.mock_client)
+        self.assertEqual(self.vault.vault_url, self.vault_url)
+
+    def test_fetch_secrets_success(self):
+        """Test successful fetch of secrets from Azure Key Vault."""
+        # Mock the client response
+        mock_secret = MagicMock()
+        mock_secret.value = '{"key": "value"}'
+        self.mock_client.get_secret.return_value = mock_secret
+
+        result = self.vault.fetch_secrets("test_secret")
+        self.assertEqual(result, '{"key": "value"}')
+        self.mock_logger.info.assert_called_with(
+            "Successfully fetched secret: test_secret"
+        )
+        self.mock_client.get_secret.assert_called_with("test_secret")
+
+    def test_fetch_secrets_resource_not_found(self):
+        """Test fetch_secrets handles ResourceNotFoundError gracefully."""
+        # Set up the client to raise ResourceNotFoundError
+        self.mock_client.get_secret.side_effect = ResourceNotFoundError(
+            "Secret not found"
+        )
+
+        result = self.vault.fetch_secrets("non_existent_secret")
+        self.assertEqual(result, "")
+        self.mock_logger.error.assert_called_with(
+            "The given secret: non_existent_secret is not found on azure vault. Please check the secret name, vault name or subscription ID."
+        )
+
+    def test_fetch_secrets_generic_exception(self):
+        """Test fetch_secrets raises exception and logs an error for generic exceptions."""
+        # Set up the client to raise a generic exception
+        exception_message = "Some generic error"
+        self.mock_client.get_secret.side_effect = Exception(exception_message)
+
+        with self.assertRaises(Exception) as context:
+            self.vault.fetch_secrets("generic_error_secret")
+        self.assertEqual(str(context.exception), exception_message)
+
+        # Extract the actual call to the logger and check its arguments
+        self.assertTrue(self.mock_logger.error.called)
+        error_call = self.mock_logger.error.call_args
+        self.assertEqual(
+            error_call[0][0],
+            f"Error fetching secret: generic_error_secret, Error: {exception_message}",
+        )

tests/test_gcp.py

@@ -0,0 +1,72 @@
+"""Tests for GCP module"""
+
+import unittest
+from unittest.mock import Mock, patch, MagicMock
+
+import google.api_core.exceptions
+import structlog
+
+from whispr.vault import SimpleVault
+from whispr.gcp import GCPVault
+
+
+class GCPVaultTestCase(unittest.TestCase):
+    """Unit tests for GCPVault class, which fetches secrets from GCP Secrets Manager."""
+
+    def setUp(self):
+        """Set up mocks for logger, GCP client, and project_id before each test."""
+        self.mock_logger = MagicMock()
+        self.mock_client = MagicMock()
+        self.project_id = "test_project_id"
+        self.vault = GCPVault(
+            logger=self.mock_logger, client=self.mock_client, project_id=self.project_id
+        )
+
+    def test_initialization(self):
+        """Test that GCPVault initializes with logger, client, and project_id correctly."""
+        self.assertEqual(self.vault.logger, self.mock_logger)
+        self.assertEqual(self.vault.client, self.mock_client)
+        self.assertEqual(self.vault.project_id, self.project_id)
+
+    def test_fetch_secrets_success(self):
+        """Test successful fetch of secrets from GCP Secrets Manager."""
+        # Mock the client response
+        mock_response = MagicMock()
+        mock_response.payload.data.decode.return_value = '{"key": "value"}'
+        self.mock_client.access_secret_version.return_value = mock_response
+
+        result = self.vault.fetch_secrets("test_secret")
+        self.assertEqual(result, '{"key": "value"}')
+        self.mock_logger.info.assert_called_with(
+            "Successfully fetched gcp secret: projects/test_project_id/secrets/test_secret/versions/latest"
+        )
+        self.mock_client.access_secret_version.assert_called_with(
+            name="projects/test_project_id/secrets/test_secret/versions/latest"
+        )
+
+    def test_fetch_secrets_not_found(self):
+        """Test fetch_secrets handles NotFound exception gracefully."""
+        # Set up the client to raise NotFound exception
+        self.mock_client.access_secret_version.side_effect = (
+            google.api_core.exceptions.NotFound("Secret not found")
+        )
+
+        result = self.vault.fetch_secrets("non_existent_secret")
+        self.assertEqual(result, "")
+        self.mock_logger.error.assert_called_with(
+            "The given secret: projects/test_project_id/secrets/non_existent_secret/versions/latest is not found on gcp vault."
+        )
+
+    def test_fetch_secrets_generic_exception(self):
+        """Test fetch_secrets handles generic exceptions gracefully."""
+        # Set up the client to raise a generic exception
+        exception_message = "Some generic error"
+        self.mock_client.access_secret_version.side_effect = Exception(
+            exception_message
+        )
+
+        result = self.vault.fetch_secrets("generic_error_secret")
+        self.assertEqual(result, "")
+        self.mock_logger.error.assert_called_with(
+            f"Error encountered while fetching secret: projects/test_project_id/secrets/generic_error_secret/versions/latest, Error: {exception_message}"
+        )

tests/test_vault.py

@@ -0,0 +1,39 @@
+"""Tests for Vault module"""
+
+import unittest
+from unittest.mock import Mock, patch
+
+import structlog
+
+from whispr.vault import SimpleVault
+
+
+class SimpleVaultTestCase(unittest.TestCase):
+    """Tests for Vault"""
+
+    def setUp(self):
+        # Mock logger and client to use in tests
+        self.mock_logger = Mock(spec=structlog.BoundLogger)
+        self.mock_client = Mock()
+
+        # Subclass SimpleVault since it's abstract, only for testing
+        class TestVault(SimpleVault):
+            def fetch_secrets(self, secret_name: str) -> str:
+                # Provide a simple implementation for the abstract method
+                return "test_secret"
+
+        self.vault = TestVault(logger=self.mock_logger, client=self.mock_client)
+
+    @patch.object(
+        SimpleVault, "__abstractmethods__", set()
+    )  # This allows instantiation of SimpleVault directly if needed
+    def test_initialization(self):
+        """Test if the SimpleVault initializes with logger and client."""
+        self.assertEqual(self.vault.logger, self.mock_logger)
+        self.assertEqual(self.vault.client, self.mock_client)
+
+    def test_fetch_secrets(self):
+        """Test the fetch_secrets method to ensure it returns the expected result."""
+        secret_name = "my_secret"
+        result = self.vault.fetch_secrets(secret_name)
+        self.assertEqual(result, "test_secret")