Analytics Module

allianceauth/analytics/__init__.py

@@ -0,0 +1 @@
+default_app_config = 'allianceauth.analytics.apps.AnalyticsConfig'

allianceauth/analytics/admin.py

@@ -0,0 +1,21 @@
+from django.contrib import admin
+
+from .models import AnalyticsIdentifier, AnalyticsPath, AnalyticsTokens
+
+
+@admin.register(AnalyticsIdentifier)
+class AnalyticsIdentifierAdmin(admin.ModelAdmin):
+    search_fields = ['identifier', ]
+    list_display = ('identifier',)
+
+
+@admin.register(AnalyticsTokens)
+class AnalyticsTokensAdmin(admin.ModelAdmin):
+    search_fields = ['name', ]
+    list_display = ('name', 'type',)
+
+
+@admin.register(AnalyticsPath)
+class AnalyticsPathAdmin(admin.ModelAdmin):
+    search_fields = ['ignore_path', ]
+    list_display = ('ignore_path',)

allianceauth/analytics/apps.py

@@ -0,0 +1,9 @@
+from django.apps import AppConfig
+
+
+class AnalyticsConfig(AppConfig):
+    name = 'allianceauth.analytics'
+    label = 'analytics'
+
+    def ready(self):
+        import allianceauth.analytics.signals

allianceauth/analytics/fixtures/disable_analytics.json

@@ -0,0 +1,21 @@
+[
+    {
+    "model": "analytics.AnalyticsTokens",
+    "pk": 1,
+    "fields": {
+        "name": "AA Team Public Google Analytics (Universal)",
+        "type": "GA-V4",
+        "token": "UA-186249766-2",
+        "send_page_views": "False",
+        "send_celery_tasks": "False",
+        "send_stats": "False"
+        }
+    },
+    {
+        "model": "analytics.AnalyticsIdentifier",
+        "pk": 1,
+        "fields": {
+            "identifier": "ab33e241fbf042b6aa77c7655a768af7"
+        }
+    }
+]

allianceauth/analytics/middleware.py

@@ -0,0 +1,41 @@
+from bs4 import BeautifulSoup
+
+from django.utils.deprecation import MiddlewareMixin
+from .models import AnalyticsTokens, AnalyticsIdentifier
+from .tasks import send_ga_tracking_web_view
+
+
+class AnalyticsMiddleware(MiddlewareMixin):
+    def process_response(self, request, response):
+        """Django Middleware: Process Page Views and creates Analytics Celery Tasks"""
+        analyticstokens = AnalyticsTokens.objects.all()
+        client_id = AnalyticsIdentifier.objects.get(id=1).identifier.hex
+        try:
+            title = BeautifulSoup(
+                response.content, "html.parser").html.head.title.text
+        except AttributeError:
+            title = ''
+        for token in analyticstokens:
+            # Check if Page View Sending is Disabled
+            if token.send_page_views is False:
+                continue
+            # Check Exclusions
+            if request.path in token.ignore_paths.all():
+                continue
+
+            tracking_id = token.token
+            locale = request.LANGUAGE_CODE
+            path = request.path
+            try:
+                useragent = request.headers["User-Agent"]
+            except KeyError:
+                useragent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Safari/537.36"
+
+            send_ga_tracking_web_view.s(tracking_id=tracking_id,
+                                        client_id=client_id,
+                                        page=path,
+                                        title=title,
+                                        locale=locale,
+                                        useragent=useragent).\
+                apply_async(priority=9)
+        return response

allianceauth/analytics/migrations/0001_initial.py

@@ -0,0 +1,42 @@
+# Generated by Django 3.1.4 on 2020-12-30 13:11
+
+from django.db import migrations, models
+import uuid
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='AnalyticsIdentifier',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('identifier', models.UUIDField(default=uuid.uuid4, editable=False)),
+            ],
+        ),
+        migrations.CreateModel(
+            name='AnalyticsPath',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('ignore_path', models.CharField(default='/example/', max_length=254)),
+            ],
+        ),
+        migrations.CreateModel(
+            name='AnalyticsTokens',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('name', models.CharField(max_length=254)),
+                ('type', models.CharField(choices=[('GA-U', 'Google Analytics Universal'), ('GA-V4', 'Google Analytics V4')], max_length=254)),
+                ('token', models.CharField(max_length=254)),
+                ('send_page_views', models.BooleanField(default=False)),
+                ('send_celery_tasks', models.BooleanField(default=False)),
+                ('send_stats', models.BooleanField(default=False)),
+                ('ignore_paths', models.ManyToManyField(blank=True, to='analytics.AnalyticsPath')),
+            ],
+        ),
+    ]

