[pre-commit] Fix style issues in test scripts under tests/l*, tests/m*, tests/n* folder

tests/lldp/test_lldp.py

@@ -17,12 +17,16 @@ def lldp_setup(duthosts, enum_rand_one_per_hwsku_frontend_hostname, patch_lldpct
     unpatch_lldpctl(localhost, duthost)
 
 
-def test_lldp(duthosts, enum_rand_one_per_hwsku_frontend_hostname, localhost, collect_techsupport_all_duts, enum_frontend_asic_index):
+def test_lldp(duthosts, enum_rand_one_per_hwsku_frontend_hostname, localhost,
+              collect_techsupport_all_duts, enum_frontend_asic_index):
     """ verify the LLDP message on DUT """
     duthost = duthosts[enum_rand_one_per_hwsku_frontend_hostname]
 
-    config_facts = duthost.asic_instance(enum_frontend_asic_index).config_facts(host=duthost.hostname, source="running")['ansible_facts']
-    lldpctl_facts = duthost.lldpctl_facts(asic_instance_id=enum_frontend_asic_index, skip_interface_pattern_list=["eth0", "Ethernet-BP", "Ethernet-IB"])['ansible_facts']
+    config_facts = duthost.asic_instance(
+        enum_frontend_asic_index).config_facts(host=duthost.hostname, source="running")['ansible_facts']
+    lldpctl_facts = duthost.lldpctl_facts(
+        asic_instance_id=enum_frontend_asic_index,
+        skip_interface_pattern_list=["eth0", "Ethernet-BP", "Ethernet-IB"])['ansible_facts']
     if not list(lldpctl_facts['lldpctl'].items()):
         pytest.fail("No LLDP neighbors received (lldpctl_facts are empty)")
     for k, v in list(lldpctl_facts['lldpctl'].items()):
@@ -46,13 +50,17 @@ def test_lldp_neighbor(duthosts, enum_rand_one_per_hwsku_frontend_hostname, loca
                 not sent since it contain invalid OIDs, bug.*",
         ])
 
-    res = duthost.shell("docker exec -i lldp lldpcli show chassis | grep \"SysDescr:\" | sed -e 's/^\\s*SysDescr:\\s*//g'")
+    res = duthost.shell(
+        "docker exec -i lldp lldpcli show chassis | grep \"SysDescr:\" | sed -e 's/^\\s*SysDescr:\\s*//g'")
     dut_system_description = res['stdout']
-    lldpctl_facts = duthost.lldpctl_facts(asic_instance_id=enum_frontend_asic_index, skip_interface_pattern_list=["eth0", "Ethernet-BP", "Ethernet-IB"])['ansible_facts']
-    config_facts = duthost.asic_instance(enum_frontend_asic_index).config_facts(host=duthost.hostname, source="running")['ansible_facts']
+    lldpctl_facts = duthost.lldpctl_facts(
+        asic_instance_id=enum_frontend_asic_index,
+        skip_interface_pattern_list=["eth0", "Ethernet-BP", "Ethernet-IB"])['ansible_facts']
+    config_facts = duthost.asic_instance(enum_frontend_asic_index).config_facts(host=duthost.hostname,
+                                                                                source="running")['ansible_facts']
     if not list(lldpctl_facts['lldpctl'].items()):
         pytest.fail("No LLDP neighbors received (lldpctl_facts are empty)")
-    # We use the MAC of mgmt port to generate chassis ID as LLDPD dose. 
+    # We use the MAC of mgmt port to generate chassis ID as LLDPD dose.
     # To be compatible with PR #3331, we keep using router MAC on T2 devices
     switch_mac = ""
     if tbinfo["topo"]["type"] != "t2":
