PEP440 versioning for admin dashboard

allianceauth/authentication/tests/test_templatetags.py

@@ -0,0 +1,268 @@
+from math import ceil
+from unittest.mock import patch
+
+from requests import RequestException
+import requests_mock
+from packaging.version import Version as Pep440Version
+
+from django.test import TestCase
+
+from allianceauth.templatetags.admin_status import (
+    status_overview,
+    _fetch_list_from_gitlab,
+    _current_notifications, 
+    _current_version_summary, 
+    _fetch_notification_issues_from_gitlab,
+    _fetch_tags_from_gitlab,
+    _latests_versions
+)
+
+MODULE_PATH = 'allianceauth.templatetags'
+
+
+def create_tags_list(tag_names: list):
+    return [{'name': str(tag_name)} for tag_name in tag_names]
+
+
+GITHUB_TAGS = create_tags_list(['v2.4.6a1', 'v2.4.5', 'v2.4.0', 'v2.0.0', 'v1.1.1'])
+GITHUB_NOTIFICATION_ISSUES = [
+    {
+        'id': 1,
+        'title': 'first issue'
+    },
+    {
+        'id': 2,
+        'title': 'second issue'
+    },
+    {
+        'id': 3,
+        'title': 'third issue'
+    },
+    {
+        'id': 4,
+        'title': 'forth issue'
+    },
+    {
+        'id': 5,
+        'title': 'fifth issue'
+    },
+    {
+        'id': 6,
+        'title': 'sixth issue'
+    },
+]
+TEST_VERSION = '2.6.5'
+
+
+class TestStatusOverviewTag(TestCase):
+
+    @patch(MODULE_PATH + '.admin_status.__version__', TEST_VERSION)
+    @patch(MODULE_PATH + '.admin_status._fetch_celery_queue_length')
+    @patch(MODULE_PATH + '.admin_status._current_version_summary')
+    @patch(MODULE_PATH + '.admin_status._current_notifications')    
+    def test_status_overview(
+        self, 
+        mock_current_notifications, 
+        mock_current_version_info,
+        mock_fetch_celery_queue_length
+    ):
+        notifications = {
+            'notifications': GITHUB_NOTIFICATION_ISSUES[:5]
+        }
+        mock_current_notifications.return_value = notifications
+            
+        version_info = {
+            'latest_major': True,
+            'latest_minor': True,
+            'latest_patch': True,
+            'current_version': TEST_VERSION,
+            'latest_major_version': '2.0.0',
+            'latest_minor_version': '2.4.0',
+            'latest_patch_version': '2.4.5',
+        }
+        mock_current_version_info.return_value = version_info
+        mock_fetch_celery_queue_length.return_value = 3
+        
+        context = {}
+        result = status_overview(context)
+        expected = {
+            'notifications': GITHUB_NOTIFICATION_ISSUES[:5],
+            'latest_major': True,
+            'latest_minor': True,
+            'latest_patch': True,
+            'current_version': TEST_VERSION,
+            'latest_major_version': '2.0.0',
+            'latest_minor_version': '2.4.0',
+            'latest_patch_version': '2.4.5',
+            'task_queue_length': 3,
+        }
+        self.assertEqual(result, expected)
+
+
+class TestNotifications(TestCase):
+
+    @requests_mock.mock()
+    def test_fetch_notification_issues_from_gitlab(self, requests_mocker):
+        url = (
+            'https://gitlab.com/api/v4/projects/allianceauth%2Fallianceauth/issues'
+            '?labels=announcement'
+        )        
+        requests_mocker.get(url, json=GITHUB_NOTIFICATION_ISSUES)
+        result = _fetch_notification_issues_from_gitlab()
+        self.assertEqual(result, GITHUB_NOTIFICATION_ISSUES)
+
+    @patch(MODULE_PATH + '.admin_status.cache')
+    def test_current_notifications_normal(self, mock_cache):
+        mock_cache.get_or_set.return_value = GITHUB_NOTIFICATION_ISSUES
+
+        result = _current_notifications()
+        self.assertEqual(result['notifications'], GITHUB_NOTIFICATION_ISSUES[:5])
+
+    @patch(MODULE_PATH + '.admin_status.cache')
+    def test_current_notifications_failed(self, mock_cache):
+        mock_cache.get_or_set.side_effect = RequestException
+
+        result = _current_notifications()
+        self.assertEqual(result['notifications'], list())
+
+
+class TestCeleryQueueLength(TestCase):
+
+    def test_get_celery_queue_length(self):
+        pass
+
+
+class TestVersionTags(TestCase):
+
+    @patch(MODULE_PATH + '.admin_status.__version__', TEST_VERSION)
+    @patch(MODULE_PATH + '.admin_status.cache')
+    def test_current_version_info_normal(self, mock_cache):
+        mock_cache.get_or_set.return_value = GITHUB_TAGS
+
+        result = _current_version_summary()
+        self.assertTrue(result['latest_major'])
+        self.assertTrue(result['latest_minor'])
+        self.assertTrue(result['latest_patch'])
+        self.assertEqual(result['latest_major_version'], '2.0.0')
+        self.assertEqual(result['latest_minor_version'], '2.4.0')
+        self.assertEqual(result['latest_patch_version'], '2.4.5')
+
+    @patch(MODULE_PATH + '.admin_status.__version__', TEST_VERSION)
+    @patch(MODULE_PATH + '.admin_status.cache')
+    def test_current_version_info_failed(self, mock_cache):
+        mock_cache.get_or_set.side_effect = RequestException
+        
+        expected = {}
+        result = _current_version_summary()
+        self.assertEqual(result, expected)
+
+    @requests_mock.mock()
+    def test_fetch_tags_from_gitlab(self, requests_mocker):
+        url = (
+            'https://gitlab.com/api/v4/projects/allianceauth%2Fallianceauth'
+            '/repository/tags'
+        )        
+        requests_mocker.get(url, json=GITHUB_TAGS)
+        result = _fetch_tags_from_gitlab()
+        self.assertEqual(result, GITHUB_TAGS)
+
+
+class TestLatestsVersion(TestCase):
+
+    def test_all_version_types_defined(self):
+        
+        tags = create_tags_list(
+            ['2.1.1', '2.1.0', '2.0.0', '2.1.1a1', '1.1.1', '1.1.0', '1.0.0']
+        )
+        major, minor, patch = _latests_versions(tags)
+        self.assertEqual(major, Pep440Version('2.0.0'))
+        self.assertEqual(minor, Pep440Version('2.1.0'))
+        self.assertEqual(patch, Pep440Version('2.1.1'))
+
+    def test_major_and_minor_not_defined_with_zero(self):
+        
+        tags = create_tags_list(
+            ['2.1.2', '2.1.1', '2.0.1', '2.1.1a1', '1.1.1', '1.1.0', '1.0.0']
+        )
+        major, minor, patch = _latests_versions(tags)
+        self.assertEqual(major, Pep440Version('2.0.1'))
+        self.assertEqual(minor, Pep440Version('2.1.1'))
+        self.assertEqual(patch, Pep440Version('2.1.2'))
+        
+    def test_can_ignore_invalid_versions(self):
+        
+        tags = create_tags_list(
+            ['2.1.1', '2.1.0', '2.0.0', '2.1.1a1', 'invalid']
+        )
+        major, minor, patch = _latests_versions(tags)
+        self.assertEqual(major, Pep440Version('2.0.0'))
+        self.assertEqual(minor, Pep440Version('2.1.0'))
+        self.assertEqual(patch, Pep440Version('2.1.1'))
+
+
+class TestFetchListFromGitlab(TestCase):
+   
+    page_size = 2
+    
+    def setUp(self):
+        self.url = (
+            'https://gitlab.com/api/v4/projects/allianceauth%2Fallianceauth'
+            '/repository/tags'
+        )
+
+    @classmethod
+    def my_callback(cls, request, context):
+        page = int(request.qs['page'][0])
+        start = (page - 1) * cls.page_size
+        end = start + cls.page_size
+        return GITHUB_TAGS[start:end]        
+    
+    @requests_mock.mock()
+    def test_can_fetch_one_page_with_header(self, requests_mocker):
+        headers = {
+            'x-total-pages': '1'
+        }
+        requests_mocker.get(self.url, json=GITHUB_TAGS, headers=headers)
+        result = _fetch_list_from_gitlab(self.url)
+        self.assertEqual(result, GITHUB_TAGS)
+        self.assertEqual(requests_mocker.call_count, 1)
+
+    @requests_mock.mock()
+    def test_can_fetch_one_page_wo_header(self, requests_mocker):        
+        requests_mocker.get(self.url, json=GITHUB_TAGS)
+        result = _fetch_list_from_gitlab(self.url)
+        self.assertEqual(result, GITHUB_TAGS)
+        self.assertEqual(requests_mocker.call_count, 1)
+
+    @requests_mock.mock()
+    def test_can_fetch_one_page_and_ignore_invalid_header(self, requests_mocker):
+        headers = {
+            'x-total-pages': 'invalid'
+        }
+        requests_mocker.get(self.url, json=GITHUB_TAGS, headers=headers)
+        result = _fetch_list_from_gitlab(self.url)
+        self.assertEqual(result, GITHUB_TAGS)
+        self.assertEqual(requests_mocker.call_count, 1)
+
+    @requests_mock.mock()
+    def test_can_fetch_multiple_pages(self, requests_mocker):     
+        total_pages = ceil(len(GITHUB_TAGS) / self.page_size)
+        headers = {
+            'x-total-pages': str(total_pages)
+        }
+        requests_mocker.get(self.url, json=self.my_callback, headers=headers)
+        result = _fetch_list_from_gitlab(self.url)
+        self.assertEqual(result, GITHUB_TAGS)
+        self.assertEqual(requests_mocker.call_count, total_pages)
+
+    @requests_mock.mock()
+    def test_can_fetch_given_number_of_pages_only(self, requests_mocker):     
+        total_pages = ceil(len(GITHUB_TAGS) / self.page_size)
+        headers = {
+            'x-total-pages': str(total_pages)
+        }
+        requests_mocker.get(self.url, json=self.my_callback, headers=headers)
+        max_pages = 2
+        result = _fetch_list_from_gitlab(self.url, max_pages=max_pages)
+        self.assertEqual(result, GITHUB_TAGS[:4])
+        self.assertEqual(requests_mocker.call_count, max_pages)

