Mathis/allianceauth
1
0
Fork 0
mirror of https://gitlab.com/allianceauth/allianceauth.git synced 2025-07-13 14:30:17 +02:00
Code Issues Packages Projects Releases Wiki Activity

Analytics UA to V4 Conversion

Browse Source
This commit is contained in:
Ariel Rin 2023-08-14 03:31:33 +00:00
parent 6b932b1188
commit a2f217ace5
18 changed files with 243 additions and 565 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
allianceauth/analytics/admin.py
View File

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

3
allianceauth/analytics/apps.py
View File

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

7
allianceauth/analytics/fixtures/disable_analytics.json
View File

@ -3,11 +3,10 @@
"model": "analytics.AnalyticsTokens", "model": "analytics.AnalyticsTokens",
"pk": 1, "pk": 1,
"fields": { "fields": {
"name": "AA Team Public Google Analytics (Universal)", "name": "AA Team Public Google Analytics (V4)",
"type": "GA-V4", "type": "GA-V4",
"token": "UA-186249766-2", "token": "G-6LYSMYK8DE",
"send_page_views": "False", "secret": "KLlpjLZ-SRGozS5f5wb_kw",
"send_celery_tasks": "False",
"send_stats": "False" "send_stats": "False"
} }
}, },

52
allianceauth/analytics/middleware.py
View File

@ -1,52 +0,0 @@
from bs4 import BeautifulSoup
from django.conf import settings
from django.utils.deprecation import MiddlewareMixin
from .models import AnalyticsTokens, AnalyticsIdentifier
from .tasks import send_ga_tracking_web_view
import re
class AnalyticsMiddleware(MiddlewareMixin):
def process_response(self, request, response):
"""Django Middleware: Process Page Views and creates Analytics Celery Tasks"""
if getattr(settings, "ANALYTICS_DISABLED", False):
return response
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
ignore = False
for ignore_path in token.ignore_paths.values():
ignore_path_regex = re.compile(ignore_path["ignore_path"])
if re.search(ignore_path_regex, request.path) is not None:
ignore = True
if ignore is True:
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

18
allianceauth/analytics/migrations/0007_analyticstokens_secret.py Normal file
View File

@ -0,0 +1,18 @@
# Generated by Django 4.0.6 on 2022-08-30 05:47
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('analytics', '0006_more_ignore_paths'),
]
operations = [
migrations.AddField(
model_name='analyticstokens',
name='secret',
field=models.CharField(blank=True, max_length=254),
),
]

64
allianceauth/analytics/migrations/0008_add_AA_GA-4_Team_Token .py Normal file
View File

@ -0,0 +1,64 @@
# Generated by Django 3.1.4 on 2020-12-30 08:53
from django.db import migrations
from django.core.exceptions import ObjectDoesNotExist
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')
AnalyticsPath = apps.get_model('analytics', 'AnalyticsPath')
token = Tokens()
try:
ua_token = Tokens.objects.get(token="UA-186249766-2")
original_send_page_views = ua_token.send_page_views
original_send_celery_tasks = ua_token.send_celery_tasks
original_send_stats = ua_token.send_stats
except ObjectDoesNotExist:
original_send_page_views = True
original_send_celery_tasks = True
original_send_stats = True
try:
user_notifications_count = AnalyticsPath.objects.get(ignore_path=r"^\/user_notifications_count\/.*",)
except ObjectDoesNotExist:
user_notifications_count = AnalyticsPath.objects.create(ignore_path=r"^\/user_notifications_count\/.*")
try:
admin = AnalyticsPath.objects.get(ignore_path=r"^\/admin\/.*")
except ObjectDoesNotExist:
admin = AnalyticsPath.objects.create(ignore_path=r"^\/admin\/.*")
try:
account_activate = AnalyticsPath.objects.get(ignore_path=r"^\/account\/activate\/.*")
except ObjectDoesNotExist:
account_activate = AnalyticsPath.objects.create(ignore_path=r"^\/account\/activate\/.*")
token.type = 'GA-V4'
token.token = 'G-6LYSMYK8DE'
token.secret = 'KLlpjLZ-SRGozS5f5wb_kw'
token.send_page_views = original_send_page_views
token.send_celery_tasks = original_send_celery_tasks
token.send_stats = original_send_stats
token.name = 'AA Team Public Google Analytics (V4)'
token.save()
token.ignore_paths.add(admin, user_notifications_count, account_activate)
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="G-6LYSMYK8DE").delete()
class Migration(migrations.Migration):
dependencies = [
('analytics', '0007_analyticstokens_secret'),
]
operations = [migrations.RunPython(
add_aa_team_token, remove_aa_team_token)]

