xref: /openbmc/openbmc-test-automation/lib/event_notification.py (revision f0b9122a)

#!/usr/bin/env python3
r"""
This is an extended user library to support Robot Selenium code.
The class contains functions which the robot framework will use
and import as a user-defined keyword.
"""
import json
import ssl

import gen_print as gp  # NOQA
import gen_valid as gv  # NOQA
import requests
import websocket


class event_notification:  # NOQA
    r"""
    Main class to subscribe and receive event notifications.
    """

    def __init__(self, host, username, password):
        r"""
        Initialize instance variables.

        Description of argument(s):
        host        The IP or host name of the system to subscribe to.
        username    The username for the host system.
        password    The password for the host system.
        """
        self.__host = host
        self.__user = username
        self.__password = password
        self.__websocket = None

    def __del__(self):
        try:
            self.__websocket.close()
        except AttributeError:
            pass

    def login(self):
        r"""
        Login and return session object.
        """
        http_header = {"Content-Type": "application/json"}
        session = requests.session()
        response = session.post(
            "https://" + self.__host + "/login",
            headers=http_header,
            json={"data": [self.__user, self.__password]},
            verify=False,
            timeout=30,
        )
        gv.valid_value(response.status_code, valid_values=[200])
        login_response = json.loads(response.text)
        gp.qprint_var(login_response)
        gv.valid_value(login_response["status"], valid_values=["ok"])
        return session

    def subscribe(self, dbus_path, enable_trace=False):
        r"""
        Subscribe to the given path and return a list of event notifications.

        For more details on "subscribe" and "events" go to
        https://github.com/openbmc/docs/blob/master/rest-api.md#event-subscription-protocol

        Example robot code:
        ${event_notifications}=  Subscribe  /xyz/openbmc_project/sensors
        Rprint Vars  event_notifications

        Example output:
        event_notifications:
          [0]:
            [interface]:             xyz.openbmc_project.Sensor.Value
            [path]:                  /xyz/openbmc_project/sensors/temperature/ambient
            [event]:                 PropertiesChanged
            [properties]:
              [Value]:               23813

        Description of argument(s):
        dbus_path              The subscribing event's path (e.g.
                               "/xyz/openbmc_project/sensors").
        enable_trace           Enable or disable trace.
        """

        session = self.login()
        cookies = session.cookies.get_dict()
        # Convert from dictionary to a string of the following format:
        # key=value;key=value...
        cookies = gp.sprint_var(
            cookies,
            fmt=gp.no_header() | gp.strip_brackets(),
            col1_width=0,
            trailing_char="",
            delim="=",
        ).replace("\n", ";")

        websocket.enableTrace(enable_trace)
        self.__websocket = websocket.create_connection(
            f"wss://{self.__host}/subscribe",
            sslopt={"cert_reqs": ssl.CERT_NONE},
            cookie=cookies,
        )
        dbus_path = [path.strip() for path in dbus_path.split(",")]
        dbus_path = {"paths": dbus_path}

        self.__websocket.send(json.dumps(dbus_path))
        event_notifications = json.loads(self.__websocket.recv())
        self.__websocket.close()
        return event_notifications