Add mypy to CI file

.github/workflows/ci-test.yml

@@ -8,6 +8,30 @@ concurrency:
   group: ${{ github.ref_name }}
   cancel-in-progress: true
 jobs:
+  mypy:
+    name: Type checks (mypy)
+    runs-on: [self-hosted]
+    container: python:3.10-slim-buster
+    steps:
+      - name: Check out code
+        uses: actions/checkout@v3
+
+      - name: Install requirements
+        shell: bash
+        run: |
+          apt update
+          pip install --upgrade pip wheel
+          pip install mypy==0.991
+
+      - name: Link repo into the correct structure and run mypy
+        shell: bash
+        run: |
+          set -eux
+          mkdir -p $WORKDIR
+          cp -a ./  $WORKDIR
+          cd $WORKDIR
+          mypy -p plugins
+
   docs:
     runs-on: [ubuntu-latest]
     container: python:3.10-slim-buster
@@ -180,4 +204,4 @@ jobs:
       - run: apt install -y smbclient
       - run: mkdir -p $WORKDIR
       - run: cp -a ./  $WORKDIR
-      - run: cd $WORKDIR/tests/integration/cleanup && ./smb_cleanup.sh ${{ secrets.SMB_SERVER }} ${{ secrets.SMB_SHARE }} "${{ secrets.SMB_USERNAME }}" ${{ secrets.SMB_PASSWORD }}
+      - run: cd $WORKDIR/tests/integration/cleanup && ./smb_cleanup.sh ${{ secrets.SMB_SERVER }} ${{ secrets.SMB_SHARE }} "${{ secrets.SMB_USERNAME }}" ${{ secrets.SMB_PASSWORD }}

Makefile

@@ -73,3 +73,8 @@ integration:  ## Run integration tests
 docs:  ## Build collection documentation
 	pip install -r docs.requirements
 	$(MAKE) -C docs -f Makefile.custom docs
+
+.PHONY: mypy
+mypy: ## Run mypy hint checker
+	pip install -r mypy.requirements
+	mypy -p plugins

mypy.requirements

@@ -0,0 +1 @@
+mypy==0.991

plugins/module_utils/arguments.py

@@ -4,10 +4,12 @@
 # GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
 
 from __future__ import absolute_import, division, print_function
+from __future__ import annotations
 
 __metaclass__ = type
 
 from ansible.module_utils.basic import env_fallback
+from typing import Any
 
 # TODO - env from /etc/environment is loaded
 # But when env is set in bash session, env seems to be lost on ssh connection to localhost.
@@ -46,5 +48,5 @@
 )
 
 
-def get_spec(*param_names):
+def get_spec(*param_names: str) -> dict[Any, Any]:
     return dict((p, SHARED_SPECS[p]) for p in param_names)

plugins/module_utils/errors.py

@@ -7,6 +7,8 @@
 
 __metaclass__ = type
 
+from typing import Any
+
 
 class ScaleComputingError(Exception):
     pass
@@ -46,7 +48,7 @@ def __init__(self, data):
 
 # In-case argument spec doesn't catch exception
 class MissingValueHypercore(ScaleComputingError):
-    def __init__(self, data):
+    def __init__(self, data: Any):
         self.message = "Missing values from hypercore API - {0}".format(data)
         super(MissingValueHypercore, self).__init__(self.message)
 

plugins/module_utils/registration.py

@@ -12,30 +12,26 @@
 from ..module_utils.utils import PayloadMapper
 from ..module_utils import errors
 from ..module_utils.rest_client import RestClient
-from ..module_utils.task_tag import TypedTaskTag
-from typing import TypedDict, Union
-
-
-# Use for type hinting.
-class TypedRegistrationToAnsible(TypedDict):
-    company_name: str
-    contact: str
-    phone: str
-    email: str
+from ..module_utils.typed_classes import (
+    TypedTaskTag,
+    TypedRegistrationFromAnsible,
+    TypedRegistrationToAnsible,
+)
+from typing import Union, Any
 
 
 class Registration(PayloadMapper):
     def __init__(
         self,
-        uuid: str = "",
-        company_name: str = "",
-        contact: str = "",
-        phone: str = "",
-        email: str = "",
-        cluster_id: str = "",
-        cluster_data: str = "",
-        cluster_data_hash: str = "",
-        cluster_data_hash_accepted: str = "",
+        uuid: Union[str, None] = None,
+        company_name: Union[str, None] = None,
+        contact: Union[str, None] = None,
+        phone: Union[str, None] = None,
+        email: Union[str, None] = None,
+        cluster_id: Union[str, None] = None,
+        cluster_data: Union[str, None] = None,
+        cluster_data_hash: Union[str, None] = None,
+        cluster_data_hash_accepted: Union[str, None] = None,
     ):
         self.uuid = uuid
         self.company_name = company_name
