Enable quality checks (#449)

* fix: enable quality checks

.github/workflows/ci.yml

@@ -14,7 +14,7 @@ jobs:
       fail-fast: false
       matrix:
         python-version: ["3.11", "3.12"]
-        toxenv: ["django42"] # "quality", "pii_check", "check_keywords"
+        toxenv: ["django42", "quality", "pii_check", "check_keywords"]
 
     services:
       mysql:

.github/workflows/trivy-code-scanning.yml

@@ -23,6 +23,7 @@ jobs:
           scan-type: "fs"
           format: "sarif"
           output: "trivy-results.sarif"
+          args: --skip-update
 
       - name: Upload Trivy scan results to GitHub Security tab
         uses: github/codeql-action/upload-sarif@v3

.gitignore

@@ -17,6 +17,7 @@ venv/
 
 .idea/
 
-
 # Helm dependencies
 /helmcharts/**/*.tgz
+
+reports/

.pycodestyle

@@ -0,0 +1,4 @@
+[pycodestyle]
+exclude = .git, .tox, migrations
+ignore = E731
+max-line-length = 120

notesapi/v1/management/commands/bulk_create_notes.py

@@ -3,45 +3,47 @@
 import os
 import random
 import uuid
-from optparse import make_option
 
 from django.core.management.base import BaseCommand, CommandError
-from django.db import transaction
 
 from notesapi.v1.models import Note
 
 
-def extract_comma_separated_list(option, opt_str, value, parser):
+def extract_comma_separated_list(option, value, parser):
     """Parse an option string as a comma separated list"""
     setattr(parser.values, option.dest, [course_id.strip() for course_id in value.split(',')])
 
 
 class Command(BaseCommand):
     args = '<total_notes>'
+
     def add_arguments(self, parser):
         parser.add_argument(
-        '--per_user',
+            '--per_user',
             action='store',
-            type='int',
+            type=int,
             default=50,
             help='number of notes that should be attributed to each user (default 50)'
-        ),
+        )
+
         parser.add_argument(
             '--course_ids',
             action='callback',
             callback=extract_comma_separated_list,
-            type='string',
+            type=str,
             default=['edX/DemoX/Demo_Course'],
             help='comma-separated list of course_ids for which notes should be randomly attributed'
-        ),
+        )
+
         parser.add_argument(
             '--batch_size',
             action='store',
-            type='int',
+            type=int,
             default=1000,
             help='number of notes that should be bulk inserted at a time - useful for getting around the maximum SQL '
                  'query size'
         )
+
     help = 'Add N random notes to the database'
 
     def handle(self, *args, **options):
@@ -58,6 +60,7 @@ def handle(self, *args, **options):
         for notes_chunk in grouper_it(note_iter(total_notes, notes_per_user, course_ids), batch_size):
             Note.objects.bulk_create(notes_chunk)
 
+
 def note_iter(total_notes, notes_per_user, course_ids):
     """
     Return an iterable of random notes data of length `total_notes`.
@@ -85,7 +88,9 @@ def weighted_get_words(weighted_num_words):
             random.choice([word_count for word_count, weight in weighted_num_words for i in range(weight)])
         )
 
-    get_new_user_id = lambda: uuid.uuid4().hex
+    def get_new_user_id():
+        return uuid.uuid4().hex
+
     user_id = get_new_user_id()
 
     for note_count in range(total_notes):
@@ -108,7 +113,6 @@ def grouper_it(iterable, batch_size):
     Return an iterator of iterators.  Each child iterator yields the
     next `batch_size`-many elements from `iterable`.
     """
-    iterator = iter(iterable)
     while True:
         chunk_it = itertools.islice(iterable, batch_size)
         try:

notesapi/v1/models.py

@@ -33,13 +33,13 @@ def create(cls, note_dict):
         if len(note_dict) == 0:
             raise ValidationError('Note must have a body.')
 
-        ranges = note_dict.get('ranges', list())
+        ranges = note_dict.get('ranges', [])
 
         if len(ranges) < 1:
             raise ValidationError('Note must contain at least one range.')
 
         note_dict['ranges'] = json.dumps(ranges)
         note_dict['user_id'] = note_dict.pop('user', None)
-        note_dict['tags'] = json.dumps(note_dict.get('tags', list()), ensure_ascii=False)
+        note_dict['tags'] = json.dumps(note_dict.get('tags', []), ensure_ascii=False)
 
         return cls(**note_dict)

notesapi/v1/search_indexes/backends/note.py

@@ -9,6 +9,7 @@
 __all__ = ('CompoundSearchFilterBackend', 'FilteringFilterBackend')
 
 
+# pylint: disable=abstract-method
 class CompoundSearchFilterBackend(CompoundSearchFilterBackendOrigin):
     """
     Extends compound search backend.

notesapi/v1/search_indexes/serializers/note.py

@@ -54,6 +54,6 @@ def get_tags(self, note):
         Return note tags.
         """
         if hasattr(note.meta, 'highlight') and hasattr(note.meta.highlight, 'tags'):
