diff --git a/src/python_testing/TC_CCTRL_2_2.py b/src/python_testing/TC_CCTRL_2_2.py index a4314e24ddf36a..edba3eec60ebfd 100644 --- a/src/python_testing/TC_CCTRL_2_2.py +++ b/src/python_testing/TC_CCTRL_2_2.py @@ -20,10 +20,11 @@ # # TODO: Skip CI for now, we don't have any way to run this. Needs setup. See test_TC_CCTRL.py +# This test requires a TH_SERVER application. Please specify with --string-arg th_server_app_path: + import ipaddress import logging import os -import pathlib import random import signal import subprocess @@ -43,15 +44,14 @@ class TC_CCTRL_2_2(MatterBaseTest): @async_test_body async def setup_class(self): super().setup_class() - # TODO: This needs to come from an arg and needs to be something available on the TH # TODO: confirm whether we can open processes like this on the TH - app = os.path.join(pathlib.Path(__file__).resolve().parent, '..', '..', 'out', - 'linux-x64-all-clusters-no-ble', 'chip-all-clusters-app') + app = self.matter_test_config.user_params.get("th_server_app_path", None) + if not app: + asserts.fail('This test requires a TH_SERVER app. Specify app path with --string-arg th_server_app_path:') self.kvs = f'kvs_{str(uuid.uuid4())}' self.port = 5543 discriminator = random.randint(0, 4095) - discriminator = 3840 passcode = 20202021 app_args = f'--secured-device-port {self.port} --discriminator {discriminator} --passcode {passcode} --KVS {self.kvs}' cmd = f'{app} {app_args}' @@ -76,7 +76,6 @@ def teardown_class(self): logging.warning("Stopping app with SIGTERM") self.app_process.send_signal(signal.SIGTERM.value) self.app_process.wait() - # TODO: Use timeout, if term doesn't work, try SIGINT os.remove(self.kvs) super().teardown_class() diff --git a/src/python_testing/test_testing/test_TC_CCNTL_2_2.py b/src/python_testing/test_testing/test_TC_CCNTL_2_2.py index 3fa2481e9fa4c3..ba884bc48f9ee4 100644 --- a/src/python_testing/test_testing/test_TC_CCNTL_2_2.py +++ b/src/python_testing/test_testing/test_TC_CCNTL_2_2.py @@ -17,6 +17,7 @@ # import base64 +import click import os import pathlib import sys @@ -29,17 +30,19 @@ from MockTestRunner import AsyncMock, MockTestRunner try: - from matter_testing_support import get_default_paa_trust_store, run_tests_no_exit + from matter_testing_support import MatterTestConfig, get_default_paa_trust_store, run_tests_no_exit except ImportError: sys.path.append(os.path.abspath( os.path.join(os.path.dirname(__file__), '..'))) - from matter_testing_support import get_default_paa_trust_store, run_tests_no_exit + from matter_testing_support import MatterTestConfig, get_default_paa_trust_store, run_tests_no_exit invoke_call_count = 0 event_call_count = 0 def dynamic_invoke_return(*args, **argv): + ''' Returns the response to a mocked SendCommand call. + ''' global invoke_call_count invoke_call_count += 1 @@ -78,6 +81,8 @@ def dynamic_invoke_return(*args, **argv): def dynamic_event_return(*args, **argv): + ''' Returns the response to a mocked ReadEvent call. + ''' global event_call_count event_call_count += 1 @@ -104,6 +109,12 @@ def dynamic_event_return(*args, **argv): def wildcard() -> Attribute.AsyncReadTransaction.ReadResponse: + ''' Returns the response to a wildcard read. + For this test, we just need descriptors and a few attributes + Tree + EP1 (Aggregator): Descriptor + - EP2 (Bridged Node): Descriptor, Bridged Device Basic Information, Ecosystem Information + ''' cc = Clusters.CommissionerControl ei = Clusters.EcosystemInformation desc = Clusters.Descriptor @@ -133,9 +144,20 @@ def wildcard() -> Attribute.AsyncReadTransaction.ReadResponse: class MyMock(MockTestRunner): - # TODO consolidate with above def run_test_with_mock(self, dynamic_invoke_return: typing.Callable, dynamic_event_return: typing.Callable, read_cache: Attribute.AsyncReadTransaction.ReadResponse, hooks=None): - ''' Effects is a list of callable functions with *args, **kwargs parameters. It can either throw an InteractionModelException or return the command response.''' + ''' Run the test using the Mocked versions of Read, SendCommand, OpenCommissioningWindow, FindOrEstablishPASESession and ReadEvent + dynamic_invoke_return: Callable function that returns the result of a SendCommand call + Function should return one of + - command response for commands with responses + - None for commands with success results + - raise InteractionModelError for error results + dynamic_event_return: Callable function that returns the result of a ReadEvent call + Function should return one of + - list of EventReadResult for successful reads + - raise InteractionModelError for error results + read_cache : Response to a Read call. For this test, this will be the wildcard read of all teh attributes + hooks : Test harness hook object if desired. + ''' self.default_controller.Read = AsyncMock(return_value=read_cache) self.default_controller.SendCommand = AsyncMock(return_value=None, side_effect=dynamic_invoke_return) # It doesn't actually matter what we return here because I'm going to catch the next pase session connection anyway @@ -147,7 +169,9 @@ def run_test_with_mock(self, dynamic_invoke_return: typing.Callable, dynamic_eve return run_tests_no_exit(self.test_class, self.config, hooks, self.default_controller, self.stack) -def main(): +@click.command() +@click.argument('th_server_app', type=click.Path(exists=True)) +def main(th_server_app: str): root = os.path.abspath(os.path.join(pathlib.Path(__file__).resolve().parent, '..', '..', '..')) print(f'root = {root}') paa_path = get_default_paa_trust_store(root) @@ -155,6 +179,9 @@ def main(): pics = {"PICS_SDK_CI_ONLY": True} test_runner = MyMock('TC_CCTRL_2_2', 'TC_CCTRL_2_2', 'test_TC_CCTRL_2_2', 1, paa_trust_store_path=paa_path, pics=pics) + config = MatterTestConfig() + config.user_params = {'th_server_app_path': th_server_app} + test_runner.set_test_config(config) test_runner.run_test_with_mock(dynamic_invoke_return, dynamic_event_return, wildcard()) test_runner.Shutdown()