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