Skip to content

Commit 64d145c

Browse files
[Linux] Add test logic to verify commissioning between Linux tv-casting-app and Linux tv-app.
1 parent 0afc132 commit 64d145c

File tree

1 file changed

+233
-29
lines changed

1 file changed

+233
-29
lines changed

scripts/tests/run_tv_casting_test.py

+233-29
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@
2020
import sys
2121
import tempfile
2222
import time
23-
from typing import List, Optional, TextIO, Tuple
23+
from typing import List, Optional, TextIO, Tuple, Union
2424

2525
import click
2626

@@ -30,6 +30,9 @@
3030
# The maximum amount of time to wait for the Linux tv-app to start before timeout.
3131
TV_APP_MAX_START_WAIT_SEC = 2
3232

33+
# The maximum amount of time to commission the Linux tv-casting-app and the tv-app before timeout.
34+
COMMISSIONING_STAGE_MAX_WAIT_SEC =3
35+
3336
# File names of logs for the Linux tv-casting-app and the Linux tv-app.
3437
LINUX_TV_APP_LOGS = 'Linux-tv-app-logs.txt'
3538
LINUX_TV_CASTING_APP_LOGS = 'Linux-tv-casting-app-logs.txt'
@@ -47,13 +50,14 @@ class ProcessManager:
4750
This class provides a context manager for safely starting and stopping a subprocess.
4851
"""
4952

50-
def __init__(self, command: List[str], stdout, stderr):
53+
def __init__(self, command: List[str], stdin, stdout, stderr):
5154
self.command = command
55+
self.stdin = stdin
5256
self.stdout = stdout
5357
self.stderr = stderr
5458

5559
def __enter__(self):
56-
self.process = subprocess.Popen(self.command, stdout=self.stdout, stderr=self.stderr, text=True)
60+
self.process = subprocess.Popen(self.command, stdin=self.stdin, stdout=self.stdout, stderr=self.stderr, text=True)
5761
return self.process
5862

5963
def __exit__(self, exception_type, exception_value, traceback):
@@ -71,9 +75,9 @@ def dump_temporary_logs_to_console(log_file_path: str):
7175
print(line.rstrip())
7276

7377

74-
def handle_discovery_failure(log_file_paths: List[str]):
75-
"""Log 'Discovery failed!' as error, dump log files to console, exit on error."""
76-
logging.error('Discovery failed!')
78+
def handle_casting_state_failure(casting_state: str, log_file_paths: List[str]):
79+
"""Log '{casting_state} failed!' as error, dump log files to console, exit on error."""
80+
logging.error(casting_state + ' failed!')
7781

7882
for log_file_path in log_file_paths:
7983
try:
@@ -84,30 +88,35 @@ def handle_discovery_failure(log_file_paths: List[str]):
8488
sys.exit(1)
8589

8690

87-
def extract_value_from_string(line: str) -> int:
88-
"""Extract and return integer value from given output string.
91+
def extract_value_from_string(line: str) -> Union[int, str]:
92+
"""Extract and return either an integer or substring from given input string.
8993
90-
The string is expected to be in the following format as it is received
94+
The input string is expected to be in the following format as it is received
9195
from the Linux tv-casting-app output:
9296
\x1b[0;34m[1713741926895] [7276:9521344] [DIS] Vendor ID: 65521\x1b[0m
9397
The integer value to be extracted here is 65521.
98+
Or:
99+
\x1b[0;34m[1714583616179] [7029:2386956] [SVR] device Name: Test TV casting app\x1b[0m
100+
The substring to be extracted here is 'Test TV casting app'.
94101
"""
95102
value = line.split(':')[-1].strip().replace('\x1b[0m', '')
96-
value = int(value)
97103

98-
return value
104+
try:
105+
value = int(value)
106+
return value
107+
except ValueError:
108+
return value
99109

100110

101-
def validate_value(expected_value: int, log_paths: List[str], line: str, value_name: str) -> Optional[str]:
102-
"""Validate a value in a string against an expected value."""
111+
def validate_value(casting_state: str, expected_value: int, log_paths: List[str], line: str, value_name: str) -> Optional[str]:
112+
"""Validate a value in a string against an expected value during a given casting state."""
103113
value = extract_value_from_string(line)
104114

