KEY INSIGHT
CI/CD pipelines for Nigerian SaaS must handle the unique requirements of multi-tenant deployments while accounting for payment processor integration testing and varying tenant configurations.
CI/CD for SaaS requires more than simple build and deploy. The pipeline must handle multi-tenant configuration, payment processor webhooks, and compliance requirements specific to Nigerian operations.
```python
import yaml
from dataclasses import dataclass
from typing import Optional
import subprocess
@dataclass
class PipelineStage:
name: str
commands: list[str]
timeout: int = 600
allow_failure: bool = False
class SaaSCICDPipeline:
"""CI/CD pipeline for multi-tenant SaaS."""
def __init__(self, config_dir: str):
self.config_dir = config_dir
self.stages = self._load_pipeline_config()
def execute_pipeline(
self,
environment: str,
skip_stages: list = None
) -> dict:
"""Execute full CI/CD pipeline."""
results = []
skip_stages = skip_stages or []
for stage_config in self.stages:
stage = PipelineStage(**stage_config)
if stage.name in skip_stages:
results.append({
'stage': stage.name,
'status': 'skipped'
})
continue
if environment == 'production' and stage.name in ['security-scan', 'integration-tests']:
pass # Required for production
elif environment == 'staging' and stage.name in ['penetration-test']:
continue # Skip in staging
try:
result = self._execute_stage(stage, environment)
results.append(result)
if not stage.allow_failure and result['status'] == 'failed':
return {'status': 'failed', 'results': results}
except Exception as e:
results.append({
'stage': stage.name,
'status': 'failed',
'error': str(e)
})
return {'status': 'failed', 'results': results}
return {'status': 'success', 'results': results}
def _execute_stage(self, stage: PipelineStage, environment: str) -> dict:
"""Execute a single pipeline stage."""
start_time = datetime.utcnow()
for command in stage.commands:
env_vars = self._get_environment_vars(environment)
result = subprocess.run(
command,
shell=True,
capture_output=True,
timeout=stage.timeout,
env=env_vars
)
if result.returncode != 0:
return {
'stage': stage.name,
'status': 'failed',
'error': result.stderr.decode(),
'duration': (datetime.utcnow() - start_time).seconds
}
return {
'stage': stage.name,
'status': 'success',
'duration': (datetime.utcnow() - start_time).seconds
}
def _get_environment_vars(self, environment: str) -> dict:
"""Get environment-specific variables."""
secrets = self._load_secrets(environment)
return {
'DATABASE_URL': secrets['database_url'],
'REDIS_URL': secrets['redis_url'],
'PAYSTACK_SECRET_KEY': secrets['paystack_key'],
'FLUTTERWAVE_SECRET_KEY': secrets['flutterwave_key'],
'ENVIRONMENT': environment,
'SENTRY_DSN': secrets.get('sentry_dsn', '')
}
```
**Testing Multi-Tenant Scenarios:**
```python
class MultiTenantTestRunner:
"""Run tests across multiple tenant configurations."""
def __init__(self, test_client, db_fixtures):
self.client = test_client
self.fixtures = db_fixtures
def run_tenant_matrix_tests(self) -> dict:
"""Test across all plan combinations."""
plans = ['free', 'starter', 'professional', 'enterprise']
features = ['api_access', 'ai_tokens', 'webhooks', 'sso']
results = []
for plan in plans:
for features_subset in self._generate_feature_combinations(features):
tenant = self._create_test_tenant(plan, features_subset)
test_results = self._run_tenant_tests(tenant)
results.append({
'plan': plan,
'features': features_subset,
'passed': test_results['passed'],
'failed': test_results['failed']
})
return {
'total': len(results),
'passed': sum(1 for r in results if r['failed'] == 0),
'failed': sum(1 for r in results if r['failed'] > 0),
'details': results
}
def _run_tenant_tests(self, tenant: Tenant) -> dict:
"""Run tests for a specific tenant configuration."""
passed = 0
failed = 0
if tenant.has_feature('api_access'):
if not self._test_api_access(tenant):
failed += 1
else:
passed += 1
if tenant.has_feature('ai_tokens'):
if not self._test_ai_integration(tenant):
failed += 1
else:
passed += 1
if tenant.has_feature('webhooks'):
if not self._test_webhook_delivery(tenant):
failed += 1
else:
passed += 1
return {'passed': passed, 'failed': failed}
```
**Payment Integration Testing:**
```python
class PaymentIntegrationTest:
"""Test payment processor integrations."""
def __init__(self, paystack_client, flutterwave_client):
self.paystack = paystack_client
self.flutterwave = flutterwave_client
def test_nigerian_payment_flow(self) -> dict:
"""Test complete payment flow with Nigerian payment methods."""
results = {}
results['paystack_card'] = self._test_paystack_card()
results['paystack_transfer'] = self._test_paystack_transfer()
results['flutterwave_card'] = self._test_flutterwave_card()
results['flutterwave_ussd'] = self._test_flutterwave_ussd()
return results
def _test_paystack_transfer(self) -> dict:
"""Test Paystack bank transfer payment."""
test_txn = {
'amount': 45000,
'email': '[email protected] ',
'currency': 'NGN',
'payment_type': 'bank_transfer'
}
initialization = self.paystack.initialize_transaction(test_txn)
authorization = self.paystack.verify_authorization(
initialization['reference'],
mock_verification=True
)
settlement = self._verify_settlement(
'paystack',
authorization['reference']
)
return {
'initialized': initialization['status'] == 'success',
'authorized': authorization['status'] == 'success',
'settled': settlement,
'currency_handled': 'NGN'
}
```
**Common Failure Modes:**
CI/CD pipelines that run full test suites against every commit create unnecessary costs and delays. Implement smart test selection that runs only relevant tests based on changed files.
```python
def determine_relevant_tests(changed_files: list) -> list:
"""Determine which tests to run based on changes."""
test_mapping = {
'services/billing': ['tests/billing/*', 'tests/payments/*'],
'services/ai': ['tests/ai/*', 'tests/models/*'],
'services/api': ['tests/api/*'],
'frontend': ['tests/frontend/*', 'tests/e2e/*'],
'shared': ['tests/*']
}
tests = set()
for changed_file in changed_files:
for module, module_tests in test_mapping.items():
if changed_file.startswith(module):
tests.update(module_tests)
tests.update(test_mapping['shared'])
return list(tests)
```