allianceauth/templates/allianceauth/admin-status/overview.html

@@ -36,7 +36,7 @@
                                 {{ current_version }}
                             </p>
                         </li>
-                        <li class="list-group-item list-group-item-{% if latest_major %}success{% else %}warning{% endif %}">
+                        <li class="list-group-item list-group-item-{% if latest_major %}success{% else %}danger{% endif %}">
                             <h5 class="list-group-item-heading">{% trans "Latest Major" %}</h5>
                             <p class="list-group-item-text">
                                 <a href="https://gitlab.com/allianceauth/allianceauth/tags" style="color:#000">
@@ -46,7 +46,7 @@
                                 {% if not latest_major %}<br>{% trans "Update available" %}{% endif %}
                             </p>
                         </li>
-                        <li class="list-group-item list-group-item-{% if latest_minor %}success{% else %}warning{% endif %}">
+                        <li class="list-group-item list-group-item-{% if latest_minor %}success{% else %}danger{% endif %}">
                             <h5 class="list-group-item-heading">{% trans "Latest Minor" %}</h5>
                             <p class="list-group-item-text">
                                 <a href="https://gitlab.com/allianceauth/allianceauth/tags" style="color:#000">
@@ -56,7 +56,7 @@
                                 {% if not latest_minor %}<br>{% trans "Update available" %}{% endif %}
                             </p>
                         </li>
