Integration Testing

Integration Testing Skill

Comprehensive guide for testing integration points across the Muni system.

Overview

Muni has complex integration points that require end-to-end testing:

Integration Points:

• Rover ↔ Depot: WebSocket (teleop, discovery, dispatch), UDP (metrics)
• CAN Bus: VESC motor controllers, MCU peripherals
• Database: PostgreSQL (dispatch service)
• File System: Session recording, sync, playback
• External Services: GPS/RTK, video streaming

Testing Approaches:

Rust Integration Testing

Test Structure

Unit tests (in same file):

// src/lib.rs or src/main.rs
#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_function() {
        assert_eq!(2 + 2, 4);
    }
}

Integration tests (separate directory):

crate/
├── src/
│   └── lib.rs
├── tests/
│   ├── common/
│   │   └── mod.rs      # Shared test utilities
│   ├── integration_test_1.rs
│   └── integration_test_2.rs
└── Cargo.toml

Async Testing with Tokio

// tests/websocket_test.rs
use tokio::net::TcpListener;
use futures_util::{SinkExt, StreamExt};

#[tokio::test]
async fn test_websocket_echo() {
    // Start test server
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();

    tokio::spawn(async move {
        let (stream, _) = listener.accept().await.unwrap();
        // Handle WebSocket...
    });

    // Connect as client
    let (ws_stream, _) = tokio_tungstenite::connect_async(
        format!("ws://{}", addr)
    ).await.unwrap();

    let (mut write, mut read) = ws_stream.split();

    // Send message
    write.send(tungstenite::Message::Text("hello".into())).await.unwrap();

    // Receive response
    let msg = read.next().await.unwrap().unwrap();
    assert_eq!(msg, tungstenite::Message::Text("hello".into()));
}

Test Fixtures and Setup

Shared test utilities:

// tests/common/mod.rs
use std::sync::Arc;
use tokio::sync::RwLock;

pub struct TestContext {
    pub state: Arc<AppState>,
    pub addr: SocketAddr,
}

impl TestContext {
    pub async fn new() -> Self {
        // Setup test state
        let state = Arc::new(AppState::new());

        // Start test server
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();

        tokio::spawn(async move {
            axum::serve(listener, create_router(state)).await.unwrap();
        });

        Self { state, addr }
    }

    pub fn url(&self, path: &str) -> String {
        format!("http://{}{}", self.addr, path)
    }
}

Using in tests:

// tests/api_test.rs
mod common;

#[tokio::test]
async fn test_health_endpoint() {
    let ctx = common::TestContext::new().await;

    let response = reqwest::get(&ctx.url("/health")).await.unwrap();
    assert_eq!(response.status(), 200);

    let body: serde_json::Value = response.json().await.unwrap();
    assert_eq!(body["status"], "ok");
}

Database Testing

In-memory database (SQLite):

use sqlx::SqlitePool;

async fn setup_test_db() -> SqlitePool {
    let pool = SqlitePool::connect(":memory:").await.unwrap();

    // Run migrations
    sqlx::migrate!("./migrations")
        .run(&pool)
        .await
        .unwrap();

    pool
}

#[tokio::test]
async fn test_database_query() {
    let pool = setup_test_db().await;

    // Insert test data
    sqlx::query("INSERT INTO zones (name) VALUES ($1)")
        .bind("Test Zone")
        .execute(&pool)
        .await
        .unwrap();

    // Query
    let zones: Vec<Zone> = sqlx::query_as("SELECT * FROM zones")
        .fetch_all(&pool)
        .await
        .unwrap();

    assert_eq!(zones.len(), 1);
    assert_eq!(zones[0].name, "Test Zone");
}

Test database (PostgreSQL via Docker):

// tests/common/mod.rs
use testcontainers::{clients::Cli, images::postgres::Postgres, Container};

pub struct TestDb<'a> {
    _container: Container<'a, Postgres>,
    pub pool: PgPool,
}

