Linting based on pylint report (#639)

src/csm_testing/lib/data_json_parser.py

@@ -26,11 +26,11 @@
 dataJson is a convenience class to work with the data.json file
 
 data.json exists as key:value file where key can be either a mac address that is inconsistent depending on the environment
-Or ... everything else - which IS consistent Storage, Default, Global, etc. 
+Or ... everything else - which IS consistent Storage, Default, Global, etc.
 Convenience functions are included to easily dig out the most likely needed data
 
 Parameters:
-dataJson(/path/to/data.json) 
+dataJson(/path/to/data.json)
 
 Exposes:
     payload: A dictionary of the entire data.json file

src/csm_testing/lib/endpoints.py

@@ -51,8 +51,8 @@
 # Pattern for 65000-65499   = 65[0-4][0-9]{2}
 # Pattern for 65500-65529   = 655[0-2][0-9]
 # Pattern for 65530-65535   = 6553[0-5]
-port_patterns = [ 
-    "[1-9][0-9]{3}", 
+port_patterns = [
+    "[1-9][0-9]{3}",
     "[1-5][0-9]{4}",
     "6[0-4][0-9]{3}",
     "65[0-4][0-9]{2}",
@@ -104,7 +104,7 @@ def load_goss_endpoints() -> Dict[str, List[EndpointTuple]]:
     if goss_endpoints_by_ncn_type != None:
         return goss_endpoints_by_ncn_type
     config_file = goss_servers_config(validate=True)
-    
+
     endpoints_by_type = { ntype: list() for ntype in NCN_TYPES }
     with open(config_file, "rt") as f:
         for line in f.readlines():

src/csm_testing/lib/grok_exporter_logger.py

@@ -26,7 +26,7 @@
 Helper functions for Goss Python automated scripts
 
 These functions relate to generating log files to be consumed by
-grok-exporter. These log files 
+grok-exporter. These log files
 """
 
 from collections import OrderedDict

src/csm_testing/tests/cfs_state_reporter_check/__main__.py

@@ -62,7 +62,7 @@ def get_systemctl_data():
     logger.info(f"Running command: {' '.join(command_line)}")
     result = subprocess.run(command_line, stdout=subprocess.PIPE, check=True)
     logger.info(f"Command completed with return code {result.returncode}")
-    
+
     # Each line of output should be of the form FieldName=<value>
     # We will load this into a dictionary, and return that
     # This could be done in fewer lines of code, but clarity is preferable
@@ -171,15 +171,15 @@ def get_field_value(field_name):
 
 def check_cfs_state_reporter_status():
     srv_exit_status, srv_load_state, srv_active_state, srv_substate, srv_result = get_service_data_fields()
-    
+
     errors=False
     if srv_exit_status != "0":
         logger.error(f"ExecMainStatus={srv_exit_status} | expected value=0")
         errors=True
     if srv_load_state != "loaded":
         logger.error(f"LoadState={srv_load_state} | expected value=loaded")
         errors=True
-        
+
     if srv_active_state != "inactive":
         logger.error(f"ActiveState={srv_active_state} | expected value=inactive")
         errors=True
@@ -195,7 +195,7 @@ def check_cfs_state_reporter_status():
     if errors:
         raise CfsTestException
 
-def main() -> int:
+def main() -> int: # pylint: disable=missing-function-docstring
     """
     Returns 0 on success, non-0 on failure.
     """

src/csm_testing/tests/check_for_unused_drives/__main__.py

@@ -30,7 +30,7 @@
 * Exactly 12 OSDs per storage node, if Gigabyte hardware
 * At least 1 OSD per storage node, if Intel hardware
 * Or that (#_osds / #_storage_nodes) has no remainder
- 
+
 The test fails if the above validation fails.
 If the system has a different hardware type or the test is unable
 to determine it, the test fails.
@@ -144,7 +144,7 @@ def get_num_osds():
         sys.exit(5)
     return num_osds
 
-def main():
+def main(): # pylint: disable=missing-function-docstring
     min_expected_osds, max_expected_osds, n_storage_nodes = parse_args()
     num_osds = get_num_osds()
     if num_osds < min_expected_osds:

src/csm_testing/tests/check_ncn_uan_ip_dns/__main__.py

@@ -33,7 +33,7 @@
 from kubernetes import client, config
 
 
-class APIRequest(object):
+class APIRequest:
     """
 
     Example use:
@@ -88,20 +88,16 @@ def __call__(self, method, route, **kwargs):
         response = http.request(method=method, url=url, headers=headers, **kwargs)
 
         if 'data' in kwargs:
-            log.debug(f"{method} {url} with headers:"
-                      f"{json.dumps(headers, indent=4)}"
-                      f"and data:"
-                      f"{json.dumps(kwargs['data'], indent=4)}")
+            log.debug("%s %s with headers: %s and data: %s", method, url,
+                      json.dumps(headers, indent=4), json.dumps(kwargs['data'], indent=4))
         elif 'json' in kwargs:
-            log.debug(f"{method} {url} with headers:"
-                      f"{json.dumps(headers, indent=4)}"
-                      f"and JSON:"
-                      f"{json.dumps(kwargs['json'], indent=4)}")
+            log.debug("%s %s with headers: %s and JSON: %s", method, url,
+                      json.dumps(headers, indent=4), json.dumps(kwargs['json'], indent=4))
         else:
-            log.debug(f"{method} {url} with headers:"
-                      f"{json.dumps(headers, indent=4)}")
-        log.debug(f"Response to {method} {url} => {response.status_code} {response.reason}"
-                  f"{response.text}")
+            log.debug("%s %s with headers: %s", method, url, json.dumps(headers, indent=4))
+
+        log.debug("Response to %s %s => %d %s %s", method, url, response.status_code,
+                  response.reason, response.text)
 
         return response
 
@@ -131,7 +127,8 @@ def token():
     token = base64.b64decode(secret['client-secret']).decode('utf-8')
 
     # create post data to keycloak istio ingress
-    token_data = {'grant_type': 'client_credentials', 'client_id': 'admin-client', 'client_secret': token}
+    token_data = {'grant_type': 'client_credentials', 'client_id': 'admin-client',
+                  'client_secret': token}
 
     # query keycloack
     token_url = '/keycloak/realms/shasta/protocol/openid-connect/token'
@@ -141,8 +138,7 @@ def token():
 
     return access_token
 
-def main():
-
+def main(): # pylint: disable=missing-function-docstring
     error_found = False
 
     bearer_token = token()
@@ -169,28 +165,29 @@ def main():
                 # print (ip)
                 if ip != '':
                     if ip in ip_set:
-                        log.error(f'Error: found duplicate IP: {ip}')
+                        log.error('Error: found duplicate IP: %s', ip)
                         error_found = True
                         nslookup_cmd = subprocess.Popen(('nslookup', ip), stdout=subprocess.PIPE,
                                                         stderr=subprocess.PIPE)
-                        output, errors = nslookup_cmd.communicate()
-                        print("output.decode('ascii')")
+                        output, _ = nslookup_cmd.communicate()
+                        print(output.decode('ascii'))
                     else:
                         ip_set.add(ip)
 
     hostname_list = []
 
-    for i in range(len(sls_hardware)):
-        if 'ExtraProperties' in sls_hardware[i]:
-            if 'Role' in sls_hardware[i]['ExtraProperties'] and (
-                    sls_hardware[i]['ExtraProperties']['Role'] == 'Application' or sls_hardware[i]['ExtraProperties'][
-                'Role'] == 'Management'):
-                hostname_list.append(sls_hardware[i]['ExtraProperties']['Aliases'][0] + '.nmn')
-                hostname_list.append(sls_hardware[i]['ExtraProperties']['Aliases'][0] + '.can')
-                hostname_list.append(sls_hardware[i]['ExtraProperties']['Aliases'][0] + '.hmn')
-                hostname_list.append(sls_hardware[i]['ExtraProperties']['Aliases'][0] + '-mgmt')
-                hostname_list.append(sls_hardware[i]['ExtraProperties']['Aliases'][0] + '.cmn')
-                hostname_list.append(sls_hardware[i]['ExtraProperties']['Aliases'][0] + '.chn')
+    for hardware in sls_hardware:
+        if 'ExtraProperties' not in hardware:
+            continue
+        if 'Role' not in hardware['ExtraProperties']:
+            continue
+        if hardware['ExtraProperties']['Role'] in {'Application', 'Management'}:
+            hostname_list.append(hardware['ExtraProperties']['Aliases'][0] + '.nmn')
+            hostname_list.append(hardware['ExtraProperties']['Aliases'][0] + '.can')
+            hostname_list.append(hardware['ExtraProperties']['Aliases'][0] + '.hmn')
+            hostname_list.append(hardware['ExtraProperties']['Aliases'][0] + '-mgmt')
+            hostname_list.append(hardware['ExtraProperties']['Aliases'][0] + '.cmn')
+            hostname_list.append(hardware['ExtraProperties']['Aliases'][0] + '.chn')
 
     for hostname in hostname_list:
 
@@ -199,8 +196,9 @@ def main():
         result = int(wc_cmd.decode('ascii').strip())
         if result > 1:
             error_found = True
-            log.error(f'ERROR: {hostname} has more than 1 DNS entry')
-            nslookup_cmd = subprocess.Popen(('nslookup', hostname), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+            log.error('ERROR: %s has more than 1 DNS entry', hostname)
+            nslookup_cmd = subprocess.Popen(('nslookup', hostname), stdout=subprocess.PIPE,
+                                             stderr=subprocess.PIPE)
             output, errors = nslookup_cmd.communicate()
             print(f"{output.decode('ascii')}")
 
@@ -212,4 +210,4 @@ def main():
         sys.exit(0)
 
 if __name__ == "__main__":
-    main()
+    main()

src/csm_testing/tests/check_remote_mac_against_configs/__main__.py

@@ -44,14 +44,14 @@ def remoteCmd(host, command):
 def get_arg_no_brackets(arg):
     return arg.strip('[').strip(']')
 
-def main() -> int:
+def main() -> int: # pylint: disable=missing-function-docstring
     # quick check to ensure we received the locations of data.json and statics.conf
     if len(sys.argv) != 3:
         print("Wrong number of arguments provided")
         return 2
-    
+
     # This version of goss sends [.Arg.*] as string with [
-    # Apparently fixed in 0.3.14 
+    # Apparently fixed in 0.3.14
     data = djp.dataJson(get_arg_no_brackets(sys.argv[1]))
     staticsFile = open(get_arg_no_brackets(sys.argv[2]),'r')
     statics = staticsFile.read()
@@ -60,24 +60,24 @@ def main() -> int:
     for server in data.ncnList:
         # get the MAC from the NCN
         mac = remoteCmd(server, getMACcommand).decode().strip()
-    
+
         # ensure that the MAC address is somewhere in data.json
         if mac in data.ncnKeys:
             # ensure that the hostname's MAC in data.json matches reality
             if data.ncnList[server] == mac:
                 passed += 1
         else:
             failed += 1
-        
+
         # ensure that the mac exists in statics.conf
         if mac in statics:
             # check statics.conf
             # should find something like: dhcp-host=b8:59:9f:2b:2e:d2,10.252.0.7,ncn-s001,infinite
             # the ip is between the first commas
             try:
-                search = statics[statics.find('dhcp-host=' + data.ncnList[server]):statics.find('\n',statics.find('dhcp-host=' + data.ncnList[server]))]    
+                search = statics[statics.find('dhcp-host=' + data.ncnList[server]):statics.find('\n',statics.find('dhcp-host=' + data.ncnList[server]))]
                 ip, hname = search[search.find(','):search.rfind(',')].split(',')[-2:]
-                if mac == data.ncnList[hname]: 
+                if mac == data.ncnList[hname]:
                     passed += 1
             except:
                 print("Error in statics.conf")

src/csm_testing/tests/check_remote_resolv_conf_against_data_json/__main__.py

@@ -23,7 +23,8 @@
 #
 # Script to check remote resolv.conf namserver value == dns-server value in data.json
 # Invocation: check-remote-resolv_conf-against-data_json.py /path/to/data.json
-# Check the count of passed tests against the number of NCNs in data.conf - and send either PASS or FAIL
+# Check the count of passed tests against the number of NCNs in data.conf - and send
+# either PASS or FAIL
 
 import subprocess, sys
 import csm_testing.lib.data_json_parser as djp
@@ -33,25 +34,26 @@
 failed = 0
 
 def remoteCmd(host, command):
-    cmd = subprocess.Popen(['ssh', '-o StrictHostKeyChecking=no', host , command], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
-
+    cmd = subprocess.Popen(['ssh', '-o StrictHostKeyChecking=no', host , command],
+                           stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+
     stdout,stderr = cmd.communicate()
     return stdout
 
-def main() -> int:
+def main() -> int: # pylint: disable=missing-function-docstring
     # quick check to ensure we received the locations of data.json and statics.conf
     if len(sys.argv) != 2:
         print("Wrong number of arguments provided")
         return 2
-    
+
     data = djp.dataJson(sys.argv[1])
     dns_server = data.getGlobalMD()['dns-server']
 
     # ensure remote MAC matches data.json (casminst-384) and statics.conf (casminst-380)
     for server in data.ncnList:
         # get the MAC from the NCN
         results = remoteCmd(server, remoteCommand).decode().strip().split()
-    
+
         if len(results) < 1:
             # result set empty
             print("No nameserver entry in /etc/resolv.conf for server " , server)

src/csm_testing/tests/compare_k8s_ncns/__main__.py

@@ -31,7 +31,8 @@
 import kubernetes
 import sys
 
-KubernetesNodeInfoFields = [ "container_runtime_version", "kube_proxy_version", "kubelet_version", "os_image" ]
+KubernetesNodeInfoFields = [ "container_runtime_version", "kube_proxy_version",
+                             "kubelet_version", "os_image" ]
 
 def print_err(msg: str) -> None:
     """
@@ -40,7 +41,7 @@ def print_err(msg: str) -> None:
     sys.stderr.write(f"ERROR: {msg}\n")
 
 
-def main() -> None:
+def main() -> None: # pylint: disable=missing-function-docstring
     print("Loading Kubernetes configuration")
     kubernetes.config.load_kube_config()
     print("Initializing Kubernetes client")

src/csm_testing/tests/console_verify_services/__main__.py

@@ -88,9 +88,9 @@ def check_services_running():
                     #  gather what information we can.
                     if c.ready != True:
                         if c.state.terminated != None:
-                            logger.error(f"Pod: {i.metadata.name} Container Terminated: {c.name}, " + 
+                            logger.error(f"Pod: {i.metadata.name} Container Terminated: {c.name}, " +
                                             f"Exit Code: {c.state.terminated.exit_code}, " +
-                                            f"Reason: {c.state.terminated.reason}, " + 
+                                            f"Reason: {c.state.terminated.reason}, " +
                                             f"Message: {c.state.terminated.message}")
                             ok = False
                         if c.state.waiting != None:
@@ -108,7 +108,7 @@ def check_services_running():
         logger.error(f"Found pods: {foundPods}")
         raise ConsoleException
 
-def main() -> int:
+def main() -> int: # pylint: disable=missing-function-docstring
     try:
         return_value = True
         logger.info("Beginning verification that all console services are running")

src/csm_testing/tests/csi_validate_dns_check/__main__.py

@@ -24,15 +24,15 @@
 import sys, os
 import csm_testing.lib.data_json_parser as dp
 
-''' Simple test to validate dns. It mimics the CSI validate check `grep -Eo 'ncn-.*-mgmt'` 
+''' Simple test to validate dns. It mimics the CSI validate check `grep -Eo 'ncn-.*-mgmt'`
     against data.json global meta-data ntp-peers.
 Counts the number of times that the ntp-peer appears in dnsmasq.leases file
 matches that to the number of ntp_peers - they should be ==
 '''
 
 dnsmasq_file = '/var/lib/misc/dnsmasq.leases'
 
-def main() -> int:
+def main() -> int: # pylint: disable=missing-function-docstring
     PASSED = 0
 
     # Assume we got the right info from goss, but JIC

src/csm_testing/tests/data_json_multiple_dns_server_entries/__main__.py

@@ -27,7 +27,7 @@
 logging.basicConfig(filename='/tmp/data_json_dns_server_test.log', level=logging.DEBUG)
 logging.info("Starting up")
 
-def main() -> int:
+def main() -> int: # pylint: disable=missing-function-docstring
     # Goss sends [.Args.datajson] as string with the brackets
     filename = sys.argv[1].strip('[').strip(']')
     logging.debug("Using file: %s", filename)

src/csm_testing/tests/data_validator/__main__.py

@@ -80,7 +80,7 @@ def user_data(data):
     for blob in data:
       if blob != "Global":
         filtered_data.append(data[blob]['user-data'])
-  
+
   return filtered_data
 
 
@@ -297,7 +297,7 @@ def boot_params(data):
     return err
 
 
-def main() -> int:
+def main() -> int: # pylint: disable=missing-function-docstring
   """
   Returns 0 on success, non-0 on failure.
   """