Fix SonarQube found issues

.github/actions/ci-setup/action.yml

@@ -18,10 +18,12 @@ runs:
         python-version: ${{ inputs.python-version }}
 
     - name: Install dependencies
+      env:
+        EXTRAS: ${{ inputs.extras }}
       run: |
         python3 -m pip install --upgrade pip
-        if [ -n "${{ inputs.extras }}" ]; then
-          pip install .[${{ inputs.extras }}]
+        if [ -n "$EXTRAS" ]; then
+          pip install ".[$EXTRAS]"
         else
           pip install .
         fi

.github/workflows/release.yml

@@ -19,7 +19,8 @@ jobs:
 
   release:
     needs: build
-    permissions: write-all
+    permissions:
+      contents: write
     runs-on: ubuntu-latest
 
     steps:

ankaios_sdk/_components/complete_state.py

@@ -66,7 +66,7 @@
 
 __all__ = ["CompleteState", "AgentAttributes"]
 
-from typing import Union
+from typing import Optional, Union
 from .._protos import _ank_base
 from .workload import Workload
 from .workload_state import WorkloadStateCollection
@@ -124,7 +124,6 @@ def __init__(
                     workload.name
                 ].CopyFrom(workload._to_proto())
             logger.debug("CompleteState initialized from workloads")
-            return
 
     def __str__(self) -> str:
         """
@@ -153,7 +152,7 @@ def get_api_version(self) -> str:
         """
         return str(self._complete_state.desiredState.apiVersion)
 
-    def get_workload(self, workload_name: str) -> Workload:
+    def get_workload(self, workload_name: str) -> Optional[Workload]:
         """
         Gets a workload from the complete state by its name.
 
@@ -256,7 +255,7 @@ def get_configs(self) -> dict:
 
         def _from_config_item(
             item: _ank_base.ConfigItem,
-        ) -> Union[str, list, dict]:
+        ) -> Optional[Union[str, list, dict]]:
             if item.HasField("String"):
                 return item.String
             if item.HasField("array"):

ankaios_sdk/_components/control_interface.py

@@ -390,7 +390,6 @@ def _handle_response(self, response: Response) -> None:
                     "CONTROL_INTERFACE_ACCEPTED. Ignoring..",
                     response.content_type,
                 )
-                return
 
         # Handle the connected state
         elif self._state == ControlInterfaceState.CONNECTED:
@@ -494,7 +493,7 @@ def write_request(self, request: Request) -> None:
             raise ConnectionClosedException(
                 "Could not write to pipe, connection closed."
             )
-        if not self._state == ControlInterfaceState.CONNECTED:
+        if self._state != ControlInterfaceState.CONNECTED:
             raise ControlInterfaceException(
                 "Could not write to pipe, not connected."
             )

ankaios_sdk/_components/manifest.py

@@ -118,10 +118,10 @@ def from_dict(manifest: dict) -> "Manifest":
         :rtype: Manifest
         """
         desired_state = _ank_base.State()
-        if "apiVersion" not in manifest.keys():
+        if "apiVersion" not in manifest:
             raise InvalidManifestException("apiVersion is missing.")
         desired_state.apiVersion = manifest["apiVersion"]
-        if "workloads" in manifest.keys():
+        if "workloads" in manifest:
             workloads = manifest["workloads"]
             for wl_name, wl_data in workloads.items():
                 try:
@@ -133,7 +133,7 @@ def from_dict(manifest: dict) -> "Manifest":
                     raise InvalidManifestException(
                         f"Error building workload {wl_name}: {e}"
                     ) from e
-        if "configs" in manifest.keys():
+        if "configs" in manifest:
             configs = manifest["configs"]
             for key, value in configs.items():
                 desired_state.configs.configs[key].CopyFrom(

ankaios_sdk/_components/response.py

@@ -80,7 +80,7 @@
 ]
 
 from dataclasses import dataclass
-from typing import Any, Union
+from typing import Any, Optional, Union
 from enum import Enum
 from .._protos import _ank_base, _control_api
 from ..exceptions import ResponseException