impl<'a> TestDb<'a> {
    pub async fn new(docker: &'a Cli) -> Self {
        let container = docker.run(Postgres::default());
        let port = container.get_host_port_ipv4(5432);

        let pool = PgPoolOptions::new()
            .connect(&format!("postgres://postgres:postgres@localhost:{}/postgres", port))
            .await
            .unwrap();

        // Run migrations
        sqlx::migrate!("./migrations").run(&pool).await.unwrap();

        Self {
            _container: container,
            pool,
        }
    }
}

#[tokio::test]
async fn test_with_postgres() {
    let docker = Cli::default();
    let db = TestDb::new(&docker).await;

    // Use db.pool for queries...
}

Mocking CAN Bus

Mock CAN interface:

// tests/mock_can.rs
use std::sync::{Arc, Mutex};

#[derive(Clone)]
pub struct MockCan {
    sent_frames: Arc<Mutex<Vec<CanFrame>>>,
}

impl MockCan {
    pub fn new() -> Self {
        Self {
            sent_frames: Arc::new(Mutex::new(Vec::new())),
        }
    }

    pub fn send(&self, frame: CanFrame) {
        self.sent_frames.lock().unwrap().push(frame);
    }

    pub fn recv(&self) -> Option<CanFrame> {
        // Return pre-configured frames or generate on-demand
        Some(CanFrame::new(0x001, &[0x01, 0x02, 0x03]))
    }

    pub fn get_sent_frames(&self) -> Vec<CanFrame> {
        self.sent_frames.lock().unwrap().clone()
    }

    pub fn clear(&self) {
        self.sent_frames.lock().unwrap().clear();
    }
}

#[tokio::test]
async fn test_motor_command() {
    let can = MockCan::new();

    // Send motor command
    send_motor_command(&can, MotorId::FrontLeft, 100).await;

    // Verify frame was sent
    let frames = can.get_sent_frames();
    assert_eq!(frames.len(), 1);
    assert_eq!(frames[0].id(), 0x001);  // VESC ID
}

Trait-based abstraction (for real/mock CAN):

#[async_trait]
pub trait CanInterface: Send + Sync {
    async fn send(&self, frame: CanFrame) -> Result<()>;
    async fn recv(&self) -> Result<CanFrame>;
}

// Real implementation
pub struct RealCan {
    socket: CanSocket,
}

#[async_trait]
impl CanInterface for RealCan {
    async fn send(&self, frame: CanFrame) -> Result<()> {
        self.socket.write_frame(&frame).await
    }

    async fn recv(&self) -> Result<CanFrame> {
        self.socket.read_frame().await
    }
}

// Mock implementation
pub struct MockCan {
    // ... as above
}

#[async_trait]
impl CanInterface for MockCan {
    async fn send(&self, frame: CanFrame) -> Result<()> {
        self.sent_frames.lock().unwrap().push(frame);
        Ok(())
    }

    async fn recv(&self) -> Result<CanFrame> {
        Ok(CanFrame::new(0x001, &[0x01, 0x02, 0x03]))
    }
}

// Usage in tests
#[tokio::test]
async fn test_with_mock_can() {
    let can: Arc<dyn CanInterface> = Arc::new(MockCan::new());
    // Test code uses CanInterface trait...
}

Testing WebSocket Protocols

WebSocket test client:

use tokio_tungstenite::{connect_async, tungstenite::Message};

