TcpDump API

A FastAPI application for generating tcpdump from Kubernetes/OpenShift pods and sending the results via email.

Setup and Installation

1. Clone the repository
2. Install dependencies:

pip install -r requirements.txt

Running Tests

Setting Up Test Environment

First, let's install the test dependencies:

pip install pytest pytest-asyncio pytest-mock httpx

Add these to your requirements.txt for development:

pytest==7.4.3
pytest-asyncio==0.21.1
pytest-mock==3.12.0
httpx==0.25.0

Test Structure

The tests are organized in the tests directory with the following structure:

tests/
├── __init__.py
├── conftest.py
├── test_k8s_client.py
├── test_email_service.py
└── test_main.py

Writing Tests

Here are examples of test files for each component:

1. tests/conftest.py:

import pytest
from fastapi.testclient import TestClient
from main import app

@pytest.fixture
def test_client():
    return TestClient(app)

@pytest.fixture
def mock_k8s_config(mocker):
    return mocker.patch('kubernetes.config.load_incluster_config')

@pytest.fixture
def mock_k8s_client(mocker):
    return mocker.patch('kubernetes.client.CoreV1Api')

2. tests/test_k8s_client.py:

import pytest
from k8s_client import get_k8s_client, create_debug_pod, delete_debug_pod

def test_get_k8s_client(mock_k8s_config, mock_k8s_client):
    client = get_k8s_client()
    assert client is not None
    mock_k8s_config.assert_called_once()

@pytest.mark.asyncio
async def test_create_debug_pod(mock_k8s_client):
    api = mock_k8s_client()
    api.read_namespaced_pod.return_value.spec.node_name = "test-node"
    
    pod_name = create_debug_pod(
        api,
        "test-namespace",
        "test-pod",
        "debug-image",
        "target-pod",
        "-i any"
    )
    
    assert pod_name == "tcpdump-debug-test-pod"
    api.create_namespaced_pod.assert_called_once()

3. tests/test_email_service.py:

import pytest
from email_service import EmailService

@pytest.mark.asyncio
async def test_send_email_with_attachment(mocker):
    # Mock aiosmtplib.send
    mock_send = mocker.patch('aiosmtplib.send')
    
    email_service = EmailService()
    result = await email_service.send_email_with_attachment(
        ["[email protected]"],
        "Test Subject",
        "Test Body",
        "test.pcap"
    )
    
    assert result is True
    mock_send.assert_called_once()

4. tests/test_main.py:

from fastapi.testclient import TestClient
import pytest
from main import app

def test_root_endpoint(test_client):
    response = test_client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "TcpDump API is running"}

def test_generate_tcpdump(test_client, mocker):
    # Mock background tasks
    mocker.patch('fastapi.BackgroundTasks.add_task')
    
    response = test_client.post(
        "/api/v1/tcpdump",
        json={
            "app_id": "test-app",
            "cluster_name": "test-cluster",
            "debug_image": "debug-image",
            "namespace": "test-namespace",
            "email_recipients": ["[email protected]"],
            "pod_name": "test-pod",
            "tcpdump_options": "-i any",
            "duration": 30
        }
    )
    
    assert response.status_code == 200
    assert response.json()["status"] == "accepted"

Running Tests

To run all tests:

pytest

To run tests with coverage report:

pytest --cov=src tests/

To run specific test files:

pytest tests/test_main.py

To run tests with detailed output:

pytest -v

Mocking External Services

1. Kubernetes API:

   • Use pytest-mock to mock the Kubernetes client
   • Mock pod operations and API responses
   • Test error scenarios and edge cases

2. Email Service:

   • Mock SMTP operations
   • Test email formatting and attachments
   • Verify error handling

3. File Operations:

   • Mock file system operations
   • Test file creation and cleanup
   • Verify proper error handling

Integration Tests

For integration tests, you can use kind or minikube to set up a local Kubernetes cluster:

# tests/test_integration.py
import pytest
from kubernetes import client, config
from main import app

@pytest.mark.integration
def test_full_tcpdump_flow():
    # Setup test namespace and pods
    config.load_kube_config()
    api = client.CoreV1Api()
    
    # Create test resources
    create_test_namespace(api)
    create_test_pod(api)
    
    # Run tcpdump operation
    response = test_client.post(
        "/api/v1/tcpdump",
        json={
            "app_id": "integration-test",
            "cluster_name": "test-cluster",
            "debug_image": "debug-image",
            "namespace": "test-namespace",
            "email_recipients": ["[email protected]"],
            "pod_name": "test-pod",
            "tcpdump_options": "-i any",
            "duration": 10
        }
    )
    
    assert response.status_code == 200
    
    # Cleanup test resources
    cleanup_test_resources(api)

Environment Variables for Tests

Create a .env.test file for test configuration:

SMTP_HOST=localhost
SMTP_PORT=2525
SMTP_USER=test
SMTP_PASSWORD=test
SENDER_EMAIL=[email protected]

Continuous Integration

Add this GitHub Actions workflow for automated testing:

name: Tests

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v2
      - name: Set up Python
        uses: actions/setup-python@v2
        with:
          python-version: '3.11'
      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt
          pip install pytest pytest-asyncio pytest-mock pytest-cov
      - name: Run tests
        run: |
          pytest --cov=src tests/

Remember to:

• Write tests before implementing features (TDD)
• Mock external dependencies
• Test error cases and edge scenarios
• Keep tests focused and independent
• Use meaningful test names and descriptions