context.py 6.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199
  1. ## Copyright (C) 2016 Intel Corporation
  2. #
  3. # SPDX-License-Identifier: MIT
  4. #
  5. import os
  6. import sys
  7. import json
  8. import time
  9. import logging
  10. import collections
  11. from oeqa.core.loader import OETestLoader
  12. from oeqa.core.runner import OETestRunner
  13. from oeqa.core.exception import OEQAMissingManifest, OEQATestNotFound
  14. class OETestContext(object):
  15. loaderClass = OETestLoader
  16. runnerClass = OETestRunner
  17. files_dir = os.path.abspath(os.path.join(os.path.dirname(
  18. os.path.abspath(__file__)), "../files"))
  19. def __init__(self, td=None, logger=None):
  20. if not type(td) is dict:
  21. raise TypeError("td isn't dictionary type")
  22. self.td = td
  23. self.logger = logger
  24. self._registry = {}
  25. self._registry['cases'] = collections.OrderedDict()
  26. def _read_modules_from_manifest(self, manifest):
  27. if not os.path.exists(manifest):
  28. raise OEQAMissingManifest("Manifest does not exist on %s" % manifest)
  29. modules = []
  30. for line in open(manifest).readlines():
  31. line = line.strip()
  32. if line and not line.startswith("#"):
  33. modules.append(line)
  34. return modules
  35. def skipTests(self, skips):
  36. if not skips:
  37. return
  38. for test in self.suites:
  39. for skip in skips:
  40. if test.id().startswith(skip):
  41. setattr(test, 'setUp', lambda: test.skipTest('Skip by the command line argument "%s"' % skip))
  42. def loadTests(self, module_paths, modules=[], tests=[],
  43. modules_manifest="", modules_required=[], filters={}):
  44. if modules_manifest:
  45. modules = self._read_modules_from_manifest(modules_manifest)
  46. self.loader = self.loaderClass(self, module_paths, modules, tests,
  47. modules_required, filters)
  48. self.suites = self.loader.discover()
  49. def runTests(self, processes=None, skips=[]):
  50. self.runner = self.runnerClass(self, descriptions=False, verbosity=2)
  51. # Dinamically skip those tests specified though arguments
  52. self.skipTests(skips)
  53. self._run_start_time = time.time()
  54. if processes:
  55. from oeqa.core.utils.concurrencytest import ConcurrentTestSuite
  56. concurrent_suite = ConcurrentTestSuite(self.suites, processes)
  57. result = self.runner.run(concurrent_suite)
  58. else:
  59. self.runner.buffer = True
  60. result = self.runner.run(self.suites)
  61. self._run_end_time = time.time()
  62. return result
  63. def listTests(self, display_type):
  64. self.runner = self.runnerClass(self, verbosity=2)
  65. return self.runner.list_tests(self.suites, display_type)
  66. class OETestContextExecutor(object):
  67. _context_class = OETestContext
  68. _script_executor = 'oe-test'
  69. name = 'core'
  70. help = 'core test component example'
  71. description = 'executes core test suite example'
  72. default_cases = [os.path.join(os.path.abspath(os.path.dirname(__file__)),
  73. 'cases/example')]
  74. default_test_data = os.path.join(default_cases[0], 'data.json')
  75. default_tests = None
  76. def register_commands(self, logger, subparsers):
  77. self.parser = subparsers.add_parser(self.name, help=self.help,
  78. description=self.description, group='components')
  79. self.default_output_log = '%s-results-%s.log' % (self.name,
  80. time.strftime("%Y%m%d%H%M%S"))
  81. self.parser.add_argument('--output-log', action='store',
  82. default=self.default_output_log,
  83. help="results output log, default: %s" % self.default_output_log)
  84. group = self.parser.add_mutually_exclusive_group()
  85. group.add_argument('--run-tests', action='store', nargs='+',
  86. default=self.default_tests,
  87. help="tests to run in <module>[.<class>[.<name>]]")
  88. group.add_argument('--list-tests', action='store',
  89. choices=('module', 'class', 'name'),
  90. help="lists available tests")
  91. if self.default_test_data:
  92. self.parser.add_argument('--test-data-file', action='store',
  93. default=self.default_test_data,
  94. help="data file to load, default: %s" % self.default_test_data)
  95. else:
  96. self.parser.add_argument('--test-data-file', action='store',
  97. help="data file to load")
  98. if self.default_cases:
  99. self.parser.add_argument('CASES_PATHS', action='store',
  100. default=self.default_cases, nargs='*',
  101. help="paths to directories with test cases, default: %s"\
  102. % self.default_cases)
  103. else:
  104. self.parser.add_argument('CASES_PATHS', action='store',
  105. nargs='+', help="paths to directories with test cases")
  106. self.parser.set_defaults(func=self.run)
  107. def _setup_logger(self, logger, args):
  108. formatter = logging.Formatter('%(asctime)s - ' + self.name + \
  109. ' - %(levelname)s - %(message)s')
  110. sh = logger.handlers[0]
  111. sh.setFormatter(formatter)
  112. fh = logging.FileHandler(args.output_log)
  113. fh.setFormatter(formatter)
  114. logger.addHandler(fh)
  115. return logger
  116. def _process_args(self, logger, args):
  117. self.tc_kwargs = {}
  118. self.tc_kwargs['init'] = {}
  119. self.tc_kwargs['load'] = {}
  120. self.tc_kwargs['list'] = {}
  121. self.tc_kwargs['run'] = {}
  122. self.tc_kwargs['init']['logger'] = self._setup_logger(logger, args)
  123. if args.test_data_file:
  124. self.tc_kwargs['init']['td'] = json.load(
  125. open(args.test_data_file, "r"))
  126. else:
  127. self.tc_kwargs['init']['td'] = {}
  128. if args.run_tests:
  129. self.tc_kwargs['load']['modules'] = args.run_tests
  130. self.tc_kwargs['load']['modules_required'] = args.run_tests
  131. else:
  132. self.tc_kwargs['load']['modules'] = []
  133. self.tc_kwargs['run']['skips'] = []
  134. self.module_paths = args.CASES_PATHS
  135. def _pre_run(self):
  136. pass
  137. def run(self, logger, args):
  138. self._process_args(logger, args)
  139. self.tc = self._context_class(**self.tc_kwargs['init'])
  140. try:
  141. self.tc.loadTests(self.module_paths, **self.tc_kwargs['load'])
  142. except OEQATestNotFound as ex:
  143. logger.error(ex)
  144. sys.exit(1)
  145. if args.list_tests:
  146. rc = self.tc.listTests(args.list_tests, **self.tc_kwargs['list'])
  147. else:
  148. self._pre_run()
  149. rc = self.tc.runTests(**self.tc_kwargs['run'])
  150. rc.logDetails()
  151. rc.logSummary(self.name)
  152. output_link = os.path.join(os.path.dirname(args.output_log),
  153. "%s-results.log" % self.name)
  154. if os.path.exists(output_link):
  155. os.remove(output_link)
  156. os.symlink(args.output_log, output_link)
  157. return rc
  158. _executor_class = OETestContextExecutor