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