depends.py 3.1 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798
  1. #
  2. # Copyright (C) 2016 Intel Corporation
  3. #
  4. # SPDX-License-Identifier: MIT
  5. #
  6. from unittest import SkipTest
  7. from oeqa.core.exception import OEQADependency
  8. from . import OETestDiscover, registerDecorator
  9. def _add_depends(registry, case, depends):
  10. module_name = case.__module__
  11. class_name = case.__class__.__name__
  12. case_id = case.id()
  13. for depend in depends:
  14. dparts = depend.split('.')
  15. if len(dparts) == 1:
  16. depend_id = ".".join((module_name, class_name, dparts[0]))
  17. elif len(dparts) == 2:
  18. depend_id = ".".join((module_name, dparts[0], dparts[1]))
  19. else:
  20. depend_id = depend
  21. if not case_id in registry:
  22. registry[case_id] = []
  23. if not depend_id in registry[case_id]:
  24. registry[case_id].append(depend_id)
  25. def _validate_test_case_depends(cases, depends):
  26. for case in depends:
  27. if not case in cases:
  28. continue
  29. for dep in depends[case]:
  30. if not dep in cases:
  31. raise OEQADependency("TestCase %s depends on %s and isn't available"\
  32. ", cases available %s." % (case, dep, str(cases.keys())))
  33. def _order_test_case_by_depends(cases, depends):
  34. def _dep_resolve(graph, node, resolved, seen):
  35. seen.append(node)
  36. for edge in graph[node]:
  37. if edge not in resolved:
  38. if edge in seen:
  39. raise OEQADependency("Test cases %s and %s have a circular" \
  40. " dependency." % (node, edge))
  41. _dep_resolve(graph, edge, resolved, seen)
  42. resolved.append(node)
  43. dep_graph = {}
  44. dep_graph['__root__'] = cases.keys()
  45. for case in cases:
  46. if case in depends:
  47. dep_graph[case] = depends[case]
  48. else:
  49. dep_graph[case] = []
  50. cases_ordered = []
  51. _dep_resolve(dep_graph, '__root__', cases_ordered, [])
  52. cases_ordered.remove('__root__')
  53. return [cases[case_id] for case_id in cases_ordered]
  54. def _skipTestDependency(case, depends):
  55. for dep in depends:
  56. found = False
  57. for test, _ in case.tc.results.successes:
  58. if test.id() == dep:
  59. found = True
  60. break
  61. if not found:
  62. raise SkipTest("Test case %s depends on %s but it didn't pass/run." \
  63. % (case.id(), dep))
  64. @registerDecorator
  65. class OETestDepends(OETestDiscover):
  66. attrs = ('depends',)
  67. def bind(self, registry, case):
  68. super(OETestDepends, self).bind(registry, case)
  69. if not registry.get('depends'):
  70. registry['depends'] = {}
  71. _add_depends(registry['depends'], case, self.depends)
  72. @staticmethod
  73. def discover(registry):
  74. if registry.get('depends'):
  75. _validate_test_case_depends(registry['cases'], registry['depends'])
  76. return _order_test_case_by_depends(registry['cases'], registry['depends'])
  77. else:
  78. return [registry['cases'][case_id] for case_id in registry['cases']]
  79. def setUpDecorator(self):
  80. _skipTestDependency(self.case, self.depends)