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