Testing Strategies

Relevant source files

The following files were used as context for generating this wiki page:

• test.sh
• tests/test_eventbus.py
• bubus-ts/tests/bridges.test.ts
• bubus-ts/tests/ts_to_python_roundtrip.test.ts
• bubus-ts/tests/performance.scenarios.js
• bubus-ts/tests/retry.test.ts
• bubus-ts/tests/eventbus_basics.test.ts
• tests/performance_runtime.py

The testing strategy for the bbus project is designed to ensure cross-language parity between Python and TypeScript implementations, high-performance runtime stability, and reliable event propagation across different bridge architectures. The suite encompasses unit testing, integration testing for bridges (NATS, Redis, Postgres), cross-runtime roundtrip validation, and rigorous performance benchmarking.

Sources: test.sh:1-35, bubus-ts/tests/ts_to_python_roundtrip.test.ts:1-20

Core Testing Architecture

The project utilizes a multi-layered testing approach coordinated via a central shell script. It separates standard functional tests from resource-intensive performance suites to prevent cross-runtime contention and flakes.

Execution Flow

1. Linting & Type Checking: Uses ruff, ty, and pyright for Python, and eslint/tsc for TypeScript.
2. Functional Testing: Runs pytest for Python and the Node.js test runner for TypeScript.
3. Cross-Runtime Validation: Executes roundtrip tests that serialize state in one language and rehydrate it in the other.
4. Performance Benchmarking: Runs dedicated performance scripts in isolated subprocesses.

Sources: test.sh:4-35

The diagram shows the sequential execution of test phases to ensure stability before performance measurement. Sources: test.sh:4-35

Functional & Unit Testing

Functional tests focus on the core lifecycle of the EventBus, including event dispatching, handler registration, and status transitions (pending -> started -> completed).

Key Test Areas

• Event Lifecycle: Validating that events transition through states and capture results correctly.
• Error Isolation: Ensuring a failure in one handler does not prevent others from executing.
• Retry Mechanisms: Testing exponential backoff, max attempts, and semaphore concurrency control in the TypeScript implementation.
• Middleware Behavior: Verifying that built-in middlewares like WALEventBusMiddleware or LoggerEventBusMiddleware correctly intercept and process events.

Sources: tests/test_eventbus.py:100-150, bubus-ts/tests/eventbus_basics.test.ts:335-360, bubus-ts/tests/retry.test.ts:40-80

Test Component	Purpose	Key Classes/Functions
Event Lifecycle	Verify status transitions and history	BaseEvent.markStarted(), EventBus.dispatch()
Concurrency	Test parallel vs serial execution	event_handler_concurrency, Semaphore
Error Handling	Validate error capture and isolation	event_errors, EventHandlerTimeoutError
Persistence	Test WAL and SQLite mirroring	WALEventBusMiddleware, SQLiteHistoryMirrorMiddleware

Sources: tests/test_eventbus.py:348-400, bubus-ts/tests/eventbus_basics.test.ts:10-50

Bridge & Integration Testing

Bridge tests verify the ability of EventBus to communicate across different transport layers. These tests often spawn external services (like Redis or NATS) or use temporary files (SQLite, JSONL) to simulate real-world deployments.

Bridge Latency & Roundtrip

The suite includes a measureWarmLatencyMs function to benchmark the overhead of different transports. It also uses a "worker" pattern where a separate process listens for events on a bridge while the main test process sends them, verifying full end-to-end roundtrip integrity.

Sources: bubus-ts/tests/bridges.test.ts:133-180, bubus-ts/tests/bridges.test.ts:203-230

The sequence shows the coordination between the test process and a background worker to verify bridge transport. Sources: bubus-ts/tests/bridges.test.ts:183-201

Cross-Runtime Parity

A critical part of the strategy is the ts_to_python_roundtrip test. This ensures that an event or a whole EventBus state serialized in TypeScript can be rehydrated and continued in Python, and vice versa.

Roundtrip Validation Process

1. Define Classes: Create identical BaseEvent subclasses in TS.
2. Serialize: Use event.toJSON() to create a JSON payload in TS.
3. Execute Python: Spawn a Python subprocess that uses BaseEvent.model_validate() to load the TS JSON and dump it back.
4. Rehydrate: Re-import the Python JSON into TS and verify all fields (IDs, timestamps, schemas) remain stable.

Sources: bubus-ts/tests/ts_to_python_roundtrip.test.ts:245-300

Performance & Stress Testing

Performance tests are executed in isolated environments to avoid resource contention. They measure latency (ms/event) and memory usage (kb/event) across various scenarios.

Performance Scenarios

• 50k-events: Processing a large stream of events on a single bus.
• Ephemeral Buses: Rapidly creating and destroying buses to test for memory leaks.
• Parallel Handlers: Dispatching one event to 50,000 parallel handlers.
• Worst-Case Forwarding: Deeply nested event propagation (Parent -> Child -> Grandchild) with timeouts and cancellations.

Sources: bubus-ts/tests/performance.scenarios.js:154-200, tests/performance_runtime.py:15-21

Memory Leak Prevention

The performance harness includes "cleanup equivalence" tests that verify if unreferenced buses are garbage collected without explicit destroy() calls, comparing WeakRef behavior across Node.js, Bun, and Deno.

Sources: bubus-ts/tests/performance.scenarios.js:51-80, bubus-ts/tests/performance.scenarios.js:345-380

The project's testing strategy ensures that bbus remains a reliable, high-performance event orchestration layer regardless of the language or transport protocol used. By combining strict type safety, cross-language roundtrips, and extensive performance monitoring, the codebase maintains high architectural integrity.