allianceauth/analytics/migrations/0002_add_AA_Team_Token.py

@@ -0,0 +1,34 @@
+# Generated by Django 3.1.4 on 2020-12-30 08:53
+
+from django.db import migrations
+
+
+def add_aa_team_token(apps, schema_editor):
+    # We can't import the Person model directly as it may be a newer
+    # version than this migration expects. We use the historical version.
+    Tokens = apps.get_model('analytics', 'AnalyticsTokens')
+    token = Tokens()
+    token.type = 'GA-U'
+    token.token = 'UA-186249766-2'
+    token.send_page_views = True
+    token.send_celery_views = True
+    token.send_stats = True
+    token.name = 'AA Team Public Google Analytics (Universal)'
+    token.save()
+
+
+def remove_aa_team_token(apps, schema_editor):
+    # Have to define some code to remove this identifier
+    # In case of migration rollback?
+    Tokens = apps.get_model('analytics', 'AnalyticsTokens')
+    token = Tokens.objects.filter(token="UA-186249766-2").delete()
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('analytics', '0001_initial'),
+        ]
+
+    operations = [migrations.RunPython(add_aa_team_token, remove_aa_team_token)
+                    ]

allianceauth/analytics/migrations/0003_Generate_Identifier.py

@@ -0,0 +1,30 @@
+# Generated by Django 3.1.4 on 2020-12-30 08:53
+
+from uuid import uuid4
+from django.db import migrations
+
+
+def generate_identifier(apps, schema_editor):
+    # We can't import the Person model directly as it may be a newer
+    # version than this migration expects. We use the historical version.
+    Identifier = apps.get_model('analytics', 'AnalyticsIdentifier')
+    identifier = Identifier()
+    identifier.id = 1
+    identifier.save()
+
+
+def zero_identifier(apps, schema_editor):
+    # Have to define some code to remove this identifier
+    # In case of migration rollback?
+    Identifier = apps.get_model('analytics', 'AnalyticsIdentifier')
+    Identifier.objects.filter(id=1).delete()
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('analytics', '0002_add_AA_Team_Token'),
+    ]
+
+    operations = [migrations.RunPython(generate_identifier, zero_identifier)
+                    ]

allianceauth/analytics/models.py

@@ -0,0 +1,38 @@
+from django.db import models
+from django.core.exceptions import ValidationError
+from django.utils.translation import gettext_lazy as _
+
+from uuid import uuid4
+
+
+class AnalyticsIdentifier(models.Model):
+
+    identifier = models.UUIDField(default=uuid4,
+                                  editable=False)
+
+    def save(self, *args, **kwargs):
+        if not self.pk and AnalyticsIdentifier.objects.exists():
+            # Force a single object
+            raise ValidationError('There is can be only one \
+                                   AnalyticsIdentifier instance')
+        self.pk = self.id = 1 # If this happens to be deleted and recreated, force it to be 1
+        return super(AnalyticsIdentifier, self).save(*args, **kwargs)
+
+
+class AnalyticsPath(models.Model):
+    ignore_path = models.CharField(max_length=254, default="/example/")
+
+
+class AnalyticsTokens(models.Model):
+
+    class Analytics_Type(models.TextChoices):
+        GA_U = 'GA-U', _('Google Analytics Universal')
+        GA_V4 = 'GA-V4', _('Google Analytics V4')
+
+    name = models.CharField(max_length=254)
+    type = models.CharField(max_length=254, choices=Analytics_Type.choices)
+    token = models.CharField(max_length=254, blank=False)
+    send_page_views = models.BooleanField(default=False)
+    send_celery_tasks = models.BooleanField(default=False)
+    send_stats = models.BooleanField(default=False)
+    ignore_paths = models.ManyToManyField(AnalyticsPath, blank=True)

allianceauth/analytics/signals.py

