ceph/src/pybind/mgr/dashboard/tests/test_notification.py

   1 # -*- coding: utf-8 -*-
   2 from __future__ import absolute_import
   3
   4 import random
   5 import time
   6 import unittest
   7
   8
   9 from ..tools import NotificationQueue
  10
  11
  12 class Listener(object):
  13     # pylint: disable=too-many-instance-attributes
  14     def __init__(self):
  15         self.type1 = []
  16         self.type1_ts = []
  17         self.type2 = []
  18         self.type2_ts = []
  19         self.type1_3 = []
  20         self.type1_3_ts = []
  21         self.all = []
  22         self.all_ts = []
  23
  24     def register(self):
  25         NotificationQueue.register(self.log_type1, 'type1', priority=90)
  26         NotificationQueue.register(self.log_type2, 'type2')
  27         NotificationQueue.register(self.log_type1_3, ['type1', 'type3'])
  28         NotificationQueue.register(self.log_all, priority=50)
  29
  30         # these should be ignored by the queue
  31         NotificationQueue.register(self.log_type1, 'type1')
  32         NotificationQueue.register(self.log_type1_3, ['type1', 'type3'])
  33         NotificationQueue.register(self.log_all)
  34
  35     def log_type1(self, val):
  36         self.type1_ts.append(time.time())
  37         self.type1.append(val)
  38
  39     def log_type2(self, val):
  40         self.type2_ts.append(time.time())
  41         self.type2.append(val)
  42
  43     def log_type1_3(self, val):
  44         self.type1_3_ts.append(time.time())
  45         self.type1_3.append(val)
  46
  47     def log_all(self, val):
  48         self.all_ts.append(time.time())
  49         self.all.append(val)
  50
  51     def clear(self):
  52         self.type1 = []
  53         self.type1_ts = []
  54         self.type2 = []
  55         self.type2_ts = []
  56         self.type1_3 = []
  57         self.type1_3_ts = []
  58         self.all = []
  59         self.all_ts = []
  60         NotificationQueue.deregister(self.log_type1, 'type1')
  61         NotificationQueue.deregister(self.log_type2, 'type2')
  62         NotificationQueue.deregister(self.log_type1_3, ['type1', 'type3'])
  63         NotificationQueue.deregister(self.log_all)
  64
  65
  66 class NotificationQueueTest(unittest.TestCase):
  67     @classmethod
  68     def setUpClass(cls):
  69         cls.listener = Listener()
  70
  71     def setUp(self):
  72         self.listener.register()
  73
  74     def tearDown(self):
  75         self.listener.clear()
  76
  77     def test_invalid_register(self):
  78         with self.assertRaises(Exception) as ctx:
  79             NotificationQueue.register(None, 1)
  80         self.assertEqual(str(ctx.exception),
  81                          "n_types param is neither a string nor a list")
  82
  83     def test_notifications(self):
  84         NotificationQueue.start_queue()
  85         NotificationQueue.new_notification('type1', 1)
  86         NotificationQueue.new_notification('type2', 2)
  87         NotificationQueue.new_notification('type3', 3)
  88         NotificationQueue.stop()
  89         self.assertEqual(self.listener.type1, [1])
  90         self.assertEqual(self.listener.type2, [2])
  91         self.assertEqual(self.listener.type1_3, [1, 3])
  92         self.assertEqual(self.listener.all, [1, 2, 3])
  93
  94         # validate priorities
  95         self.assertLessEqual(self.listener.type1_3_ts[0], self.listener.all_ts[0])
  96         self.assertLessEqual(self.listener.all_ts[0], self.listener.type1_ts[0])
  97         self.assertLessEqual(self.listener.type2_ts[0], self.listener.all_ts[1])
  98         self.assertLessEqual(self.listener.type1_3_ts[1], self.listener.all_ts[2])
  99
 100     def test_notifications2(self):
 101         NotificationQueue.start_queue()
 102         for i in range(0, 600):
 103             typ = "type{}".format(i % 3 + 1)
 104             if random.random() < 0.5:
 105                 time.sleep(0.002)
 106             NotificationQueue.new_notification(typ, i)
 107         NotificationQueue.stop()
 108         for i in range(0, 600):
 109             typ = i % 3 + 1
 110             if typ == 1:
 111                 self.assertIn(i, self.listener.type1)
 112                 self.assertIn(i, self.listener.type1_3)
 113             elif typ == 2:
 114                 self.assertIn(i, self.listener.type2)
 115             elif typ == 3:
 116                 self.assertIn(i, self.listener.type1_3)
 117             self.assertIn(i, self.listener.all)
 118
 119         self.assertEqual(len(self.listener.type1), 200)
 120         self.assertEqual(len(self.listener.type2), 200)
 121         self.assertEqual(len(self.listener.type1_3), 400)
 122         self.assertEqual(len(self.listener.all), 600)
 123
 124     def test_deregister(self):
 125         NotificationQueue.start_queue()
 126         NotificationQueue.new_notification('type1', 1)
 127         NotificationQueue.new_notification('type3', 3)
 128         NotificationQueue.stop()
 129         self.assertEqual(self.listener.type1, [1])
 130         self.assertEqual(self.listener.type1_3, [1, 3])
 131
 132         NotificationQueue.start_queue()
 133         NotificationQueue.deregister(self.listener.log_type1_3, ['type1'])
 134         NotificationQueue.new_notification('type1', 4)
 135         NotificationQueue.new_notification('type3', 5)
 136         NotificationQueue.stop()
 137         self.assertEqual(self.listener.type1, [1, 4])
 138         self.assertEqual(self.listener.type1_3, [1, 3, 5])