Skip to content

Commit 95fa9af

Browse files
arkqarkq
committed
Reuse AppServer from TC_MCORE_FS_1_1
1 parent aa41a5a commit 95fa9af

File tree

2 files changed

+75
-84
lines changed

2 files changed

+75
-84
lines changed

‎src/python_testing/TC_MCORE_FS_1_2.py

+37-42
Original file line numberDiff line numberDiff line change
@@ -24,11 +24,9 @@
2424
import os
2525
import queue
2626
import secrets
27-
import signal
2827
import struct
29-
import subprocess
28+
import tempfile
3029
import time
31-
import uuid
3230
from dataclasses import dataclass
3331

3432
import chip.clusters as Clusters
@@ -37,6 +35,7 @@
3735
from matter_testing_support import MatterBaseTest, TestStep, async_test_body, default_matter_test_main, type_matches
3836
from mobly import asserts
3937
from TC_SC_3_6 import AttributeChangeAccumulator
38+
from TC_MCORE_FS_1_1 import AppServer
4039

4140
# Length of `w0s` and `w1s` elements
4241
WS_LENGTH = NIST256p.baselen + 8
@@ -63,44 +62,48 @@ class TC_MCORE_FS_1_2(MatterBaseTest):
6362
@async_test_body
6463
async def setup_class(self):
6564
super().setup_class()
65+
6666
self._partslist_subscription = None
67-
self._app_th_server_process = None
68-
self._th_server_kvs = None
67+
self.th_server = None
68+
self.storage = None
69+
70+
th_server_port = self.user_params.get("th_server_port", 5543)
71+
th_server_app = self.user_params.get("th_server_app_path", None)
72+
if not th_server_app:
73+
asserts.fail('This test requires a TH_SERVER app. Specify app path with --string-arg th_server_app_path:<path_to_app>')
74+
if not os.path.exists(th_server_app):
75+
asserts.fail(f'The path {th_server_app} does not exist')
76+
77+
# Create a temporary storage directory for keeping KVS files.
78+
self.storage = tempfile.TemporaryDirectory(prefix=self.__class__.__name__)
79+
logging.info("Temporary storage directory: %s", self.storage.name)
80+
81+
self.th_server_port = th_server_port
82+
self.th_server_setup_params = _SetupParameters(
83+
setup_qr_code="MT:-24J0AFN00KA0648G00",
84+
manual_code=34970112332,
85+
discriminator=3840,
86+
passcode=20202021)
87+
88+
# Start the TH_SERVER_NO_UID app.
89+
self.th_server = AppServer(
90+
th_server_app,
91+
storage_dir=self.storage.name,
92+
port=self.th_server_port,
93+
discriminator=self.th_server_setup_params.discriminator,
94+
passcode=self.th_server_setup_params.passcode)
95+
self.th_server.start()
6996

7097
def teardown_class(self):
7198
if self._partslist_subscription is not None:
7299
self._partslist_subscription.Shutdown()
73100
self._partslist_subscription = None
74-
75-
if self._app_th_server_process is not None:
76-
logging.warning("Stopping app with SIGTERM")
77-
self._app_th_server_process.send_signal(signal.SIGTERM.value)
78-
self._app_th_server_process.wait()
79-
80-
if self._th_server_kvs is not None:
81-
os.remove(self._th_server_kvs)
101+
if self.th_server is not None:
102+
self.th_server.terminate()
103+
if self.storage is not None:
104+
self.storage.cleanup()
82105
super().teardown_class()
83106