@@ -0,0 +1,50 @@
+from allianceauth.analytics.tasks import analytics_event
+from celery.signals import task_failure, task_success
+
+import logging
+logger = logging.getLogger(__name__)
+
+
+@task_failure.connect
+def process_failure_signal(
+                        exception, traceback,
+                        sender, task_id, signal,
+                        args, kwargs, einfo, **kw):
+    logger.debug("Celery task_failure signal %s" % sender.__class__.__name__)
+
+    category = sender.__module__
+
+    if 'allianceauth.analytics' not in category:
+        if category.endswith(".tasks"):
+            category = category[:-6]
+
+        action = sender.__name__
+
+        label = f"{exception.__class__.__name__}: {str(exception)}"
+
+        analytics_event(category=category,
+                        action=action,
+                        label=label)
+
+
+@task_success.connect
+def celery_success_signal(sender, result=None, **kw):
+    logger.debug("Celery task_success signal %s" % sender.__class__.__name__)
+
+    category = sender.__module__
+
+    if 'allianceauth.analytics' not in category:
+        if category.endswith(".tasks"):
+            category = category[:-6]
+
+        action = sender.__name__
+        label = "Success"
+
+        value = 0
+        if isinstance(result, int):
+            value = result
+
+        analytics_event(category=category,
+                        action=action,
+                        label=label,
+                        value=value)

allianceauth/analytics/tasks.py

