Fixes some pylint issues

framework/python/src/api/api.py

@@ -248,7 +248,7 @@ async def start_test_run(self, request: Request, response: Response):
       return self._generate_msg(False,"Configured interfaces are not " +
                                 "ready for use. Ensure required interfaces " +
                                 "are connected.")
-    
+
     device.test_modules = body_json["device"]["test_modules"]
 
     LOGGER.info("Starting Testrun with device target " +
@@ -389,6 +389,8 @@ async def delete_report(self, request: Request, response: Response):
     try:
       body_json = json.loads(body_raw)
     except JSONDecodeError as e:
+      LOGGER.error("An error occurred whilst decoding JSON")
+      LOGGER.debug(e)
       response.status_code = status.HTTP_400_BAD_REQUEST
       return self._generate_msg(False,
                                 "Invalid request received")
@@ -647,6 +649,8 @@ async def update_profile(self, request: Request, response: Response):
       req_raw = (await request.body()).decode("UTF-8")
       req_json = json.loads(req_raw)
     except JSONDecodeError as e:
+      LOGGER.error("An error occurred whilst decoding JSON")
+      LOGGER.debug(e)
       response.status_code = status.HTTP_400_BAD_REQUEST
       return self._generate_msg(False,
                                 "Invalid request received")
@@ -656,9 +660,9 @@ async def update_profile(self, request: Request, response: Response):
     if not valid_profile:
       response.status_code = status.HTTP_400_BAD_REQUEST
       return self._generate_msg(False, "Invalid profile request received")
-    
+
     profile_name = req_json.get("name")
-    
+
     # Check if profile exists
     profile = self.get_session().get_profile(profile_name)
 
@@ -674,15 +678,17 @@ async def update_profile(self, request: Request, response: Response):
     else:
       # Update existing profile
       profile = self.get_session().update_profile(req_json)
-      
+
       if profile is not None:
         response.status_code = status.HTTP_200_OK
         return self._generate_msg(True, "Successfully updated that profile")
       LOGGER.error("An error occurred whilst updating a profile")
 
     response.status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
-    return self._generate_msg(False, "An error occurred whilst creating or updating a profile")
-
+    return self._generate_msg(
+      False,
+      "An error occurred whilst creating or updating a profile")
+
   async def delete_profile(self, request: Request, response: Response):
 
     LOGGER.debug("Received profile delete request")
@@ -691,16 +697,18 @@ async def delete_profile(self, request: Request, response: Response):
       req_raw = (await request.body()).decode("UTF-8")
       req_json = json.loads(req_raw)
     except JSONDecodeError as e:
+      LOGGER.error("An error occurred whilst decoding JSON")
+      LOGGER.debug(e)
       response.status_code = status.HTTP_400_BAD_REQUEST
       return self._generate_msg(False,
                                 "Invalid request received")
 
     # Check name included in request
-    if 'name' not in req_json:
+    if "name" not in req_json:
       response.status_code = status.HTTP_400_BAD_REQUEST
       return self._generate_msg(False,
                                 "Invalid request received")
-    
+
     # Get profile name
     profile_name = req_json.get("name")
 
@@ -712,15 +720,16 @@ async def delete_profile(self, request: Request, response: Response):
       response.status_code = status.HTTP_404_NOT_FOUND
       return self._generate_msg(False,
                                 "A profile with that name could not be found")
-    
+
     # Attempt to delete the profile
     success = self.get_session().delete_profile(profile)
 
     if not success:
       response.status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
-      return self._generate_msg(False,
-                                "An error occurred whilst deleting that profile")
-
+      return self._generate_msg(
+        False,
+        "An error occurred whilst deleting that profile")
+
     return self._generate_msg(True,
                               "Successfully deleted that profile")
 

framework/python/src/common/risk_profile.py

@@ -16,6 +16,7 @@
 from datetime import datetime
 
 class RiskProfile():
+  """Python representation of a risk profile"""
 
   def __init__(self, json_data):
     self.name = json_data['name']
@@ -38,4 +39,4 @@ def to_json(self):
       'status': self.status,
       'questions': self.questions
     }
-    return json
+    return json

framework/python/src/common/session.py

@@ -375,7 +375,6 @@ def _load_profiles(self):
           risk_profile = RiskProfile(json_data)
           risk_profile.status = self.check_profile_status(risk_profile)
           self._profiles.append(risk_profile)
-
 
     except Exception as e:
       LOGGER.error('An error occurred whilst loading risk profiles')
@@ -386,13 +385,13 @@ def get_profiles_format(self):
 
   def get_profiles(self):
     return self._profiles
