black: re-format
black and isort are enabled in the openbmc-build-scripts on Python files
to have a consistent formatting. Re-run the formatter on the whole
repository.
Signed-off-by: Patrick Williams <patrick@stwcx.xyz>
Change-Id: I944f1915ece753f72a3fa654902d445a9749d0f9
diff --git a/lib/event_notification.py b/lib/event_notification.py
index 363caf4..53c010e 100755
--- a/lib/event_notification.py
+++ b/lib/event_notification.py
@@ -1,14 +1,15 @@
#!/usr/bin/env python3
-import requests
-import websocket
import json
import ssl
-import gen_valid as gv
+
import gen_print as gp
+import gen_valid as gv
+import requests
+import websocket
-class event_notification():
+class event_notification:
r"""
Main class to subscribe and receive event notifications.
"""
@@ -36,20 +37,22 @@
r"""
Login and return session object.
"""
- http_header = {'Content-Type': 'application/json'}
+ 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)
+ 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'])
+ 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.
@@ -79,15 +82,21 @@
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", ";")
+ 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("wss://{host}/subscribe".format(host=self.__host),
- sslopt={"cert_reqs": ssl.CERT_NONE},
- cookie=cookies)
- dbus_path = [path.strip() for path in dbus_path.split(',')]
+ self.__websocket = websocket.create_connection(
+ "wss://{host}/subscribe".format(host=self.__host),
+ 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))