Enable event notification automation base code
Resolves openbmc/openbmc-test-automation#1901
Change-Id: Ieea4c1770b2f676cc9b77804d8c1f44b9a771c63
Signed-off-by: Anusha Dathatri <adathatr@in.ibm.com>
diff --git a/bin/event_notification_util.py b/bin/event_notification_util.py
new file mode 100755
index 0000000..66cf970
--- /dev/null
+++ b/bin/event_notification_util.py
@@ -0,0 +1,57 @@
+#!/usr/bin/env python
+
+r"""
+See help text for details.
+"""
+
+import sys
+
+save_dir_path = sys.path.pop(0)
+
+modules = ['gen_arg', 'gen_print', 'gen_valid', 'event_notification']
+for module in modules:
+ exec("from " + module + " import *")
+
+sys.path.insert(0, save_dir_path)
+
+parser = argparse.ArgumentParser(
+ usage='%(prog)s [OPTIONS]',
+ description="%(prog)s will subscribe and receive event notifications when "
+ + "properties change for the given dbus path.",
+ formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+ prefix_chars='-+')
+parser.add_argument(
+ '--host',
+ default='',
+ help='The host name or IP of the system to subscribe to.')
+parser.add_argument(
+ '--username',
+ default='root',
+ help='The username for the host system.')
+parser.add_argument(
+ '--password',
+ default='',
+ help='The password for the host system.')
+parser.add_argument(
+ '--dbus_path',
+ default='',
+ help='The path to be monitored (e.g. "/xyz/openbmc_project/sensors").')
+parser.add_argument(
+ '--enable_trace',
+ choices=[0, 1],
+ default=0,
+ help='Indicates that trace needs to be enabled.')
+
+
+# Populate stock_list with options we want.
+stock_list = [("test_mode", 0), ("quiet", 0), ("debug", 0)]
+
+
+def main():
+ gen_setup()
+ my_event = event_notification(host, username, password)
+ event_notifications = my_event.subscribe(dbus_path, enable_trace)
+ print_var(event_notifications, fmt=[no_header(), strip_brackets()])
+
+
+main()
diff --git a/lib/event_notification.py b/lib/event_notification.py
new file mode 100755
index 0000000..dac6c5e
--- /dev/null
+++ b/lib/event_notification.py
@@ -0,0 +1,96 @@
+#!/usr/bin/env python
+
+import requests
+import websocket
+import json
+import ssl
+import gen_valid as gv
+import gen_print as gp
+
+
+class event_notification():
+ 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
+
+ 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 subcribing 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("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))
+ event_notifications = json.loads(self.__websocket.recv())
+ self.__websocket.close()
+ return event_notifications
diff --git a/tests/test_event_notification.robot b/tests/test_event_notification.robot
new file mode 100644
index 0000000..1aeb17d
--- /dev/null
+++ b/tests/test_event_notification.robot
@@ -0,0 +1,58 @@
+*** Settings ***
+Documentation Event notification test cases.
+
+Library ../lib/gen_cmd.py
+Library ../lib/var_funcs.py
+Library ../lib/gen_robot_valid.py
+Library ../lib/gen_robot_keyword.py
+Resource ../lib/resource.robot
+Resource ../lib/openbmc_ffdc.robot
+Resource ../lib/rest_client.robot
+
+Test Setup Printn
+
+Test Teardown FFDC On Test Case Fail
+
+*** Test Cases ***
+
+Subscribe And Verify Event Notification
+ [Documentation] Subscribe and verify event notification.
+ [Tags] Subscribe_And_Verify_Event_Notification
+ [Teardown] Run Keyword And Ignore Error Kill Cmd ${popen}
+
+ ${cmd_buf}= Catenate event_notification_util.py --quiet=1 --host=${OPENBMC_HOST}
+ ... --password=${OPENBMC_PASSWORD} --dbus_path=${CONTROL_HOST_URI}power_cap
+ ${popen}= Shell Cmd ${cmd_buf} return_stderr=1 fork=1
+ Rprint Vars popen.pid
+ Qprint Timen Allow child event_notification_util.py job to begin to wait for the event notification.
+ Run Key U Sleep \ 5 seconds
+
+ # Get current reading for debug.
+ ${original_power_cap_settings}= Read Properties ${CONTROL_HOST_URI}power_cap quiet=1
+ Rprint Vars original_power_cap_settings
+
+ # Set power limit out of range.
+ ${power_cap}= Evaluate random.randint(1000, 3000) modules=random
+ Rprint Vars original_power_cap_settings power_cap
+ ${data}= Create Dictionary data=${power_cap}
+ Write Attribute ${CONTROL_HOST_URI}power_cap PowerCap data=${data}
+
+ Qprint Timen Wait for child event_notification_util.py job to see the event notification.
+ Run Key U Sleep \ 5 seconds
+
+ Qprint Timen Retrieving output from spawned event_notification_util.py job.
+ ${rc} ${stdout} ${stderr}= Kill Cmd ${popen}
+ Run Keyword If ${rc} Log to Console ${stderr}
+ Valid Value rc [0]
+ ${event_notification}= Key Value Outbuf To Dict ${stdout} process_indent=1
+ Rprint Vars event_notification
+
+ # Example output:
+ # interface: xyz.openbmc_project.Control.Power.Cap
+ # path: /xyz/openbmc_project/control/host0/power_cap
+ # event: PropertiesChanged
+ # properties:
+ # PowerCap: 1318
+
+ Valid Value event_notification['event'] ['PropertiesChanged']
+ Valid Value event_notification['properties']['powercap'] ['${power_cap}']
\ No newline at end of file