@@ -68,20 +76,23 @@ def test_lldp_neighbor(duthosts, enum_rand_one_per_hwsku_frontend_hostname, loca
     for k, v in list(lldpctl_facts['lldpctl'].items()):
         try:
             hostip = v['chassis']['mgmt-ip']
-        except:
+        except Exception:
             logger.info("Neighbor device {} does not sent management IP via lldp".format(v['chassis']['name']))
             hostip = nei_meta[v['chassis']['name']]['mgmt_addr']
 
-        nei_lldp_facts = localhost.lldp_facts(host=hostip, version='v2c', community=eos['snmp_rocommunity'])['ansible_facts']
+        nei_lldp_facts = localhost.lldp_facts(
+            host=hostip, version='v2c', community=eos['snmp_rocommunity'])['ansible_facts']
         neighbor_interface = v['port']['ifname']
         # Verify the published DUT system name field is correct
         assert nei_lldp_facts['ansible_lldp_facts'][neighbor_interface]['neighbor_sys_name'] == duthost.hostname
         # Verify the published DUT chassis id field is not empty
         assert nei_lldp_facts['ansible_lldp_facts'][neighbor_interface]['neighbor_chassis_id'] == \
-               "0x%s" % (switch_mac.replace(':', ''))
+            "0x%s" % (switch_mac.replace(':', ''))
         # Verify the published DUT system description field is correct
         assert nei_lldp_facts['ansible_lldp_facts'][neighbor_interface]['neighbor_sys_desc'] == dut_system_description
         # Verify the published DUT port id field is correct
-        assert nei_lldp_facts['ansible_lldp_facts'][neighbor_interface]['neighbor_port_id'] == config_facts['PORT'][k]['alias']
+        assert nei_lldp_facts['ansible_lldp_facts'][neighbor_interface]['neighbor_port_id'] == \
+            config_facts['PORT'][k]['alias']
         # Verify the published DUT port description field is correct
-        assert nei_lldp_facts['ansible_lldp_facts'][neighbor_interface]['neighbor_port_desc'] == config_facts['PORT'][k]['description']
+        assert nei_lldp_facts['ansible_lldp_facts'][neighbor_interface]['neighbor_port_desc'] == \
+            config_facts['PORT'][k]['description']

tests/log_fidelity/test_bgp_shutdown.py

@@ -3,18 +3,19 @@
 
 from tests.common.plugins.loganalyzer.loganalyzer import LogAnalyzer, LogAnalyzerError
 
-logger=logging.getLogger(__name__)
+logger = logging.getLogger(__name__)
 
 pytestmark = [
     pytest.mark.topology('any')
 ]
 
+
 def check_syslog(duthost, prefix, trigger_action, expected_log, restore_action):
     loganalyzer = LogAnalyzer(ansible_host=duthost, marker_prefix=prefix)
-    loganalyzer.expect_regex=[expected_log]
+    loganalyzer.expect_regex = [expected_log]
 
     try:
-        marker=loganalyzer.init()
+        marker = loganalyzer.init()
         duthost.command(trigger_action)
         logger.info("Check for expected log {} in syslog".format(expected_log))
         loganalyzer.analyze(marker)
@@ -26,12 +27,12 @@ def check_syslog(duthost, prefix, trigger_action, expected_log, restore_action):
     finally:
         duthost.command(restore_action)
 
+
 def test_bgp_shutdown(duthosts, enum_rand_one_per_hwsku_frontend_hostname):
-    duthost=duthosts[enum_rand_one_per_hwsku_frontend_hostname]
+    duthost = duthosts[enum_rand_one_per_hwsku_frontend_hostname]
 
     BGP_DOWN_EXPECTED_LOG_MESSAGE = "admin state is set to 'down'"
     BGP_DOWN_COMMAND = "config bgp shutdown all"
     BGP_UP_COMMAND = "config bgp startup all"
 
     check_syslog(duthost, "bgp_shutdown", BGP_DOWN_COMMAND, BGP_DOWN_EXPECTED_LOG_MESSAGE, BGP_UP_COMMAND)
-

tests/macsec/macsec_config_helper.py

@@ -17,10 +17,11 @@
     'disable_macsec_port'
 ]
 