84-
async def _create_th_server(self, port):
85-
# These are default testing values
86-
setup_params = _SetupParameters(setup_qr_code="MT:-24J0AFN00KA0648G00",
87-
manual_code=34970112332, discriminator=3840, passcode=20202021)
88-
kvs = f'kvs_{str(uuid.uuid4())}'
89-
90-
cmd = [self._th_server_app_path]
91-
cmd.extend(['--secured-device-port', str(port)])
92-
cmd.extend(['--discriminator', str(setup_params.discriminator)])
93-
cmd.extend(['--passcode', str(setup_params.passcode)])
94-
cmd.extend(['--KVS', kvs])
95-
96-
# TODO: Determine if we want these logs cooked or pushed to somewhere else
97-
logging.info("Starting TH_SERVER")
98-
self._app_th_server_process = subprocess.Popen(cmd)
99-
self._th_server_kvs = kvs
100-
logging.info("Started TH_SERVER")
101-
time.sleep(3)
102-
return setup_params
103-
104107
def _ask_for_vendor_commissioning_ux_operation(self, setup_params: _SetupParameters):
105108
self.wait_for_user_input(
106109
prompt_msg=f"Using the DUT vendor's provided interface, commission the ICD device using the following parameters:\n"
@@ -115,7 +118,6 @@ def steps_TC_MCORE_FS_1_2(self) -> list[TestStep]:
115118
steps = [TestStep(1, "TH subscribes to PartsList attribute of the Descriptor cluster of DUT_FSA endpoint 0."),
116119
TestStep(2, "Follow manufacturer provided instructions to have DUT_FSA commission TH_SERVER"),
117120
TestStep(3, "TH waits up to 30 seconds for subscription report from the PartsList attribute of the Descriptor to contain new endpoint"),
118-
119121
TestStep(4, "TH uses DUT to open commissioning window to TH_SERVER"),
120122
TestStep(5, "TH commissions TH_SERVER"),
121123
TestStep(6, "TH reads all attributes in Basic Information cluster from TH_SERVER directly"),
@@ -134,12 +136,6 @@ async def test_TC_MCORE_FS_1_2(self):
134136

135137
min_report_interval_sec = self.user_params.get("min_report_interval_sec", 0)
136138
max_report_interval_sec = self.user_params.get("max_report_interval_sec", 30)
137-
th_server_port = self.user_params.get("th_server_port", 5543)
138-
self._th_server_app_path = self.user_params.get("th_server_app_path", None)
139-
if not self._th_server_app_path:
140-
asserts.fail('This test requires a TH_SERVER app. Specify app path with --string-arg th_server_app_path:<path_to_app>')
141-
if not os.path.exists(self._th_server_app_path):
142-
asserts.fail(f'The path {self._th_server_app_path} does not exist')
143139

144140
self.step(1)
145141
# Subscribe to the PartsList
@@ -164,8 +160,7 @@ async def test_TC_MCORE_FS_1_2(self):
164160
asserts.assert_true(type_matches(step_1_dut_parts_list, list), "PartsList is expected to be a list")
165161

166162
self.step(2)
167-
setup_params = await self._create_th_server(th_server_port)
168-
self._ask_for_vendor_commissioning_ux_operation(setup_params)
163+
self._ask_for_vendor_commissioning_ux_operation(self.th_server_setup_params)
169164

170165
self.step(3)
171166
report_waiting_timeout_delay_sec = 30

‎src/python_testing/TC_MCORE_FS_1_5.py

+38-42
Original file line numberDiff line numberDiff line change
@@ -22,11 +22,9 @@
2222
import os
2323
import queue
2424
import secrets
25-
import signal
2625
import struct
27-
import subprocess
26+
import tempfile
2827
import time
29-
import uuid
3028
from dataclasses import dataclass
3129

3230
import chip.clusters as Clusters
@@ -35,6 +33,7 @@
3533
from matter_testing_support import MatterBaseTest, TestStep, async_test_body, default_matter_test_main, type_matches
3634
from mobly import asserts
3735
from TC_SC_3_6 import AttributeChangeAccumulator
36+
from TC_MCORE_FS_1_1 import AppServer
3837

3938
# Length of `w0s` and `w1s` elements
4039
WS_LENGTH = NIST256p.baselen + 8
@@ -61,49 +60,53 @@ class TC_MCORE_FS_1_5(MatterBaseTest):
6160
@async_test_body
6261
async def setup_class(self):
6362
super().setup_class()
63+
6464
self._partslist_subscription = None
6565
self._cadmin_subscription = None
66-
self._app_th_server_process = None
67-
self._th_server_kvs = None
66+
self.th_server = None
67+
self.storage = None
68+
69+
th_server_port = self.user_params.get("th_server_port", 5543)
70+
th_server_app = self.user_params.get("th_server_app_path", None)
71+
if not th_server_app:
72+
asserts.fail('This test requires a TH_SERVER app. Specify app path with --string-arg th_server_app_path:<path_to_app>')
73+
if not os.path.exists(th_server_app):
74+
asserts.fail(f'The path {th_server_app} does not exist')
75+
76+
# Create a temporary storage directory for keeping KVS files.
77+
self.storage = tempfile.TemporaryDirectory(prefix=self.__class__.__name__)
78+
logging.info("Temporary storage directory: %s", self.storage.name)
79+
80+
self.th_server_port = th_server_port
81+
# These are default testing values
82+
self.th_server_setup_params = _SetupParameters(
83+
setup_qr_code="MT:-24J0AFN00KA0648G00",
84+
manual_code=34970112332,
85+
discriminator=3840,
86+
passcode=20202021)
87+
88+
# Start the TH_SERVER_NO_UID app.
89+
self.th_server = AppServer(
90+
th_server_app,
91+
storage_dir=self.storage.name,
92+
port=self.th_server_port,
93+
discriminator=self.th_server_setup_params.discriminator,
94+
passcode=self.th_server_setup_params.passcode)
95+
self.th_server.start()
6896

6997
def teardown_class(self):
7098
if self._partslist_subscription is not None:
7199
self._partslist_subscription.Shutdown()
72100
self._partslist_subscription = None
73-
74101
if self._cadmin_subscription is not None:
75102
self._cadmin_subscription.Shutdown()
76103
self._cadmin_subscription = None
77-
78-
if self._app_th_server_process is not None:
79-
logging.warning("Stopping app with SIGTERM")
80-
self._app_th_server_process.send_signal(signal.SIGTERM.value)
81-
self._app_th_server_process.wait()
82-
83-
if self._th_server_kvs is not None:
84-
os.remove(self._th_server_kvs)
104+
if self.th_server is not None:
105+
self.th_server.terminate()
106+
if self.storage is not None:
107+
self.storage.cleanup()
85108
super().teardown_class()
86109

87-
async def _create_th_server(self, port):
88-
# These are default testing values
89-
setup_params = _SetupParameters(setup_qr_code="MT:-24J0AFN00KA0648G00",
90-
manual_code=34970112332, discriminator=3840, passcode=20202021)
91-
kvs = f'kvs_{str(uuid.uuid4())}'
92-
93-
cmd = [self._th_server_app_path]
94-
cmd.extend(['--secured-device-port', str(port)])
95-
cmd.extend(['--discriminator', str(setup_params.discriminator)])
96-
cmd.extend(['--passcode', str(setup_params.passcode)])
97-
cmd.extend(['--KVS', kvs])
98-
99-
# TODO: Determine if we want these logs cooked or pushed to somewhere else
100-
logging.info("Starting TH_SERVER")
101-
self._app_th_server_process = subprocess.Popen(cmd)
102-
self._th_server_kvs = kvs
103-
logging.info("Started TH_SERVER")
104-
time.sleep(3)
105-
return setup_params
106-
107110
def _ask_for_vendor_commissioning_ux_operation(self, setup_params: _SetupParameters):
108111
self.wait_for_user_input(
109112
prompt_msg=f"Using the DUT vendor's provided interface, commission the ICD device using the following parameters:\n"
@@ -139,12 +142,6 @@ async def test_TC_MCORE_FS_1_5(self):
139142

140143
min_report_interval_sec = 0
141144
max_report_interval_sec = 30
142-
th_server_port = self.user_params.get("th_server_port", 5543)
143-
self._th_server_app_path = self.user_params.get("th_server_app_path", None)
144-
if not self._th_server_app_path:
145-
asserts.fail('This test requires a TH_SERVER app. Specify app path with --string-arg th_server_app_path:<path_to_app>')
146-
if not os.path.exists(self._th_server_app_path):
147-
asserts.fail(f'The path {self._th_server_app_path} does not exist')
148145

149146
self.step(1)
150147
# Subscribe to the PartsList
@@ -169,8 +166,7 @@ async def test_TC_MCORE_FS_1_5(self):
169166
asserts.assert_true(type_matches(step_1_dut_parts_list, list), "PartsList is expected to be a list")
170167

171168
self.step(2)
172-
setup_params = await self._create_th_server(th_server_port)
173-
self._ask_for_vendor_commissioning_ux_operation(setup_params)
169+
self._ask_for_vendor_commissioning_ux_operation(self.th_server_setup_params)
174170

175171
self.step(3)
176172
report_waiting_timeout_delay_sec = 30

0 commit comments

Comments
 (0)