-  
+
   def get_profile(self, name):
     for profile in self._profiles:
       if profile.name.lower() == name.lower():
         return profile
     return None
-   
+
   def validate_profile(self, profile_json):
 
     # Check name field is present
@@ -405,8 +404,10 @@ def validate_profile(self, profile_json):
 
     # Check all questions are present
     for format_q in self.get_profiles_format():
-      if self._get_profile_question(profile_json, format_q.get('question')) is None:
-        LOGGER.error('Missing question: ' + format_q.get('question'))
+      if self._get_profile_question(
+        profile_json, format_q.get('question')) is None:
+        LOGGER.error(
+          'Missing question: ' + format_q.get('question'))
         return False
 
     return True
@@ -444,11 +445,12 @@ def update_profile(self, profile_json):
 
           # Check answer is present
           if 'answer' not in profile_question:
-            LOGGER.error("Missing answer for question: " + question.get('question'))
+            LOGGER.error(
+              'Missing answer for question: ' + question.get('question'))
             all_questions_answered = False
 
         else:
-          LOGGER.error("Missing question: " + question.get('question'))
+          LOGGER.error('Missing question: ' + question.get('question'))
           all_questions_answered = False
 
       if not all_questions_answered:
@@ -481,9 +483,11 @@ def update_profile(self, profile_json):
       risk_profile.questions = profile_json.get('questions')
 
     # Write file to disk
-    with open(os.path.join(PROFILES_DIR, risk_profile.name + '.json'), 'w') as f:
+    with open(os.path.join(
+      PROFILES_DIR, risk_profile.name + '.json'), 'w',
+      encoding='utf-8') as f:
       f.write(json.dumps(risk_profile.to_json()))
-    
+
     return risk_profile
 
   def check_profile_status(self, profile):
@@ -499,7 +503,7 @@ def check_profile_status(self, profile):
         profile.status = 'Expired'
 
     return profile.status
-  
+
   def delete_profile(self, profile):
 
     try:

framework/python/src/net_orc/network_orchestrator.py

@@ -538,7 +538,7 @@ def _start_network_service(self, net_module):
           # DNS configuration (/etc/resolv.conf)  Re-add when/if
           # this network is utilized and DNS issue is resolved
           #network=PRIVATE_DOCKER_NET,
-          network_mode="none",
+          network_mode='none',
           privileged=True,
           detach=True,
           mounts=net_module.mounts,

modules/devices/faux-dev/python/src/util.py

@@ -26,21 +26,23 @@ def run_command(cmd, logger, output=True):
   by any return code from the process other than zero."""
 
   success = False
-  process = subprocess.Popen(shlex.split(cmd),
-                             stdout=subprocess.PIPE,
-                             stderr=subprocess.PIPE)
-  stdout, stderr = process.communicate()
+  with subprocess.Popen(
+    shlex.split(cmd),
+    stdout=subprocess.PIPE,
+    stderr=subprocess.PIPE) as process:
 
-  if process.returncode != 0:
-    err_msg = f'{stderr.strip()}. Code: {process.returncode}'
-    logger.error('Command Failed: ' + cmd)
-    logger.error('Error: ' + err_msg)
-  else:
-    success = True
-    logger.debug('Command succeeded: ' + cmd)
-  if output:
-    out = stdout.strip().decode('utf-8')
-    logger.debug('Command output: ' + out)
-    return success, out
-  else:
-    return success, None
+    stdout, stderr = process.communicate()
+
+    if process.returncode != 0:
+      err_msg = f'{stderr.strip()}. Code: {process.returncode}'
+      logger.error('Command Failed: ' + cmd)
+      logger.error('Error: ' + err_msg)
+    else:
+      success = True
+      logger.debug('Command succeeded: ' + cmd)
+    if output:
+      out = stdout.strip().decode('utf-8')
+      logger.debug('Command output: ' + out)
+      return success, out
+    else:
+      return success, None

modules/test/tls/python/src/run.py

@@ -66,4 +66,4 @@ def run():
 
 
 if __name__ == '__main__':
-  run()
+  run()

modules/test/tls/python/src/tls_util.py

@@ -771,4 +771,4 @@ def is_ecdh_and_ecdsa(self, ciphers):
     for cipher in ciphers:
       ecdh |= 'ECDH' in cipher
       ecdsa |= 'ECDSA' in cipher
-    return {'ecdh': ecdh, 'ecdsa': ecdsa}
+    return {'ecdh': ecdh, 'ecdsa': ecdsa}