28
allianceauth/analytics/migrations/0009_remove_analyticstokens_ignore_paths_and_more.py Normal file
View File

@ -0,0 +1,28 @@
# Generated by Django 4.0.10 on 2023-05-08 05:24
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
('analytics', '0008_add_AA_GA-4_Team_Token '),
]
operations = [
migrations.RemoveField(
model_name='analyticstokens',
name='ignore_paths',
),
migrations.RemoveField(
model_name='analyticstokens',
name='send_celery_tasks',
),
migrations.RemoveField(
model_name='analyticstokens',
name='send_page_views',
),
migrations.DeleteModel(
name='AnalyticsPath',
),
]

15
allianceauth/analytics/models.py
View File

@ -7,22 +7,19 @@ from uuid import uuid4
class AnalyticsIdentifier(models.Model): class AnalyticsIdentifier(models.Model):
identifier = models.UUIDField(default=uuid4, identifier = models.UUIDField(
editable=False) default=uuid4,
editable=False)
def save(self, *args, **kwargs): def save(self, *args, **kwargs):
if not self.pk and AnalyticsIdentifier.objects.exists(): if not self.pk and AnalyticsIdentifier.objects.exists():
# Force a single object # Force a single object
raise ValidationError('There is can be only one \ raise ValidationError('There is can be only one \
AnalyticsIdentifier instance') AnalyticsIdentifier instance')
self.pk = self.id = 1 # If this happens to be deleted and recreated, force it to be 1 self.pk = self.id = 1 # If this happens to be deleted and recreated, force it to be 1
return super().save(*args, **kwargs) return super().save(*args, **kwargs)
class AnalyticsPath(models.Model):
ignore_path = models.CharField(max_length=254, default="/example/", help_text="Regex Expression, If matched no Analytics Page View is sent")
class AnalyticsTokens(models.Model): class AnalyticsTokens(models.Model):
class Analytics_Type(models.TextChoices): class Analytics_Type(models.TextChoices):
@ -32,7 +29,5 @@ class AnalyticsTokens(models.Model):
name = models.CharField(max_length=254) name = models.CharField(max_length=254)
type = models.CharField(max_length=254, choices=Analytics_Type.choices) type = models.CharField(max_length=254, choices=Analytics_Type.choices)
token = models.CharField(max_length=254, blank=False) token = models.CharField(max_length=254, blank=False)
send_page_views = models.BooleanField(default=False) secret = models.CharField(max_length=254, blank=True)
send_celery_tasks = models.BooleanField(default=False)
send_stats = models.BooleanField(default=False) send_stats = models.BooleanField(default=False)
ignore_paths = models.ManyToManyField(AnalyticsPath, blank=True)

55
allianceauth/analytics/signals.py
View File

@ -1,55 +0,0 @@
import logging
from celery.signals import task_failure, task_success
from django.conf import settings
from allianceauth.analytics.tasks import analytics_event
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__)
if getattr(settings, "ANALYTICS_DISABLED", False):
return
category = sender.__module__
if 'allianceauth.analytics' not in category:
if category.endswith(".tasks"):
category = category[:-6]
action = sender.__name__
label = f"{exception.__class__.__name__}"
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__)
if getattr(settings, "ANALYTICS_DISABLED", False):
return
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)

205
allianceauth/analytics/tasks.py
View File

