Testing

Rivven uses a multi-layered testing strategy to ensure reliability, performance, and security.

Test Structure

Tests: 1685+ passing
├── Unit Tests       (~1300) - Per-crate tests in src/
├── Integration      (~285)  - rivven-integration-tests/
├── Property-based   (~50)   - proptest/quickcheck
├── Benchmarks       (~50)   - criterion micro-benchmarks
└── Feature-gated    (~113)  - Optional feature tests

Running Tests

Quick Start

# Run all tests
cargo test --workspace

# Run with all features enabled
cargo test --workspace --all-features

# Run specific crate tests
cargo test -p rivven-core
cargo test -p rivven-cdc
cargo test -p rivven-integration-tests

# Run integration tests only
cargo test -p rivven-integration-tests --test '*'

Integration Tests

Integration tests require Docker for testcontainers:

# Ensure Docker is running
docker info

# Run integration tests
cargo test -p rivven-integration-tests

# Run specific test categories
cargo test -p rivven-integration-tests --test cdc_postgres
cargo test -p rivven-integration-tests --test cdc_mysql
cargo test -p rivven-integration-tests --test cdc_mariadb
cargo test -p rivven-integration-tests --test kafka_connector
cargo test -p rivven-integration-tests --test mqtt_connector
cargo test -p rivven-integration-tests --test client_protocol
cargo test -p rivven-integration-tests --test e2e_pipeline
cargo test -p rivven-integration-tests --test cluster_consensus
cargo test -p rivven-integration-tests --test security
cargo test -p rivven-integration-tests --test chaos
cargo test -p rivven-integration-tests --test stress
cargo test -p rivven-integration-tests --test consumer_groups
cargo test -p rivven-integration-tests --test connect_e2e
cargo test -p rivven-integration-tests --test durability
cargo test -p rivven-integration-tests --test rbac
cargo test -p rivven-integration-tests --test tls

# Run with verbose output
RUST_LOG=info cargo test -p rivven-integration-tests -- --nocapture

Integration Test Categories

CDC PostgreSQL Tests (cdc_postgres.rs)

Tests PostgreSQL Change Data Capture functionality using testcontainers:

CDC MySQL Tests (cdc_mysql.rs)

Tests MySQL Change Data Capture functionality using binary log replication:

CDC MariaDB Tests (cdc_mariadb.rs)

Tests MariaDB Change Data Capture functionality with MariaDB-specific features:

Client Protocol Tests (client_protocol.rs)

Tests the client-server wire protocol:

End-to-End Pipeline Tests (e2e_pipeline.rs)

Tests complete data pipelines:

Cluster Consensus Tests (cluster_consensus.rs)

Tests distributed cluster behavior:

Security Tests (security.rs)

Tests security boundaries and attack resistance:

Chaos Engineering Tests (chaos.rs)

Tests system resilience under failure conditions:

Stress Tests (stress.rs)

Tests system behavior under high-load conditions:

# Run stress tests
cargo test -p rivven-integration-tests --test stress

# Run with soak test (takes ~1 minute)
cargo test -p rivven-integration-tests --test stress -- --ignored

Consumer Group Tests (consumer_groups.rs)

Tests consumer group coordination and delivery guarantees:

Connector End-to-End Tests (connect_e2e.rs)

Tests complete connector pipelines with simulated sources and sinks:

Kafka Connector Tests (kafka_connector.rs)

Tests Kafka connector integration using testcontainers with a real Kafka broker:

# Run Kafka connector tests
cargo test -p rivven-integration-tests --test kafka_connector

Infrastructure Tests:

Source Connector Tests:

Sink Connector Tests:

Batch Operations Tests:

MQTT Connector Tests (mqtt_connector.rs)

Tests MQTT connector integration using testcontainers with Mosquitto broker:

# Run MQTT connector tests
cargo test -p rivven-integration-tests --test mqtt_connector

Infrastructure Tests:

Source Connector Tests:

QoS Level Tests:

Advanced Feature Tests:

Edge Case Tests:

Durability Tests (durability.rs)

Tests data persistence and recovery across broker restarts:

RBAC Integration Tests (rbac.rs)

Real end-to-end tests for authentication and authorization at the broker level using TestSecureBroker:

Authentication Tests:

Authorization Tests:

Session Management Tests:

Edge Case Tests:

# Run RBAC tests
cargo test -p rivven-integration-tests --test rbac

TLS Integration Tests (tls.rs)

Tests TLS/mTLS functionality at the broker level using self-signed certificates:

Basic TLS Connection Tests:

Message Transmission Tests:

mTLS Mode Tests:

Connection Tests:

TLS Configuration Tests:

# Run TLS tests
cargo test -p rivven-integration-tests --test tls

# Run observability tests
cargo test -p rivven-integration-tests --test observability

Observability Tests (observability.rs)

Tests for metrics, monitoring, and health endpoints:

Health Endpoint Tests:

Prometheus Metrics Tests:

JSON Metrics Tests:

Metrics Accuracy Tests:

Performance Tests:

Edge Case Tests:

# Run observability tests
cargo test -p rivven-integration-tests --test observability

