Function notifications_received_via_websocket

tests/test_clnrest.py:266–311

Return the list of notifications received by the websocket client. We try to connect to the websocket server running at `base_url` with `http_session` parameters. Then we create an invoice on the node: - if we were effectively connected, we received an `invoice_creation` notific

(l1, base_url, http_session, rpc_method='invoice', rpc_params=[100000, 'label', 'description'], expect_error=None)

# unused 'message'".

def notifications_received_via_websocket(l1, base_url, http_session, rpc_method='invoice', rpc_params=[100000, 'label', 'description'], expect_error=None):
    """Return the list of notifications received by the websocket client.

    We try to connect to the websocket server running at `base_url`
    with `http_session` parameters.  Then we create an invoice on the node:

    - if we were effectively connected, we received an `invoice_creation`
      notification via websocket that should be in the list of notifications
      we return.
    - if we couldn't connect to the websocket server, the notification list
      we return is empty."""
    http_session.headers.update({"upgrade": "websocket"})
    sio = socketio.Client(http_session=http_session)
    notifications = []
    connection_error_msg = None

    @sio.event
    def message(data):
        notifications.append(data)

    @sio.event
    def connect_error(data):
        nonlocal connection_error_msg
        connection_error_msg = str(data)

    try:
        sio.connect(base_url)
    except socketio.exceptions.ConnectionError as e:
        # Use the detailed error message if available
        error_str = connection_error_msg if connection_error_msg else str(e)

        if expect_error and expect_error in error_str:
            return notifications
        else:
            raise ValueError(f"Expected error code `{expect_error}`, got `{error_str}` instead")
    except Exception:
        raise
    if expect_error:
        raise Exception(f"did not raise expected error {expect_error}")
    time.sleep(2)
    # trigger notification by calling method
    rpc_call = getattr(l1.rpc, rpc_method)
    rpc_call(*rpc_params)
    time.sleep(2)
    sio.disconnect()
    return notifications


def test_websocket_no_rune(node_factory):