Merge pull request RedHatInsights#1013 from petracihalova/add-type-ch…

…ecker

[RHCLOUD-30835] Add type checker

.dockerignore

@@ -12,6 +12,7 @@
 !tests/
 !requirements.txt
 !Makefile
+!mypy.ini
 
 # keep the test files out of the final image
 #**/*test*

Dockerfile-pr-check

@@ -4,6 +4,7 @@ USER root
 COPY Pipfile .
 COPY Pipfile.lock .
 COPY tox.ini .
+COPY mypy.ini .
 COPY requirements.txt .
 COPY Makefile .
 
@@ -18,5 +19,5 @@ RUN pipenv install --deploy && \
 #copy the src files
 COPY . .
 
-#Add command to run tox 
-CMD ["tox"]
+#Add command to run tox
+CMD ["tox"]

Makefile

@@ -30,6 +30,7 @@ Please use \`make <target>' where <target> is one of:
   html                     create html documentation for the project
   lint                     run linting against the project
   format                   format linting errors found by lint task
+  typecheck                run type check
 
 --- Commands using local services ---
   create-test-db-file      create a Postgres DB dump file for RBAC
@@ -95,6 +96,9 @@ lint:
 format:
 	black -t py39 -l 119 rbac tests
 
+typecheck:
+	mypy --install-types --non-interactive rbac
+
 reinitdb:
 	make start-db
 	make reset-db

Pipfile

@@ -48,6 +48,8 @@ flake8 = "==6.0.0"
 flake8-docstrings = "==1.7.0"
 flake8-import-order = "==0.18.2"
 flake8-quotes = "==3.3.2"
+mypy = "==1.8.0"
+types-pyyaml = "==6.0.12.12"
 
 [requires]
 python_version = "3.9"

mypy.ini

@@ -0,0 +1,4 @@
+[mypy]
+python_version = 3.9
+ignore_missing_imports = True
+disable_error_code = annotation-unchecked, var-annotated, misc, index

rbac/api/models.py

@@ -72,5 +72,5 @@ class User:
     is_active = True
     org_id = None
     # Service account properties.
-    client_id: str = None
+    client_id: str = ""
     is_service_account: bool = False

rbac/management/authorization/token_validator.py

@@ -67,8 +67,7 @@ def _get_json_web_keyset(self) -> KeySet:
         except Exception as e:
             logger.debug(
                 "Fetching the JSON Web Key Set from Redis raised an exception, attempting to fetch the keys from the"
-                " OIDC configuration instead. Raised error: ",
-                e,
+                f" OIDC configuration instead. Raised error: {e}"
             )
 
         if jwks_certificates_response:
@@ -79,7 +78,7 @@ def _get_json_web_keyset(self) -> KeySet:
             if not status.is_success(oidc_response.status_code):
                 logger.error(
                     f"Unable to get the OIDC configuration payload when attempting to validate a JWT token. Response"
-                    f" code: {oidc_response.status_code}. Response body: {oidc_response.content}"
+                    f" code: {oidc_response.status_code!r}. Response body: {oidc_response.content!r}"
                 )
                 raise UnableMeetPrerequisitesError()
 
@@ -90,7 +89,7 @@ def _get_json_web_keyset(self) -> KeySet:
             if not jwks_uri:
                 logger.error(
                     f"Unable to extract the JWKs' URI when attempting to validate a JWT token. Actual payload:"
-                    f"{oidc_response.content}"
+                    f"{oidc_response.content!r}"
                 )
                 raise UnableMeetPrerequisitesError()
 
@@ -117,7 +116,7 @@ def _get_json_web_keyset(self) -> KeySet:
         try:
             return KeySet.import_key_set(jwks_certificates_response)
         except Exception as e:
-            logger.error("Unable to import IT's public keys to validate the token: {token:%s}".format(token=str(e)))
+            logger.error(f"Unable to import IT's public keys to validate the token: {e}")
             raise UnableMeetPrerequisitesError()
 
     def validate_token(self, request: Request) -> str:

rbac/management/group/view.py

@@ -17,6 +17,7 @@
 
 """View for group management."""
 import logging
+from typing import Iterable
 
 import requests
 from django.conf import settings
@@ -397,9 +398,9 @@ def add_service_accounts(
         user: User,
         group: Group,
         bearer_token: str,
-        service_accounts: [dict],
-        account_name: str = None,
-        org_id: str = None,
+        service_accounts: Iterable[dict],
+        account_name: str = "",
+        org_id: str = "",
     ) -> Group:
         """Process the list of service accounts and add them to the group."""
         # Fetch all the user's service accounts from IT. If we are on a development or testing environment, we might
@@ -1061,7 +1062,7 @@ def obtain_roles_with_exclusion(self, request, group):
         return roles.exclude(uuid__in=roles_for_group)
 
     def remove_service_accounts(
-        self, user: User, group: Group, service_accounts: [str], account_name: str = None, org_id: str = None
+        self, user: User, group: Group, service_accounts: Iterable[str], account_name: str = "", org_id: str = ""
     ) -> None:
         """Remove the given service accounts from the tenant."""
         # Log our intention.
@@ -1127,7 +1128,7 @@ def user_has_permission_act_on_service_account(self, user: User, service_account
             is_user_owner = user.username == owner
 
         # Check if the user has the "User Access administrator" permission. Leaving the RAW query here
-        username: str = user.username
+        username: str = user.username  # type: ignore
         query = (
             "SELECT EXISTS ( "
             "SELECT "

rbac/management/principal/it_service.py

@@ -17,6 +17,7 @@
 """Class to manage interactions with the IT service accounts service."""
 import logging
 import time
+from typing import Tuple
 
 import requests
 from django.conf import settings
@@ -154,7 +155,7 @@ def request_service_accounts(self, bearer_token: str) -> list[dict]:
 
         return service_accounts
 
-    def get_service_accounts(self, user: User, bearer_token: str, options: dict = {}) -> (list[dict], int):
+    def get_service_accounts(self, user: User, bearer_token: str, options: dict = {}) -> Tuple[list[dict], int]:
         """Request and returns the service accounts for the given tenant."""
         # We might want to bypass calls to the IT service on ephemeral or test environments.
         it_service_accounts: list[dict] = []
@@ -173,15 +174,14 @@ def get_service_accounts(self, user: User, bearer_token: str, options: dict = {}
         # - Admin only
         # - Email
         # - Status
-        usernames: [str] = []
+        usernames: list[str] = []
         specified_usernames = options.get("usernames")
         if specified_usernames:
             usernames = specified_usernames.split(",")
 
         # If "match_criteria" is specified, only the first username is taken into account.
         match_criteria = options.get("match_criteria")
         if match_criteria:
-            match_criteria: str = options["match_criteria"]
             username = usernames[0]
 
             if match_criteria == "partial":
@@ -222,7 +222,7 @@ def get_service_accounts(self, user: User, bearer_token: str, options: dict = {}
 
         # Filter the incoming service accounts. Also, transform them to the payload we will
         # be returning.
-        service_accounts: [dict] = self._merge_principals_it_service_accounts(
+        service_accounts: list[dict] = self._merge_principals_it_service_accounts(
             service_account_principals=sap_dict, it_service_accounts=it_service_accounts, options=options
         )
 
@@ -279,7 +279,7 @@ def get_service_accounts_group(self, group: Group, bearer_token: str, options: d
 
         # Filter the incoming service accounts. Also, transform them to the payload we will
         # be returning.
-        service_accounts: [dict] = self._merge_principals_it_service_accounts(
+        service_accounts: list[dict] = self._merge_principals_it_service_accounts(
             service_account_principals=sap_dict, it_service_accounts=it_service_accounts, options=options
         )
 
@@ -348,7 +348,7 @@ def _merge_principals_it_service_accounts(
         self, service_account_principals: dict[str, dict], it_service_accounts: list[dict], options: dict
     ) -> list[dict]:
         """Merge the database principals with the service account principals and return the response payload."""
-        service_accounts: [dict] = []
+        service_accounts: list[dict] = []
 
         # If the "username_only" parameter was set, we should only return that for the user.
         username_only = options.get("username_only")
@@ -358,10 +358,10 @@ def _merge_principals_it_service_accounts(
                 sa_principal = service_account_principals[it_service_account["clientID"]]
 
                 if username_only and username_only == "true":
-                    service_accounts.append({"username": sa_principal.username})
+                    service_accounts.append({"username": sa_principal.username})  # type: ignore
                 else:
                     # Get the principal's username from the database and set it in the response for the user.
-                    it_service_account["username"] = sa_principal.username
+                    it_service_account["username"] = sa_principal.username  # type: ignore
 
                     service_accounts.append(it_service_account)
             # If we cannot find a requested service account to IT in the database, we simply

requirements.txt

@@ -7,7 +7,7 @@ billiard==4.2.0; python_version >= '3.7'
 boto3==1.24.24; python_version >= '3.7'
 botocore==1.27.96; python_version >= '3.7'
 celery==5.3.0b2; python_version >= '3.7'
-certifi==2023.11.17; python_version >= '3.6'
+certifi==2024.2.2; python_version >= '3.6'
 cffi==1.16.0; platform_python_implementation != 'PyPy'
 charset-normalizer==3.3.2; python_full_version >= '3.7.0'
 click==8.1.7; python_version >= '3.7'

tox.ini

@@ -1,5 +1,5 @@
 [tox]
-envlist = lint, py39
+envlist = lint, py39, mypy
 skipsdist = True
 
 [flake8]
@@ -55,3 +55,9 @@ setenv =
 commands =
   flake8 rbac
   black --check -t py39 -l 119 rbac tests --diff
+
+[testenv:mypy]
+deps =
+  mypy
+commands =
+  mypy --install-types --non-interactive rbac