mirror of
https://github.com/qodo-ai/pr-agent.git
synced 2025-07-03 04:10:49 +08:00
feat: add AWS Secrets Manager Integration
This commit is contained in:
@ -203,6 +203,21 @@ For example: `GITHUB.WEBHOOK_SECRET` --> `GITHUB__WEBHOOK_SECRET`
|
||||
7. Go back to steps 8-9 of [Method 5](#run-as-a-github-app) with the function url as your Webhook URL.
|
||||
The Webhook URL would look like `https://<LAMBDA_FUNCTION_URL>/api/v1/github_webhooks`
|
||||
|
||||
### Using AWS Secrets Manager (Recommended)
|
||||
|
||||
For production Lambda deployments, use AWS Secrets Manager instead of environment variables:
|
||||
|
||||
1. Create a secret in AWS Secrets Manager with your configuration
|
||||
2. Add IAM permissions for `secretsmanager:GetSecretValue`
|
||||
3. Set the secret ARN in your Lambda environment:
|
||||
|
||||
```bash
|
||||
AWS_SECRETS_MANAGER__SECRET_ARN=arn:aws:secretsmanager:region:account:secret:name
|
||||
CONFIG__SECRET_PROVIDER=aws_secrets_manager
|
||||
```
|
||||
|
||||
For detailed setup instructions, see [AWS Secrets Manager Integration](../usage-guide/aws_secrets_manager.md).
|
||||
|
||||
---
|
||||
|
||||
## AWS CodeCommit Setup
|
||||
|
@ -249,4 +249,15 @@ ignore_pr_authors = ["my-special-bot-user", ...]
|
||||
Where the `ignore_pr_authors` is a list of usernames that you want to ignore.
|
||||
|
||||
!!! note
|
||||
There is one specific case where bots will receive an automatic response - when they generated a PR with a _failed test_. In that case, the [`ci_feedback`](https://qodo-merge-docs.qodo.ai/tools/ci_feedback/) tool will be invoked.
|
||||
There is one specific case where bots will receive an automatic response - when they generated a PR with a _failed test_. In that case, the [`ci_feedback`](https://qodo-merge-docs.qodo.ai/tools/ci_feedback/) tool will be invoked.
|
||||
|
||||
## Secret Management
|
||||
|
||||
For production deployments, consider using external secret management:
|
||||
|
||||
- **AWS Secrets Manager**: Recommended for AWS Lambda deployments
|
||||
- **Google Cloud Storage**: For Google Cloud environments
|
||||
|
||||
External secret providers automatically override environment variables at startup, providing enhanced security for sensitive information like API keys and webhook secrets.
|
||||
|
||||
See [Configuration Options](configuration_options.md#secret-providers) for setup details.
|
||||
|
111
docs/docs/usage-guide/aws_secrets_manager.md
Normal file
111
docs/docs/usage-guide/aws_secrets_manager.md
Normal file
@ -0,0 +1,111 @@
|
||||
# AWS Secrets Manager Integration
|
||||
|
||||
Securely manage sensitive information such as API keys and webhook secrets when running PR-Agent in AWS Lambda environments.
|
||||
|
||||
## Overview
|
||||
|
||||
AWS Secrets Manager integration allows you to:
|
||||
|
||||
- Store sensitive configuration in AWS Secrets Manager instead of environment variables
|
||||
- Automatically retrieve and apply secrets at application startup
|
||||
- Improve security for Lambda deployments
|
||||
- Centrally manage secrets across multiple environments
|
||||
|
||||
## Prerequisites
|
||||
|
||||
- AWS Lambda deployment of PR-Agent
|
||||
- AWS Secrets Manager access permissions
|
||||
- Boto3 library (already included in PR-Agent dependencies)
|
||||
|
||||
## Configuration
|
||||
|
||||
### Step 1: Create Secret in AWS Secrets Manager
|
||||
|
||||
Create a secret in AWS Secrets Manager with JSON format:
|
||||
|
||||
```json
|
||||
{
|
||||
"openai.key": "sk-...",
|
||||
"github.webhook_secret": "your-webhook-secret",
|
||||
"github.user_token": "ghp_...",
|
||||
"gitlab.personal_access_token": "glpat-..."
|
||||
}
|
||||
```
|
||||
|
||||
### Step 2: Configure PR-Agent
|
||||
|
||||
Add the following to your configuration:
|
||||
|
||||
```toml
|
||||
# configuration.toml
|
||||
[config]
|
||||
secret_provider = "aws_secrets_manager"
|
||||
|
||||
# .secrets.toml or environment variables
|
||||
[aws_secrets_manager]
|
||||
secret_arn = "arn:aws:secretsmanager:us-east-1:123456789012:secret:pr-agent-secrets-AbCdEf"
|
||||
region_name = "" # Optional: specific region (defaults to Lambda's region)
|
||||
```
|
||||
|
||||
### Step 3: Set IAM Permissions
|
||||
|
||||
Your Lambda execution role needs the following permissions:
|
||||
|
||||
```json
|
||||
{
|
||||
"Version": "2012-10-17",
|
||||
"Statement": [
|
||||
{
|
||||
"Effect": "Allow",
|
||||
"Action": ["secretsmanager:GetSecretValue"],
|
||||
"Resource": "arn:aws:secretsmanager:region:account:secret:pr-agent/*"
|
||||
}
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
## Environment Variable Mapping
|
||||
|
||||
Secrets Manager keys should use dot notation that maps to configuration sections:
|
||||
|
||||
| Secret Key | Configuration Section | Environment Variable |
|
||||
| ----------------------- | --------------------- | ------------------------ |
|
||||
| `openai.key` | `[openai]` | `OPENAI__KEY` |
|
||||
| `github.webhook_secret` | `[github]` | `GITHUB__WEBHOOK_SECRET` |
|
||||
| `github.user_token` | `[github]` | `GITHUB__USER_TOKEN` |
|
||||
|
||||
## Fallback Behavior
|
||||
|
||||
If AWS Secrets Manager is unavailable or misconfigured:
|
||||
|
||||
- PR-Agent will fall back to environment variables
|
||||
- A debug log message will be recorded
|
||||
- No service interruption occurs
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### Common Issues
|
||||
|
||||
1. **Permission Denied**: Ensure Lambda execution role has `secretsmanager:GetSecretValue` permission
|
||||
2. **Secret Not Found**: Verify the secret ARN is correct and exists in the specified region
|
||||
3. **JSON Parse Error**: Ensure the secret value is valid JSON format
|
||||
4. **Connection Issues**: Check network connectivity and AWS region settings
|
||||
|
||||
### Debug Logging
|
||||
|
||||
Enable debug logging to troubleshoot:
|
||||
|
||||
```toml
|
||||
[config]
|
||||
log_level = "DEBUG"
|
||||
```
|
||||
|
||||
Check CloudWatch logs for warning/error messages related to AWS Secrets Manager access.
|
||||
|
||||
## Security Best Practices
|
||||
|
||||
1. Use least-privilege IAM policies
|
||||
2. Rotate secrets regularly
|
||||
3. Use separate secrets for different environments
|
||||
4. Monitor CloudTrail for secret access
|
||||
5. Enable secret versioning for rollback capability
|
@ -16,6 +16,7 @@ nav:
|
||||
- Introduction: 'usage-guide/introduction.md'
|
||||
- Enabling a Wiki: 'usage-guide/enabling_a_wiki.md'
|
||||
- Configuration File: 'usage-guide/configuration_options.md'
|
||||
- AWS Secrets Manager: 'usage-guide/aws_secrets_manager.md'
|
||||
- Usage and Automation: 'usage-guide/automations_and_usage.md'
|
||||
- Managing Mail Notifications: 'usage-guide/mail_notifications.md'
|
||||
- Changing a Model: 'usage-guide/changing_a_model.md'
|
||||
|
@ -81,3 +81,65 @@ def _find_pyproject() -> Optional[Path]:
|
||||
pyproject_path = _find_pyproject()
|
||||
if pyproject_path is not None:
|
||||
get_settings().load_file(pyproject_path, env=f'tool.{PR_AGENT_TOML_KEY}')
|
||||
|
||||
|
||||
def apply_secrets_manager_config():
|
||||
"""
|
||||
Retrieve configuration from AWS Secrets Manager and override existing settings
|
||||
"""
|
||||
try:
|
||||
from pr_agent.secret_providers import get_secret_provider
|
||||
from pr_agent.log import get_logger
|
||||
|
||||
secret_provider = get_secret_provider()
|
||||
if not secret_provider:
|
||||
return
|
||||
|
||||
# Execute only when AWS Secrets Manager specific method is available
|
||||
if (hasattr(secret_provider, 'get_all_secrets') and
|
||||
get_settings().get("CONFIG.SECRET_PROVIDER") == 'aws_secrets_manager'):
|
||||
try:
|
||||
secrets = secret_provider.get_all_secrets()
|
||||
if secrets:
|
||||
apply_secrets_to_config(secrets)
|
||||
get_logger().info("Applied AWS Secrets Manager configuration")
|
||||
except Exception as e:
|
||||
get_logger().error(f"Failed to apply AWS Secrets Manager config: {e}")
|
||||
except Exception as e:
|
||||
# Fail silently when secret provider is not configured
|
||||
try:
|
||||
from pr_agent.log import get_logger
|
||||
get_logger().debug(f"Secret provider not configured: {e}")
|
||||
except:
|
||||
# Fail completely silently if log module is not available
|
||||
pass
|
||||
|
||||
|
||||
def apply_secrets_to_config(secrets: dict):
|
||||
"""
|
||||
Apply secret dictionary to configuration
|
||||
Configuration override with same pattern as Google Cloud Storage
|
||||
"""
|
||||
try:
|
||||
from pr_agent.log import get_logger
|
||||
except:
|
||||
# Do nothing if logging is not available
|
||||
def get_logger():
|
||||
class DummyLogger:
|
||||
def debug(self, msg): pass
|
||||
return DummyLogger()
|
||||
|
||||
for key, value in secrets.items():
|
||||
if '.' in key: # nested key like "openai.key"
|
||||
parts = key.split('.')
|
||||
if len(parts) == 2:
|
||||
section, setting = parts
|
||||
# Convert case to match Dynaconf pattern
|
||||
section_upper = section.upper()
|
||||
setting_upper = setting.upper()
|
||||
|
||||
# Set only when no existing value (prioritize environment variables)
|
||||
current_value = get_settings().get(f"{section_upper}.{setting_upper}")
|
||||
if current_value is None or current_value == "":
|
||||
get_settings().set(f"{section_upper}.{setting_upper}", value)
|
||||
get_logger().debug(f"Set {section}.{setting} from AWS Secrets Manager")
|
||||
|
@ -13,5 +13,12 @@ def get_secret_provider():
|
||||
return GoogleCloudStorageSecretProvider()
|
||||
except Exception as e:
|
||||
raise ValueError(f"Failed to initialize google_cloud_storage secret provider {provider_id}") from e
|
||||
elif provider_id == 'aws_secrets_manager':
|
||||
try:
|
||||
from pr_agent.secret_providers.aws_secrets_manager_provider import \
|
||||
AWSSecretsManagerProvider
|
||||
return AWSSecretsManagerProvider()
|
||||
except Exception as e:
|
||||
raise ValueError(f"Failed to initialize aws_secrets_manager secret provider {provider_id}") from e
|
||||
else:
|
||||
raise ValueError("Unknown SECRET_PROVIDER")
|
||||
|
79
pr_agent/secret_providers/aws_secrets_manager_provider.py
Normal file
79
pr_agent/secret_providers/aws_secrets_manager_provider.py
Normal file
@ -0,0 +1,79 @@
|
||||
import json
|
||||
import boto3
|
||||
from botocore.exceptions import ClientError
|
||||
|
||||
from pr_agent.config_loader import get_settings
|
||||
from pr_agent.log import get_logger
|
||||
from pr_agent.secret_providers.secret_provider import SecretProvider
|
||||
|
||||
|
||||
class AWSSecretsManagerProvider(SecretProvider):
|
||||
def __init__(self):
|
||||
try:
|
||||
# AWS credentials are automatically retrieved from environment variables or IAM roles
|
||||
# Region configuration is flexible like Google Cloud Storage pattern
|
||||
region_name = get_settings().get("aws_secrets_manager.region_name") or \
|
||||
get_settings().get("aws.AWS_REGION_NAME")
|
||||
if region_name:
|
||||
self.client = boto3.client('secretsmanager', region_name=region_name)
|
||||
else:
|
||||
self.client = boto3.client('secretsmanager')
|
||||
|
||||
# Require secret_arn similar to Google Cloud Storage pattern
|
||||
self.secret_arn = get_settings().aws_secrets_manager.secret_arn
|
||||
|
||||
except Exception as e:
|
||||
get_logger().error(f"Failed to initialize AWS Secrets Manager Provider: {e}")
|
||||
raise e
|
||||
|
||||
def get_secret(self, secret_name: str) -> str:
|
||||
"""
|
||||
Retrieve individual secret by name (for webhook tokens)
|
||||
Same error handling pattern as Google Cloud Storage
|
||||
"""
|
||||
try:
|
||||
response = self.client.get_secret_value(SecretId=secret_name)
|
||||
return response['SecretString']
|
||||
except Exception as e:
|
||||
get_logger().warning(f"Failed to get secret {secret_name} from AWS Secrets Manager: {e}")
|
||||
return ""
|
||||
|
||||
def get_all_secrets(self) -> dict:
|
||||
"""
|
||||
Retrieve all secrets for configuration override
|
||||
AWS Secrets Manager specific method (not available in Google Cloud Storage)
|
||||
"""
|
||||
try:
|
||||
response = self.client.get_secret_value(SecretId=self.secret_arn)
|
||||
return json.loads(response['SecretString'])
|
||||
except Exception as e:
|
||||
get_logger().error(f"Failed to get secrets from AWS Secrets Manager {self.secret_arn}: {e}")
|
||||
return {}
|
||||
|
||||
def store_secret(self, secret_name: str, secret_value: str):
|
||||
"""
|
||||
Same error handling pattern as Google Cloud Storage
|
||||
"""
|
||||
try:
|
||||
# Update existing secret
|
||||
self.client.update_secret(
|
||||
SecretId=secret_name,
|
||||
SecretString=secret_value
|
||||
)
|
||||
except ClientError as e:
|
||||
if e.response['Error']['Code'] == 'ResourceNotFoundException':
|
||||
# Create new secret if it doesn't exist
|
||||
try:
|
||||
self.client.create_secret(
|
||||
Name=secret_name,
|
||||
SecretString=secret_value
|
||||
)
|
||||
except Exception as create_error:
|
||||
get_logger().error(f"Failed to store secret {secret_name} in AWS Secrets Manager: {create_error}")
|
||||
raise create_error
|
||||
else:
|
||||
get_logger().error(f"Failed to store secret {secret_name} in AWS Secrets Manager: {e}")
|
||||
raise e
|
||||
except Exception as e:
|
||||
get_logger().error(f"Failed to store secret {secret_name} in AWS Secrets Manager: {e}")
|
||||
raise e
|
@ -5,6 +5,20 @@ from starlette_context.middleware import RawContextMiddleware
|
||||
|
||||
from pr_agent.servers.github_app import router
|
||||
|
||||
# Execute AWS Secrets Manager configuration override at module load time
|
||||
# Initialize with same pattern as Google Cloud Storage provider
|
||||
try:
|
||||
from pr_agent.config_loader import apply_secrets_manager_config
|
||||
apply_secrets_manager_config()
|
||||
except Exception as e:
|
||||
# Handle initialization failure silently (fallback to environment variables)
|
||||
try:
|
||||
from pr_agent.log import get_logger
|
||||
get_logger().debug(f"AWS Secrets Manager initialization failed, falling back to environment variables: {e}")
|
||||
except:
|
||||
# Fail completely silently if log module is not available
|
||||
pass
|
||||
|
||||
middleware = [Middleware(RawContextMiddleware)]
|
||||
app = FastAPI(middleware=middleware)
|
||||
app.include_router(router)
|
||||
|
@ -121,4 +121,18 @@ api_base = ""
|
||||
[aws]
|
||||
AWS_ACCESS_KEY_ID = ""
|
||||
AWS_SECRET_ACCESS_KEY = ""
|
||||
AWS_REGION_NAME = ""
|
||||
AWS_REGION_NAME = ""
|
||||
|
||||
# AWS Secrets Manager (for secure secret management in Lambda environments)
|
||||
[aws_secrets_manager]
|
||||
secret_arn = "" # The ARN of the AWS Secrets Manager secret containing PR-Agent configuration
|
||||
region_name = "" # Optional: specific AWS region (defaults to AWS_REGION_NAME or Lambda region)
|
||||
|
||||
# AWS Secrets Manager secret should contain JSON with configuration overrides:
|
||||
# Example secret value:
|
||||
# {
|
||||
# "openai.key": "sk-...",
|
||||
# "github.webhook_secret": "your-webhook-secret",
|
||||
# "github.user_token": "ghp_...",
|
||||
# "gitlab.personal_access_token": "glpat-..."
|
||||
# }
|
@ -39,7 +39,7 @@ allow_dynamic_context=true
|
||||
max_extra_lines_before_dynamic_context = 10 # will try to include up to 10 extra lines before the hunk in the patch, until we reach an enclosing function or class
|
||||
patch_extra_lines_before = 5 # Number of extra lines (+3 default ones) to include before each hunk in the patch
|
||||
patch_extra_lines_after = 1 # Number of extra lines (+3 default ones) to include after each hunk in the patch
|
||||
secret_provider=""
|
||||
secret_provider="" # "" (disabled), "google_cloud_storage", or "aws_secrets_manager" for secure secret management
|
||||
cli_mode=false
|
||||
ai_disclaimer_title="" # Pro feature, title for a collapsible disclaimer to AI outputs
|
||||
ai_disclaimer="" # Pro feature, full text for the AI disclaimer
|
||||
|
102
tests/unittest/test_aws_secrets_manager_provider.py
Normal file
102
tests/unittest/test_aws_secrets_manager_provider.py
Normal file
@ -0,0 +1,102 @@
|
||||
import json
|
||||
import pytest
|
||||
from unittest.mock import MagicMock, patch
|
||||
from botocore.exceptions import ClientError
|
||||
|
||||
from pr_agent.secret_providers.aws_secrets_manager_provider import AWSSecretsManagerProvider
|
||||
|
||||
|
||||
class TestAWSSecretsManagerProvider:
|
||||
|
||||
def _provider(self):
|
||||
"""Create provider following existing pattern"""
|
||||
with patch('pr_agent.secret_providers.aws_secrets_manager_provider.get_settings') as mock_get_settings, \
|
||||
patch('pr_agent.secret_providers.aws_secrets_manager_provider.boto3.client') as mock_boto3_client:
|
||||
|
||||
settings = MagicMock()
|
||||
settings.get.side_effect = lambda k, d=None: {
|
||||
'aws_secrets_manager.secret_arn': 'arn:aws:secretsmanager:us-east-1:123456789012:secret:test-secret',
|
||||
'aws_secrets_manager.region_name': 'us-east-1',
|
||||
'aws.AWS_REGION_NAME': 'us-east-1'
|
||||
}.get(k, d)
|
||||
settings.aws_secrets_manager.secret_arn = 'arn:aws:secretsmanager:us-east-1:123456789012:secret:test-secret'
|
||||
mock_get_settings.return_value = settings
|
||||
|
||||
# Mock boto3 client
|
||||
mock_client = MagicMock()
|
||||
mock_boto3_client.return_value = mock_client
|
||||
|
||||
provider = AWSSecretsManagerProvider()
|
||||
provider.client = mock_client # Set client directly for testing
|
||||
return provider, mock_client
|
||||
|
||||
# Positive test cases
|
||||
def test_get_secret_success(self):
|
||||
provider, mock_client = self._provider()
|
||||
mock_client.get_secret_value.return_value = {'SecretString': 'test-secret-value'}
|
||||
|
||||
result = provider.get_secret('test-secret-name')
|
||||
assert result == 'test-secret-value'
|
||||
mock_client.get_secret_value.assert_called_once_with(SecretId='test-secret-name')
|
||||
|
||||
def test_get_all_secrets_success(self):
|
||||
provider, mock_client = self._provider()
|
||||
secret_data = {'openai.key': 'sk-test', 'github.webhook_secret': 'webhook-secret'}
|
||||
mock_client.get_secret_value.return_value = {'SecretString': json.dumps(secret_data)}
|
||||
|
||||
result = provider.get_all_secrets()
|
||||
assert result == secret_data
|
||||
|
||||
# Negative test cases (following Google Cloud Storage pattern)
|
||||
def test_get_secret_failure(self):
|
||||
provider, mock_client = self._provider()
|
||||
mock_client.get_secret_value.side_effect = Exception("AWS error")
|
||||
|
||||
result = provider.get_secret('nonexistent-secret')
|
||||
assert result == "" # Confirm empty string is returned
|
||||
|
||||
def test_get_all_secrets_failure(self):
|
||||
provider, mock_client = self._provider()
|
||||
mock_client.get_secret_value.side_effect = Exception("AWS error")
|
||||
|
||||
result = provider.get_all_secrets()
|
||||
assert result == {} # Confirm empty dictionary is returned
|
||||
|
||||
def test_store_secret_update_existing(self):
|
||||
provider, mock_client = self._provider()
|
||||
mock_client.update_secret.return_value = {}
|
||||
|
||||
provider.store_secret('test-secret', 'test-value')
|
||||
mock_client.update_secret.assert_called_once_with(
|
||||
SecretId='test-secret',
|
||||
SecretString='test-value'
|
||||
)
|
||||
|
||||
def test_store_secret_create_new(self):
|
||||
provider, mock_client = self._provider()
|
||||
mock_client.update_secret.side_effect = ClientError(
|
||||
{'Error': {'Code': 'ResourceNotFoundException'}}, 'UpdateSecret'
|
||||
)
|
||||
mock_client.create_secret.return_value = {}
|
||||
|
||||
provider.store_secret('new-secret', 'test-value')
|
||||
mock_client.create_secret.assert_called_once_with(
|
||||
Name='new-secret',
|
||||
SecretString='test-value'
|
||||
)
|
||||
|
||||
def test_init_failure_invalid_config(self):
|
||||
with patch('pr_agent.secret_providers.aws_secrets_manager_provider.get_settings') as mock_get_settings:
|
||||
settings = MagicMock()
|
||||
settings.aws_secrets_manager.secret_arn = None # Configuration error
|
||||
mock_get_settings.return_value = settings
|
||||
|
||||
with pytest.raises(Exception):
|
||||
AWSSecretsManagerProvider()
|
||||
|
||||
def test_store_secret_failure(self):
|
||||
provider, mock_client = self._provider()
|
||||
mock_client.update_secret.side_effect = Exception("AWS error")
|
||||
|
||||
with pytest.raises(Exception):
|
||||
provider.store_secret('test-secret', 'test-value')
|
120
tests/unittest/test_config_loader_secrets.py
Normal file
120
tests/unittest/test_config_loader_secrets.py
Normal file
@ -0,0 +1,120 @@
|
||||
import pytest
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from pr_agent.config_loader import apply_secrets_manager_config, apply_secrets_to_config
|
||||
|
||||
|
||||
class TestConfigLoaderSecrets:
|
||||
|
||||
def test_apply_secrets_manager_config_success(self):
|
||||
with patch('pr_agent.config_loader.get_secret_provider') as mock_get_provider, \
|
||||
patch('pr_agent.config_loader.apply_secrets_to_config') as mock_apply_secrets, \
|
||||
patch('pr_agent.config_loader.get_settings') as mock_get_settings:
|
||||
|
||||
# Mock secret provider
|
||||
mock_provider = MagicMock()
|
||||
mock_provider.get_all_secrets.return_value = {'openai.key': 'sk-test'}
|
||||
mock_get_provider.return_value = mock_provider
|
||||
|
||||
# Mock settings
|
||||
settings = MagicMock()
|
||||
settings.get.return_value = "aws_secrets_manager"
|
||||
mock_get_settings.return_value = settings
|
||||
|
||||
apply_secrets_manager_config()
|
||||
|
||||
mock_apply_secrets.assert_called_once_with({'openai.key': 'sk-test'})
|
||||
|
||||
def test_apply_secrets_manager_config_no_provider(self):
|
||||
with patch('pr_agent.config_loader.get_secret_provider') as mock_get_provider:
|
||||
mock_get_provider.return_value = None
|
||||
|
||||
# Confirm no exception is raised
|
||||
apply_secrets_manager_config()
|
||||
|
||||
def test_apply_secrets_manager_config_not_aws(self):
|
||||
with patch('pr_agent.config_loader.get_secret_provider') as mock_get_provider, \
|
||||
patch('pr_agent.config_loader.get_settings') as mock_get_settings:
|
||||
|
||||
# Mock Google Cloud Storage provider
|
||||
mock_provider = MagicMock()
|
||||
mock_get_provider.return_value = mock_provider
|
||||
|
||||
# Mock settings (Google Cloud Storage)
|
||||
settings = MagicMock()
|
||||
settings.get.return_value = "google_cloud_storage"
|
||||
mock_get_settings.return_value = settings
|
||||
|
||||
# Confirm execution is skipped for non-AWS Secrets Manager
|
||||
apply_secrets_manager_config()
|
||||
|
||||
# Confirm get_all_secrets is not called
|
||||
assert not hasattr(mock_provider, 'get_all_secrets') or \
|
||||
not mock_provider.get_all_secrets.called
|
||||
|
||||
def test_apply_secrets_to_config_nested_keys(self):
|
||||
with patch('pr_agent.config_loader.get_settings') as mock_get_settings:
|
||||
settings = MagicMock()
|
||||
settings.get.return_value = None # No existing value
|
||||
settings.set = MagicMock()
|
||||
mock_get_settings.return_value = settings
|
||||
|
||||
secrets = {
|
||||
'openai.key': 'sk-test',
|
||||
'github.webhook_secret': 'webhook-secret'
|
||||
}
|
||||
|
||||
apply_secrets_to_config(secrets)
|
||||
|
||||
# Confirm settings are applied correctly
|
||||
settings.set.assert_any_call('OPENAI.KEY', 'sk-test')
|
||||
settings.set.assert_any_call('GITHUB.WEBHOOK_SECRET', 'webhook-secret')
|
||||
|
||||
def test_apply_secrets_to_config_existing_value_preserved(self):
|
||||
with patch('pr_agent.config_loader.get_settings') as mock_get_settings:
|
||||
settings = MagicMock()
|
||||
settings.get.return_value = "existing-value" # Existing value present
|
||||
settings.set = MagicMock()
|
||||
mock_get_settings.return_value = settings
|
||||
|
||||
secrets = {'openai.key': 'sk-test'}
|
||||
|
||||
apply_secrets_to_config(secrets)
|
||||
|
||||
# Confirm settings are not overridden when existing value present
|
||||
settings.set.assert_not_called()
|
||||
|
||||
def test_apply_secrets_to_config_single_key(self):
|
||||
with patch('pr_agent.config_loader.get_settings') as mock_get_settings:
|
||||
settings = MagicMock()
|
||||
settings.get.return_value = None
|
||||
settings.set = MagicMock()
|
||||
mock_get_settings.return_value = settings
|
||||
|
||||
secrets = {'simple_key': 'simple_value'}
|
||||
|
||||
apply_secrets_to_config(secrets)
|
||||
|
||||
# Confirm non-dot notation keys are ignored
|
||||
settings.set.assert_not_called()
|
||||
|
||||
def test_apply_secrets_to_config_multiple_dots(self):
|
||||
with patch('pr_agent.config_loader.get_settings') as mock_get_settings:
|
||||
settings = MagicMock()
|
||||
settings.get.return_value = None
|
||||
settings.set = MagicMock()
|
||||
mock_get_settings.return_value = settings
|
||||
|
||||
secrets = {'section.subsection.key': 'value'}
|
||||
|
||||
apply_secrets_to_config(secrets)
|
||||
|
||||
# Confirm keys with multiple dots are ignored
|
||||
settings.set.assert_not_called()
|
||||
|
||||
def test_apply_secrets_manager_config_exception_handling(self):
|
||||
with patch('pr_agent.config_loader.get_secret_provider') as mock_get_provider:
|
||||
mock_get_provider.side_effect = Exception("Provider error")
|
||||
|
||||
# Confirm processing continues even when exception occurs
|
||||
apply_secrets_manager_config() # Confirm no exception is raised
|
69
tests/unittest/test_secret_provider_factory.py
Normal file
69
tests/unittest/test_secret_provider_factory.py
Normal file
@ -0,0 +1,69 @@
|
||||
import pytest
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from pr_agent.secret_providers import get_secret_provider
|
||||
|
||||
|
||||
class TestSecretProviderFactory:
|
||||
|
||||
def test_get_secret_provider_none_when_not_configured(self):
|
||||
with patch('pr_agent.secret_providers.get_settings') as mock_get_settings:
|
||||
settings = MagicMock()
|
||||
settings.get.return_value = None
|
||||
mock_get_settings.return_value = settings
|
||||
|
||||
result = get_secret_provider()
|
||||
assert result is None
|
||||
|
||||
def test_get_secret_provider_google_cloud_storage(self):
|
||||
with patch('pr_agent.secret_providers.get_settings') as mock_get_settings:
|
||||
settings = MagicMock()
|
||||
settings.get.return_value = "google_cloud_storage"
|
||||
settings.config.secret_provider = "google_cloud_storage"
|
||||
mock_get_settings.return_value = settings
|
||||
|
||||
with patch('pr_agent.secret_providers.google_cloud_storage_secret_provider.GoogleCloudStorageSecretProvider') as MockProvider:
|
||||
mock_instance = MagicMock()
|
||||
MockProvider.return_value = mock_instance
|
||||
|
||||
result = get_secret_provider()
|
||||
assert result is mock_instance
|
||||
MockProvider.assert_called_once()
|
||||
|
||||
def test_get_secret_provider_aws_secrets_manager(self):
|
||||
with patch('pr_agent.secret_providers.get_settings') as mock_get_settings:
|
||||
settings = MagicMock()
|
||||
settings.get.return_value = "aws_secrets_manager"
|
||||
settings.config.secret_provider = "aws_secrets_manager"
|
||||
mock_get_settings.return_value = settings
|
||||
|
||||
with patch('pr_agent.secret_providers.aws_secrets_manager_provider.AWSSecretsManagerProvider') as MockProvider:
|
||||
mock_instance = MagicMock()
|
||||
MockProvider.return_value = mock_instance
|
||||
|
||||
result = get_secret_provider()
|
||||
assert result is mock_instance
|
||||
MockProvider.assert_called_once()
|
||||
|
||||
def test_get_secret_provider_unknown_provider(self):
|
||||
with patch('pr_agent.secret_providers.get_settings') as mock_get_settings:
|
||||
settings = MagicMock()
|
||||
settings.get.return_value = "unknown_provider"
|
||||
settings.config.secret_provider = "unknown_provider"
|
||||
mock_get_settings.return_value = settings
|
||||
|
||||
with pytest.raises(ValueError, match="Unknown SECRET_PROVIDER"):
|
||||
get_secret_provider()
|
||||
|
||||
def test_get_secret_provider_initialization_error(self):
|
||||
with patch('pr_agent.secret_providers.get_settings') as mock_get_settings:
|
||||
settings = MagicMock()
|
||||
settings.get.return_value = "aws_secrets_manager"
|
||||
settings.config.secret_provider = "aws_secrets_manager"
|
||||
mock_get_settings.return_value = settings
|
||||
|
||||
with patch('pr_agent.secret_providers.aws_secrets_manager_provider.AWSSecretsManagerProvider') as MockProvider:
|
||||
MockProvider.side_effect = Exception("Initialization failed")
|
||||
|
||||
with pytest.raises(ValueError, match="Failed to initialize aws_secrets_manager secret provider"):
|
||||
get_secret_provider()
|
Reference in New Issue
Block a user