@@ -56,7 +52,7 @@ def get(cls, rest_client: RestClient) -> Union[Registration, None]:
         return None
 
     @classmethod
-    def from_hypercore(cls, hypercore_data: dict) -> Registration:
+    def from_hypercore(cls, hypercore_data: dict[Any, Any]) -> Registration:
         try:
             obj = cls()
             obj.uuid = hypercore_data["uuid"]
@@ -73,15 +69,15 @@ def from_hypercore(cls, hypercore_data: dict) -> Registration:
             raise errors.MissingValueHypercore(e)
 
     @classmethod
-    def from_ansible(cls, ansible_data: dict) -> Registration:
+    def from_ansible(cls, ansible_data: TypedRegistrationFromAnsible) -> Registration:
         obj = cls()
         obj.company_name = ansible_data.get("company_name", None)
         obj.contact = ansible_data.get("contact", None)
         obj.phone = ansible_data.get("phone", None)
         obj.email = ansible_data.get("email", None)
         return obj
 
-    def to_hypercore(self) -> dict:
+    def to_hypercore(self) -> dict[Any, Any]:
         hypercore_dict = dict()
         if self.company_name:
             hypercore_dict["companyName"] = self.company_name

plugins/module_utils/rest_client.py

@@ -4,12 +4,17 @@
 # GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
 
 from __future__ import absolute_import, division, print_function
+from __future__ import annotations
 
 from . import errors
 from . import utils
+from ..module_utils.client import Client
+from ..module_utils.typed_classes import TypedTaskTag
 
 __metaclass__ = type
 
+from typing import Any, Union
+
 
 def _query(original=None):
     # Make sure the query isn't equal to None
@@ -18,10 +23,12 @@ def _query(original=None):
 
 
 class RestClient:
-    def __init__(self, client):
+    def __init__(self, client: Client):
         self.client = client
 
-    def list_records(self, endpoint, query=None, timeout=None):
+    def list_records(
+        self, endpoint: str, query: dict[Any, Any] = None, timeout: float = None
+    ) -> list[Any]:
         """Results are obtained so that first off, all records are obtained and
         then filtered manually"""
         try:
@@ -30,7 +37,13 @@ def list_records(self, endpoint, query=None, timeout=None):
             raise errors.ScaleComputingError(f"Request timed out: {e}")
         return utils.filter_results(records, query)
 
-    def get_record(self, endpoint, query=None, must_exist=False, timeout=None):
+    def get_record(
+        self,
+        endpoint: str,
+        query: dict[Any, Any] = None,
+        must_exist: bool = False,
+        timeout: float = None,
+    ) -> Union[dict[Any, Any], None]:
         records = self.list_records(endpoint=endpoint, query=query, timeout=timeout)
         if len(records) > 1:
             raise errors.ScaleComputingError(
@@ -46,35 +59,39 @@ def get_record(self, endpoint, query=None, must_exist=False, timeout=None):
             )
         return records[0] if records else None
 
-    def create_record(self, endpoint, payload, check_mode, timeout=None):
+    def create_record(
+        self, endpoint, payload, check_mode, timeout=None
+    ) -> TypedTaskTag:
         if check_mode:
             return utils.MOCKED_TASK_TAG
         try:
-            response = self.client.post(
+            response: TypedTaskTag = self.client.post(
                 endpoint, payload, query=_query(), timeout=timeout
             ).json
         except TimeoutError as e:
             raise errors.ScaleComputingError(f"Request timed out: {e}")
         return response
 
-    def update_record(self, endpoint, payload, check_mode, record=None, timeout=None):
+    def update_record(
+        self, endpoint, payload, check_mode, record=None, timeout=None
+    ) -> TypedTaskTag:
         # No action is possible when updating a record
         if check_mode:
             return utils.MOCKED_TASK_TAG
         try:
-            response = self.client.patch(
+            response: TypedTaskTag = self.client.patch(
                 endpoint, payload, query=_query(), timeout=timeout
             ).json
         except TimeoutError as e:
             raise errors.ScaleComputingError(f"Request timed out: {e}")
         return response
 
-    def delete_record(self, endpoint, check_mode, timeout=None):
+    def delete_record(self, endpoint, check_mode, timeout=None) -> TypedTaskTag:
         # No action is possible when deleting a record
         if check_mode:
             return utils.MOCKED_TASK_TAG
         try:
-            response = self.client.delete(endpoint, timeout=timeout).json
+            response: TypedTaskTag = self.client.delete(endpoint, timeout=timeout).json
         except TimeoutError as e:
             raise errors.ScaleComputingError(f"Request timed out: {e}")
         return response

plugins/module_utils/role.py

@@ -5,24 +5,30 @@
 
 
 from __future__ import absolute_import, division, print_function
+from __future__ import annotations
 
 __metaclass__ = type
 
 from ..module_utils.utils import PayloadMapper
 from ..module_utils.rest_client import RestClient
+from ..module_utils.typed_classes import TypedDNSConfigToAnsible
+
+from typing import Union, Any
 
 
 class Role(PayloadMapper):
-    def __init__(self, uuid, name):
+    def __init__(self, uuid: str, name: str):
         self.uuid = uuid
         self.name = name
 
     @classmethod
-    def from_ansible(cls):
+    def from_ansible(cls, ansible_data: dict[Any, Any]) -> None:
         pass
 
     @classmethod
-    def from_hypercore(cls, hypercore_data):
+    def from_hypercore(
+        cls, hypercore_data: Union[dict[Any, Any], None]
+    ) -> Union[Role, None]:
         if not hypercore_data:
             # In case for get_record, return None if no result is found
             return None
@@ -31,20 +37,22 @@ def from_hypercore(cls, hypercore_data):
             name=hypercore_data["name"],
         )
 
-    def to_hypercore(self):
+    def to_hypercore(self) -> None:
         pass
 
-    def to_ansible(self):
+    def to_ansible(self) -> TypedDNSConfigToAnsible:
         return dict(
             uuid=self.uuid,
             name=self.name,
         )
 
-    def __eq__(self, other):
+    def __eq__(self, other: object) -> bool:
         """
         One User is equal to another if it has ALL attributes exactly the same.
         This method is used only in tests.
         """
+        if not isinstance(other, Role):
+            return NotImplemented
         return all(
             (
                 self.uuid == other.uuid,
@@ -53,15 +61,19 @@ def __eq__(self, other):
         )
 
     @classmethod
-    def get_role_from_uuid(cls, role_uuid, rest_client: RestClient, must_exist=False):
+    def get_role_from_uuid(
+        cls, role_uuid: str, rest_client: RestClient, must_exist: bool = False
+    ) -> Union[Role, None]:
         hypercore_dict = rest_client.get_record(
             "/rest/v1/Role/{0}".format(role_uuid), must_exist=must_exist
         )
         role = cls.from_hypercore(hypercore_dict)
         return role
 
     @classmethod
-    def get_role_from_name(cls, role_name, rest_client: RestClient, must_exist=False):
+    def get_role_from_name(
+        cls, role_name: str, rest_client: RestClient, must_exist: bool = False
+    ) -> Union[Role, None]:
         hypercore_dict = rest_client.get_record(
             "/rest/v1/Role", {"name": role_name}, must_exist=must_exist
         )

plugins/module_utils/task_tag.py

@@ -11,18 +11,15 @@
 from time import sleep
 
 from ..module_utils import errors
-from typing import TypedDict
-
-
-# Use for type hinting.
-class TypedTaskTag(TypedDict):
-    createdUUID: str
-    taskTag: str
+from ..module_utils.rest_client import RestClient
+from ..module_utils.typed_classes import TypedTaskTag
 
 
 class TaskTag:
     @classmethod
-    def wait_task(cls, rest_client, task, check_mode=False):
+    def wait_task(
+        cls, rest_client: RestClient, task: TypedTaskTag, check_mode: bool = False
+    ):
         if check_mode:
             return
         if type(task) != dict: