diff --git a/.flake8 b/.flake8 index fe0075bf560c49..9ea57bd941d574 100644 --- a/.flake8 +++ b/.flake8 @@ -9,7 +9,6 @@ exclude = third_party # TODO: Remove the paths below when all bugs are fixed src/tools/chip-cert/* src/test_driver/mbed/* - src/test_driver/linux-cirque/* build/chip/java/tests/* build/chip/linux/* build/config/linux/* diff --git a/src/test_driver/linux-cirque/CommissioningFailureOnReportTest.py b/src/test_driver/linux-cirque/CommissioningFailureOnReportTest.py index 4f657235af1a53..b4c346bf3e45eb 100755 --- a/src/test_driver/linux-cirque/CommissioningFailureOnReportTest.py +++ b/src/test_driver/linux-cirque/CommissioningFailureOnReportTest.py @@ -17,9 +17,7 @@ import logging import os -import pprint import sys -import time from helper.CHIPTestBase import CHIPVirtualHome @@ -83,8 +81,11 @@ def run_controller_test(self): if device['type'] == 'MobileDevice'] for server in server_ids: - self.execute_device_cmd(server, "CHIPCirqueDaemon.py -- run gdb -return-child-result -q -ex \"set pagination off\" -ex run -ex \"bt 25\" --args {} --thread --discriminator {}".format( - os.path.join(CHIP_REPO, "out/debug/standalone/chip-all-clusters-app"), TEST_DISCRIMINATOR)) + self.execute_device_cmd( + server, + ("CHIPCirqueDaemon.py -- run gdb -return-child-result -q -ex \"set pagination off\" " + "-ex run -ex \"bt 25\" --args {} --thread --discriminator {}").format( + os.path.join(CHIP_REPO, "out/debug/standalone/chip-all-clusters-app"), TEST_DISCRIMINATOR)) self.reset_thread_devices(server_ids) @@ -97,7 +98,8 @@ def run_controller_test(self): self.execute_device_cmd(req_device_id, "pip3 install {}".format(os.path.join( CHIP_REPO, "out/debug/linux_x64_gcc/controller/python/chip_repl-0.0-py3-none-any.whl"))) - command = "gdb -return-child-result -q -ex run -ex bt --args python3 {} -t 150 -a {} --paa-trust-store-path {} --fail-on-report".format( + command = ("gdb -return-child-result -q -ex run -ex bt --args python3 " + "{} -t 150 -a {} --paa-trust-store-path {} --fail-on-report").format( os.path.join( CHIP_REPO, "src/controller/python/test/test_scripts/commissioning_failure_test.py"), ethernet_ip, os.path.join(CHIP_REPO, MATTER_DEVELOPMENT_PAA_ROOT_CERTS)) diff --git a/src/test_driver/linux-cirque/CommissioningFailureTest.py b/src/test_driver/linux-cirque/CommissioningFailureTest.py index 8cabb4deba54cc..dbb8dfebc09838 100755 --- a/src/test_driver/linux-cirque/CommissioningFailureTest.py +++ b/src/test_driver/linux-cirque/CommissioningFailureTest.py @@ -17,9 +17,7 @@ import logging import os -import pprint import sys -import time from helper.CHIPTestBase import CHIPVirtualHome @@ -83,8 +81,11 @@ def run_controller_test(self): if device['type'] == 'MobileDevice'] for server in server_ids: - self.execute_device_cmd(server, "CHIPCirqueDaemon.py -- run gdb -return-child-result -q -ex \"set pagination off\" -ex run -ex \"bt 25\" --args {} --thread --discriminator {}".format( - os.path.join(CHIP_REPO, "out/debug/standalone/chip-all-clusters-app"), TEST_DISCRIMINATOR)) + self.execute_device_cmd( + server, + ("CHIPCirqueDaemon.py -- run gdb -return-child-result -q -ex \"set pagination off\" " + "-ex run -ex \"bt 25\" --args {} --thread --discriminator {}").format( + os.path.join(CHIP_REPO, "out/debug/standalone/chip-all-clusters-app"), TEST_DISCRIMINATOR)) self.reset_thread_devices(server_ids) diff --git a/src/test_driver/linux-cirque/CommissioningTest.py b/src/test_driver/linux-cirque/CommissioningTest.py index b2c3a654deb505..adc6c21656dc53 100755 --- a/src/test_driver/linux-cirque/CommissioningTest.py +++ b/src/test_driver/linux-cirque/CommissioningTest.py @@ -17,9 +17,7 @@ import logging import os -import pprint import sys -import time from helper.CHIPTestBase import CHIPVirtualHome @@ -99,8 +97,11 @@ def run_controller_test(self): servers[1]['nodeid'] = 2 for server in servers: - self.execute_device_cmd(server['id'], "CHIPCirqueDaemon.py -- run gdb -return-child-result -q -ex \"set pagination off\" -ex run -ex \"bt 25\" --args {} --thread --discriminator {}".format( - os.path.join(CHIP_REPO, "out/debug/standalone/chip-all-clusters-app"), server['discriminator'])) + self.execute_device_cmd( + server['id'], + ("CHIPCirqueDaemon.py -- run gdb -return-child-result -q -ex \"set pagination off\" " + "-ex run -ex \"bt 25\" --args {} --thread --discriminator {}").format( + os.path.join(CHIP_REPO, "out/debug/standalone/chip-all-clusters-app"), server['discriminator'])) self.reset_thread_devices([server['id'] for server in servers]) @@ -113,7 +114,8 @@ def run_controller_test(self): self.execute_device_cmd(req_device_id, "pip3 install {}".format(os.path.join( CHIP_REPO, "out/debug/linux_x64_gcc/controller/python/chip_repl-0.0-py3-none-any.whl"))) - command = "gdb -return-child-result -q -ex run -ex bt --args python3 {} -t 150 -a {} --paa-trust-store-path {} --discriminator {} --nodeid {}".format( + command = ("gdb -return-child-result -q -ex run -ex bt --args python3 " + "{} -t 150 -a {} --paa-trust-store-path {} --discriminator {} --nodeid {}").format( os.path.join( CHIP_REPO, "src/controller/python/test/test_scripts/commissioning_test.py"), servers[0]['ip'], @@ -125,7 +127,8 @@ def run_controller_test(self): self.assertEqual(ret['return_code'], '0', "Test failed: non-zero return code") - command = "gdb -return-child-result -q -ex run -ex bt --args python3 {} -t 150 --paa-trust-store-path {} --discriminator {} --setup-payload {} --nodeid {}".format( + command = ("gdb -return-child-result -q -ex run -ex bt --args python3 " + "{} -t 150 --paa-trust-store-path {} --discriminator {} --setup-payload {} --nodeid {}").format( os.path.join( CHIP_REPO, "src/controller/python/test/test_scripts/commissioning_test.py"), os.path.join(CHIP_REPO, MATTER_DEVELOPMENT_PAA_ROOT_CERTS), diff --git a/src/test_driver/linux-cirque/CommissioningWindowTest.py b/src/test_driver/linux-cirque/CommissioningWindowTest.py index 16fad7ebdff8bd..7c1a646c58c52e 100755 --- a/src/test_driver/linux-cirque/CommissioningWindowTest.py +++ b/src/test_driver/linux-cirque/CommissioningWindowTest.py @@ -17,9 +17,7 @@ import logging import os -import pprint import sys -import time from helper.CHIPTestBase import CHIPVirtualHome @@ -88,7 +86,9 @@ def run_controller_test(self): servers[0]['nodeid'] = 1 for server in servers: - self.execute_device_cmd(server['id'], "CHIPCirqueDaemon.py -- run gdb -return-child-result -q -ex \"set pagination off\" -ex run -ex \"bt 25\" --args {} --thread --discriminator {}".format( + self.execute_device_cmd(server['id'], + ("CHIPCirqueDaemon.py -- run gdb -return-child-result -q -ex \"set pagination off\" " + "-ex run -ex \"bt 25\" --args {} --thread --discriminator {}").format( os.path.join(CHIP_REPO, "out/debug/standalone/chip-all-clusters-app"), server['discriminator'])) self.reset_thread_devices([server['id'] for server in servers]) @@ -102,7 +102,8 @@ def run_controller_test(self): self.execute_device_cmd(req_device_id, "pip3 install {}".format(os.path.join( CHIP_REPO, "out/debug/linux_x64_gcc/controller/python/chip_repl-0.0-py3-none-any.whl"))) - command = "gdb -return-child-result -q -ex run -ex bt --args python3 {} -t 150 --address {} --paa-trust-store-path {}".format( + command = ("gdb -return-child-result -q -ex run -ex bt " + "--args python3 {} -t 150 --address {} --paa-trust-store-path {}").format( os.path.join( CHIP_REPO, "src/controller/python/test/test_scripts/commissioning_window_test.py"), servers[0]['ip'], diff --git a/src/test_driver/linux-cirque/EchoOverTcpTest.py b/src/test_driver/linux-cirque/EchoOverTcpTest.py index 5cb0c0b1c7a111..b5a3a2471a9e48 100755 --- a/src/test_driver/linux-cirque/EchoOverTcpTest.py +++ b/src/test_driver/linux-cirque/EchoOverTcpTest.py @@ -18,7 +18,6 @@ import logging import os import sys -import time from helper.CHIPTestBase import CHIPVirtualHome @@ -83,8 +82,10 @@ def run_data_model_test(self): req_device_id = req_ids[0] for id in resp_ids: - self.execute_device_cmd(id, "CHIPCirqueDaemon.py -- run gdb -batch -return-child-result -q -ex run -ex bt --args {}".format( - os.path.join(CHIP_REPO, "out/debug/linux_x64_gcc/chip-echo-responder --tcp"))) + self.execute_device_cmd( + id, + "CHIPCirqueDaemon.py -- run gdb -batch -return-child-result -q -ex run -ex bt --args {}".format( + os.path.join(CHIP_REPO, "out/debug/linux_x64_gcc/chip-echo-responder --tcp"))) command = "gdb -return-child-result -q -ex run -ex bt --args " + \ os.path.join( diff --git a/src/test_driver/linux-cirque/EchoTest.py b/src/test_driver/linux-cirque/EchoTest.py index 4f13027958591a..b76812dbf3aa4f 100755 --- a/src/test_driver/linux-cirque/EchoTest.py +++ b/src/test_driver/linux-cirque/EchoTest.py @@ -18,7 +18,6 @@ import logging import os import sys -import time from helper.CHIPTestBase import CHIPVirtualHome diff --git a/src/test_driver/linux-cirque/FailsafeTest.py b/src/test_driver/linux-cirque/FailsafeTest.py index dbcddda426849e..3f75873cd5d683 100755 --- a/src/test_driver/linux-cirque/FailsafeTest.py +++ b/src/test_driver/linux-cirque/FailsafeTest.py @@ -17,9 +17,7 @@ import logging import os -import pprint import sys -import time from helper.CHIPTestBase import CHIPVirtualHome @@ -83,8 +81,11 @@ def run_controller_test(self): if device['type'] == 'MobileDevice'] for server in server_ids: - self.execute_device_cmd(server, "CHIPCirqueDaemon.py -- run gdb -return-child-result -q -ex \"set pagination off\" -ex run -ex \"bt 25\" --args {} --thread --discriminator {}".format( - os.path.join(CHIP_REPO, "out/debug/standalone/chip-all-clusters-app"), TEST_DISCRIMINATOR)) + self.execute_device_cmd( + server, + ("CHIPCirqueDaemon.py -- run gdb -return-child-result -q -ex \"set pagination off\" " + "-ex run -ex \"bt 25\" --args {} --thread --discriminator {}").format( + os.path.join(CHIP_REPO, "out/debug/standalone/chip-all-clusters-app"), TEST_DISCRIMINATOR)) self.reset_thread_devices(server_ids) diff --git a/src/test_driver/linux-cirque/InteractionModelTest.py b/src/test_driver/linux-cirque/InteractionModelTest.py index fe0848d1bc76f1..3884f0a125cd36 100755 --- a/src/test_driver/linux-cirque/InteractionModelTest.py +++ b/src/test_driver/linux-cirque/InteractionModelTest.py @@ -18,7 +18,6 @@ import logging import os import sys -import time from helper.CHIPTestBase import CHIPVirtualHome diff --git a/src/test_driver/linux-cirque/ManualTest.py b/src/test_driver/linux-cirque/ManualTest.py index 0032df00b8ff2f..ba71b0fa8c9080 100755 --- a/src/test_driver/linux-cirque/ManualTest.py +++ b/src/test_driver/linux-cirque/ManualTest.py @@ -20,7 +20,7 @@ import os import sys import time -from optparse import OptionParser, OptionValueError +from optparse import OptionParser from helper.CHIPTestBase import CHIPVirtualHome diff --git a/src/test_driver/linux-cirque/MobileDeviceTest.py b/src/test_driver/linux-cirque/MobileDeviceTest.py index c5d29a1c2feff0..84ab53b9c15514 100755 --- a/src/test_driver/linux-cirque/MobileDeviceTest.py +++ b/src/test_driver/linux-cirque/MobileDeviceTest.py @@ -17,9 +17,7 @@ import logging import os -import pprint import sys -import time from helper.CHIPTestBase import CHIPVirtualHome @@ -83,8 +81,11 @@ def run_controller_test(self): if device['type'] == 'MobileDevice'] for server in server_ids: - self.execute_device_cmd(server, "CHIPCirqueDaemon.py -- run gdb -batch -return-child-result -q -ex \"set pagination off\" -ex run -ex \"thread apply all bt\" --args {} --thread --discriminator {}".format( - os.path.join(CHIP_REPO, "out/debug/standalone/chip-all-clusters-app"), TEST_DISCRIMINATOR)) + self.execute_device_cmd( + server, + ("CHIPCirqueDaemon.py -- run gdb -batch -return-child-result -q -ex \"set pagination off\" " + "-ex run -ex \"thread apply all bt\" --args {} --thread --discriminator {}").format( + os.path.join(CHIP_REPO, "out/debug/standalone/chip-all-clusters-app"), TEST_DISCRIMINATOR)) self.reset_thread_devices(server_ids) @@ -97,7 +98,8 @@ def run_controller_test(self): self.execute_device_cmd(req_device_id, "pip3 install {}".format(os.path.join( CHIP_REPO, "out/debug/linux_x64_gcc/controller/python/chip_repl-0.0-py3-none-any.whl"))) - command = "gdb -batch -return-child-result -q -ex run -ex \"thread apply all bt\" --args python3 {} -t 240 -a {} --paa-trust-store-path {}".format( + command = ("gdb -batch -return-child-result -q -ex run -ex \"thread apply all bt\" " + "--args python3 {} -t 240 -a {} --paa-trust-store-path {}").format( os.path.join( CHIP_REPO, "src/controller/python/test/test_scripts/mobile-device-test.py"), ethernet_ip, os.path.join(CHIP_REPO, MATTER_DEVELOPMENT_PAA_ROOT_CERTS)) diff --git a/src/test_driver/linux-cirque/OnOffClusterTest.py b/src/test_driver/linux-cirque/OnOffClusterTest.py index 701bd80dfaea81..0bd78f2c8a5246 100755 --- a/src/test_driver/linux-cirque/OnOffClusterTest.py +++ b/src/test_driver/linux-cirque/OnOffClusterTest.py @@ -115,8 +115,11 @@ def run_data_model_test(self): for device_id in server_ids: self.logger.info("checking device log for {}".format( self.get_device_pretty_id(device_id))) - self.assertTrue(self.sequenceMatch(self.get_device_log(device_id).decode('utf-8'), ["Thread initialized.", "LightingManager::InitiateAction(ON_ACTION)", "LightingManager::InitiateAction(OFF_ACTION)"]), - "Datamodel test failed: cannot find matching string from device {}".format(device_id)) + self.assertTrue(self.sequenceMatch(self.get_device_log(device_id).decode('utf-8'), [ + "Thread initialized.", + "LightingManager::InitiateAction(ON_ACTION)", + "LightingManager::InitiateAction(OFF_ACTION)" + ]), "Datamodel test failed: cannot find matching string from device {}".format(device_id)) if __name__ == "__main__": diff --git a/src/test_driver/linux-cirque/PythonCommissioningTest.py b/src/test_driver/linux-cirque/PythonCommissioningTest.py index 052474a48c8f47..c3f70c85760bac 100755 --- a/src/test_driver/linux-cirque/PythonCommissioningTest.py +++ b/src/test_driver/linux-cirque/PythonCommissioningTest.py @@ -17,9 +17,7 @@ import logging import os -import pprint import sys -import time from helper.CHIPTestBase import CHIPVirtualHome diff --git a/src/test_driver/linux-cirque/SplitCommissioningTest.py b/src/test_driver/linux-cirque/SplitCommissioningTest.py index 623e4355dee79d..ad4475fd92437d 100755 --- a/src/test_driver/linux-cirque/SplitCommissioningTest.py +++ b/src/test_driver/linux-cirque/SplitCommissioningTest.py @@ -17,9 +17,7 @@ import logging import os -import pprint import sys -import time from helper.CHIPTestBase import CHIPVirtualHome @@ -91,8 +89,10 @@ def run_controller_test(self): if device['type'] == 'MobileDevice'] for server in server_ids: - self.execute_device_cmd(server, "CHIPCirqueDaemon.py -- run gdb -return-child-result -q -ex \"set pagination off\" -ex run -ex \"bt 25\" --args {} --thread".format( - os.path.join(CHIP_REPO, "out/debug/standalone/chip-all-clusters-app"))) + self.execute_device_cmd(server, + ("CHIPCirqueDaemon.py -- run gdb -return-child-result -q -ex " + "\"set pagination off\" -ex run -ex \"bt 25\" --args {} --thread").format( + os.path.join(CHIP_REPO, "out/debug/standalone/chip-all-clusters-app"))) self.reset_thread_devices(server_ids) @@ -105,7 +105,8 @@ def run_controller_test(self): self.execute_device_cmd(req_device_id, "pip3 install {}".format(os.path.join( CHIP_REPO, "out/debug/linux_x64_gcc/controller/python/chip_repl-0.0-py3-none-any.whl"))) - command = "gdb -return-child-result -q -ex run -ex bt --args python3 {} -t 150 --address1 {} --address2 {} --paa-trust-store-path {}".format( + command = ("gdb -return-child-result -q -ex run -ex bt --args python3 " + "{} -t 150 --address1 {} --address2 {} --paa-trust-store-path {}").format( os.path.join( CHIP_REPO, "src/controller/python/test/test_scripts/split_commissioning_test.py"), ethernet_ips[0], ethernet_ips[1], os.path.join(CHIP_REPO, MATTER_DEVELOPMENT_PAA_ROOT_CERTS)) diff --git a/src/test_driver/linux-cirque/helper/CHIPTestBase.py b/src/test_driver/linux-cirque/helper/CHIPTestBase.py index 957c9187aa465f..62489b24bfc360 100644 --- a/src/test_driver/linux-cirque/helper/CHIPTestBase.py +++ b/src/test_driver/linux-cirque/helper/CHIPTestBase.py @@ -17,7 +17,6 @@ import ipaddress import json -import logging import os import re import sys @@ -74,12 +73,12 @@ def run_test(self, save_logs=True): if save_logs: try: self.save_device_logs() - except: + except Exception: test_ret = TestResult.SYSTEM_FAILURE traceback.print_exc(file=sys.stderr) try: self.destroy_home() - except: + except requests.exceptions.RequestException: test_ret = TestResult.SYSTEM_FAILURE traceback.print_exc(file=sys.stderr) return test_ret @@ -101,7 +100,7 @@ def execute_device_cmd(self, device_id, cmd, stream=False): ret_struct = ret.json() command_ret_code = ret_struct.get('return_code', None) - if command_ret_code == None: + if command_ret_code is None: # could be 0 self.logger.error("cannot get command return code") raise Exception("cannot get command return code") @@ -110,7 +109,7 @@ def execute_device_cmd(self, device_id, cmd, stream=False): ret_struct.get('return_code', 'Unknown')) ) command_output = ret_struct.get('output', None) - if command_output == None: + if command_output is None: # could be empty string self.logger.error("cannot get command output") raise Exception("cannot get command output") @@ -213,7 +212,9 @@ def connect_to_thread_network(self): otInitCommands = [ "ot-ctl thread stop", "ot-ctl ifconfig down", - "ot-ctl dataset set active 0e080000000000010000000300000d35060004001fffe00208dead00beef00cafe0708fd01234567890abc051000112233445566778899aabbccddeeff030a4f70656e546872656164010212340410ad463152f9622c7297ec6c6c543a63e70c0302a0ff", + ("ot-ctl dataset set active 0e080000000000010000000300000d35060004001fffe00208d" + "ead00beef00cafe0708fd01234567890abc051000112233445566778899aabbccddeeff030a4f" + "70656e546872656164010212340410ad463152f9622c7297ec6c6c543a63e70c0302a0ff"), "ot-ctl ifconfig up", "ot-ctl thread start", "ot-ctl dataset active", # Emit @@ -264,7 +265,8 @@ def get_device_thread_ip(self, device_id): continue if not ipaddr.is_private: continue - if re.match("fd[0-9a-f]{2}:[0-9a-f]{4}:[0-9a-f]{4}:[0-9a-f]{4}:0000:00ff:fe00:[0-9a-f]{4}", ipaddr.exploded) != None: + if re.match(("fd[0-9a-f]{2}:[0-9a-f]{4}:[0-9a-f]" + "{4}:[0-9a-f]{4}:0000:00ff:fe00:[0-9a-f]{4}"), ipaddr.exploded) is not None: continue self.logger.info("Get Mesh-Local EID: {}".format(ipstr)) return str(ipaddr) @@ -293,19 +295,19 @@ def assertTrue(self, exp, note=None): assert(Not)Equal python unittest style functions that raise exceptions when condition not met ''' - if not exp == True: + if not (exp is True): if note: self.logger.error(note) raise AssertionError def assertFalse(self, exp, note=None): - if not exp == False: + if not (exp is False): if note: self.logger.error(note) raise AssertionError def assertEqual(self, val1, val2, note=None): - if not val1 == val2: + if not (val1 == val2): if note: self.logger.error(note) raise AssertionError @@ -366,7 +368,7 @@ def initialize_home(self): def save_device_logs(self): timestamp = int(time.time()) log_dir = os.environ.get("DEVICE_LOG_DIR", None) - if log_dir != None and not os.path.exists(log_dir): + if log_dir is not None and not os.path.exists(log_dir): os.makedirs("logs") for device in self.non_ap_devices: @@ -409,7 +411,7 @@ def kill_existing_wpa_supplicant(self, device_id): def get_device_pretty_name(self, device_id): device_obj = self.device_config.get(device_id, None) - if device_obj != None: + if device_obj is not None: return device_obj['type'] return ""