#[tokio::test]
async fn test_dispatch_protocol() {
    // Start service
    let ctx = TestContext::new().await;
    let ws_url = format!("ws://{}/ws", ctx.addr);

    // Connect
    let (ws_stream, _) = connect_async(&ws_url).await.unwrap();
    let (mut write, mut read) = ws_stream.split();

    // Send registration
    let register = serde_json::json!({
        "type": "register",
        "rover_id": "test-rover"
    });
    write.send(Message::Text(register.to_string().into())).await.unwrap();

    // Receive task assignment
    let msg = read.next().await.unwrap().unwrap();
    let task: TaskMessage = serde_json::from_str(&msg.to_text().unwrap()).unwrap();

    assert_eq!(task.r#type, "task");
    assert!(!task.waypoints.is_empty());

    // Send progress
    let progress = serde_json::json!({
        "type": "progress",
        "task_id": task.task_id,
        "progress": 50
    });
    write.send(Message::Text(progress.to_string().into())).await.unwrap();

    // Cleanup
    write.close().await.unwrap();
}

Test Timeouts

use tokio::time::{timeout, Duration};

#[tokio::test]
async fn test_with_timeout() {
    let result = timeout(Duration::from_secs(5), async {
        // Test code that might hang
        expensive_operation().await
    }).await;

    assert!(result.is_ok(), "Test timed out");
}

Parallel Test Execution

cargo-nextest (faster test runner):

# Install
cargo install cargo-nextest

# Run tests in parallel
cargo nextest run

# Run specific test
cargo nextest run test_name

# Show output
cargo nextest run --nocapture

Test isolation:

// Use unique ports for each test
use std::sync::atomic::{AtomicU16, Ordering};

static NEXT_PORT: AtomicU16 = AtomicU16::new(50000);

fn get_test_port() -> u16 {
    NEXT_PORT.fetch_add(1, Ordering::SeqCst)
}

#[tokio::test]
async fn test_parallel_1() {
    let port = get_test_port();
    let listener = TcpListener::bind(format!("127.0.0.1:{}", port)).await.unwrap();
    // ...
}

#[tokio::test]
async fn test_parallel_2() {
    let port = get_test_port();
    let listener = TcpListener::bind(format!("127.0.0.1:{}", port)).await.unwrap();
    // ...
}

TypeScript/React Testing

Test Setup (Vitest)

vitest.config.ts:

import { defineConfig } from 'vitest/config';
import react from '@vitejs/plugin-react';

export default defineConfig({
  plugins: [react()],
  test: {
    environment: 'jsdom',
    setupFiles: ['./src/test/setup.ts'],
    globals: true,
  },
});

test/setup.ts:

import '@testing-library/jest-dom';
import { cleanup } from '@testing-library/react';
import { afterEach } from 'vitest';

// Cleanup after each test
afterEach(() => {
  cleanup();
});

Component Testing

// src/components/RoverCard.test.tsx
import { render, screen } from '@testing-library/react';
import { describe, it, expect } from 'vitest';
import { RoverCard } from './RoverCard';

describe('RoverCard', () => {
  it('renders rover information', () => {
    const rover = {
      id: 'rover-1',
      name: 'Frog Zero',
      online: true,
      batteryVoltage: 48.5,
    };

    render(<RoverCard rover={rover} />);

    expect(screen.getByText('Frog Zero')).toBeInTheDocument();
    expect(screen.getByText('Online')).toBeInTheDocument();
    expect(screen.getByText('48.5V')).toBeInTheDocument();
  });

  it('shows offline status when rover is offline', () => {
    const rover = {
      id: 'rover-1',
      name: 'Frog Zero',
      online: false,
      batteryVoltage: 0,
    };

    render(<RoverCard rover={rover} />);

    expect(screen.getByText('Offline')).toBeInTheDocument();
  });
});

Testing Hooks

// src/hooks/useWebSocket.test.ts
import { renderHook, waitFor } from '@testing-library/react';
import { describe, it, expect, vi } from 'vitest';
import { useWebSocket } from './useWebSocket';
import WS from 'jest-websocket-mock';

describe('useWebSocket', () => {
  it('connects to WebSocket server', async () => {
    const server = new WS('ws://localhost:1234');

    const { result } = renderHook(() =>
      useWebSocket('ws://localhost:1234')
    );

    await waitFor(() => {
      expect(result.current.status).toBe('connected');
    });

    server.close();
  });

  it('receives messages', async () => {
    const server = new WS('ws://localhost:1234');
    const

    renderHook(() =>
      useWebSocket('ws://localhost:1234', { onMessage })
    );

    server.send({ type: 'update', data: 'test' });

    await waitFor(() => {
      expect(onMessage).toHaveBeenCalledWith({ type: 'update', data: 'test' });
    });

    server.close();
  });
});

Mocking Zustand Store

// src/test/mocks/store.ts
import { create } from 'zustand';
import { AppState } from '@/store';

export const createMockStore = (initialState?: Partial<AppState>) => {
  return create<AppState>(() => ({
    rovers: [],
    selectedRover: null,
    teleopActive: false,
    ...initialState,
    // Actions
    setRovers: vi.fn(),
    selectRover: vi.fn(),
    startTeleop: vi.fn(),
  }));
};

// In tests
import { createMockStore } from '@/test/mocks/store';

it('displays selected rover', () => {
  const mockStore = createMockStore({
    selectedRover: {
      id: 'rover-1',
      name: 'Frog Zero',
      online: true,
    },
  });

  render(
    <StoreProvider store={mockStore}>
      <RoverDetails />
    </StoreProvider>
  );

  expect(screen.getByText('Frog Zero')).toBeInTheDocument();
});

E2E Testing with Playwright

playwright.config.ts:

import { defineConfig } from '@playwright/test';

export default defineConfig({
  testDir: './e2e',
  use: {
    baseURL: 'http://localhost:5173',
    screenshot: 'only-on-failure',
    video: 'retain-on-failure',
  },
  webServer: {
    command: 'npm run dev',
    port: 5173,
    reuseExistingServer: !process.env.CI,
  },
});

e2e/teleop.spec.ts:

import { test, expect } from '@playwright/test';

test('teleop flow', async ({ page }) => {
  await page.goto('/');

  // Select rover
  await page.click('[data-testid="rover-card-rover-1"]');
  await expect(page).toHaveURL(/\/rover\/rover-1/);

  // Start teleop
  await page.click('[data-testid="start-teleop-button"]');
  await expect(page.locator('[data-testid="teleop-active"]')).toBeVisible();

  // Send command (keyboard)
  await page.keyboard.press('KeyW');
  await page.waitForTimeout(100);

  // Stop teleop
  await page.click('[data-testid="stop-teleop-button"]');
  await expect(page.locator('[data-testid="teleop-active"]')).not.toBeVisible();
});

Docker Test Environments

Test Compose File

docker-compose.test.yml:

version: '3.8'

services:
  postgres-test:
    image: postgres:16
    environment:
      POSTGRES_PASSWORD: test
      POSTGRES_DB: test_db
    ports:
      - "5433:5432"
    tmpfs:
      - /var/lib/postgresql/data  # In-memory (fast, ephemeral)

  redis-test:
    image: redis:7
    ports:
      - "6380:6379"

  test-runner:
    build:
      context: .
      dockerfile: Dockerfile.test
    depends_on:
      - postgres-test
      - redis-test
    environment:
      DATABASE_URL: postgres://postgres:test@postgres-test:5432/test_db
      REDIS_URL: redis://redis-test:6379
    command: cargo test

Running tests:

# Run tests in Docker
docker compose -f docker-compose.test.yml up --abort-on-container-exit

# Cleanup
docker compose -f docker-compose.test.yml down -v

Test Fixtures with Docker

// tests/common/docker.rs
use testcontainers::{clients::Cli, Container, RunnableImage};
use testcontainers::images::postgres::Postgres;

pub struct TestEnv<'a> {
    pub postgres: Container<'a, Postgres>,
    pub pool: PgPool,
}

