Testing Guide

Guide for testing the NorthBuilt RAG System components.

Overview

The NorthBuilt RAG System is designed for AWS-native deployment. There is no traditional “local development” environment since the system relies on AWS managed services (Bedrock, Secrets Manager, DynamoDB, etc.).

Instead, we test components in two ways:

1. Unit Tests - Test individual functions with mocked dependencies
2. AWS Integration Tests - Test deployed Lambda functions in AWS

Unit Testing

Python Unit Tests

Test Lambda functions with mocked AWS services.

Setup

cd lambda/chat
python3 -m venv venv
source venv/bin/activate
pip3 install -r requirements.txt
pip3 install pytest pytest-cov moto boto3

Example Test

# lambda/chat/test_handler.py
import json
import pytest
from unittest.mock import Mock, patch, MagicMock
from handler import lambda_handler, validate_request


@pytest.fixture
def sample_event():
    """Sample API Gateway event"""
    return {
        'body': json.dumps({'query': 'What is this system?', 'max_results': 5}),
        'requestContext': {
            'authorizer': {
                'claims': {
                    'sub': 'user-123',
                    'email': 'user@example.com'
                }
            }
        },
        'headers': {
            'Authorization': 'Bearer fake-token'
        }
    }


@pytest.fixture
def sample_context():
    """Mock Lambda context"""
    context = Mock()
    context.request_id = 'test-request-id'
    context.get_remaining_time_in_millis = Mock(return_value=30000)
    return context


class TestChatHandler:
    """Test chat Lambda handler"""

    def test_validate_request_success(self, sample_event):
        """Test successful request validation"""
        valid, error = validate_request(sample_event)
        assert valid is True
        assert error is None

    def test_validate_request_missing_query(self):
        """Test validation with missing query"""
        event = {'body': json.dumps({})}
        valid, error = validate_request(event)
        assert valid is False
        assert 'query' in error.lower()

    @patch('handler.retrieve_from_knowledge_base')
    @patch('handler.invoke_bedrock')
    def test_lambda_handler_success(self, mock_bedrock, mock_kb, sample_event, sample_context):
        """Test successful Lambda invocation"""
        # Mock Knowledge Base retrieval response
        mock_kb.return_value = {
            'documents': [
                {
                    'content': 'System documentation content',
                    'metadata': {'title': 'System Overview', 'client_name': 'TestClient'}
                }
            ]
        }

        # Mock Bedrock LLM response
        mock_bedrock.return_value = {
            'answer': 'This is a RAG system for searching company knowledge.',
            'sources': [1]
        }

        response = lambda_handler(sample_event, sample_context)

        assert response['statusCode'] == 200
        body = json.loads(response['body'])
        assert 'answer' in body
        assert 'sources' in body
        assert len(body['sources']) > 0

    def test_lambda_handler_invalid_request(self, sample_context):
        """Test Lambda with invalid request"""
        event = {'body': '{}'}
        response = lambda_handler(event, sample_context)

        assert response['statusCode'] == 400
        body = json.loads(response['body'])
        assert 'error' in body

    @patch('handler.retrieve_from_knowledge_base')
    def test_lambda_handler_kb_error(self, mock_kb, sample_event, sample_context):
        """Test Lambda when Knowledge Base retrieval fails"""
        mock_kb.side_effect = Exception('Knowledge Base retrieval failed')

        response = lambda_handler(sample_event, sample_context)

        assert response['statusCode'] == 500
        body = json.loads(response['body'])
        assert 'error' in body

Running Tests

# Run all tests
pytest

# Run with coverage
pytest --cov=handler --cov-report=html --cov-report=term

# Run specific test
pytest test_handler.py::TestChatHandler::test_lambda_handler_success

# Run with verbose output
pytest -v

# Open coverage report
open htmlcov/index.html

Mocking AWS Services

Use moto library to mock AWS services:

import boto3
from moto import mock_dynamodb, mock_secretsmanager


@mock_secretsmanager
def test_get_secret():
    """Test getting secret from Secrets Manager"""
    # Create mock secret
    client = boto3.client('secretsmanager', region_name='us-east-1')
    client.create_secret(
        Name='test-secret',
        SecretString='{"api_key": "test-key"}'
    )

    # Test function that uses secret
    from handler import get_api_key
    api_key = get_api_key('test-secret')
    assert api_key == 'test-key'


@mock_dynamodb
def test_save_classification():
    """Test saving classification to DynamoDB"""
    # Create mock table
    dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
    table = dynamodb.create_table(
        TableName='test-classify',
        KeySchema=[{'AttributeName': 'document_id', 'KeyType': 'HASH'}],
        AttributeDefinitions=[{'AttributeName': 'document_id', 'AttributeType': 'S'}],
        BillingMode='PAY_PER_REQUEST'
    )

    # Test function
    from handler import save_classification
    save_classification('doc-123', ['technical', 'support'])

    # Verify
    response = table.get_item(Key={'document_id': 'doc-123'})
    assert response['Item']['document_id'] == 'doc-123'

AWS Integration Testing

Test deployed Lambda functions in AWS.

Prerequisites

# Ensure AWS credentials configured
aws sts get-caller-identity

# Ensure Lambda functions deployed
aws lambda list-functions --query 'Functions[?starts_with(FunctionName, `nb-rag-sys`)].FunctionName'

Invoke Lambda Functions

Chat Lambda

# Create test event
cat > /tmp/chat-event.json << 'EOF'
{
  "body": "{\"query\": \"What is the system architecture?\", \"max_results\": 3}",
  "requestContext": {
    "authorizer": {
      "claims": {
        "sub": "test-user-123",
        "email": "test@example.com"
      }
    }
  },
  "headers": {
    "Authorization": "Bearer test-token"
  }
}
EOF