@@ -114,7 +114,7 @@ def __init__(self, message_buffer: bytes) -> None:
         """
         self.buffer = message_buffer
         self._response = None
-        self.content_type: ResponseType = None
+        self.content_type: Optional[ResponseType] = None
         self.content = None
 
         self._parse_response()
@@ -175,23 +175,7 @@ def _from_proto(self) -> None:
                     _proto=self._response.completeStateResponse.completeState
                 )
         elif self._response.HasField("UpdateStateSuccess"):
-            update_state_msg = self._response.UpdateStateSuccess
-            self.content_type = ResponseType.UPDATE_STATE_SUCCESS
-            self.content = UpdateStateSuccess()
-            for workload in update_state_msg.addedWorkloads:
-                workload_name, workload_id, agent_name = workload.split(".")
-                self.content.added_workloads.append(
-                    WorkloadInstanceName(
-                        agent_name, workload_name, workload_id
-                    )
-                )
-            for workload in update_state_msg.deletedWorkloads:
-                workload_name, workload_id, agent_name = workload.split(".")
-                self.content.deleted_workloads.append(
-                    WorkloadInstanceName(
-                        agent_name, workload_name, workload_id
-                    )
-                )
+            self._parse_update_state_success()
         elif self._response.HasField("logEntriesResponse"):
             self.content_type = ResponseType.LOGS_ENTRY
             self.content = []
@@ -222,12 +206,31 @@ def _from_proto(self) -> None:
         else:
             raise ResponseException("Invalid response type.")
 
-    def get_request_id(self) -> str:
+    def _parse_update_state_success(self) -> None:
+        """
+        Parses the UpdateStateSuccess response type and
+        populates content_type and content.
+        """
+        update_state_msg = self._response.UpdateStateSuccess
+        self.content_type = ResponseType.UPDATE_STATE_SUCCESS
+        self.content = UpdateStateSuccess()
+        for workload in update_state_msg.addedWorkloads:
+            workload_name, workload_id, agent_name = workload.split(".")
+            self.content.added_workloads.append(
+                WorkloadInstanceName(agent_name, workload_name, workload_id)
+            )
+        for workload in update_state_msg.deletedWorkloads:
+            workload_name, workload_id, agent_name = workload.split(".")
+            self.content.deleted_workloads.append(
+                WorkloadInstanceName(agent_name, workload_name, workload_id)
+            )
+
+    def get_request_id(self) -> Optional[str]:
         """
         Gets the request id of the response.
 
-        :returns: The request id of the response.
-        :rtype: str
+        :returns: The request id of the response, or None if not applicable.
+        :rtype: Optional[str]
         """
         if self.content_type in [
             ResponseType.CONTROL_INTERFACE_ACCEPTED,

ankaios_sdk/_components/workload.py

@@ -409,6 +409,21 @@ def _add_mask(self, mask: str) -> None:
         if self._main_mask not in self._masks and mask not in self._masks:
             self._masks.append(mask)
 
+    def _control_interface_access_to_dict(self) -> dict:
+        """
+        Convert the control interface access rules to a dictionary.
+
+        :returns: The dictionary representation of the control
+            interface access rules.
+        :rtype: dict
+        """
+        result = {"allowRules": [], "denyRules": []}
+        for rule in self._workload.controlInterfaceAccess.allowRules:
+            result["allowRules"].append(AccessRightRule(rule).to_dict())
+        for rule in self._workload.controlInterfaceAccess.denyRules:
+            result["denyRules"].append(AccessRightRule(rule).to_dict())
+        return result
+
     # pylint: disable=too-many-branches
     def to_dict(self) -> dict:
         """
@@ -440,20 +455,9 @@ def to_dict(self) -> dict:
         if self._workload.tags:
             for key, value in self._workload.tags.tags.items():
                 workload_dict["tags"].update({key: value})
-        workload_dict["controlInterfaceAccess"] = {}
-        if self._workload.controlInterfaceAccess:
-            workload_dict["controlInterfaceAccess"]["allowRules"] = []
-            for rule in self._workload.controlInterfaceAccess.allowRules:
-                access_rule = AccessRightRule(rule)
-                workload_dict["controlInterfaceAccess"]["allowRules"].append(
-                    access_rule.to_dict()
-                )
-            workload_dict["controlInterfaceAccess"]["denyRules"] = []
-            for rule in self._workload.controlInterfaceAccess.denyRules:
-                access_rule = AccessRightRule(rule)
-                workload_dict["controlInterfaceAccess"]["denyRules"].append(
-                    access_rule.to_dict()
-                )
+        workload_dict["controlInterfaceAccess"] = (
+            self._control_interface_access_to_dict()
+        )
         workload_dict["configs"] = {}
         for alias, name in self._workload.configs.configs.items():
             workload_dict["configs"][alias] = name
