Function test_multiple_sockets

tests/test_threaded_stream.py:170–183

Test managing multiple sockets

(manager)

@pytest.mark.asyncio
async def test_multiple_sockets(manager):
    """Test managing multiple sockets"""
    socket_names = ["socket1", "socket2", "socket3"]

    # Start multiple sockets
    for name in socket_names:
        manager._socket_running[name] = True

    # Stop all sockets
    manager.stop()

    # Verify all sockets are stopped
    for name in socket_names:
        assert manager._socket_running[name] is False


@pytest.mark.asyncio