11 # Add the current directory to the PLUGINDIR so we can use the plugin
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 (
24 gi
.require_version('RwKeyspec', '1.0')
25 from gi
.repository
.RwKeyspec
import quoted_key
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
33 def configure_suite(cls
, rwmain
):
37 def start_test_tasklet(cls
):
38 cls
.rwmain
.add_tasklet(
40 os
.path
.dirname(os
.path
.realpath(__file__
)),
41 'reprotesttasklet-python'
43 'reprotesttasklet-python'
47 def configure_schema(cls
):
48 return RwVnfrYang
.get_schema()
51 def configure_timeout(cls
):
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
)
59 @rift.test
.dts
.async_test
60 def test_tasklet_logging(self
):
61 self
.start_test_tasklet()
63 # The logtesttasklet signals being done, by moving into DTS Running state
64 yield from self
.wait_for_tasklet_running("reprotesttasklet-python")
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"),
74 self
.log
.debug("got metrics result: %s", metrics
)
77 self
.loop
.create_task(reader())
80 yield from asyncio
.sleep(.001, loop
=self
.loop
)
84 runner
= xmlrunner
.XMLTestRunner(output
=os
.environ
["RIFT_MODULE_TEST"])
86 parser
= argparse
.ArgumentParser()
87 parser
.add_argument('-v', '--verbose', action
='store_true')
88 args
, _
= parser
.parse_known_args()
90 RwLogTestCase
.log_level
= logging
.DEBUG
if args
.verbose
else logging
.WARN
92 unittest
.main(testRunner
=runner
)
94 if __name__
== '__main__':