reports.py 6.3 KB

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