# Invoke Lambda
aws lambda invoke \
  --function-name nb-rag-sys-chat \
  --payload file:///tmp/chat-event.json \
  --cli-binary-format raw-in-base64-out \
  /tmp/chat-response.json

# View response
cat /tmp/chat-response.json | jq .

Classify Lambda

# Create test event
cat > /tmp/classify-event.json << 'EOF'
{
  "source": "fathom",
  "data": {
    "document": {
      "participants": ["user@clientdomain.com", "agent@company.com"]
    }
  }
}
EOF

# Invoke Lambda
aws lambda invoke \
  --function-name nb-rag-sys-classify \
  --payload file:///tmp/classify-event.json \
  --cli-binary-format raw-in-base64-out \
  /tmp/classify-response.json

# View response
cat /tmp/classify-response.json | jq .

Webhook Lambda

# Create test webhook event
cat > /tmp/webhook-event.json << 'EOF'
{
  "headers": {
    "x-api-key": "test-api-key"
  },
  "body": "{\"event\": \"test\", \"data\": {\"id\": \"123\", \"title\": \"Test\"}}"
}
EOF

# Invoke Lambda
aws lambda invoke \
  --function-name nb-rag-sys-webhook-fathom \
  --payload file:///tmp/webhook-event.json \
  --cli-binary-format raw-in-base64-out \
  /tmp/webhook-response.json

# View response
cat /tmp/webhook-response.json | jq .

End-to-End API Testing

Test full API flow with actual HTTP requests.

Get JWT Token

# Authenticate and get JWT token
# (Replace with your actual Cognito user pool details)
TOKEN=$(curl -s -X POST "https://nb-rag-sys-auth.auth.us-east-1.amazoncognito.com/oauth2/token" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "grant_type=client_credentials" \
  -d "client_id=YOUR_CLIENT_ID" \
  -d "client_secret=YOUR_CLIENT_SECRET" \
  | jq -r '.access_token')

echo "Token: $TOKEN"

Test Chat Endpoint

# Get API Gateway URL
API_URL=$(terraform output -raw api_endpoint)

# Send chat request
curl -X POST "${API_URL}/chat" \
  -H "Authorization: Bearer ${TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{
    "query": "What are the main components of the system?",
    "max_results": 5
  }' | jq .

Test Webhook Endpoint

# Test Fathom webhook
curl -X POST "${API_URL}/webhooks/fathom" \
  -H "x-api-key: YOUR_FATHOM_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "event": "video.processed",
    "data": {
      "video_id": "test-123",
      "title": "Test Video",
      "transcript": "This is a test transcript."
    }
  }'

Load Testing

Use artillery or locust for load testing.

Install Artillery

npm install -g artillery

Create Load Test

# load-test.yml
config:
  target: "https://YOUR_API_GATEWAY_URL"
  phases:
    - duration: 60
      arrivalRate: 5  # 5 requests per second
  variables:
    jwt_token: "YOUR_JWT_TOKEN"

scenarios:
  - name: "Chat Query"
    flow:
      - post:
          url: "/chat"
          headers:
            Authorization: "Bearer "
            Content-Type: "application/json"
          json:
            query: "What is the system architecture?"
            max_results: 5

Run Load Test

artillery run load-test.yml

Terraform Testing

Validate Configuration

cd terraform

# Format check
terraform fmt -check -recursive

# Validate syntax
terraform validate

# Plan (dry run)
terraform plan

Terraform Tests

# Run Terraform tests (if available)
terraform test

Code Quality

Python Linting

# Install tools
pip3 install flake8 pylint black mypy

# Format code
black lambda/

# Lint
flake8 lambda/

# Type checking
mypy lambda/

Pre-Commit Hooks

# Install pre-commit
pip3 install pre-commit

# Install hooks
pre-commit install

# Run manually
pre-commit run --all-files

.pre-commit-config.yaml

repos:
  - repo: https://github.com/psf/black
    rev: 23.9.1
    hooks:
      - id: black
        language_version: python3.14

  - repo: https://github.com/pycqa/flake8
    rev: 6.1.0
    hooks:
      - id: flake8
        args: ['--max-line-length=100']

  - repo: https://github.com/antonbabenko/pre-commit-terraform
    rev: v1.83.5
    hooks:
      - id: terraform_fmt
      - id: terraform_validate

Continuous Integration

Tests run automatically in GitHub Actions on every push.

View Test Results

# List recent workflow runs
gh run list --workflow=test.yml

# View specific run
gh run view RUN_ID

# View logs
gh run view RUN_ID --log

Local CI Simulation

Run the same tests that CI runs:

# Install act (runs GitHub Actions locally)
brew install act

# Run tests workflow
act -W .github/workflows/test.yml

Troubleshooting Tests

Test Fails Locally But Passes in CI

Cause: Environment differences

Solution:

# Check Python version
python3 --version

# Check installed packages
pip3 freeze

# Clean and reinstall
rm -rf venv
python3 -m venv venv
source venv/bin/activate
pip3 install -r requirements.txt

Mock Not Working

Cause: Boto3 client created before mock

Solution:

# Bad - client created before mock
client = boto3.client('dynamodb')

@mock_dynamodb
def test_something():
    # This won't work
    pass

# Good - client created inside mock
@mock_dynamodb
def test_something():
    client = boto3.client('dynamodb')  # Created inside mock
    # This works

Lambda Invocation Timeout

Cause: Lambda function actually running (not mocked)

Solution:

# Check Lambda timeout
aws lambda get-function-configuration \
  --function-name nb-rag-sys-chat \
  --query 'Timeout'

# Increase timeout if needed
aws lambda update-function-configuration \
  --function-name nb-rag-sys-chat \
  --timeout 90

---

Last updated: 2025-12-30