-
 logger = logging.getLogger(__name__)
 
-def set_macsec_profile(host, port, profile_name, priority, cipher_suite, primary_cak, primary_ckn, policy, send_sci, rekey_period = 0):
+
+def set_macsec_profile(host, port, profile_name, priority, cipher_suite,
+                       primary_cak, primary_ckn, policy, send_sci, rekey_period=0):
     if isinstance(host, EosHost):
         eos_cipher_suite = {
             "GCM-AES-128": "aes128-gcm",
@@ -36,7 +37,7 @@ def set_macsec_profile(host, port, profile_name, priority, cipher_suite, primary
         if send_sci == 'true':
             lines.append('sci')
         host.eos_config(
-            lines = lines,
+            lines=lines,
             parents=['mac security', 'profile {}'.format(profile_name)])
         return
 
@@ -56,9 +57,12 @@ def set_macsec_profile(host, port, profile_name, priority, cipher_suite, primary
     host.command(cmd)
     if send_sci == "false":
         # The MAC address of SONiC host is locally administrated
-        # So, LLDPd will use an arbitrary fixed value (00:60:08:69:97:ef) as the source MAC address of LLDP packet (https://lldpd.github.io/usage.html)
-        # But the MACsec driver in Linux used by SONiC VM has a bug that cannot handle the packet with different source MAC address to SCI if the send_sci = false
-        # So, if send_sci = false and the neighbor device is SONiC VM, LLDPd need to use the real MAC address as the source MAC address
+        # So, LLDPd will use an arbitrary fixed value (00:60:08:69:97:ef)
+        # as the source MAC address of LLDP packet (https://lldpd.github.io/usage.html)
+        # But the MACsec driver in Linux used by SONiC VM has a bug that
+        # cannot handle the packet with different source MAC address to SCI if the send_sci = false
+        # So, if send_sci = false and the neighbor device is SONiC VM,
+        # LLDPd need to use the real MAC address as the source MAC address
         host.command("lldpcli configure system bond-slave-src-mac-type real")
 
 
@@ -69,14 +73,15 @@ def delete_macsec_profile(host, port, profile_name):
             parents=['mac security'])
         return
 
-    #if port is None, the macsec profile is deleted from all namespaces if multi-asic
+    # if port is None, the macsec profile is deleted from all namespaces if multi-asic
     if host.is_multi_asic and port is None:
         for ns in host.get_asic_namespace_list():
             CMD_PREFIX = "-n {}".format(ns) if ns is not None else " "
             cmd = "sonic-db-cli {} CONFIG_DB DEL 'MACSEC_PROFILE|{}'".format(CMD_PREFIX, profile_name)
             host.command(cmd)
     else:
-        cmd = "sonic-db-cli {} CONFIG_DB DEL 'MACSEC_PROFILE|{}'".format(getns_prefix(host, port), profile_name)
+        cmd = ("sonic-db-cli {} CONFIG_DB DEL 'MACSEC_PROFILE|{}'"
+               .format(getns_prefix(host, port), profile_name))
         host.command(cmd)
 
 
@@ -99,9 +104,10 @@ def enable_macsec_port(host, port, profile_name):
         time.sleep(2)
         host.command("sudo config portchannel {} member add {} {}".format(getns_prefix(host, port), pc["name"], port))
 
-    # wait after macsec enable 
+    # wait after macsec enable
     time.sleep(2)
 
+
 def disable_macsec_port(host, port):
     if isinstance(host, EosHost):
         host.eos_config(
@@ -121,9 +127,10 @@ def disable_macsec_port(host, port):
         time.sleep(2)
         host.command("sudo config portchannel {} member add {} {}".format(getns_prefix(host, port), pc["name"], port))
 
-    # wait after macsec disable 
+    # wait after macsec disable
     time.sleep(2)
 
+
 def enable_macsec_feature(duthost, macsec_nbrhosts):
     nbrhosts = macsec_nbrhosts
     num_asics = duthost.num_asics()
@@ -162,7 +169,7 @@ def cleanup_macsec_configuration(duthost, ctrl_links, profile_name):
     wait_all_complete(timeout=300)
 
     logger.info("Cleanup macsec configuration step2: delete macsec profile")
-    # Delete the macsec profile once after it is removed from all interfaces. if we pass port as None, 
+    # Delete the macsec profile once after it is removed from all interfaces. if we pass port as None,
     # the profile is removed from the DB in all namespaces.
     submit_async_task(delete_macsec_profile, (duthost, None, profile_name))
 
@@ -186,13 +193,15 @@ def setup_macsec_configuration(duthost, ctrl_links, profile_name, default_priori
     # 1. Set macsec profile
     i = 0
     for dut_port, nbr in list(ctrl_links.items()):
-        submit_async_task(set_macsec_profile, (duthost, dut_port, profile_name, default_priority,
-                       cipher_suite, primary_cak, primary_ckn, policy, send_sci, rekey_period))
+        submit_async_task(set_macsec_profile,
+                          (duthost, dut_port, profile_name, default_priority,
+                           cipher_suite, primary_cak, primary_ckn, policy, send_sci, rekey_period))
         if i % 2 == 0:
             priority = default_priority - 1
         else:
             priority = default_priority + 1
-        submit_async_task(set_macsec_profile, (nbr["host"], nbr["port"], profile_name, priority,
+        submit_async_task(set_macsec_profile,
+                          (nbr["host"], nbr["port"], profile_name, priority,
                            cipher_suite, primary_cak, primary_ckn, policy, send_sci, rekey_period))
         i += 1
     wait_all_complete(timeout=180)
@@ -208,7 +217,7 @@ def setup_macsec_configuration(duthost, ctrl_links, profile_name, default_priori
     for dut_port, nbr in list(ctrl_links.items()):
         wait_until(20, 3, 0,
                    lambda: duthost.iface_macsec_ok(dut_port) and
-                           nbr["host"].iface_macsec_ok(nbr["port"]))
+                   nbr["host"].iface_macsec_ok(nbr["port"]))
 
     # Enabling macsec may cause link flap, which impacts LACP, BGP, etc
     # protocols. To hold some time for protocol recovery.

tests/macsec/macsec_platform_helper.py

@@ -35,7 +35,7 @@ def sonic_db_cli(host, cmd):
     return ast.literal_eval(host.shell(cmd)["stdout_lines"][0])
 
 
-def get_all_ifnames(host, asic = None):
+def get_all_ifnames(host, asic=None):
     cmd_prefix = " "
     if host.is_multi_asic and asic is not None:
         ns = host.get_namespace_from_asic_id(asic.asic_index)

tests/macsec/test_controlplane.py

@@ -2,16 +2,12 @@
 import pytest
 import logging
 import re
-import scapy.all as scapy
-import ptf.testutils as testutils
-from collections import Counter
 
 from tests.common.utilities import wait_until
 from tests.common.devices.eos import EosHost
-from tests.common import config_reload
-from .macsec_helper import *
-from .macsec_config_helper import *
-from .macsec_platform_helper import *
+from .macsec_helper import check_wpa_supplicant_process, check_appl_db, check_mka_session,\
+                           get_mka_session, get_sci, get_appl_db, get_ipnetns_prefix
+from .macsec_platform_helper import get_platform, get_macsec_ifname
 
 logger = logging.getLogger(__name__)
 

tests/macsec/test_dataplane.py

@@ -6,12 +6,10 @@
 import ptf.testutils as testutils
 from collections import Counter
 
-from tests.common.utilities import wait_until
 from tests.common.devices.eos import EosHost
-from tests.common import config_reload
-from .macsec_helper import *
-from .macsec_config_helper import *
-from .macsec_platform_helper import *
+from .macsec_helper import create_pkt, create_exp_pkt, check_macsec_pkt,\
+                           get_ipnetns_prefix, get_macsec_sa_name, get_macsec_counters
+from .macsec_platform_helper import get_portchannel, find_portchannel_from_member
 
 logger = logging.getLogger(__name__)
 
@@ -24,7 +22,8 @@
 class TestDataPlane():
     BATCH_COUNT = 10
 
-    def test_server_to_neighbor(self, duthost, ctrl_links, downstream_links, upstream_links, ptfadapter, wait_mka_establish):
+    def test_server_to_neighbor(self, duthost, ctrl_links, downstream_links,
+                                upstream_links, ptfadapter, wait_mka_establish):
         ptfadapter.dataplane.set_qlen(TestDataPlane.BATCH_COUNT * 100)
 
         down_link = list(downstream_links.values())[0]
@@ -60,7 +59,7 @@ def test_server_to_neighbor(self, duthost, ctrl_links, downstream_links, upstrea
             logging.info(payload)
             # Source mac address is not useful in this test case and we use an arbitrary mac address as the source
             pkt = create_pkt(
-                "00:01:02:03:04:05", dut_macaddress, "1.2.3.4", up_host_ip, bytes(payload,encoding='utf8'))
+                "00:01:02:03:04:05", dut_macaddress, "1.2.3.4", up_host_ip, bytes(payload, encoding='utf8'))
             exp_pkt = create_exp_pkt(pkt, pkt[scapy.IP].ttl - 1)
 
             fail_message = ""
@@ -164,7 +163,7 @@ def test_counters(self, duthost, ctrl_links, upstream_links, rekey_period, wait_
         ret = duthost.command(
             "{} ping -c {} -s {} {}".format(get_ipnetns_prefix(duthost, port_name), PKT_NUM, PKT_OCTET, nbr_ip_addr))
         assert not ret['failed']
-        sleep(10) # wait 10s for polling counters
+        sleep(10)   # wait 10s for polling counters
 
         # Sum up end counter
         egress_end_counters = Counter()

tests/macsec/test_deployment.py

@@ -1,18 +1,9 @@
-from time import sleep
 import pytest
 import logging
-import re
-import scapy.all as scapy
-import ptf.testutils as testutils
-from collections import Counter
 
 from tests.common.utilities import wait_until
-from tests.common.devices.eos import EosHost
 from tests.common import config_reload
-from .macsec_helper import *
-from .macsec_config_helper import *
-from .macsec_platform_helper import *
-
+from .macsec_helper import check_appl_db
 logger = logging.getLogger(__name__)
 
 pytestmark = [

tests/macsec/test_fault_handling.py

@@ -1,17 +1,12 @@
 from time import sleep
 import pytest
 import logging
-import re
-import scapy.all as scapy
-import ptf.testutils as testutils
-from collections import Counter
 
 from tests.common.utilities import wait_until
 from tests.common.devices.eos import EosHost
-from tests.common import config_reload
-from .macsec_helper import *
-from .macsec_config_helper import *
-from .macsec_platform_helper import *
+from .macsec_helper import get_appl_db
+from .macsec_config_helper import disable_macsec_port, enable_macsec_port, delete_macsec_profile, set_macsec_profile
+from .macsec_platform_helper import get_eth_ifname, find_portchannel_from_member, get_portchannel
 
 logger = logging.getLogger(__name__)
 
@@ -107,8 +102,8 @@ def test_mismatch_macsec_configuration(self, duthost, unctrl_links,
 
         # Wait till macsec session has gone down.
         wait_until(20, 3, 0,
-            lambda: not duthost.iface_macsec_ok(port_name) and
-                    not nbr["host"].iface_macsec_ok(nbr["port"]))
+                   lambda: not duthost.iface_macsec_ok(port_name) and
+                   not nbr["host"].iface_macsec_ok(nbr["port"]))
 
         # Set a wrong cak to the profile
         primary_cak = "0" * len(primary_cak)