17
17
18
18
# This test requires a TH_SERVER application. Please specify with --string-arg th_server_app_path:<path_to_app>
19
19
20
+ # See https://github.com/project-chip/connectedhomeip/blob/master/docs/testing/python.md#defining-the-ci-test-arguments
21
+ # for details about the block below.
22
+ #
23
+ # === BEGIN CI TEST ARGUMENTS ===
24
+ # test-runner-runs: run1
25
+ # test-runner-run/run1/app: examples/fabric-admin/scripts/fabric-sync-app.py
26
+ # test-runner-run/run1/app-args: --app-admin=${FABRIC_ADMIN_APP} --app-bridge=${FABRIC_BRIDGE_APP} --stdin-pipe=dut-fsa-stdin --discriminator=1234
27
+ # test-runner-run/run1/factoryreset: true
28
+ # test-runner-run/run1/script-args: --PICS src/app/tests/suites/certification/ci-pics-values --storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --string-arg th_server_app_path:${ALL_CLUSTERS_APP} --trace-to json:${TRACE_TEST_JSON}.json --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto
29
+ # test-runner-run/run1/script-start-delay: 5
30
+ # test-runner-run/run1/quiet: true
31
+ # === END CI TEST ARGUMENTS ===
32
+
20
33
import logging
21
34
import os
22
35
import random
23
- import signal
24
- import subprocess
36
+ import tempfile
25
37
import time
26
- import uuid
27
38
28
39
import chip .clusters as Clusters
29
40
from chip import ChipDeviceCtrl
41
+ from chip .testing .tasks import Subprocess
30
42
from matter_testing_support import MatterBaseTest , TestStep , async_test_body , default_matter_test_main
31
43
from mobly import asserts
32
44
33
45
46
+ class AppServer (Subprocess ):
47
+ """Wrapper class for starting an application server in a subprocess."""
48
+
49
+ # Prefix for log messages from the application server.
50
+ PREFIX = "[SERVER]"
51
+
52
+ def __init__ (self , app : str , storage_dir : str , discriminator : int , passcode : int , port : int = 5540 ):
53
+ storage_kvs_dir = tempfile .mkstemp (dir = storage_dir , prefix = "kvs-app-" )[1 ]
54
+ # Start the server application with dedicated KVS storage.
55
+ super ().__init__ (app , "--KVS" , storage_kvs_dir ,
56
+ '--secured-device-port' , str (port ),
57
+ "--discriminator" , str (discriminator ),
58
+ "--passcode" , str (passcode ),
59
+ prefix = self .PREFIX )
60
+
61
+ def start (self ):
62
+ # Start process and block until it prints the expected output.
63
+ super ().start (expected_output = "Server initialization complete" )
64
+
65
+
34
66
class TC_MCORE_FS_1_1 (MatterBaseTest ):
35
67
36
68
@async_test_body
37
69
async def setup_class (self ):
38
70
super ().setup_class ()
39
- self .app_process = None
40
- app = self .user_params .get ("th_server_app_path" , None )
41
- if not app :
42
- asserts .fail ('This test requires a TH_SERVER app. Specify app path with --string-arg th_server_app_path:<path_to_app>' )
43
-
44
- self .kvs = f'kvs_{ str (uuid .uuid4 ())} '
45
- self .port = 5543
46
- discriminator = random .randint (0 , 4095 )
47
- passcode = 20202021
48
- cmd = [app ]
49
- cmd .extend (['--secured-device-port' , str (5543 )])
50
- cmd .extend (['--discriminator' , str (discriminator )])
51
- cmd .extend (['--passcode' , str (passcode )])
52
- cmd .extend (['--KVS' , self .kvs ])
53
- # TODO: Determine if we want these logs cooked or pushed to somewhere else
54
- logging .info ("Starting application to acts mock a server portion of TH_FSA" )
55
- self .app_process = subprocess .Popen (cmd )
56
- logging .info ("Started application to acts mock a server portion of TH_FSA" )
57
- time .sleep (3 )
71
+
72
+ self .th_server = None
73
+ self .storage = None
74
+
75
+ th_server_app = self .user_params .get ("th_server_app_path" , None )
76
+ if not th_server_app :
77
+ asserts .fail ("This test requires a TH_SERVER app. Specify app path with --string-arg th_server_app_path:<path_to_app>" )
78
+ if not os .path .exists (th_server_app ):
79
+ asserts .fail (f"The path { th_server_app } does not exist" )
80
+
81
+ # Create a temporary storage directory for keeping KVS files.
82
+ self .storage = tempfile .TemporaryDirectory (prefix = self .__class__ .__name__ )
83
+ logging .info ("Temporary storage directory: %s" , self .storage .name )
84
+
85
+ self .th_server_port = 5543
86
+ self .th_server_discriminator = random .randint (0 , 4095 )
87
+ self .th_server_passcode = 20202021
88
+
89
+ # Start the TH_SERVER_NO_UID app.
90
+ self .th_server = AppServer (
91
+ th_server_app ,
92
+ storage_dir = self .storage .name ,
93
+ port = self .th_server_port ,
94
+ discriminator = self .th_server_discriminator ,
95
+ passcode = self .th_server_passcode )
96
+ self .th_server .start ()
58
97
59
98
logging .info ("Commissioning from separate fabric" )
60
99
# Create a second controller on a new fabric to communicate to the server
@@ -63,25 +102,24 @@ async def setup_class(self):
63
102
paa_path = str (self .matter_test_config .paa_trust_store_path )
64
103
self .TH_server_controller = new_fabric_admin .NewController (nodeId = 112233 , paaTrustStorePath = paa_path )
65
104
self .server_nodeid = 1111
66
- await self .TH_server_controller .CommissionOnNetwork (nodeId = self .server_nodeid , setupPinCode = passcode , filterType = ChipDeviceCtrl .DiscoveryFilterType .LONG_DISCRIMINATOR , filter = discriminator )
105
+ await self .TH_server_controller .CommissionOnNetwork (
106
+ nodeId = self .server_nodeid ,
107
+ setupPinCode = self .th_server_passcode ,
108
+ filterType = ChipDeviceCtrl .DiscoveryFilterType .LONG_DISCRIMINATOR ,
109
+ filter = self .th_server_discriminator )
67
110
logging .info ("Commissioning TH_SERVER complete" )
68
111
69
112
def teardown_class (self ):
70
- # In case the th_server_app_path does not exist, then we failed the test
71
- # and there is nothing to remove
72
- if self .app_process is not None :
73
- logging .warning ("Stopping app with SIGTERM" )
74
- self .app_process .send_signal (signal .SIGTERM .value )
75
- self .app_process .wait ()
76
-
77
- if os .path .exists (self .kvs ):
78
- os .remove (self .kvs )
113
+ if self .th_server is not None :
114
+ self .th_server .terminate ()
115
+ if self .storage is not None :
116
+ self .storage .cleanup ()
79
117
super ().teardown_class ()
80
118
81
119
def steps_TC_MCORE_FS_1_1 (self ) -> list [TestStep ]:
82
120
steps = [TestStep (1 , "Enable Fabric Synchronization on DUT_FSA using the manufacturer specified mechanism." , is_commissioning = True ),
83
121
TestStep (2 , "Commission DUT_FSA onto TH_FSA fabric." ),
84
- TestStep (3 , "Reverse Commision Commission TH_FSAs onto DUT_FSA fabric." ),
122
+ TestStep (3 , "Reverse Commission TH_FSAs onto DUT_FSA fabric." ),
85
123
TestStep ("3a" , "TH_FSA sends RequestCommissioningApproval" ),
86
124
TestStep ("3b" , "TH_FSA sends CommissionNode" ),
87
125
TestStep ("3c" , "DUT_FSA commissions TH_FSA" )]
0 commit comments