SIANXVMT-127: adds manage command that exports python env json

README.md

@@ -65,15 +65,19 @@ urlpatterns = [
 Usage
 -----
 
-The plugin registers some custom REST endpoints which can be used for
-monitoring. Make sure that the **ANX\_MONITORING\_ACCESS\_TOKEN** is
-defined, since this is used for authorization. The endpoints will return
-a 401 HTTP\_STATUS code if the token is not define or invalid, and a 200
-status code otherwise.
+The plugin registers some custom REST endpoints and django management commands 
+which can be used for monitoring. Make sure that the 
+**ANX\_MONITORING\_ACCESS\_TOKEN** is defined, since this is used for 
+authorization of the endpoints. The endpoints will return a 401 HTTP\_STATUS 
+code if the token is not define or invalid, and a 200 status code otherwise.
 
 ### Version monitoring
 
-Returns all a list with platform and module information.
+Returns all a list with platform and module information. Data can be retrieved 
+via endpoint or django management command.
+
+**Command:** `./manage.py anxmonitormodules` output is same as endpoints response
+body.
 
 **URL:** `/anxapi/v1/modules/?access_token=custom_access_token`
 

anexia_monitoring/core.py

@@ -0,0 +1,45 @@
+import asyncio
+import sys
+
+from updatable import get_package_update_list, get_parsed_environment_package_list
+
+
+async def get_python_env_info() -> dict:
+    runtime = {
+        'platform': 'python',
+        'platform_version': sys.version,
+        'framework': 'django',
+        'framework_installed_version': None,
+        'framework_newest_version': None,
+    }
+    modules = []
+    packages = get_parsed_environment_package_list()
+
+    for package in packages:
+        package['_data'] = asyncio.create_task(
+            get_package_update_list(package['package'], package['version'])
+        )
+
+    for package in packages:
+        package_data = await package['_data']
+
+        modules.append({
+            'name': package['package'],
+            'installed_version': package['version'],
+            'installed_version_licences': [
+                package_data['current_release_license'],
+            ],
+            'newest_version': package_data['latest_release'],
+            'newest_version_licences': [
+                package_data['latest_release_license'],
+            ],
+        })
+
+        if package['package'] == 'Django':
+            runtime['framework_installed_version'] = package['version']
+            runtime['framework_newest_version'] = package_data['latest_release']
+
+    return {
+        'runtime': runtime,
+        'modules': modules,
+    }

anexia_monitoring/management/commands/anxmonitormodules.py

@@ -0,0 +1,15 @@
+import asyncio
+import json
+
+from django.core.management.base import BaseCommand, CommandError
+from anexia_monitoring.core import get_python_env_info
+
+
+class Command(BaseCommand):
+    help = "Outputs JSON object describing python runtime and modules. Same as /anxapi/vi/modules"
+
+    def handle(self, *args, **kwargs):
+        modules_dict = asyncio.run(get_python_env_info())
+        modules_json = json.dumps(modules_dict, indent=4)
+        self.stdout.write(modules_json)
+        self.stdout.flush()

anexia_monitoring/views.py

@@ -1,15 +1,14 @@
 import asyncio
 import sys
 
-from updatable import get_package_update_list, get_parsed_environment_package_list
-
 from django.http import JsonResponse, HttpResponse
 from django.contrib.auth import get_user_model
 from django.db import connections
 from django.dispatch import receiver
 from django.views.generic import View
 from django.conf import settings
 
+from .core import get_python_env_info
 from .decorators import access_token_check
 from .events import monitor_up_check
 
@@ -18,6 +17,7 @@ class BaseView(View):
     """
     Base view
     """
+
     @staticmethod
     def add_access_control_headers(response):
         """
@@ -42,49 +42,9 @@ class MonitorModulesView(BaseView):
     of the module. It also contains information about the runtime (python and django version).
     """
 
-    async def get_response_data(self):
-        runtime = {
-            'platform': 'python',
-            'platform_version': sys.version,
-            'framework': 'django',
-            'framework_installed_version': None,
-            'framework_newest_version': None,
-        }
-        modules = []
-        packages = get_parsed_environment_package_list()
-
-        for package in packages:
-            package['_data'] = asyncio.create_task(
-                get_package_update_list(package['package'], package['version'])
-            )
-
-        for package in packages:
-            package_data = await package['_data']
-
-            modules.append({
-                'name': package['package'],
-                'installed_version': package['version'],
-                'installed_version_licences': [
-                    package_data['current_release_license'],
-                ],
-                'newest_version': package_data['latest_release'],
-                'newest_version_licences': [
-                    package_data['latest_release_license'],
-                ],
-            })
-
-            if package['package'] == 'Django':
-                runtime['framework_installed_version'] = package['version']
-                runtime['framework_newest_version'] = package_data['latest_release']
-
-        return {
-            'runtime': runtime,
-            'modules': modules,
-        }
-
     @access_token_check
     def get(self, request, *args, **kwargs):
-        response = JsonResponse(asyncio.run(self.get_response_data()))
+        response = JsonResponse(asyncio.run(get_python_env_info()))
         self.add_access_control_headers(response)
         return response
 
@@ -139,7 +99,6 @@ def anx_monitoring_test_db_connections(sender, **kwargs):
         for connection_key in connections:
             connections[connection_key].cursor()
 
-
 if getattr(settings, 'ANX_MONITORING_TEST_QUERY_USERS', True):
     @receiver(monitor_up_check)
     def anx_monitoring_test_query_users(sender, **kwargs):

tests/test_project/test_project/tests.py

@@ -1,6 +1,10 @@
+from io import StringIO
+import json
+
 from django.test import TestCase, Client
 from django.urls import reverse
 from django.conf import settings
+from django.core.management import call_command
 
 client = Client()
 
@@ -10,6 +14,7 @@ class MonitoringAPITests(TestCase):
     Test for monitoring endpoints. The endpoints are configured in the main urls.py and come
     from the anexia_monitoring package. The access token must be defined in the django settings.
     """
+
     def test_monitoring_endpoint(self):
         """
         Ensure the monitoring endpoint can be called
@@ -37,3 +42,23 @@ def test_up_endpoint(self):
 
         response = client.get("%s?access_token=%s" % (url, settings.ANX_MONITORING_ACCESS_TOKEN), format='json')
         self.assertEqual(response.status_code, 200)
+
+
+class MngmtCommandModulesTests(TestCase):
+    """
+    Tests django management command `anxmonitormodules`.
+    """
+
+    def is_json(self, json_str):
+        try:
+            json.loads(json_str)
+        except ValueError as e:
+            return False
+        return True
+
+    def test_command_success(self):
+        out = StringIO()
+        call_command('anxmonitormodules', stdout=out)
+        output = out.getvalue().strip()
+
+        self.assertTrue(self.is_json(output))