@@ -0,0 +1,207 @@
+import requests
+import logging
+from django.conf import settings
+from django.apps import apps
+from celery import shared_task
+from allianceauth import __version__
+from .models import AnalyticsTokens, AnalyticsIdentifier
+from .utils import (
+    install_stat_addons,
+    install_stat_tokens,
+    install_stat_users)
+
+logger = logging.getLogger(__name__)
+
+BASE_URL = "https://www.google-analytics.com/"
+
+DEBUG_URL = f"{BASE_URL}debug/collect"
+COLLECTION_URL = f"{BASE_URL}collect"
+
+if getattr(settings, "ANALYTICS_ENABLE_DEBUG", False) and settings.DEBUG:
+    # Force sending of analytics data during in a debug/test environemt
+    # Usefull for developers working on this feature.
+    logger.warning(
+                "You have 'ANALYTICS_ENABLE_DEBUG' Enabled! "
+                "This debug instance will send analytics data!")
+    DEBUG_URL = COLLECTION_URL
+
+ANALYTICS_URL = COLLECTION_URL
+
+if settings.DEBUG is True:
+    ANALYTICS_URL = DEBUG_URL
+
+
+def analytics_event(category: str,
+                    action: str,
+                    label: str,
+                    value: int = 0,
+                    event_type: str = 'Celery'):
+    """
+    Send a Google Analytics Event for each token stored
+    Includes check for if its enabled/disabled
+
+    Parameters
+    -------
+    `category` (str): Celery Namespace
+    `action` (str): Task Name
+    `label` (str): Optional, Task Success/Exception
+    `value` (int): Optional, If bulk, Query size, can be a binary True/False
+    `event_type` (str): Optional, Celery or Stats only, Default to Celery
+    """
+    analyticstokens = AnalyticsTokens.objects.all()
+    client_id = AnalyticsIdentifier.objects.get(id=1).identifier.hex
+    for token in analyticstokens:
+        if event_type == 'Celery':
+            allowed = token.send_celery_tasks
+        elif event_type == 'Stats':
+            allowed = token.send_stats
+        else:
+            allowed = False
+
+        if allowed is True:
+            tracking_id = token.token
+            send_ga_tracking_celery_event.s(tracking_id=tracking_id,
+                                            client_id=client_id,
+                                            category=category,
+                                            action=action,
+                                            label=label,
+                                            value=value).\
+                apply_async(priority=9)
+
+
+@shared_task()
+def analytics_daily_stats():
+    """Celery Task: Do not call directly
+
+    Gathers a series of daily statistics and sends analytics events containing them"""
+    users = install_stat_users()
+    tokens = install_stat_tokens()
+    addons = install_stat_addons()
+    logger.debug("Running Daily Analytics Upload")
+
+    analytics_event(category='allianceauth.analytics',
+                    action='send_install_stats',
+                    label='existence',
+                    value=1,
+                    event_type='Stats')
+    analytics_event(category='allianceauth.analytics',
+                    action='send_install_stats',
+                    label='users',
+                    value=users,
+                    event_type='Stats')
+    analytics_event(category='allianceauth.analytics',
+                    action='send_install_stats',
+                    label='tokens',
+                    value=tokens,
+                    event_type='Stats')
+    analytics_event(category='allianceauth.analytics',
+                    action='send_install_stats',
+                    label='addons',
+                    value=addons,
+                    event_type='Stats')
+
+    for appconfig in apps.get_app_configs():
+        analytics_event(category='allianceauth.analytics',
+                        action='send_extension_stats',
+                        label=appconfig.label,
+                        value=1,
+                        event_type='Stats')
+
+
+@shared_task()
+def send_ga_tracking_web_view(
+                            tracking_id: str,
+                            client_id: str,
+                            page: str,
+                            title: str,
+                            locale: str,
+                            useragent: str) -> requests.Response:
+
+    """Celery Task: Do not call directly
+
+    Sends Page View events to GA, Called only via analytics.middleware
+
+    Parameters
+    ----------
+    `tracking_id` (str): Unique Server Identifier
+    `client_id` (str): GA Token
+    `page` (str): Page Path
+    `title` (str): Page Title
+    `locale` (str): Browser Language
+    `useragent` (str): Browser UserAgent
+
+    Returns
+    -------
+    requests.Reponse Object
+    """
+    headers = {"User-Agent": useragent}
+
+    payload = {
+            'v': '1',
+            'tid': tracking_id,
+            'cid': client_id,
+            't': 'pageview',
+            'dp': page,
+            'dt': title,
+            'ul': locale,
+            'ua': useragent,
+            'aip': 1,
+            'an': "allianceauth",
+            'av': __version__
+            }
+
+    response = requests.post(
+                        ANALYTICS_URL, data=payload,
+                        timeout=5, headers=headers)
+    logger.debug(f"Analytics Page View HTTP{response.status_code}")
+    return response
+
+
+@shared_task()
+def send_ga_tracking_celery_event(
+                                tracking_id: str,
+                                client_id: str,
+                                category: str,
+                                action: str,
+                                label: str,
+                                value: int) -> requests.Response:
+    """Celery Task: Do not call directly
+
+    Sends Page View events to GA, Called only via analytics.middleware
+
+    Parameters
+    ----------
+    `tracking_id` (str): Unique Server Identifier
+    `client_id` (str): GA Token
+    `category` (str): Celery Namespace
+    `action` (str): Task Name
+    `label` (str): Optional, Task Success/Exception
+    `value` (int): Optional, If bulk, Query size, can be a binary True/False
+
+    Returns
+    -------
+    requests.Reponse Object
+    """
+
+    headers = {
+        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Safari/537.36"}
+
+    payload = {
+            'v': '1',
+            'tid': tracking_id,
+            'cid': client_id,
+            't': 'event',
+            'ec': category,
+            'ea': action,
+            'el': label,
+            'ev': value,
+            'aip': 1,
+            'an': "allianceauth",
+            'av': __version__
+            }
+
+    response = requests.post(
+                        ANALYTICS_URL, data=payload,
+                        timeout=5, headers=headers)
+    logger.debug(f"Analytics Celery/Stats Event HTTP{response.status_code}")
+    return response

allianceauth/analytics/tests/test_middleware.py

@@ -0,0 +1,23 @@
+from allianceauth.analytics.middleware import AnalyticsMiddleware
+from unittest.mock import Mock
+
+from django.test.testcases import TestCase
+
+
+class TestAnalyticsMiddleware(TestCase):
+
+    def setUp(self):
+        self.middleware = AnalyticsMiddleware()
+        self.request = Mock()
+        self.request.headers = {
+                "User-Agent": "AUTOMATED TEST"
+            }
+        self.request.path = '/testURL/'
+        self.request.session = {}
+        self.request.LANGUAGE_CODE = 'en'
+        self.response = Mock()
+        self.response.content = 'hello world'
+
+    def test_middleware(self):
+        response = self.middleware.process_response(self.request, self.response)
+        self.assertEqual(self.response, response)

allianceauth/analytics/tests/test_models.py

@@ -0,0 +1,26 @@
+from allianceauth.analytics.models import AnalyticsIdentifier
+from django.core.exceptions import ValidationError
+
+from django.test.testcases import TestCase
+
+from uuid import UUID, uuid4
+
+
+# Identifiers
+uuid_1 = "ab33e241fbf042b6aa77c7655a768af7"
+uuid_2 = "7aa6bd70701f44729af5e3095ff4b55c"
+
+
+class TestAnalyticsIdentifier(TestCase):
+
+    def test_identifier_random(self):
+        self.assertNotEqual(AnalyticsIdentifier.objects.get(), uuid4)
+
+    def test_identifier_singular(self):
+        AnalyticsIdentifier.objects.all().delete()
+        AnalyticsIdentifier.objects.create(identifier=uuid_1)
+        # Yeah i have multiple asserts here, they all do the same thing
+        with self.assertRaises(ValidationError):
+            AnalyticsIdentifier.objects.create(identifier=uuid_2)
+        self.assertEqual(AnalyticsIdentifier.objects.count(), 1)
+        self.assertEqual(AnalyticsIdentifier.objects.get(pk=1).identifier, UUID(uuid_1))

allianceauth/analytics/tests/test_tasks.py

@@ -0,0 +1,119 @@
+from allianceauth.analytics.tasks import (
+    analytics_event,
+    send_ga_tracking_celery_event,
+    send_ga_tracking_web_view)
+from django.test.testcases import TestCase
+
+
+class TestAnalyticsTasks(TestCase):
+    def test_analytics_event(self):
+        analytics_event(
+                        category='allianceauth.analytics',
+                        action='send_tests',
+                        label='test',
+                        value=1,
+                        event_type='Stats')
+
+    def test_send_ga_tracking_web_view_sent(self):
+        # This test sends if the event SENDS to google
+        # Not if it was successful
+        tracking_id = 'UA-186249766-2'
+        client_id = 'ab33e241fbf042b6aa77c7655a768af7'
+        page = '/index/'
+        title = 'Hello World'
+        locale = 'en'
+        useragent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Safari/537.36"
+        response = send_ga_tracking_web_view(
+                                            tracking_id,
+                                            client_id,
+                                            page,
+                                            title,
+                                            locale,
+                                            useragent)
+        self.assertEqual(response.status_code, 200)
+
+    def test_send_ga_tracking_web_view_success(self):
+        tracking_id = 'UA-186249766-2'
+        client_id = 'ab33e241fbf042b6aa77c7655a768af7'
+        page = '/index/'
+        title = 'Hello World'
+        locale = 'en'
+        useragent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Safari/537.36"
+        json_response = send_ga_tracking_web_view(
+                                                tracking_id,
+                                                client_id,
+                                                page,
+                                                title,
+                                                locale,
+                                                useragent).json()
+        self.assertTrue(json_response["hitParsingResult"][0]["valid"])
+
+    def test_send_ga_tracking_web_view_invalid_token(self):
+        tracking_id = 'UA-IntentionallyBadTrackingID-2'
+        client_id = 'ab33e241fbf042b6aa77c7655a768af7'
+        page = '/index/'
+        title = 'Hello World'
+        locale = 'en'
+        useragent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Safari/537.36"
+        json_response = send_ga_tracking_web_view(
+                                            tracking_id,
+                                            client_id,
+                                            page,
+                                            title,
+                                            locale,
+                                            useragent).json()
+        self.assertFalse(json_response["hitParsingResult"][0]["valid"])
+        self.assertEqual(json_response["hitParsingResult"][0]["parserMessage"][1]["description"], "The value provided for parameter 'tid' is invalid. Please see http://goo.gl/a8d4RP#tid for details.")
+
+        # [{'valid': False, 'parserMessage': [{'messageType': 'INFO', 'description': 'IP Address from this hit was anonymized to 1.132.110.0.', 'messageCode': 'VALUE_MODIFIED'}, {'messageType': 'ERROR', 'description': "The value provided for parameter 'tid' is invalid. Please see http://goo.gl/a8d4RP#tid for details.", 'messageCode': 'VALUE_INVALID', 'parameter': 'tid'}], 'hit': '/debug/collect?v=1&tid=UA-IntentionallyBadTrackingID-2&cid=ab33e241fbf042b6aa77c7655a768af7&t=pageview&dp=/index/&dt=Hello World&ul=en&ua=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Safari/537.36&aip=1&an=allianceauth&av=2.9.0a2'}]
+
+    def test_send_ga_tracking_celery_event_sent(self):
+        tracking_id = 'UA-186249766-2'
+        client_id = 'ab33e241fbf042b6aa77c7655a768af7'
+        category = 'test'
+        action = 'test'
+        label = 'test'
+        value = '1'
+        response = send_ga_tracking_celery_event(
+                                                tracking_id,
+                                                client_id,
+                                                category,
+                                                action,
+                                                label,
+                                                value)
+        self.assertEqual(response.status_code, 200)
+
+    def test_send_ga_tracking_celery_event_success(self):
+        tracking_id = 'UA-186249766-2'
+        client_id = 'ab33e241fbf042b6aa77c7655a768af7'
+        category = 'test'
+        action = 'test'
+        label = 'test'
+        value = '1'
+        json_response = send_ga_tracking_celery_event(
+                                                    tracking_id,
+                                                    client_id,
+                                                    category,
+                                                    action,
+                                                    label,
+                                                    value).json()
+        self.assertTrue(json_response["hitParsingResult"][0]["valid"])
+
+    def test_send_ga_tracking_celery_event_invalid_token(self):
+        tracking_id = 'UA-IntentionallyBadTrackingID-2'
+        client_id = 'ab33e241fbf042b6aa77c7655a768af7'
+        category = 'test'
+        action = 'test'
+        label = 'test'
+        value = '1'
+        json_response = send_ga_tracking_celery_event(
+                                                    tracking_id,
+                                                    client_id,
+                                                    category,
+                                                    action,
+                                                    label,
+                                                    value).json()
+        self.assertFalse(json_response["hitParsingResult"][0]["valid"])
+        self.assertEqual(json_response["hitParsingResult"][0]["parserMessage"][1]["description"], "The value provided for parameter 'tid' is invalid. Please see http://goo.gl/a8d4RP#tid for details.")
+
+        # [{'valid': False, 'parserMessage': [{'messageType': 'INFO', 'description': 'IP Address from this hit was anonymized to 1.132.110.0.', 'messageCode': 'VALUE_MODIFIED'}, {'messageType': 'ERROR', 'description': "The value provided for parameter 'tid' is invalid. Please see http://goo.gl/a8d4RP#tid for details.", 'messageCode': 'VALUE_INVALID', 'parameter': 'tid'}], 'hit': '/debug/collect?v=1&tid=UA-IntentionallyBadTrackingID-2&cid=ab33e241fbf042b6aa77c7655a768af7&t=pageview&dp=/index/&dt=Hello World&ul=en&ua=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Safari/537.36&aip=1&an=allianceauth&av=2.9.0a2'}]

allianceauth/analytics/tests/test_utils.py

@@ -0,0 +1,55 @@
+from django.apps import apps
+from allianceauth.authentication.models import User
+from esi.models import Token
+from allianceauth.analytics.utils import install_stat_users, install_stat_tokens, install_stat_addons
+
+from django.test.testcases import TestCase
+
+
+def create_testdata():
+    User.objects.all().delete()
+    User.objects.create_user(
+        'user_1'
+        'abc@example.com',
+        'password'
+    )
+    User.objects.create_user(
+        'user_2'
+        'abc@example.com',
+        'password'
+    )
+    #Token.objects.all().delete()
+    #Token.objects.create(
+    #            character_id=101,
+    #            character_name='character1',
+    #            access_token='my_access_token'
+    #)
+    #Token.objects.create(
+    #            character_id=102,
+    #            character_name='character2',
+    #            access_token='my_access_token'
+    #)
+
+
+class TestAnalyticsUtils(TestCase):
+
+    def test_install_stat_users(self):
+        create_testdata()
+        expected = 2
+
+        users = install_stat_users()
+        self.assertEqual(users, expected)
+
+    #def test_install_stat_tokens(self):
+    #    create_testdata()
+    #    expected = 2
+    #
+    #   tokens = install_stat_tokens()
+    #   self.assertEqual(tokens, expected)
+
+    def test_install_stat_addons(self):
+        # this test does what its testing...
+        # but helpful for existing as a sanity check
+        expected = len(list(apps.get_app_configs()))
+        addons = install_stat_addons()
+        self.assertEqual(addons, expected)

allianceauth/analytics/utils.py

@@ -0,0 +1,36 @@
+from django.apps import apps
+from allianceauth.authentication.models import User
+from esi.models import Token
+
+
+def install_stat_users() -> int:
+    """Count and Return the number of User accounts
+
+    Returns
+    -------
+    int
+        The Number of User objects"""
+    users = User.objects.count()
+    return users
+
+
+def install_stat_tokens() -> int:
+    """Count and Return the number of ESI Tokens Stored
+
+    Returns
+    -------
+    int
+        The Number of Token Objects"""
+    tokens = Token.objects.count()
+    return tokens
+
+
+def install_stat_addons() -> int:
+    """Count and Return the number of Django Applications Installed
+
+    Returns
+    -------
+    int
+        The Number of Installed Apps"""
+    addons = len(list(apps.get_app_configs()))
+    return addons