Future Considerations and Potential Enhancements

While the proposed design addresses the core requirements, several areas could be explored for future enhancements to further increase the library's power and flexibility.

The current design focuses on basic subprotocol acceptance via the falcon.asgi.WebSocket.accept(subprotocol=...) method, which could be exposed in on_connect. Future versions could explore more structured ways to handle different subprotocols, perhaps allowing WebSocketResource subclasses to specialize in particular subprotocols or offering mechanisms to select a handler based on the negotiated subprotocol.

Investigating support for WebSocket compression extensions (RFC 7692, e.g., permessage-deflate) could improve bandwidth efficiency. Falcon's WebSocket.accept() method does not currently list explicit parameters for negotiating compression extensions.[^2] This would likely depend on the capabilities of the underlying ASGI server (e.g., Uvicorn, Daphne) and might require lower-level access to the ASGI connection scope or specific ASGI extensions.

The WebSocketConnectionManager currently proposes room-based grouping. More sophisticated targeting mechanisms could be added, such as:

• Targeting all connections belonging to a specific authenticated user ID (across multiple devices/tabs).

• Allowing connections to be tagged with arbitrary labels for flexible group definitions. This would likely involve a more complex internal structure for the WebSocketConnectionManager.

The initial design of the WebSocketConnectionManager is for in-process operation, suitable for single-server deployments or those using sticky sessions. For applications requiring horizontal scaling across multiple server instances, this in-process manager becomes a limitation, as broadcasts would only reach clients connected to the same instance.

A significant enhancement would be to make the WebSocketConnectionManager pluggable, allowing different backend implementations. For example:

