xref: /openbmc/openbmc-test-automation/lib/event_notification.py (revision 249fbcc0be4641e2d43987badc764e1f73a6ebeb) 
1#!/usr/bin/env python3
2r"""
3This is an extended user library to support Robot Selenium code.
4The class contains functions which the robot framework will use
5and import as a user-defined keyword.
6"""
7
8import json
9import ssl
10
11import gen_print as gp  # NOQA
12import gen_valid as gv  # NOQA
13import requests
14import websocket
15
16
17class event_notification:  # NOQA
18    r"""
19    Main class to subscribe and receive event notifications.
20    """
21
22    def __init__(self, host, username, password):
23        r"""
24        Initialize instance variables.
25
26        Description of argument(s):
27        host        The IP or host name of the system to subscribe to.
28        username    The username for the host system.
29        password    The password for the host system.
30        """
31        self.__host = host
32        self.__user = username
33        self.__password = password
34        self.__websocket = None
35
36    def __del__(self):
37        try:
38            self.__websocket.close()
39        except AttributeError:
40            pass
41
42    def login(self):
43        r"""
44        Login and return session object.
45        """
46        http_header = {"Content-Type": "application/json"}
47        session = requests.session()
48        response = session.post(
49            "https://" + self.__host + "/login",
50            headers=http_header,
51            json={"data": [self.__user, self.__password]},
52            verify=False,
53            timeout=30,
54        )
55        gv.valid_value(response.status_code, valid_values=[200])
56        login_response = json.loads(response.text)
57        gp.qprint_var(login_response)
58        gv.valid_value(login_response["status"], valid_values=["ok"])
59        return session
60
61    def subscribe(self, dbus_path, enable_trace=False):
62        r"""
63        Subscribe to the given path and return a list of event notifications.
64
65        For more details on "subscribe" and "events" go to
66        https://github.com/openbmc/docs/blob/master/rest-api.md#event-subscription-protocol
67
68        Example robot code:
69        ${event_notifications}=  Subscribe  /xyz/openbmc_project/sensors
70        Rprint Vars  event_notifications
71
72        Example output:
73        event_notifications:
74          [0]:
75            [interface]:             xyz.openbmc_project.Sensor.Value
76            [path]:                  /xyz/openbmc_project/sensors/temperature/ambient
77            [event]:                 PropertiesChanged
78            [properties]:
79              [Value]:               23813
80
81        Description of argument(s):
82        dbus_path              The subscribing event's path (e.g.
83                               "/xyz/openbmc_project/sensors").
84        enable_trace           Enable or disable trace.
85        """
86
87        session = self.login()
88        cookies = session.cookies.get_dict()
89        # Convert from dictionary to a string of the following format:
90        # key=value;key=value...
91        cookies = gp.sprint_var(
92            cookies,
93            fmt=gp.no_header() | gp.strip_brackets(),
94            col1_width=0,
95            trailing_char="",
96            delim="=",
97        ).replace("\n", ";")
98
99        websocket.enableTrace(enable_trace)
100        self.__websocket = websocket.create_connection(
101            f"wss://{self.__host}/subscribe",
102            sslopt={"cert_reqs": ssl.CERT_NONE},
103            cookie=cookies,
104        )
105        dbus_path = [path.strip() for path in dbus_path.split(",")]
106        dbus_path = {"paths": dbus_path}
107
108        self.__websocket.send(json.dumps(dbus_path))
109        event_notifications = json.loads(self.__websocket.recv())
110        self.__websocket.close()
111        return event_notifications
112