rwlaunchpad/plugins/rwmonparam/test/utest_aggregator.py

   1 #!/usr/bin/env python3
   2
   3 #
   4 #   Copyright 2016 RIFT.IO Inc
   5 #
   6 #   Licensed under the Apache License, Version 2.0 (the "License");
   7 #   you may not use this file except in compliance with the License.
   8 #   You may obtain a copy of the License at
   9 #
  10 #       http://www.apache.org/licenses/LICENSE-2.0
  11 #
  12 #   Unless required by applicable law or agreed to in writing, software
  13 #   distributed under the License is distributed on an "AS IS" BASIS,
  14 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15 #   See the License for the specific language governing permissions and
  16 #   limitations under the License.
  17 #
  18
  19
  20 import asyncio
  21 import base64
  22 import logging
  23 import os
  24 import sys
  25 import tornado.escape
  26 import tornado.platform.asyncio
  27 import tornado.testing
  28 import tornado.web
  29 import unittest
  30 import xmlrunner
  31
  32 import rift.tasklets.rwmonparam.aggregator as aggregator
  33
  34
  35 from gi.repository import VnfrYang
  36
  37 logger = logging.getLogger("mon_params_test.py")
  38
  39
  40 class TestAggregator(unittest.TestCase):
  41
  42     def test_int_aggregator(self):
  43         int_agg = aggregator.IntValueAggregator("SUM", [1, 2, 3])
  44         self.assertEqual(int_agg.aggregate(), ("value_integer", 6))
  45
  46         int_agg = aggregator.IntValueAggregator("AVERAGE", [1, 2, 3])
  47         self.assertEqual(int_agg.aggregate(), ("value_integer", 2))
  48
  49         int_agg = aggregator.IntValueAggregator("MAXIMUM", [1, 2, 3])
  50         self.assertEqual(int_agg.aggregate(), ("value_integer", 3))
  51
  52         int_agg = aggregator.IntValueAggregator("MINIMUM", [1, 2, 3])
  53         self.assertEqual(int_agg.aggregate(), ("value_integer", 1))
  54
  55         int_agg = aggregator.IntValueAggregator("COUNT", [1, 2, 3])
  56         self.assertEqual(int_agg.aggregate(), ("value_integer", 3))
  57
  58     def test_decimal_aggregator(self):
  59         int_agg = aggregator.DecimalValueAggregator("SUM", [1.1, 2, 3])
  60         self.assertEqual(int_agg.aggregate(), ("value_decimal", 6.1))
  61
  62         int_agg = aggregator.DecimalValueAggregator("AVERAGE", [1, 2, 3])
  63         self.assertEqual(int_agg.aggregate(), ("value_decimal", 2.0))
  64
  65         int_agg = aggregator.DecimalValueAggregator("MAXIMUM", [1, 2, 3.3])
  66         self.assertEqual(int_agg.aggregate(), ("value_decimal", 3.3))
  67
  68         int_agg = aggregator.DecimalValueAggregator("MINIMUM", [1.1, 2, 3.3])
  69         self.assertEqual(int_agg.aggregate(), ("value_decimal", 1.1))
  70
  71         int_agg = aggregator.DecimalValueAggregator("COUNT", [1.1, 2, 3.3])
  72         self.assertEqual(int_agg.aggregate(), ("value_decimal", 3))
  73
  74
  75 def main(argv=sys.argv[1:]):
  76
  77     # The unittest framework requires a program name, so use the name of this
  78     # file instead (we do not want to have to pass a fake program name to main
  79     # when this is called from the interpreter).
  80     unittest.main(
  81             argv=[__file__] + argv,
  82             testRunner=xmlrunner.XMLTestRunner(output=os.environ["RIFT_MODULE_TEST"])
  83             )
  84
  85 if __name__ == '__main__':
  86     main()
  87