• An implementation using a message bus like Redis Pub/Sub (similar to Django Channels' channel layer 5) would enable inter-process communication, allowing a message published on one instance to be delivered to WebSockets connected to other instances. Designing the WebSocketConnectionManager with a clear interface (e.g., an Abstract Base Class) from the outset would facilitate this evolution, allowing the library to start simple but provide a clear upgrade path for distributed deployments without requiring a fundamental rewrite of the application logic that interacts with the manager.

Providing first-class testing utilities keeps Falcon-Pachinko approachable for teams who rely on tight feedback loops. The design includes two complementary helpers—WebSocketTestClient and WebSocketSimulator—plus a pytest fixture that encapsulates common setup. Together they support both end-to-end exercises against a running ASGI server and hermetic unit tests that spy on the framework internals.

6.5.1. WebSocketTestClient

The WebSocketTestClient offers a lightweight façade over an existing WebSocket client library, reducing wheel reinvention while presenting a Falcon-flavoured API. It wraps websockets.connect() from the websockets project because that library already provides a reliable asyncio client with excellent RFC coverage.

• Instantiation: WebSocketTestClient(app_url: str, *, headers: dict | None, subprotocols: list[str] | None, allow_insecure: bool = False) stores the base URL and defaults, requiring opt-in for local ws:// use.

• Connection Context: async with client.connect(path) opens a connection with websockets.connect(f"{base}{path}", extra_headers=..., subprotocols=...) and yields a WebSocketSession helper. Context exit guarantees closure even if assertions fail.

• Session API: The session mirrors the server-facing WebSocketLike protocol so tests read naturally:

  async with WebSocketTestClient(
      "wss://example.com", allow_insecure=True
  ).connect("/ws/chat") as session:
      await session.send_json({"type": "ping"})
      reply = await session.receive_json()

JSON helpers rely on msgspec for encoding/decoding to maintain parity with runtime validation. Additional passthroughs expose send_bytes, receive, and close for full coverage.

• Trace Collection: Optional hooks capture a chronological log of outbound and inbound frames. Each TraceEvent is annotated with a monotonically increasing index alongside the direction, payload kind, and decoded payload. The log integrates with pytest's assertion introspection, making it easy to debug sequencing issues.

By delegating network semantics to websockets, the client stays thin while still supporting TLS, custom headers, and subprotocol negotiation.

6.5.2. WebSocketSimulator

Where the test client targets black-box integration, the WebSocketSimulator acts as an injectable dependency that emulates the WebSocketLike contract. It enables white-box testing of router behaviour without needing an ASGI server.

• Construction: WebSocketSimulator() accepts optional queues for inbound and outbound payloads plus dependency hooks. When mounted, the simulator is passed into a WebSocketResource in place of a real connection.

• Interface: It implements the same async methods as falcon.asgi.WebSocket (accept, close, send_media, receive_media), but internally uses asyncio queues so tests can inject messages or spy on emitted frames. Helper shortcuts (send_json, pop_sent) keep tests concise.

• Spying and Injection: Tests enqueue inbound messages via await simulator.push_message({...}) before driving the resource's receive loop. Outbound frames are recorded and made available through simulator.sent_messages, enabling assertions about order, payload, or metadata. Because the simulator is dependency-injectable, resources can be exercised in isolation alongside fake connection managers or hook managers.

• Integration with Falcon-Pachinko: The router exposes an optional simulator_factory parameter. When supplied, WebSocketRouter calls this factory to obtain a simulator whenever a connection is created during tests. Production deployments omit the factory and continue using the real ASGI websocket instance. This pattern mirrors Falcon's existing HTTP testing hooks and keeps the simulator out of the hot path in production.

6.5.3. Pytest Fixture

To streamline usage, the documentation prescribes a pytest fixture that initialises the simulator alongside a configured router. A representative fixture looks like:

@pytest.fixture
async def websocket_simulator(app, event_loop):
    sim = WebSocketSimulator()
    router = WebSocketRouter(simulator_factory=lambda *_: sim)
    app.add_route("/ws", router)
    async with sim.connected() as connection:
        yield connection

• Lifecycle Management: The fixture ensures accept() is invoked before yielding and that close() is called after the test, preventing dangling tasks or leaked queues.

• Behavioural Testing: Higher-level fixtures can compose the simulator with the connection manager to validate broadcast flows, or parameterise initial inbound frames to exercise specific message handlers.

• Extensibility: Because the simulator is injectable, teams can swap in variants that add latency simulation, failure injection, or statistics gathering without altering production code.

These utilities provide a cohesive story that spans unit, integration, and behavioural testing while keeping the production runtime free from testing concerns.

The following class diagram summarises how the harness composes simulators, routers, and the original Falcon websocket stub when exercised from tests.

classDiagram
    class SimulatorConnection {
        +path: str
        +router: WebSocketRouter
        +simulator: WebSocketSimulator
        +request: object
        +websocket: _OriginalWebSocket
        -_json_decoder: msjson.Decoder
        +accepted: bool
        +closed: bool
        +close_code: int | None
        +sent_messages: list[object]
        +pop_sent(): object
        +pop_sent_json(payload_type): object
        +push_json(payload): None
        +push_text(message): None
        +push_bytes(payload): None
    }
    class _OriginalWebSocket {
        +accepted: bool
        +closed: bool
        +close_code: int | None
        +subprotocol: str | None
        +sent: list[object]
        +accept(subprotocol): None
        +close(code): None
        +send_media(data): None
        +receive_media(): object
    }
    class _HarnessSimulator {
        +bind_original(original): None
        +accept(subprotocol): None
        +close(code): None
    }
    _HarnessSimulator --|> WebSocketSimulator
    SimulatorConnection o-- _OriginalWebSocket
    SimulatorConnection o-- WebSocketRouter
    SimulatorConnection o-- WebSocketSimulator
    class SimulatorRouterHarness {
        +app: falcon.asgi.App
        +router: WebSocketRouter
        +mount(prefix): None
        +connect(path, initial_inbound): AsyncIterator[SimulatorConnection]
    }
    SimulatorRouterHarness o-- WebSocketRouter
    SimulatorRouterHarness o-- SimulatorConnection
    class _TestRequest {
        +path: str
        +path_template: str
        +context: SimpleNamespace
    }
    SimulatorConnection o-- _TestRequest

sequenceDiagram
    actor Developer
    participant TC as TestClient
    participant WSSim as WebSocketSimulator
    participant WSConn as WebSocketConnection
    participant App as ASGIAppUnderTest

    Developer->>TC: client.connect("/ws/chat")
    activate TC
    TC->>WSConn: new WebSocketSession()
    TC-->>Developer: session
    deactivate TC

    Developer->>WSSim: simulator.push_message(data)
    activate WSSim
    activate App
    WSSim->>App: Inject frame (router receive loop)
    deactivate WSSim

    Developer->>WSConn: session.send_json(data)
    activate WSConn
    WSConn->>App: Send JSON data
    App-->>WSConn: (optional ack/response data)
    WSConn-->>Developer: (return from send)
    deactivate WSConn

    deactivate App

    Developer->>WSSim: simulator.sent_messages
    WSSim-->>Developer: Inspect recorded frames

    Developer->>WSConn: await session.close()

As a long-term vision, the structured nature of WebSocketResource and the on_{tag} handlers could potentially be leveraged to automatically generate a stub or a basic AsyncAPI document. This would further bridge the gap between design artifacts and implementation, though it represents a significantly more complex undertaking.