@ -3,7 +3,6 @@ import logging
from django.conf import settings from django.conf import settings
from django.apps import apps from django.apps import apps
from celery import shared_task from celery import shared_task
from allianceauth import __version__
from .models import AnalyticsTokens, AnalyticsIdentifier from .models import AnalyticsTokens, AnalyticsIdentifier
from .utils import ( from .utils import (
install_stat_addons, install_stat_addons,
@ -12,14 +11,14 @@ from .utils import (
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
BASE_URL = "https://www.google-analytics.com/" BASE_URL = "https://www.google-analytics.com"
DEBUG_URL = f"{BASE_URL}debug/collect" DEBUG_URL = f"{BASE_URL}/debug/mp/collect"
COLLECTION_URL = f"{BASE_URL}collect" COLLECTION_URL = f"{BASE_URL}/mp/collect"
if getattr(settings, "ANALYTICS_ENABLE_DEBUG", False) and settings.DEBUG: if getattr(settings, "ANALYTICS_ENABLE_DEBUG", False) and settings.DEBUG:
# Force sending of analytics data during in a debug/test environemt # Force sending of analytics data during in a debug/test environment
# Usefull for developers working on this feature. # Useful for developers working on this feature.
logger.warning( logger.warning(
"You have 'ANALYTICS_ENABLE_DEBUG' Enabled! " "You have 'ANALYTICS_ENABLE_DEBUG' Enabled! "
"This debug instance will send analytics data!") "This debug instance will send analytics data!")
@ -31,40 +30,38 @@ if settings.DEBUG is True:
ANALYTICS_URL = DEBUG_URL ANALYTICS_URL = DEBUG_URL
def analytics_event(category: str, def analytics_event(namespace: str,
action: str, task: str,
label: str, label: str = "",
value: int = 0, result: str = "",
value: int = 1,
event_type: str = 'Celery'): event_type: str = 'Celery'):
""" """
Send a Google Analytics Event for each token stored Send a Google Analytics Event for each token stored
Includes check for if its enabled/disabled Includes check for if its enabled/disabled
Args: Args:
`category` (str): Celery Namespace `namespace` (str): Celery Namespace
`action` (str): Task Name `task` (str): Task Name
`label` (str): Optional, Task Success/Exception `label` (str): Optional, additional task label
`value` (int): Optional, If bulk, Query size, can be a binary True/False `result` (str): Optional, Task Success/Exception
`value` (int): Optional, If bulk, Query size, can be a Boolean
`event_type` (str): Optional, Celery or Stats only, Default to Celery `event_type` (str): Optional, Celery or Stats only, Default to Celery
""" """
analyticstokens = AnalyticsTokens.objects.all() for token in AnalyticsTokens.objects.filter(type='GA-V4'):
client_id = AnalyticsIdentifier.objects.get(id=1).identifier.hex if event_type == 'Stats':
for token in analyticstokens:
if event_type == 'Celery':
allowed = token.send_celery_tasks
elif event_type == 'Stats':
allowed = token.send_stats allowed = token.send_stats
else: else:
allowed = False allowed = False
if allowed is True: if allowed is True:
tracking_id = token.token
send_ga_tracking_celery_event.s( send_ga_tracking_celery_event.s(
tracking_id=tracking_id, measurement_id=token.token,
client_id=client_id, secret=token.secret,
category=category, namespace=namespace,
action=action, task=task,
label=label, label=label,
result=result,
value=value).apply_async(priority=9) value=value).apply_async(priority=9)
@ -72,136 +69,104 @@ def analytics_event(category: str,
def analytics_daily_stats(): def analytics_daily_stats():
"""Celery Task: Do not call directly """Celery Task: Do not call directly
Gathers a series of daily statistics and sends analytics events containing them Gathers a series of daily statistics
Sends analytics events containing them
""" """
users = install_stat_users() users = install_stat_users()
tokens = install_stat_tokens() tokens = install_stat_tokens()
addons = install_stat_addons() addons = install_stat_addons()
logger.debug("Running Daily Analytics Upload") logger.debug("Running Daily Analytics Upload")
analytics_event(category='allianceauth.analytics', analytics_event(namespace='allianceauth.analytics',
action='send_install_stats', task='send_install_stats',
label='existence', label='existence',
value=1, value=1,
event_type='Stats') event_type='Stats')
analytics_event(category='allianceauth.analytics', analytics_event(namespace='allianceauth.analytics',
action='send_install_stats', task='send_install_stats',
label='users', label='users',
value=users, value=users,
event_type='Stats') event_type='Stats')
analytics_event(category='allianceauth.analytics', analytics_event(namespace='allianceauth.analytics',
action='send_install_stats', task='send_install_stats',
label='tokens', label='tokens',
value=tokens, value=tokens,
event_type='Stats') event_type='Stats')
analytics_event(category='allianceauth.analytics', analytics_event(namespace='allianceauth.analytics',
action='send_install_stats', task='send_install_stats',
label='addons', label='addons',
value=addons, value=addons,
event_type='Stats') event_type='Stats')
for appconfig in apps.get_app_configs(): for appconfig in apps.get_app_configs():
analytics_event(category='allianceauth.analytics', analytics_event(namespace='allianceauth.analytics',
action='send_extension_stats', task='send_extension_stats',
label=appconfig.label, label=appconfig.label,
value=1, value=1,
event_type='Stats') 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() @shared_task()
def send_ga_tracking_celery_event( def send_ga_tracking_celery_event(
tracking_id: str, measurement_id: str,
client_id: str, secret: str,
category: str, namespace: str,
action: str, task: str,
label: str, label: str = "",
value: int) -> requests.Response: result: str = "",
value: int = 1):
"""Celery Task: Do not call directly """Celery Task: Do not call directly
Sends Page View events to GA, Called only via analytics.middleware Sends an events to GA
Parameters Parameters
---------- ----------
`tracking_id` (str): Unique Server Identifier `measurement_id` (str): GA Token
`client_id` (str): GA Token `secret` (str): GA Authentication Secret
`category` (str): Celery Namespace `namespace` (str): Celery Namespace
`action` (str): Task Name `task` (str): Task Name
`label` (str): Optional, Task Success/Exception `label` (str): Optional, additional task label
`result` (str): Optional, Task Success/Exception
`value` (int): Optional, If bulk, Query size, can be a binary True/False `value` (int): Optional, If bulk, Query size, can be a binary True/False
Returns
-------
requests.Reponse Object
""" """
headers = { parameters = {
"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"} 'measurement_id': measurement_id,
'api_secret': secret
}
payload = { payload = {
'v': '1', 'client_id': AnalyticsIdentifier.objects.get(id=1).identifier.hex,
'tid': tracking_id, "user_properties": {
'cid': client_id, "allianceauth_version": {
't': 'event', "value": "allianceauth_version"
'ec': category,
'ea': action,
'el': label,
'ev': value,
'aip': 1,
'an': "allianceauth",
'av': __version__
} }
},
response = requests.post( 'non_personalized_ads': True,
ANALYTICS_URL, data=payload, "events": [{
timeout=5, headers=headers) "name": "celery_event",
logger.debug(f"Analytics Celery/Stats Event HTTP{response.status_code}") "params": {
return response "namespace": namespace,
"task": task,
'result': result,
'label': label,
"value": value
}
}]
}
try:
response = requests.post(
ANALYTICS_URL,
params=parameters,
json=payload,
timeout=10)
response.raise_for_status()
logger.debug(
f"Analytics Celery/Stats Event HTTP{response.status_code}")
return response.status_code
except requests.exceptions.HTTPError as e:
logger.debug(e)
return response.status_code
except requests.exceptions.ConnectionError as e:
logger.debug(e)
return "Failed"

109
allianceauth/analytics/tests/test_integration.py
View File

@ -1,109 +0,0 @@
from unittest.mock import patch
from urllib.parse import parse_qs
import requests_mock
from django.test import override_settings
from allianceauth.analytics.tasks import ANALYTICS_URL
from allianceauth.eveonline.tasks import update_character
from allianceauth.tests.auth_utils import AuthUtils
from allianceauth.utils.testing import NoSocketsTestCase
@override_settings(CELERY_ALWAYS_EAGER=True)
@requests_mock.mock()
class TestAnalyticsForViews(NoSocketsTestCase):
@override_settings(ANALYTICS_DISABLED=False)
def test_should_run_analytics(self, requests_mocker):
# given
requests_mocker.post(ANALYTICS_URL)
user = AuthUtils.create_user("Bruce Wayne")
self.client.force_login(user)
# when
response = self.client.get("/dashboard/")
# then
self.assertEqual(response.status_code, 200)
self.assertTrue(requests_mocker.called)
@override_settings(ANALYTICS_DISABLED=True)
def test_should_not_run_analytics(self, requests_mocker):
# given
requests_mocker.post(ANALYTICS_URL)
user = AuthUtils.create_user("Bruce Wayne")
self.client.force_login(user)
# when
response = self.client.get("/dashboard/")
# then
self.assertEqual(response.status_code, 200)
self.assertFalse(requests_mocker.called)
@override_settings(CELERY_ALWAYS_EAGER=True)
@requests_mock.mock()
class TestAnalyticsForTasks(NoSocketsTestCase):
@override_settings(ANALYTICS_DISABLED=False)
@patch("allianceauth.eveonline.models.EveCharacter.objects.update_character")
def test_should_run_analytics_for_successful_task(
self, requests_mocker, mock_update_character
):
# given
requests_mocker.post(ANALYTICS_URL)
user = AuthUtils.create_user("Bruce Wayne")
character = AuthUtils.add_main_character_2(user, "Bruce Wayne", 1001)
# when
update_character.delay(character.character_id)
# then
self.assertTrue(mock_update_character.called)
self.assertTrue(requests_mocker.called)
payload = parse_qs(requests_mocker.last_request.text)
self.assertListEqual(payload["el"], ["Success"])
@override_settings(ANALYTICS_DISABLED=True)
@patch("allianceauth.eveonline.models.EveCharacter.objects.update_character")
def test_should_not_run_analytics_for_successful_task(
self, requests_mocker, mock_update_character
):
# given
requests_mocker.post(ANALYTICS_URL)
user = AuthUtils.create_user("Bruce Wayne")
character = AuthUtils.add_main_character_2(user, "Bruce Wayne", 1001)
# when
update_character.delay(character.character_id)
# then
self.assertTrue(mock_update_character.called)
self.assertFalse(requests_mocker.called)
@override_settings(ANALYTICS_DISABLED=False)
@patch("allianceauth.eveonline.models.EveCharacter.objects.update_character")
def test_should_run_analytics_for_failed_task(
self, requests_mocker, mock_update_character
):
# given
requests_mocker.post(ANALYTICS_URL)
mock_update_character.side_effect = RuntimeError
user = AuthUtils.create_user("Bruce Wayne")
character = AuthUtils.add_main_character_2(user, "Bruce Wayne", 1001)
# when
update_character.delay(character.character_id)
# then
self.assertTrue(mock_update_character.called)
self.assertTrue(requests_mocker.called)
payload = parse_qs(requests_mocker.last_request.text)
self.assertNotEqual(payload["el"], ["Success"])
@override_settings(ANALYTICS_DISABLED=True)
@patch("allianceauth.eveonline.models.EveCharacter.objects.update_character")
def test_should_not_run_analytics_for_failed_task(
self, requests_mocker, mock_update_character
):
# given
requests_mocker.post(ANALYTICS_URL)
mock_update_character.side_effect = RuntimeError
user = AuthUtils.create_user("Bruce Wayne")
character = AuthUtils.add_main_character_2(user, "Bruce Wayne", 1001)
# when
update_character.delay(character.character_id)
# then
self.assertTrue(mock_update_character.called)
self.assertFalse(requests_mocker.called)

24
allianceauth/analytics/tests/test_middleware.py
View File

@ -1,24 +0,0 @@
from allianceauth.analytics.middleware import AnalyticsMiddleware
from unittest.mock import Mock
from django.http import HttpResponse
from django.test.testcases import TestCase
class TestAnalyticsMiddleware(TestCase):
def setUp(self):
self.middleware = AnalyticsMiddleware(HttpResponse)
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)

3
allianceauth/analytics/tests/test_models.py
View File

@ -23,4 +23,5 @@ class TestAnalyticsIdentifier(TestCase):
with self.assertRaises(ValidationError): with self.assertRaises(ValidationError):
AnalyticsIdentifier.objects.create(identifier=uuid_2) AnalyticsIdentifier.objects.create(identifier=uuid_2)
self.assertEqual(AnalyticsIdentifier.objects.count(), 1) self.assertEqual(AnalyticsIdentifier.objects.count(), 1)
self.assertEqual(AnalyticsIdentifier.objects.get(pk=1).identifier, UUID(uuid_1)) self.assertEqual(AnalyticsIdentifier.objects.get(
pk=1).identifier, UUID(uuid_1))

201
allianceauth/analytics/tests/test_tasks.py
View File

@ -4,12 +4,11 @@ from django.test.utils import override_settings
from allianceauth.analytics.tasks import ( from allianceauth.analytics.tasks import (
analytics_event, analytics_event,
send_ga_tracking_celery_event, send_ga_tracking_celery_event)
send_ga_tracking_web_view)
from allianceauth.utils.testing import NoSocketsTestCase from allianceauth.utils.testing import NoSocketsTestCase
GOOGLE_ANALYTICS_DEBUG_URL = 'https://www.google-analytics.com/debug/collect' GOOGLE_ANALYTICS_DEBUG_URL = 'https://www.google-analytics.com/debug/mp/collect'
@override_settings(CELERY_ALWAYS_EAGER=True, CELERY_EAGER_PROPAGATES_EXCEPTIONS=True) @override_settings(CELERY_ALWAYS_EAGER=True, CELERY_EAGER_PROPAGATES_EXCEPTIONS=True)
@ -18,195 +17,53 @@ class TestAnalyticsTasks(NoSocketsTestCase):
def test_analytics_event(self, requests_mocker): def test_analytics_event(self, requests_mocker):
requests_mocker.register_uri('POST', GOOGLE_ANALYTICS_DEBUG_URL) requests_mocker.register_uri('POST', GOOGLE_ANALYTICS_DEBUG_URL)
analytics_event( analytics_event(
category='allianceauth.analytics', namespace='allianceauth.analytics',
action='send_tests', task='send_tests',
label='test', label='test',
value=1, value=1,
event_type='Stats') result="Success",
event_type='Stats')
def test_send_ga_tracking_web_view_sent(self, requests_mocker):
"""This test sends if the event SENDS to google.
Not if it was successful.
"""
# given
requests_mocker.register_uri('POST', GOOGLE_ANALYTICS_DEBUG_URL)
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"
# when
response = send_ga_tracking_web_view(
tracking_id,
client_id,
page,
title,
locale,
useragent)
# then
self.assertEqual(response.status_code, 200)
def test_send_ga_tracking_web_view_success(self, requests_mocker):
# given
requests_mocker.register_uri(
'POST',
GOOGLE_ANALYTICS_DEBUG_URL,
json={"hitParsingResult":[{'valid': True}]}
)
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"
# when
json_response = send_ga_tracking_web_view(
tracking_id,
client_id,
page,
title,
locale,
useragent).json()
# then
self.assertTrue(json_response["hitParsingResult"][0]["valid"])
def test_send_ga_tracking_web_view_invalid_token(self, requests_mocker):
# given
requests_mocker.register_uri(
'POST',
GOOGLE_ANALYTICS_DEBUG_URL,
json={
"hitParsingResult":[
{
'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'
}
]
}
)
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"
# when
json_response = send_ga_tracking_web_view(
tracking_id,
client_id,
page,
title,
locale,
useragent).json()
# then
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, requests_mocker): def test_send_ga_tracking_celery_event_sent(self, requests_mocker):
# given # given
requests_mocker.register_uri('POST', GOOGLE_ANALYTICS_DEBUG_URL) requests_mocker.register_uri('POST', GOOGLE_ANALYTICS_DEBUG_URL)
tracking_id = 'UA-186249766-2' token = 'G-6LYSMYK8DE'
client_id = 'ab33e241fbf042b6aa77c7655a768af7' secret = 'KLlpjLZ-SRGozS5f5wb_kw',
category = 'test' category = 'test'
action = 'test' action = 'test'
label = 'test' label = 'test'
value = '1' value = '1'
# when # when
response = send_ga_tracking_celery_event( task = send_ga_tracking_celery_event(
tracking_id, token,
client_id, secret,
category, category,
action, action,
label, label,
value) value)
# then # then
self.assertEqual(response.status_code, 200) self.assertEqual(task, 200)
def test_send_ga_tracking_celery_event_success(self, requests_mocker): def test_send_ga_tracking_celery_event_success(self, requests_mocker):
# given # given
requests_mocker.register_uri( requests_mocker.register_uri(
'POST', 'POST',
GOOGLE_ANALYTICS_DEBUG_URL, GOOGLE_ANALYTICS_DEBUG_URL,
json={"hitParsingResult":[{'valid': True}]} json={"validationMessages": []}
) )
tracking_id = 'UA-186249766-2' token = 'G-6LYSMYK8DE'
client_id = 'ab33e241fbf042b6aa77c7655a768af7' secret = 'KLlpjLZ-SRGozS5f5wb_kw',
category = 'test' category = 'test'
action = 'test' action = 'test'
label = 'test' label = 'test'
value = '1' value = '1'
# when # when
json_response = send_ga_tracking_celery_event( task = send_ga_tracking_celery_event(
tracking_id, token,
client_id, secret,
category, category,
action, action,
label, label,
value).json() value)
# then # then
self.assertTrue(json_response["hitParsingResult"][0]["valid"]) self.assertTrue(task, 200)
def test_send_ga_tracking_celery_event_invalid_token(self, requests_mocker):
# given
requests_mocker.register_uri(
'POST',
GOOGLE_ANALYTICS_DEBUG_URL,
json={
"hitParsingResult":[
{
'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'
}
]
}
)
tracking_id = 'UA-IntentionallyBadTrackingID-2'
client_id = 'ab33e241fbf042b6aa77c7655a768af7'
category = 'test'
action = 'test'
label = 'test'
value = '1'
# when
json_response = send_ga_tracking_celery_event(
tracking_id,
client_id,
category,
action,
label,
value).json()
# then
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."
)

12
allianceauth/analytics/tests/test_utils.py
View File

@ -18,17 +18,17 @@ def create_testdata():
'abc@example.com', 'abc@example.com',
'password' 'password'
) )
#Token.objects.all().delete() # Token.objects.all().delete()
#Token.objects.create( # Token.objects.create(
# character_id=101, # character_id=101,
# character_name='character1', # character_name='character1',
# access_token='my_access_token' # access_token='my_access_token'
#) # )
#Token.objects.create( # Token.objects.create(
# character_id=102, # character_id=102,
# character_name='character2', # character_name='character2',
# access_token='my_access_token' # access_token='my_access_token'
#) # )
class TestAnalyticsUtils(TestCase): class TestAnalyticsUtils(TestCase):
@ -40,7 +40,7 @@ class TestAnalyticsUtils(TestCase):
users = install_stat_users() users = install_stat_users()
self.assertEqual(users, expected) self.assertEqual(users, expected)
#def test_install_stat_tokens(self): # def test_install_stat_tokens(self):
# create_testdata() # create_testdata()
# expected = 2 # expected = 2
# #