-                        <li class="list-group-item list-group-item-{% if latest_patch %}success{% else %}danger{% endif %}">
+                        <li class="list-group-item list-group-item-{% if latest_patch %}success{% else %}warning{% endif %}">
                             <h5 class="list-group-item-heading">{% trans "Latest Patch" %}</h5>
                             <p class="list-group-item-text">
                                 <a href="https://gitlab.com/allianceauth/allianceauth/tags" style="color:#000">

allianceauth/templatetags/admin_status.py

@@ -1,60 +1,59 @@
-import requests
 import logging
+
+import requests
 import amqp.exceptions
-import semantic_version as semver
+from packaging.version import Version as Pep440Version, InvalidVersion
+from celery.app import app_or_default
+
 from django import template
 from django.conf import settings
 from django.core.cache import cache
-from celery.app import app_or_default
+
 from allianceauth import __version__
 
+
 register = template.Library()
 
-TAG_CACHE_TIME = 10800  # 3 hours
+# cache timers
+TAG_CACHE_TIME = 3600  # 1 hours
 NOTIFICATION_CACHE_TIME = 300  # 5 minutes
+# timeout for all requests
+REQUESTS_TIMEOUT = 5    # 5 seconds
+# max pages to be fetched from gitlab
+MAX_PAGES = 50
 