impl<'a> TestEnv<'a> {
    pub async fn new(docker: &'a Cli) -> Self {
        let postgres = docker.run(Postgres::default());
        let port = postgres.get_host_port_ipv4(5432);

        let pool = PgPoolOptions::new()
            .connect(&format!("postgres://postgres:postgres@localhost:{}/postgres", port))
            .await
            .unwrap();

        sqlx::migrate!("./migrations").run(&pool).await.unwrap();

        Self { postgres, pool }
    }

    pub async fn seed_data(&self) {
        sqlx::query("INSERT INTO zones (name) VALUES ('Test Zone')")
            .execute(&self.pool)
            .await
            .unwrap();
    }
}

Python Testing (splat-worker)

pytest Setup

tests/conftest.py:

import pytest
import asyncio

@pytest.fixture
def event_loop():
    """Create event loop for async tests."""
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

@pytest.fixture
def sample_point_cloud():
    """Load sample point cloud data."""
    import numpy as np
    points = np.random.rand(1000, 3)
    colors = np.random.rand(1000, 3)
    return points, colors

tests/test_splat.py:

import pytest
from splat_worker import process_session

@pytest.mark.asyncio
async def test_process_session(sample_point_cloud):
    points, colors = sample_point_cloud

    result = await process_session(points, colors)

    assert result is not None
    assert result.splat_count > 0
    assert result.output_path.exists()

