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