Initial work on reports

Author: jeremystretch

.gitignore

@@ -1,6 +1,8 @@
 *.pyc
 /netbox/netbox/configuration.py
 /netbox/netbox/ldap_config.py
+/netbox/reports/*
+!/netbox/reports/__init__.py
 /netbox/static
 .idea
 /*.sh

netbox/extras/constants.py

@@ -62,3 +62,17 @@
     (ACTION_DELETE, 'deleted'),
     (ACTION_BULK_DELETE, 'bulk deleted'),
 )
+
+# Report logging levels
+LOG_DEFAULT = 0
+LOG_SUCCESS = 10
+LOG_INFO = 20
+LOG_WARNING = 30
+LOG_FAILURE = 40
+LOG_LEVEL_CODES = {
+    LOG_DEFAULT: 'default',
+    LOG_SUCCESS: 'success',
+    LOG_INFO: 'info',
+    LOG_WARNING: 'warning',
+    LOG_FAILURE: 'failure',
+}

netbox/extras/management/commands/runreport.py

@@ -0,0 +1,47 @@
+from __future__ import unicode_literals
+import importlib
+import inspect
+
+from django.core.management.base import BaseCommand, CommandError
+from django.utils import timezone
+
+from extras.reports import Report
+
+
+class Command(BaseCommand):
+    help = "Run a report to validate data in NetBox"
+
+    def add_arguments(self, parser):
+        parser.add_argument('reports', nargs='+', help="Report(s) to run")
+        # parser.add_argument('--verbose', action='store_true', default=False, help="Print all logs")
+
+    def handle(self, *args, **options):
+
+        # Gather all reports to be run
+        reports = []
+        for module_name in options['reports']:
+            try:
+                report_module = importlib.import_module('reports.report_{}'.format(module_name))
+            except ImportError:
+                self.stdout.write(
+                    "Report '{}' not found. Ensure that the report has been saved as 'report_{}.py' in the reports "
+                    "directory.".format(module_name, module_name)
+                )
+                return
+            for name, cls in inspect.getmembers(report_module, inspect.isclass):
+                if cls in Report.__subclasses__():
+                    reports.append((name, cls))
+
+        # Run reports
+        for name, report in reports:
+            self.stdout.write("[{:%H:%M:%S}] Running report {}...".format(timezone.now(), name))
+            report = report()
+            report.run()
+            status = self.style.ERROR('FAILED') if report.failed else self.style.SUCCESS('SUCCESS')
+            self.stdout.write("[{:%H:%M:%S}] {}: {}".format(timezone.now(), name, status))
+            for test_name, attrs in report.results.items():
+                self.stdout.write("    {}: {} success, {} info, {} warning, {} failed".format(
+                    test_name, attrs['success'], attrs['info'], attrs['warning'], attrs['failed']
+                ))
+
+        self.stdout.write("[{:%H:%M:%S}] Finished".format(timezone.now()))

netbox/extras/reports.py

@@ -0,0 +1,100 @@
+from collections import OrderedDict
+
+from django.utils import timezone
+
+from .constants import LOG_DEFAULT, LOG_FAILURE, LOG_INFO, LOG_LEVEL_CODES, LOG_SUCCESS, LOG_WARNING
+
+
+class Report(object):
+    """
+    NetBox users can extend this object to write custom reports to be used for validating data within NetBox. Each
+    report must have one or more test methods named `test_*`.
+
+    The `results` attribute of a completed report will take the following form:
+
+    {
+        'test_bar': {
+            'failures': 42,
+            'log': [
+                (<datetime>, <level>, <object>, <message>),
+                ...
+            ]
+        },
+        'test_foo': {
+            'failures': 0,
+            'log': [
+                (<datetime>, <level>, <object>, <message>),
+                ...
+            ]
+        }
+    }
+    """
+    results = OrderedDict()
+    active_test = None
+    failed = False
+
+    def __init__(self):
+
+        # Compile test methods and initialize results skeleton
+        test_methods = []
+        for method in dir(self):
+            if method.startswith('test_') and callable(getattr(self, method)):
+                test_methods.append(method)
+                self.results[method] = OrderedDict([
+                    ('success', 0),
+                    ('info', 0),
+                    ('warning', 0),
+                    ('failed', 0),
+                    ('log', []),
+                ])
+        if not test_methods:
+            raise Exception("A report must contain at least one test method.")
+        self.test_methods = test_methods
+
+    def _log(self, obj, message, level=LOG_DEFAULT):
+        """
+        Log a message from a test method. Do not call this method directly; use one of the log_* wrappers below.
+        """
+        if level not in LOG_LEVEL_CODES:
+            raise Exception("Unknown logging level: {}".format(level))
+        logline = [timezone.now(), level, obj, message]
+        self.results[self.active_test]['log'].append(logline)
+
+    def log_success(self, obj, message=None):
+        """
+        Record a successful test against an object. Logging a message is optional.
+        """
+        if message:
+            self._log(obj, message, level=LOG_SUCCESS)
+        self.results[self.active_test]['success'] += 1
+
+    def log_info(self, obj, message):
+        """
+        Log an informational message.
+        """
+        self._log(obj, message, level=LOG_INFO)
+        self.results[self.active_test]['info'] += 1
+
+    def log_warning(self, obj, message):
+        """
+        Log a warning.
+        """
+        self._log(obj, message, level=LOG_WARNING)
+        self.results[self.active_test]['warning'] += 1
+
+    def log_failure(self, obj, message):
+        """
+        Log a failure. Calling this method will automatically mark the report as failed.
+        """
+        self._log(obj, message, level=LOG_FAILURE)
+        self.results[self.active_test]['failed'] += 1
+        self.failed = True
+
+    def run(self):
+        """
+        Run the report. Each test method will be executed in order.
+        """
+        for method_name in self.test_methods:
+            self.active_test = method_name
+            test_method = getattr(self, method_name)
+            test_method()