Compression Tests (compression.rs)

Tests for LZ4 and Zstd compression algorithms:

Basic Compression Tests:

Compression Ratio Tests:

Throughput Tests:

Large Payload Tests:

Multi-Message Tests:

Compression Level Tests:

Edge Case Tests:

# Run compression tests
cargo test -p rivven-integration-tests --test compression

Test Infrastructure

Fixtures (fixtures.rs)

The test infrastructure provides reusable fixtures:

use rivven_integration_tests::fixtures::*;

// Embedded broker for fast tests
let broker = TestBroker::start().await?;
let addr = broker.connection_string();

// Secure broker with authentication enabled (for RBAC tests)
let secure_broker = TestSecureBroker::start().await?;
let admin = secure_broker.admin_credentials();

// TLS broker with self-signed certificates (for TLS tests)
let tls_broker = TestTlsBroker::start().await?;
let client_tls_config = tls_broker.client_tls_config();

// Metrics broker with observability endpoints (for observability tests)
let metrics_broker = TestMetricsBroker::start().await?;
let metrics_url = metrics_broker.metrics_url();      // /metrics (Prometheus)
let json_url = metrics_broker.json_metrics_url();    // /metrics/json
let health_url = metrics_broker.health_url();        // /health

// PostgreSQL container for CDC tests
let pg = TestPostgres::start().await?;
let conn = pg.connect().await?;

// Multi-node cluster for consensus tests
let cluster = TestCluster::start(3).await?;

Helpers (helpers.rs)

Common test utilities:

use rivven_integration_tests::helpers::*;

// Initialize tracing for test debugging (also initializes TLS crypto provider)
init_tracing();

// Wait for condition with timeout
wait_for(|| async { broker.is_ready() }, Duration::from_secs(30)).await?;

// Generate unique test identifiers
let topic = unique_topic_name("test");
let group = unique_group_id("consumer");

Test Data Generation

use rivven_integration_tests::fixtures::test_data::*;

// Generate test events
let event = TestEvent::new("order.created", json!({"id": 123}));

// Generate message batches
let messages = generate_messages(1000);
let large_batch = generate_large_messages(100, 1024 * 1024);

Property-Based Testing

Property-based tests verify invariants across random inputs:

# Run property tests
cargo test -p rivven-core -- --test-threads=1 proptest

# Increase case count for thorough testing
PROPTEST_CASES=10000 cargo test -p rivven-core proptest

Example properties tested:

• Message serialization roundtrips
• Partition assignment determinism
• Compression ratio bounds
• Offset ordering invariants

Coverage

Generate coverage reports with cargo-tarpaulin:

# Install tarpaulin
cargo install cargo-tarpaulin

# Generate coverage report
cargo tarpaulin --workspace --out Html

# View report
open tarpaulin-report.html

Current coverage targets:

• Core crates: >80%
• CDC crate: >75%
• Protocol crate: >90%

CI Integration

Tests run automatically in GitHub Actions:

# .github/workflows/ci.yml
- name: Run tests
  run: cargo test --workspace --all-features

- name: Run integration tests
  run: cargo test -p rivven-integration-tests
  env:
    DOCKER_HOST: unix:///var/run/docker.sock

Required CI Secrets

For integration tests with external services:

Writing New Tests

Unit Test Guidelines

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_function_does_expected_thing() {
        // Arrange
        let input = create_input();
        
        // Act
        let result = function_under_test(input);
        
        // Assert
        assert_eq!(result, expected);
    }
}

Integration Test Guidelines

use rivven_integration_tests::prelude::*;

#[tokio::test]
async fn test_feature_works_end_to_end() {
    // Use fixtures for setup
    let broker = TestBroker::start().await.unwrap();
    let client = broker.client().await.unwrap();
    
    // Create unique resources to avoid test interference
    let topic = unique_topic_name("feature");
    client.create_topic(&topic, 3).await.unwrap();
    
    // Test the feature
    client.produce(&topic, "key", "value").await.unwrap();
    
    // Verify results
    let messages = client.consume(&topic, 0, 1).await.unwrap();
    assert_eq!(messages.len(), 1);
    
    // Cleanup happens automatically via Drop
}

Test Naming Conventions

• Unit tests: test_<function>_<scenario>_<expected>
• Integration tests: <category>_<feature>_<behavior>
• Property tests: prop_<invariant>
• Benchmarks: bench_<operation>_<variant>

Troubleshooting

Docker Issues

# Check Docker is running
docker info

# Check for port conflicts
lsof -i :5432  # PostgreSQL
lsof -i :9092  # Broker

# Clean up orphaned containers
docker container prune -f
docker volume prune -f

Test Timeouts

# Increase test timeout
RUST_TEST_TIME_UNIT=60000 cargo test

# Run tests serially to reduce contention
cargo test -- --test-threads=1

Debugging Failures

# Enable trace logging
RUST_LOG=trace cargo test -- --nocapture

# Run specific failing test
cargo test -p rivven-integration-tests test_name -- --exact --nocapture

See Also

• Architecture — System design
• CDC Configuration — CDC setup
• Security — Security features
• Getting Started — Quick start guide