| #!/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 |