105115
if value != expected_value:
106116
logging.error(f'{value_name} does not match the expected value!')
107117
logging.error(f'Expected {value_name}: {expected_value}')
108118
logging.error(line.rstrip('\n'))
109-
handle_discovery_failure(log_paths)
110-
return None
119+
handle_casting_state_failure(casting_state, log_paths)
111120

112121
# Return the line containing the valid value.
113122
return line.rstrip('\n')
@@ -120,7 +129,7 @@ def start_up_tv_app_success(tv_app_process: subprocess.Popen, linux_tv_app_log_f
120129
while True:
121130
# Check if the time elapsed since the start wait time exceeds the maximum allowed startup time for the TV app.
122131
if time.time() - start_wait_time > TV_APP_MAX_START_WAIT_SEC:
123-
logging.error("The Linux tv-app process did not start successfully within the timeout.")
132+
logging.error('The Linux tv-app process did not start successfully within the timeout.')
124133
return False
125134

126135
tv_app_output_line = tv_app_process.stdout.readline()
@@ -134,7 +143,170 @@ def start_up_tv_app_success(tv_app_process: subprocess.Popen, linux_tv_app_log_f
134143
return True
135144

136145

137-
def parse_output_for_valid_commissioner(tv_casting_app_info: Tuple[subprocess.Popen, TextIO], log_paths: List[str]):
146+
def initiate_cast_request_success(tv_casting_app_info: Tuple[subprocess.Popen, TextIO], valid_discovered_commissioner_number: str) -> bool:
147+
"""Initiate commissioning between Linux tv-casting-app and tv-app by sending `cast request {valid_discovered_commissioner_number}` via Linux tv-casting-app process."""
148+
tv_casting_app_process, linux_tv_casting_app_log_file = tv_casting_app_info
149+
150+
start_wait_time = time.time()
151+
152+
while True:
153+
# Check if we exceeded the maximum wait time for initiating 'cast request' from the Linux tv-casting-app to the Linux tv-app.
154+
if time.time() - start_wait_time > COMMISSIONING_STAGE_MAX_WAIT_SEC:
155+
logging.error('The command `cast request ' + valid_discovered_commissioner_number +
156+
'` was not sent to the Linux tv-casting-app process within the timeout.')
157+
return False
158+
159+
tv_casting_app_output_line = tv_casting_app_process.stdout.readline()
160+
if tv_casting_app_output_line:
161+
linux_tv_casting_app_log_file.write(tv_casting_app_output_line)
162+
linux_tv_casting_app_log_file.flush()
163+
164+
if 'cast request 0' in tv_casting_app_output_line:
165+
tv_casting_app_process.stdin.write('cast request ' + valid_discovered_commissioner_number + '\n')
166+
tv_casting_app_process.stdin.flush()
167+
# Move to the next line otherwise we will keep entering this code block
168+
next_line = tv_casting_app_process.stdout.readline()
169+
linux_tv_casting_app_log_file.write(next_line)
170+
linux_tv_casting_app_log_file.flush()
171+
logging.info('Sent `' + next_line.rstrip('\n') + '` to the Linux tv-casting-app process.')
172+
return True
173+
174+
175+
def extract_device_info(tv_casting_app_info: Tuple[subprocess.Popen, TextIO]) -> Tuple[Optional[str], Optional[str], Optional[str]]:
176+
"""Extract device information from the 'Identification Declaration' block in the Linux tv-casting-app output."""
177+
tv_casting_app_process, linux_tv_casting_app_log_file = tv_casting_app_info
178+
179+
device_name = None
180+
vendor_id = None
181+
product_id = None
182+
183+
for line in tv_casting_app_process.stdout:
184+
linux_tv_casting_app_log_file.write(line)
185+
linux_tv_casting_app_log_file.flush()
186+
187+
if 'device Name' in line:
188+
device_name = extract_value_from_string(line)
189+
elif 'vendor id' in line:
190+
vendor_id = extract_value_from_string(line)
191+
elif 'product id' in line:
192+
product_id = extract_value_from_string(line)
193+
194+
if device_name and vendor_id and product_id:
195+
break
196+
197+
return device_name, vendor_id, product_id
198+
199+
200+
def validate_tv_app_device_info(tv_app_info, device_name, vendor_id, product_id, log_paths) -> bool:
201+
"""Validate device information from the 'Identification Declaration' block from the Linux tv-app output against the expected values."""
202+
tv_app_process, linux_tv_app_log_file = tv_app_info
203+
204+
parsing_identification_block = False
205+
start_wait_time = time.time()
206+
207+
while True:
208+
# Check if we exceeded the maximum wait time for validating the device information from the Linux tv-app to the corresponding values from the Linux tv-app.
209+
if time.time() - start_wait_time > COMMISSIONING_STAGE_MAX_WAIT_SEC:
210+
logging.erro('The device information from the Linux tv-app output was not validated against the corresponding values from the Linux tv-casting-app output within the timeout.')
211+
return False
212+
213+
tv_app_line = tv_app_process.stdout.readline()
214+
215+
if tv_app_line:
216+
linux_tv_app_log_file.write(tv_app_line)
217+
linux_tv_app_log_file.flush()
218+
219+
if 'Identification Declaration Start' in tv_app_line:
220+
logging.info('"Identification Declaration" block from the Linux tv-app output:')
221+
logging.info(tv_app_line.rstrip('\n'))
222+
parsing_identification_block = True
223+
elif parsing_identification_block:
224+
logging.info(tv_app_line.rstrip('\n'))
225+
if 'device Name' in tv_app_line:
226+
validate_value('Commissioning', device_name, log_paths, tv_app_line, 'device Name')
227+
elif 'vendor id' in tv_app_line:
228+
validate_value('Commissioning', vendor_id, log_paths, tv_app_line, 'vendor id')
229+
elif 'product id' in tv_app_line:
230+
validate_value('Commissioning', product_id, log_paths, tv_app_line, 'product id')
231+
elif 'Identification Declaration End' in tv_app_line:
232+
parsing_identification_block = False
233+
return True
234+
235+
236+
def approve_tv_casting_request(tv_app_info: Tuple[subprocess.Popen, TextIO], log_paths: List[str]) -> bool:
237+
"""Approve the TV casting request from the Linux tv-casting-app to the Linux tv-app by sending `controller ux ok` via Linux tv-app process."""
238+
tv_app_process, linux_tv_app_log_file = tv_app_info
239+
240+
start_wait_time = time.time()
241+
242+
while True:
243+
# Check if we exceeded the maximum wait time for sending 'controller ux ok' from the Linux tv-app to the Linux tv-casting-app.
244+
if time.time() - start_wait_time > COMMISSIONING_STAGE_MAX_WAIT_SEC:
245+
logging.error('The cast request from the Linux tv-casting-app to the Linux tv-app was not approved within the timeout.')
246+
return False
247+
248+
tv_app_line = tv_app_process.stdout.readline()
249+
250+
if tv_app_line:
251+
linux_tv_app_log_file.write(tv_app_line)
252+
linux_tv_app_log_file.flush()
253+
254+
if 'PROMPT USER: Test TV casting app is requesting permission to cast to this TV, approve?' in tv_app_line:
255+
logging.info(tv_app_line.rstrip('\n'))
256+
elif 'Via Shell Enter: controller ux ok|cancel' in tv_app_line:
257+
logging.info(tv_app_line.rstrip('\n'))
258+
259+
tv_app_process.stdin.write('controller ux ok\n')
260+
tv_app_process.stdin.flush()
261+
262+
tv_app_line = tv_app_process.stdout.readline()
263+
linux_tv_app_log_file.write(tv_app_line)
264+
linux_tv_app_log_file.flush()
265+
266+
logging.info('Sent `' + tv_app_line.rstrip('\n') + '` to the Linux tv-app process.')
267+
return True
268+
269+
270+
def validate_commissioning_success(tv_casting_app_info: Tuple[subprocess.Popen, TextIO], tv_app_info: Tuple[subprocess.Popen, TextIO], log_paths: List[str]) -> bool:
271+
"""Parse output of Linux tv-casting-app and Linux tv-app output for strings indicating commissioning status."""
272+
tv_casting_app_process, linux_tv_casting_app_log_file = tv_casting_app_info
273+
tv_app_process, linux_tv_app_log_file = tv_app_info
274+
275+
start_wait_time = time.time()
276+
277+
while True:
278+
# Check if we exceeded the maximum wait time for validating commissioning success between the Linux tv-casting-app and the Linux tv-app.
279+
if time.time() - start_wait_time > COMMISSIONING_STAGE_MAX_WAIT_SEC:
280+
logging.error('The commissioning between the Linux tv-casting-app process and the Linux tv-app process did not complete successfully within the timeout.')
281+
return False
282+
283+
tv_casting_line = tv_casting_app_process.stdout.readline()
284+
tv_app_line = tv_app_process.stdout.readline()
285+
286+
if tv_casting_line:
287+
linux_tv_casting_app_log_file.write(tv_casting_line)
288+
linux_tv_casting_app_log_file.flush()
289+
290+
if 'Commissioning completed successfully' in tv_casting_line:
291+
logging.info('Commissioning success noted on the Linux tv-casting-app output:')
292+
logging.info(tv_casting_line.rstrip('\n'))
293+
continue_reading_tv_casting_app_output = False
294+
elif 'Commissioning failed' in tv_casting_line:
295+
logging.error('Commissioning failed noted on the Linux tv-casting-app output:')
296+
logging.error(tv_casting_line.rstrip('\n'))
297+
return False
298+
299+
if tv_app_line:
300+
linux_tv_app_log_file.write(tv_app_line)
301+
linux_tv_app_log_file.flush()
302+
303+
if 'PROMPT USER: commissioning success' in tv_app_line:
304+
logging.info('Commissioning success noted on the Linux tv-app output:')
305+
logging.info(tv_app_line.rstrip('\n'))
306+
return True
307+
308+
309+
def test_discovery_fn(tv_casting_app_info: Tuple[subprocess.Popen, TextIO], log_paths: List[str]) -> Optional[str]:
138310
"""Parse the output of the Linux tv-casting-app to find a valid commissioner."""
139311
tv_casting_app_process, linux_tv_casting_app_log_file = tv_casting_app_info
140312

@@ -151,21 +323,21 @@ def parse_output_for_valid_commissioner(tv_casting_app_info: Tuple[subprocess.Po
151323
# Fail fast if "No commissioner discovered" string found.
152324
if "No commissioner discovered" in line:
153325
logging.error(line.rstrip('\n'))
154-
handle_discovery_failure(log_paths)
326+
handle_casting_state_failure('Discovery', log_paths)
155327

156328
elif "Discovered Commissioner" in line:
157329
valid_discovered_commissioner = line.rstrip('\n')
158330

159331
elif valid_discovered_commissioner:
160332
# Continue parsing the output for the information of interest under 'Discovered Commissioner'
161333
if 'Vendor ID:' in line:
162-
valid_vendor_id = validate_value(VENDOR_ID, log_paths, line, 'Vendor ID')
334+
valid_vendor_id = validate_value('Discovery', VENDOR_ID, log_paths, line, 'Vendor ID')
163335

164336
elif 'Product ID:' in line:
165-
valid_product_id = validate_value(PRODUCT_ID, log_paths, line, 'Product ID')
337+
valid_product_id = validate_value('Discovery', PRODUCT_ID, log_paths, line, 'Product ID')
166338

167339
elif 'Device Type:' in line:
168-
valid_device_type = validate_value(DEVICE_TYPE_CASTING_VIDEO_PLAYER, log_paths, line, 'Device Type')
340+
valid_device_type = validate_value('Discovery', DEVICE_TYPE_CASTING_VIDEO_PLAYER, log_paths, line, 'Device Type')
169341

170342
# A valid commissioner has VENDOR_ID, PRODUCT_ID, and DEVICE TYPE in its list of entries.
171343
if valid_vendor_id and valid_product_id and valid_device_type:
@@ -177,12 +349,33 @@ def parse_output_for_valid_commissioner(tv_casting_app_info: Tuple[subprocess.Po
177349
logging.info('Discovery success!')
178350
break
179351

352+
return valid_discovered_commissioner
353+
354+
355+
def test_commissioning_fn(valid_discovered_commissioner_number, tv_casting_app_info: Tuple[subprocess.Popen, TextIO], tv_app_info: Tuple[subprocess.Popen, TextIO], log_paths: List[str]):
356+
"""Test commissioning between Linux tv-casting-app and Linux tv-app."""
357+
358+
if not initiate_cast_request_success(tv_casting_app_info, valid_discovered_commissioner_number):
359+
handle_casting_state_failure('Commissioning', log_paths)
360+
361+
# Extract the values from the 'Identification Declaration' block in the tv-casting-app output that we want to validate against.
362+
device_name, vendor_id, product_id = extract_device_info(tv_casting_app_info)
363+
364+
if not validate_tv_app_device_info(tv_app_info, device_name, vendor_id, product_id, log_paths):
365+
handle_casting_state_failure('Commissioning', log_paths)
366+
367+
if not approve_tv_casting_request(tv_app_info, log_paths):
368+
handle_casting_state_failure('Commissioning', log_paths)
369+
370+
if not validate_commissioning_success(tv_casting_app_info, tv_app_info, log_paths):
371+
handle_casting_state_failure('Commissioning', log_paths)
372+
180373

181374
@click.command()
182375
@click.option('--tv-app-rel-path', type=str, default='out/tv-app/chip-tv-app', help='Path to the Linux tv-app executable.')
183376
@click.option('--tv-casting-app-rel-path', type=str, default='out/tv-casting-app/chip-tv-casting-app', help='Path to the Linux tv-casting-app executable.')
184-
def test_discovery_fn(tv_app_rel_path, tv_casting_app_rel_path):
185-
"""Test if the Linux tv-casting-app is able to discover the Linux tv-app.
377+
def test_casting_fn(tv_app_rel_path, tv_casting_app_rel_path):
378+
"""Test if the Linux tv-casting-app is able to discover and commission the Linux tv-app as part of casting.
186379
187380
Default paths for the executables are provided but can be overridden via command line arguments.
188381
For example: python3 run_tv_casting_test.py --tv-app-rel-path=path/to/tv-app
@@ -203,23 +396,34 @@ def test_discovery_fn(tv_app_rel_path, tv_casting_app_rel_path):
203396

204397
tv_app_abs_path = os.path.abspath(tv_app_rel_path)
205398
# Run the Linux tv-app subprocess.
206-
with ProcessManager(disable_stdout_buffering_cmd + [tv_app_abs_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE) as tv_app_process:
399+
with ProcessManager(disable_stdout_buffering_cmd + [tv_app_abs_path], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) as tv_app_process:
207400

208401
if not start_up_tv_app_success(tv_app_process, linux_tv_app_log_file):
209-
handle_discovery_failure([linux_tv_app_log_path])
402+
handle_casting_state_failure('Discovery', [linux_tv_app_log_path])
210403

211404
tv_casting_app_abs_path = os.path.abspath(tv_casting_app_rel_path)
212405
# Run the Linux tv-casting-app subprocess.
213-
with ProcessManager(disable_stdout_buffering_cmd + [tv_casting_app_abs_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE) as tv_casting_app_process:
406+
with ProcessManager(disable_stdout_buffering_cmd + [tv_casting_app_abs_path], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) as tv_casting_app_process:
214407
log_paths = [linux_tv_app_log_path, linux_tv_casting_app_log_path]
215408
tv_casting_app_info = (tv_casting_app_process, linux_tv_casting_app_log_file)
216-
parse_output_for_valid_commissioner(tv_casting_app_info, log_paths)
409+
tv_app_info = (tv_app_process, linux_tv_app_log_file)
410+
valid_discovered_commissioner = test_discovery_fn(tv_casting_app_info, log_paths)
411+
412+
if not valid_discovered_commissioner:
413+
handle_casting_state_failure('Discovery', log_paths)
414+
415+
# We need the valid discovered commissioner number to continue with commissioning.
416+
# Example string: \x1b[0;32m[1714582264602] [77989:2286038] [SVR] Discovered Commissioner #0\x1b[0m
417+
# The value '0' will be extracted from the string.
418+
valid_discovered_commissioner_number = valid_discovered_commissioner.split('#')[-1].replace('\x1b[0m', '')
419+
420+
test_commissioning_fn(valid_discovered_commissioner_number, tv_casting_app_info, tv_app_info, log_paths)
217421

218422

219423
if __name__ == '__main__':
220424

221425
# Start with a clean slate by removing any previously cached entries.
222426
os.system('rm -f /tmp/chip_*')
223427

224-
# Test discovery between the Linux tv-casting-app and the tv-app.
225-
test_discovery_fn()
428+
# Test casting (discovery and commissioning) between the Linux tv-casting-app and the tv-app.
429+
test_casting_fn()

0 commit comments

Comments
 (0)