-            return [i for i in note.meta.highlight.tags]
+            return list(note.meta.highlight.tags)
 
-        return [i for i in note.tags] if note.tags else []
+        return list(note.tags) if note.tags else []

notesapi/v1/tests/test_update_index.py

@@ -1,10 +1,10 @@
 from unittest import skipIf
 
+import factory
 from django.conf import settings
 from django.core.management import call_command
-from django.urls import reverse
 from django.db.models import signals
-import factory
+from django.urls import reverse
 
 from .test_views import BaseAnnotationViewTests
 
@@ -51,7 +51,7 @@ def test_delete(self):
 
         # Delete first note.
         url = reverse('api:v1:annotations_detail', kwargs={'annotation_id': first_note['id']})
-        response = self.client.delete(url, self.headers)
+        response = self.client.delete(url, self.headers)  # pylint: disable=unused-variable
 
         # Delete second note.
         url = reverse('api:v1:annotations_detail', kwargs={'annotation_id': second_note['id']})

notesapi/v1/tests/test_views.py

@@ -1,17 +1,15 @@
-import sys
 import unittest
 from calendar import timegm
 from datetime import datetime, timedelta
+from unittest.mock import patch
 from urllib import parse
 
+import ddt
+import jwt
 from django.conf import settings
 from django.core.management import call_command
 from django.test.utils import override_settings
 from django.urls import reverse
-from unittest.mock import patch
-
-import ddt
-import jwt
 from rest_framework import status
 from rest_framework.test import APITestCase
 
@@ -21,9 +19,9 @@
 TEST_OTHER_USER = "test_other_user_id"
 
 if not settings.ES_DISABLED:
-    from notesapi.v1.search_indexes.documents import NoteDocument
+    from notesapi.v1.search_indexes.documents import NoteDocument  # pylint: disable=unused-import
 else:
-    def call_command(*args, **kwargs):
+    def call_command(*args, **kwargs):  # pylint: disable=function-redefined
         pass
 
 
@@ -111,7 +109,7 @@ def get_annotations(self, query_parameters=None, expected_status=200):
         self.assertEqual(expected_status, response.status_code)
         return response.data
 
-    # pylint: disable=too-many-arguments
+    # pylint: disable=too-many-positional-arguments
     def verify_pagination_info(
             self, response,
             total_annotations,
@@ -161,7 +159,7 @@ def get_page_value(url, current_page):
         self.assertEqual(get_page_value(response['next'], response['current_page']), next_page)
         self.assertEqual(response['start'], start)
 
-    # pylint: disable=too-many-arguments
+    # pylint: disable=too-many-positional-arguments
     def verify_list_view_pagination(
             self,
             query_parameters,
@@ -176,7 +174,6 @@ def verify_list_view_pagination(
         """
         Verify pagination information for AnnotationListView
         """
-        total_annotations = total_annotations
         for i in range(total_annotations):
             self._create_annotation(text=f'annotation {i}')
 
@@ -192,7 +189,7 @@ def verify_list_view_pagination(
             start=start
         )
 
-    # pylint: disable=too-many-arguments
+    # pylint: disable=too-many-positional-arguments
     def verify_search_view_pagination(
             self,
             query_parameters,
@@ -207,7 +204,6 @@ def verify_search_view_pagination(
         """
         Verify pagination information for AnnotationSearchView
         """
-        total_annotations = total_annotations
         for i in range(total_annotations):
             self._create_annotation(text=f'annotation {i}')
 
@@ -370,15 +366,15 @@ def test_create_maximum_allowed(self):
         # if user tries to create note in a different course it should succeed
         kwargs = {'course_id': 'test-course-id-2'}
         response = self._create_annotation(**kwargs)
-        self.assertTrue('id' in response)
+        self.assertIn('id', response)
 
         # if another user to tries to create note in first course it should succeed
         token = get_id_token(TEST_OTHER_USER)
         self.client.credentials(HTTP_X_ANNOTATOR_AUTH_TOKEN=token)
         self.headers = {'user': TEST_OTHER_USER}
         kwargs = {'user': TEST_OTHER_USER}
         response = self._create_annotation(**kwargs)
-        self.assertTrue('id' in response)
+        self.assertIn('id', response)
 
     def test_read_all_no_annotations(self):
         """
@@ -437,7 +433,7 @@ def test_read_all_no_query_param(self):
         {'page': 2, 'annotations_per_page': 10, 'previous_page': 1, 'next_page': 3, 'start': 10},
         {'page': 3, 'annotations_per_page': 3, 'previous_page': 2, 'next_page': None, 'start': 20}
     )
-    # pylint: disable=too-many-arguments
+    # pylint: disable=too-many-positional-arguments
     def test_pagination_multiple_pages(self, page, annotations_per_page, previous_page, next_page, start):
         """
         Verify that pagination info is correct when we have data spanned on multiple pages.
@@ -1081,7 +1077,7 @@ def test_search_highlight_tag(self):
         {'page': 2, 'annotations_per_page': 10, 'previous_page': 1, 'next_page': 3, 'start': 10},
         {'page': 3, 'annotations_per_page': 3, 'previous_page': 2, 'next_page': None, 'start': 20}
     )
-    # pylint: disable=too-many-arguments
+    # pylint: disable=too-many-positional-arguments
     def test_pagination_multiple_pages(self, page, annotations_per_page, previous_page, next_page, start):
         """
         Verify that pagination info is correct when we have data spanned on multiple pages.
@@ -1221,7 +1217,7 @@ def test_no_token(self):
         """
         403 when no token is provided
         """
-        self.client._credentials = {}
+        self.client._credentials = {}  # pylint: disable=protected-access
         response = self.client.get(self.url, self.headers)
         self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
 

notesapi/v1/utils.py

@@ -42,7 +42,7 @@ def dict_to_querydict(dict_):
             query_dict.setlist(name, value)
         else:
             query_dict.appendlist(name, value)
-    query_dict._mutable = False
+    query_dict._mutable = False  # pylint: disable=protected-access
     return query_dict
 
 

notesapi/v1/views.py

@@ -1,3 +1,4 @@
+#  pylint:disable=possibly-used-before-assignment
 import json
 import logging
 import newrelic.agent
@@ -171,9 +172,10 @@ def initiate_es_specific_state_if_is_enabled(self):
         Should be called in the class `__init__` method.
         """
         if not settings.ES_DISABLED:
-            self.client = connections.get_connection(self.document._get_using())
-            self.index = self.document._index._name
-            self.mapping = self.document._doc_type.mapping.properties.name
+            self.client = connections.get_connection(self.document._get_using())  # pylint: disable=protected-access
+            self.index = self.document._index._name  # pylint: disable=protected-access
+            self.mapping = self.document._doc_type.mapping.properties.name  # pylint: disable=protected-access
+            # pylint: disable=protected-access
             self.search = Search(using=self.client, index=self.index, doc_type=self.document._doc_type.name)
 
     @property
@@ -216,9 +218,13 @@ def paginator(self):
         """
         if not hasattr(self, '_paginator'):
             if self.pagination_class is None:
-                self._paginator = None
+                self._paginator = None  # pylint: disable=attribute-defined-outside-init
             else:
-                self._paginator = self.pagination_class() if self.is_es_disabled else ESNotesPagination()
+                self._paginator = (  # pylint: disable=attribute-defined-outside-init
+                    self.pagination_class()
+                    if self.is_es_disabled
+                    else ESNotesPagination()
+                )
 
         return self._paginator
 
@@ -279,7 +285,7 @@ def build_query_params_state(self):
             else:
                 self.query_params['user'] = self.params['user']
 
-    def get(self, *args, **kwargs):  # pylint: disable=unused-argument
+    def get(self, *args, **kwargs):
         """
         Search annotations in most appropriate storage
         """
@@ -294,7 +300,7 @@ class AnnotationRetireView(GenericAPIView):
     Administrative functions for the notes service.
     """
 
-    def post(self, *args, **kwargs):  # pylint: disable=unused-argument
+    def post(self, *args, **kwargs):
         """
         Delete all annotations for a user.
         """
@@ -422,7 +428,7 @@ class AnnotationListView(GenericAPIView):
 
     serializer_class = NoteSerializer
 
-    def get(self, *args, **kwargs):  # pylint: disable=unused-argument
+    def get(self, *args, **kwargs):
         """
         Get paginated list of all annotations.
         """
@@ -440,7 +446,7 @@ def get(self, *args, **kwargs):  # pylint: disable=unused-argument
         response = self.get_paginated_response(serializer.data)
         return response
 
-    def post(self, *args, **kwargs):  # pylint: disable=unused-argument
+    def post(self, *args, **kwargs):
         """
         Create a new annotation.
 
@@ -549,7 +555,7 @@ class AnnotationDetailView(APIView):
             * HTTP_204_NO_CONTENT is returned
     """
 
-    def get(self, *args, **kwargs):  # pylint: disable=unused-argument
+    def get(self, *args, **kwargs):
         """
         Get an existing annotation.
         """
@@ -563,7 +569,7 @@ def get(self, *args, **kwargs):  # pylint: disable=unused-argument
         serializer = NoteSerializer(note)
         return Response(serializer.data)
 
-    def put(self, *args, **kwargs):  # pylint: disable=unused-argument
+    def put(self, *args, **kwargs):
         """
         Update an existing annotation.
         """
@@ -587,7 +593,7 @@ def put(self, *args, **kwargs):  # pylint: disable=unused-argument
         serializer = NoteSerializer(note)
         return Response(serializer.data)
 
-    def delete(self, *args, **kwargs):  # pylint: disable=unused-argument
+    def delete(self, *args, **kwargs):
         """
         Delete an annotation.
         """