@@ -462,7 +466,30 @@ def to_dict(self) -> dict:
             workload_dict["files"].append(File._from_proto(file).to_dict())
         return workload_dict
 
-    # pylint: disable=too-many-branches
+    @staticmethod
+    def _apply_control_interface_access_from_dict(
+        workload, cia_dict: dict
+    ):
+        """
+        Apply control interface access rules from a dictionary
+        to a workload builder.
+
+        :param workload: The workload builder to apply rules to.
+        :param cia_dict: The control interface access dictionary.
+        :type cia_dict: dict
+
+        :returns: The updated workload builder.
+        """
+        for rule in cia_dict.get("allowRules", []):
+            workload = workload.add_allow_state_rule(
+                rule["operation"], rule["filterMask"]
+            )
+        for rule in cia_dict.get("denyRules", []):
+            workload = workload.add_deny_state_rule(
+                rule["operation"], rule["filterMask"]
+            )
+        return workload
+
     @staticmethod
     def _from_dict(workload_name: str, dict_workload: dict) -> "Workload":
         """
@@ -485,34 +512,19 @@ def _from_dict(workload_name: str, dict_workload: dict) -> "Workload":
             workload = workload.runtime_config(dict_workload["runtimeConfig"])
         if "restartPolicy" in dict_workload:
             workload = workload.restart_policy(dict_workload["restartPolicy"])
-        if "dependencies" in dict_workload:
-            for dep_key, dep_value in dict_workload["dependencies"].items():
-                workload = workload.add_dependency(dep_key, dep_value)
-        if "tags" in dict_workload:
-            for key, value in dict_workload["tags"].items():
-                workload = workload.add_tag(key, value)
-        if "controlInterfaceAccess" in dict_workload:
-            if "allowRules" in dict_workload["controlInterfaceAccess"]:
-                for rule in dict_workload["controlInterfaceAccess"][
-                    "allowRules"
-                ]:
-                    workload = workload.add_allow_state_rule(
-                        rule["operation"], rule["filterMask"]
-                    )
-            if "denyRules" in dict_workload["controlInterfaceAccess"]:
-                for rule in dict_workload["controlInterfaceAccess"][
-                    "denyRules"
-                ]:
-                    workload = workload.add_deny_state_rule(
-                        rule["operation"], rule["filterMask"]
-                    )
-        if "configs" in dict_workload:
-            for alias, name in dict_workload["configs"].items():
-                workload = workload.add_config(alias, name)
-        if "files" in dict_workload:
-            for file in dict_workload["files"]:
-                workload = workload.add_file(File._from_dict(file))
-
+        for dep_key, dep_value in dict_workload.get(
+            "dependencies", {}
+        ).items():
+            workload = workload.add_dependency(dep_key, dep_value)
+        for key, value in dict_workload.get("tags", {}).items():
+            workload = workload.add_tag(key, value)
+        workload = Workload._apply_control_interface_access_from_dict(
+            workload, dict_workload.get("controlInterfaceAccess", {})
+        )
+        for alias, name in dict_workload.get("configs", {}).items():
+            workload = workload.add_config(alias, name)
+        for file in dict_workload.get("files", []):
+            workload = workload.add_file(File._from_dict(file))
         return workload.build()
 
     def _to_proto(self) -> _ank_base.Workload:

ankaios_sdk/_components/workload_state.py

@@ -244,9 +244,9 @@ def __init__(self, state: _ank_base.ExecutionState) -> None:
         :param state: The execution state to interpret.
         :type state: _ank_base.ExecutionState
         """
-        self.state: WorkloadStateEnum = None
-        self.substate: WorkloadSubStateEnum = None
-        self.additional_info: str = None
+        self.state: Optional[WorkloadStateEnum] = None
+        self.substate: Optional[WorkloadSubStateEnum] = None
+        self.additional_info: Optional[str] = None
 
         self._interpret_state(state)
 

ankaios_sdk/_protos/__init__.py

@@ -25,10 +25,7 @@
     Used for exchanging messages with the control interface.
 """
 
-try:
-    import ankaios_sdk._protos.ank_base_pb2 as _ank_base
-    import ankaios_sdk._protos.control_api_pb2 as _control_api
-except ImportError as r:
-    raise r
+import ankaios_sdk._protos.ank_base_pb2 as _ank_base
+import ankaios_sdk._protos.control_api_pb2 as _control_api
 
 __all__ = ["_ank_base", "_control_api"]