rwlaunchpad/plugins/rwmonitor/test/repro_tasklet_test.py

   1 #!/usr/bin/env python3
   2 import argparse
   3 import asyncio
   4 import gi
   5 import logging
   6 import os
   7 import tempfile
   8 import unittest
   9 import xmlrunner
  10
  11 # Add the current directory to the PLUGINDIR so we can use the plugin
  12 # file added here.
  13 os.environ["PLUGINDIR"] += (":" + os.path.dirname(os.path.realpath(__file__)))
  14 gi.require_version("RwDts", "1.0")
  15 gi.require_version("RwVnfrYang", "1.0")
  16 from gi.repository import (
  17     RwDts,
  18     RwVnfrYang,
  19 )
  20
  21 import rift.tasklets
  22 import rift.test.dts
  23
  24 gi.require_version('RwKeyspec', '1.0')
  25 from gi.repository.RwKeyspec import quoted_key
  26
  27 class RwLogTestCase(rift.test.dts.AbstractDTSTest):
  28     # Disable the log_utest_mode so that log messages actually get logged
  29     # using the rwlog handler since that is what we are testing here.
  30     log_utest_mode = False
  31
  32     @classmethod
  33     def configure_suite(cls, rwmain):
  34         pass
  35
  36     @classmethod
  37     def start_test_tasklet(cls):
  38         cls.rwmain.add_tasklet(
  39                 os.path.join(
  40                     os.path.dirname(os.path.realpath(__file__)),
  41                     'reprotesttasklet-python'
  42                     ),
  43                 'reprotesttasklet-python'
  44                 )
  45
  46     @classmethod
  47     def configure_schema(cls):
  48         return RwVnfrYang.get_schema()
  49
  50     @classmethod
  51     def configure_timeout(cls):
  52         return 1000000
  53
  54     def configure_test(self, loop, test_id):
  55         self.log.debug("STARTING - %s", self.id())
  56         self.tinfo = self.new_tinfo(self.id())
  57         self.dts = rift.tasklets.DTS(self.tinfo, self.schema, self.loop)
  58
  59     @rift.test.dts.async_test
  60     def test_tasklet_logging(self):
  61         self.start_test_tasklet()
  62
  63         # The logtesttasklet signals being done, by moving into DTS Running state
  64         yield from self.wait_for_tasklet_running("reprotesttasklet-python")
  65         @asyncio.coroutine
  66         def reader():
  67             while True:
  68                 res_iter = yield from self.dts.query_read("D,/vnfr:vnfr-catalog/vnfr:vnfr[vnfr:id={}]/vnfr:vdur[vnfr:id={}]/rw-vnfr:nfvi-metrics".format(
  69                     quoted_key("a7f30def-0942-4425-8454-1ffe02b7db1e"), quoted_key("a7f30def-0942-4425-8454-1ffe02b7db1e"),
  70                     ))
  71                 for ent in res_iter:
  72                     res = yield from ent
  73                     metrics = res.result
  74                     self.log.debug("got metrics result: %s", metrics)
  75
  76         for _ in range(20):
  77             self.loop.create_task(reader())
  78
  79         while True:
  80             yield from asyncio.sleep(.001, loop=self.loop)
  81
  82
  83 def main():
  84     runner = xmlrunner.XMLTestRunner(output=os.environ["RIFT_MODULE_TEST"])
  85
  86     parser = argparse.ArgumentParser()
  87     parser.add_argument('-v', '--verbose', action='store_true')
  88     args, _ = parser.parse_known_args()
  89
  90     RwLogTestCase.log_level = logging.DEBUG if args.verbose else logging.WARN
  91
  92     unittest.main(testRunner=runner)
  93
  94 if __name__ == '__main__':
  95     main()