2
allianceauth/authentication/task_statistics/tests/test_signals.py
View File

@ -17,7 +17,7 @@ from allianceauth.eveonline.tasks import update_character
@override_settings( @override_settings(
CELERY_ALWAYS_EAGER=True,ALLIANCEAUTH_DASHBOARD_TASK_STATISTICS_DISABLED=False CELERY_ALWAYS_EAGER=True, ALLIANCEAUTH_DASHBOARD_TASK_STATISTICS_DISABLED=False
) )
class TestTaskSignals(TestCase): class TestTaskSignals(TestCase):
fixtures = ["disable_analytics"] fixtures = ["disable_analytics"]

1
allianceauth/project_template/project_name/settings/base.py
View File

@ -75,7 +75,6 @@ MIDDLEWARE = [
'django.contrib.auth.middleware.AuthenticationMiddleware', 'django.contrib.auth.middleware.AuthenticationMiddleware',
'django.contrib.messages.middleware.MessageMiddleware', 'django.contrib.messages.middleware.MessageMiddleware',
'django.middleware.clickjacking.XFrameOptionsMiddleware', 'django.middleware.clickjacking.XFrameOptionsMiddleware',
'allianceauth.analytics.middleware.AnalyticsMiddleware',
] ]
ROOT_URLCONF = 'allianceauth.urls' ROOT_URLCONF = 'allianceauth.urls'

1
allianceauth/services/modules/discord/tests/test_integration.py
View File

@ -377,6 +377,7 @@ class TestTasks(NoSocketsTestCase):
@patch(MODULE_PATH + '.models.DISCORD_GUILD_ID', TEST_GUILD_ID) @patch(MODULE_PATH + '.models.DISCORD_GUILD_ID', TEST_GUILD_ID)
@requests_mock.Mocker() @requests_mock.Mocker()
class StateTestCase(NoSocketsTestCase): class StateTestCase(NoSocketsTestCase):
fixtures = ['disable_analytics.json']
def setUp(self): def setUp(self):
clear_cache() clear_cache()