Rerun Recording Validation

Validate Recording Structure

// tests/recording_test.rs
use rerun::RecordingStream;

#[tokio::test]
async fn test_session_recording() {
    let rec = RecordingStream::new("test", rerun::default_store()).unwrap();

    // Log some data
    rec.log("world/robot/position", &rerun::Position3D::new(1.0, 2.0, 3.0)).unwrap();

    // Save recording
    rec.save("test.rrd").unwrap();

    // Validate file exists and is not empty
    let metadata = std::fs::metadata("test.rrd").unwrap();
    assert!(metadata.len() > 0);

    // Cleanup
    std::fs::remove_file("test.rrd").unwrap();
}

Playback Validation

# Play back recording
rerun test.rrd

# Validate with Python
python -c "import rerun as rr; rr.load('test.rrd'); print('Valid')"

CI/CD Integration

GitHub Actions

. github/workflows/test.yml:

name: Tests

on: [push, pull_request]

jobs:
  test-rust:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:16
        env:
          POSTGRES_PASSWORD: test
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
        ports:
          - 5432:5432

    steps:
      - uses: actions/checkout@v4

      - name: Setup Rust
        uses: actions-rs/toolchain@v1
        with:
          toolchain: stable

      - name: Run tests
        run: cargo test
        env:
          DATABASE_URL: postgres://postgres:test@localhost/postgres

  test-typescript:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Setup Node
        uses: actions/setup-node@v4
        with:
          node-version: '20'

      - name: Install dependencies
        run: npm ci
        working-directory: depot/console

      - name: Run tests
        run: npm test
        working-directory: depot/console

Testing Checklist

Before Committing

• All unit tests pass: cargo test
• Integration tests pass: cargo test --test '*'
• TypeScript tests pass: npm test
• Linting passes: cargo clippy, npm run lint
• Format is correct: cargo fmt --check

Before Deploying

• All tests pass in CI
• Manual smoke tests completed
• Database migrations tested
• Rollback procedure verified
• Monitoring dashboards checked

Test Coverage Goals

• Critical paths: 100% (safety systems, CAN communication)
• Core functionality: >80% (state machine, control logic)
• Utilities: >60% (helpers, converters)
• UI components: >50% (interactive elements)

Common Testing Patterns

Testing Error Handling

#[tokio::test]
async fn test_error_handling() {
    let result = function_that_fails().await;

    assert!(result.is_err());
    assert_eq!(
        result.unwrap_err().to_string(),
        "Expected error message"
    );
}

Testing with Timeouts

use tokio::time::{timeout, Duration};

#[tokio::test]
async fn test_operation_completes_in_time() {
    let result = timeout(
        Duration::from_secs(5),
        slow_operation()
    ).await;

    assert!(result.is_ok(), "Operation timed out");
}

Snapshot Testing

use insta::assert_debug_snapshot;

#[test]
fn test_serialization() {
    let data = ComplexStruct { /* ... */ };
    assert_debug_snapshot!(data);
}

References

• Tokio testing guide
• SQLx testing
• Testing Library docs
• Vitest documentation
• Playwright documentation
• testcontainers-rs