Issue #92 - Adding script for testing new notification options

test_scripts/notification_tests.py

@@ -0,0 +1,118 @@
+import ait.gui
+import ait.core
+import mock
+from collections import defaultdict
+from time import sleep
+from ait.core import cfg, tlm, limits
+
+"""
+This script tests that notifications are triggered at the threshold
+and frequency specified in the configs. To run the test, execute the script,
+wait up to 10 seconds for it to run, and then quit the process using Ctrl-C.
+"""
+
+
+@mock.patch('ait.core.notify.trigger_notification')
+def enable_monitoring_test(mock_notify):
+    try:
+        set_notif_options(thrshld=3, freq=2)
+        print_notif_options()
+
+        ait.gui.init('localhost', 8080)
+        ait.gui.enable_monitoring()
+
+        sleep(1)
+        add_telemetry()
+
+        ait.gui.wait()
+        sleep(1)
+        ait.gui.cleanup()
+
+    except KeyboardInterrupt:
+        ait.core.log.info('Received Ctrl-C.  Stopping AIT GUI.')
+        ait.gui.cleanup()
+
+    except Exception as e:
+        ait.core.log.error('AIT GUI error: %s' % str(e))
+
+    # define expected calls
+    # value 0 should notify 3 times (msg 3, 5, 7)
+    call_list = [mock.mock.call(
+                  'limit-error',
+                  'Field Voltage_A error out of limit with value 0')] * 3
+    # value 50 should notify 3 times (msg 3, 5, 7)
+    call_list.extend([mock.mock.call(
+                      'limit-error',
+                      'Field Voltage_A error out of limit with value 50')] * 3)
+
+    print('Notification was triggered {} times.'
+          .format(mock_notify.call_count))
+
+    assert mock_notify.call_args_list == call_list
+
+
+def add_telemetry():
+    # error value
+    pkt.Voltage_A = 0
+    for i in range(8):
+        ait.gui.Sessions.addTelemetry(pkt_defn.uid, pkt._data)
+    sleep(1)
+
+    # good value - no notifs
+    pkt.Voltage_A = 20
+    for i in range(2):
+        ait.gui.Sessions.addTelemetry(pkt_defn.uid, pkt._data)
+    sleep(1)
+
+    # error value
+    pkt.Voltage_A = 50
+    for i in range(8):
+        ait.gui.Sessions.addTelemetry(pkt_defn.uid, pkt._data)
+
+
+def get_packet_and_defn():
+    first_stream = ait.config.get('gui.telemetry')[0]
+    stream = cfg.AitConfig(config=first_stream).get('stream')
+    name = stream.get('name', '<unnamed>')
+    pkt_defn = tlm.getDefaultDict().get(name, None)
+    pkt = tlm.Packet(pkt_defn)
+
+    return pkt_defn, pkt
+
+
+def get_limit_dict():
+    limit_dict = defaultdict(dict)
+    for k, v in limits.getDefaultDict().iteritems():
+        packet, field = k.split('.')
+        limit_dict[packet][field] = v
+
+    return limit_dict
+
+
+def print_notif_options():
+    thrshld = ait.config.get('notifications.options.threshold')
+    freq = ait.config.get('notifications.options.frequency')
+    print('Threshold and frequency are {} and {} respectively.'
+           .format(thrshld, freq))
+
+
+def set_notif_options(thrshld=None, freq=None):
+    pathvars = {}
+    if thrshld:
+        pathvars['notifications.options.threshold'] = thrshld
+        print('Changing notif threshold to {}.'.format(thrshld))
+    if freq:
+        pathvars['notifications.options.frequency'] = freq
+        print('Changing notif freq to {}.'.format(freq))
+
+    ait.config.addPathVariables(pathvars)
+    ait.config.reload()
+
+
+if __name__ == "__main__":
+    global pkt_defn, pkt, limit_dict, limit_defn
+    pkt_defn, pkt = get_packet_and_defn()
+    limit_dict = get_limit_dict()
+    limit_defn = limit_dict[pkt_defn.name]['Voltage_A']
+
+    enable_monitoring_test()