reports.py 5.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189
  1. from __future__ import unicode_literals
  2. import importlib
  3. import inspect
  4. import pkgutil
  5. from collections import OrderedDict
  6. from django.conf import settings
  7. from django.utils import timezone
  8. from .constants import LOG_DEFAULT, LOG_FAILURE, LOG_INFO, LOG_LEVEL_CODES, LOG_SUCCESS, LOG_WARNING
  9. from .models import ReportResult
  10. def is_report(obj):
  11. """
  12. Returns True if the given object is a Report.
  13. """
  14. if obj in Report.__subclasses__():
  15. return True
  16. return False
  17. def get_report(module_name, report_name):
  18. """
  19. Return a specific report from within a module.
  20. """
  21. module = importlib.import_module('reports.{}'.format(module_name))
  22. report = getattr(module, report_name, None)
  23. if report is None:
  24. return None
  25. return report()
  26. def get_reports():
  27. """
  28. Compile a list of all reports available across all modules in the reports path. Returns a list of tuples:
  29. [
  30. (module_name, (report, report, report, ...)),
  31. (module_name, (report, report, report, ...)),
  32. ...
  33. ]
  34. """
  35. module_list = []
  36. # Iterate through all modules within the reports path. These are the user-created files in which reports are
  37. # defined.
  38. for importer, module_name, is_pkg in pkgutil.walk_packages([settings.REPORTS_ROOT]):
  39. module = importlib.import_module('reports.{}'.format(module_name))
  40. report_list = [cls() for _, cls in inspect.getmembers(module, is_report)]
  41. module_list.append((module_name, report_list))
  42. return module_list
  43. class Report(object):
  44. """
  45. NetBox users can extend this object to write custom reports to be used for validating data within NetBox. Each
  46. report must have one or more test methods named `test_*`.
  47. The `_results` attribute of a completed report will take the following form:
  48. {
  49. 'test_bar': {
  50. 'failures': 42,
  51. 'log': [
  52. (<datetime>, <level>, <object>, <message>),
  53. ...
  54. ]
  55. },
  56. 'test_foo': {
  57. 'failures': 0,
  58. 'log': [
  59. (<datetime>, <level>, <object>, <message>),
  60. ...
  61. ]
  62. }
  63. }
  64. """
  65. description = None
  66. def __init__(self):
  67. self._results = OrderedDict()
  68. self.active_test = None
  69. self.failed = False
  70. # Compile test methods and initialize results skeleton
  71. test_methods = []
  72. for method in dir(self):
  73. if method.startswith('test_') and callable(getattr(self, method)):
  74. test_methods.append(method)
  75. self._results[method] = OrderedDict([
  76. ('success', 0),
  77. ('info', 0),
  78. ('warning', 0),
  79. ('failure', 0),
  80. ('log', []),
  81. ])
  82. if not test_methods:
  83. raise Exception("A report must contain at least one test method.")
  84. self.test_methods = test_methods
  85. @property
  86. def module(self):
  87. return self.__module__.rsplit('.', 1)[1]
  88. @property
  89. def name(self):
  90. return self.__class__.__name__
  91. @property
  92. def full_name(self):
  93. return '.'.join([self.module, self.name])
  94. def _log(self, obj, message, level=LOG_DEFAULT):
  95. """
  96. Log a message from a test method. Do not call this method directly; use one of the log_* wrappers below.
  97. """
  98. if level not in LOG_LEVEL_CODES:
  99. raise Exception("Unknown logging level: {}".format(level))
  100. self._results[self.active_test]['log'].append((
  101. timezone.now().isoformat(),
  102. LOG_LEVEL_CODES.get(level),
  103. str(obj) if obj else None,
  104. obj.get_absolute_url() if getattr(obj, 'get_absolute_url', None) else None,
  105. message,
  106. ))
  107. def log(self, message):
  108. """
  109. Log a message which is not associated with a particular object.
  110. """
  111. self._log(None, message, level=LOG_DEFAULT)
  112. def log_success(self, obj, message=None):
  113. """
  114. Record a successful test against an object. Logging a message is optional.
  115. """
  116. if message:
  117. self._log(obj, message, level=LOG_SUCCESS)
  118. self._results[self.active_test]['success'] += 1
  119. def log_info(self, obj, message):
  120. """
  121. Log an informational message.
  122. """
  123. self._log(obj, message, level=LOG_INFO)
  124. self._results[self.active_test]['info'] += 1
  125. def log_warning(self, obj, message):
  126. """
  127. Log a warning.
  128. """
  129. self._log(obj, message, level=LOG_WARNING)
  130. self._results[self.active_test]['warning'] += 1
  131. def log_failure(self, obj, message):
  132. """
  133. Log a failure. Calling this method will automatically mark the report as failed.
  134. """
  135. self._log(obj, message, level=LOG_FAILURE)
  136. self._results[self.active_test]['failure'] += 1
  137. self.failed = True
  138. def run(self):
  139. """
  140. Run the report and return its results. Each test method will be executed in order.
  141. """
  142. for method_name in self.test_methods:
  143. self.active_test = method_name
  144. test_method = getattr(self, method_name)
  145. test_method()
  146. # Delete any previous ReportResult and create a new one to record the result.
  147. ReportResult.objects.filter(report=self.full_name).delete()
  148. result = ReportResult(report=self.full_name, failed=self.failed, data=self._results)
  149. result.save()
  150. self.result = result
  151. # Perform any post-run tasks
  152. self.post_run()
  153. def post_run(self):
  154. """
  155. Extend this method to include any tasks which should execute after the report has been run.
  156. """
  157. pass