+GITLAB_AUTH_REPOSITORY_TAGS_URL = (
+    'https://gitlab.com/api/v4/projects/allianceauth%2Fallianceauth/repository/tags'
+)
+GITLAB_AUTH_ANNOUNCEMENT_ISSUES_URL = (
+    'https://gitlab.com/api/v4/projects/allianceauth%2Fallianceauth/issues'
+    '?labels=announcement'
+)
 
 logger = logging.getLogger(__name__)
 
 
-def get_git_tags():
-    request = requests.get('https://gitlab.com/api/v4/projects/allianceauth%2Fallianceauth/repository/tags')
-    request.raise_for_status()
-    return request.json()
-
-
-def get_notification_issues():
-    # notification
-    request = requests.get(
-        'https://gitlab.com/api/v4/projects/allianceauth%2Fallianceauth/issues?labels=announcement')
-    request.raise_for_status()
-    return request.json()
-
-
 @register.inclusion_tag('allianceauth/admin-status/overview.html', takes_context=True)
 def status_overview(context):
     response = {
-        'notifications': list(),
+        'notifications': list(),        
-        'latest_major': True,
-        'latest_minor': True,
-        'latest_patch': True,
         'current_version': __version__,
         'task_queue_length': -1,
     }
-
+    response.update(_current_notifications())
-    response.update(get_notifications())
+    response.update(_current_version_summary())
-    response.update(get_version_info())
+    response.update({'task_queue_length': _fetch_celery_queue_length()})
-    response.update({'task_queue_length': get_celery_queue_length()})
-
     return response
 
 
-def get_celery_queue_length():
+def _fetch_celery_queue_length():
     try:
         app = app_or_default(None)
         with app.connection_or_acquire() as conn:
             return conn.default_channel.queue_declare(
-                queue=getattr(settings, 'CELERY_DEFAULT_QUEUE', 'celery'), passive=True).message_count
+                queue=getattr(settings, 'CELERY_DEFAULT_QUEUE', 'celery'), 
+                passive=True
+            ).message_count
     except amqp.exceptions.ChannelError:
         # Queue doesn't exist, probably empty
         return 0
@@ -63,72 +62,111 @@ def get_celery_queue_length():
     return -1
 
 
-def get_notifications():
+def _current_notifications() -> dict:
-    response = {
+    """returns the newest 5 announcement issues"""    
-        'notifications': list(),
-    }
     try:
-        notifications = cache.get_or_set('gitlab_notification_issues', get_notification_issues,
+        notifications = cache.get_or_set(
-                                         NOTIFICATION_CACHE_TIME)
+            'gitlab_notification_issues', 
-        # Limit notifications to those posted by repo owners and members
+            _fetch_notification_issues_from_gitlab,
-        response['notifications'] += notifications[:5]
+            NOTIFICATION_CACHE_TIME
+        )        
+        top_notifications = notifications[:5]
     except requests.RequestException:
         logger.exception('Error while getting gitlab notifications')
+        top_notifications = []
+    
+    response = {
+        'notifications': top_notifications,
+    }
     return response
 
 
-def get_version_info():
+def _fetch_notification_issues_from_gitlab() -> list:    
-    response = {
+    return _fetch_list_from_gitlab(GITLAB_AUTH_ANNOUNCEMENT_ISSUES_URL, max_pages=10)
-        'latest_major': True,
+
-        'latest_minor': True,
+
-        'latest_patch': True,
+def _current_version_summary() -> dict:
-        'current_version': __version__,
+    """returns the current version info"""    
-    }
     try:
-        tags = cache.get_or_set('git_release_tags', get_git_tags, TAG_CACHE_TIME)
+        tags = cache.get_or_set(
-        current_ver = semver.Version.coerce(__version__)
+            'git_release_tags', _fetch_tags_from_gitlab, TAG_CACHE_TIME
-
+        )
-        # Set them all to the current version to start
-        # If the server has only earlier or the same version
-        # then this will become the major/minor/patch versions
-        latest_major = current_ver
-        latest_minor = current_ver
-        latest_patch = current_ver
-
-        response.update({
-            'latest_major_version': str(latest_major),
-            'latest_minor_version': str(latest_minor),
-            'latest_patch_version': str(latest_patch),
-        })
-
-        for tag in tags:
-            tag_name = tag.get('name')
-            if tag_name[0] == 'v':
-                # Strip 'v' off front of verison if it exists
-                tag_name = tag_name[1:]
-            try:
-                tag_ver = semver.Version.coerce(tag_name)
-            except ValueError:
-                tag_ver = semver.Version('0.0.0', partial=True)
-            if tag_ver > current_ver:
-                if latest_major is None or tag_ver > latest_major:
-                    latest_major = tag_ver
-                    response['latest_major_version'] = tag_name
-                if tag_ver.major > current_ver.major:
-                    response['latest_major'] = False
-                elif tag_ver.major == current_ver.major:
-                    if latest_minor is None or tag_ver > latest_minor:
-                        latest_minor = tag_ver
-                        response['latest_minor_version'] = tag_name
-                    if tag_ver.minor > current_ver.minor:
-                        response['latest_minor'] = False
-                    elif tag_ver.minor == current_ver.minor:
-                        if latest_patch is None or tag_ver > latest_patch:
-                            latest_patch = tag_ver
-                            response['latest_patch_version'] = tag_name
-                        if tag_ver.patch > current_ver.patch:
-                            response['latest_patch'] = False
-
     except requests.RequestException:
         logger.exception('Error while getting gitlab release tags')
+        return {}
+        
+    latest_major_version, latest_minor_version, latest_patch_version = \
+        _latests_versions(tags)    
+    current_version = Pep440Version(__version__)
+
+    has_latest_major = \
+        current_version >= latest_major_version if latest_major_version else False
+    has_latest_minor = \
+        current_version >= latest_minor_version if latest_minor_version else False
+    has_latest_patch = \
+        current_version >= latest_patch_version if latest_patch_version else False
+
+    response = {        
+        'latest_major': has_latest_major,
+        'latest_minor': has_latest_minor,
+        'latest_patch': has_latest_patch,
+        'current_version': str(current_version),
+        'latest_major_version': str(latest_major_version),
+        'latest_minor_version': str(latest_minor_version),
+        'latest_patch_version': str(latest_patch_version)
+    }        
     return response
+
+
+def _fetch_tags_from_gitlab():
+    return _fetch_list_from_gitlab(GITLAB_AUTH_REPOSITORY_TAGS_URL)
+
+
+def _latests_versions(tags: list) -> tuple:
+    """returns latests version from given tags list
+    
+    Non-compliant tags will be ignored
+    """
+    versions = list()
+    for tag in tags:
+        try:
+            version = Pep440Version(tag.get('name'))            
+        except InvalidVersion:
+            pass
+        else:
+            if not version.is_prerelease:
+                versions.append(version)
+    
+    latest_version = latest_patch_version = max(versions)
+    latest_major_version = min([
+        v for v in versions if v.major == latest_version.major
+    ])
+    latest_minor_version = min([
+        v for v in versions 
+        if v.major == latest_version.major and v.minor == latest_version.minor
+    ])
+    
+    return latest_major_version, latest_minor_version, latest_patch_version
+
+
+def _fetch_list_from_gitlab(url: str, max_pages: int = MAX_PAGES):
+    """returns a list from the GitLab API. Supports pageing"""
+    result = list()    
+    for page in range(1, max_pages + 1):
+        request = requests.get(
+            url, params={'page': page}, timeout=REQUESTS_TIMEOUT
+        )
+        request.raise_for_status()
+        result += request.json()
+        if 'x-total-pages' in request.headers:
+            try:
+                total_pages = int(request.headers['x-total-pages'])
+            except ValueError:
+                total_pages = None
+        else:
+            total_pages = None
+            
+        if not total_pages or page >= total_pages:
+            break
+    
+    return result

setup.py

@@ -16,6 +16,7 @@ install_requires = [
     'python-slugify>=1.2',
     'requests-oauthlib',
     'semantic_version',
+    'packaging>=20.1,<21',
 
     'redis>=3.3.1,<4.0.0',
     'celery>=4.3.0,<5.0.0',