| #!/usr/bin/python |
| # |
| """This script uses the UWP Device Portal API [1] to run unit tests on the Xbox. |
| |
| This module is designed to be integrated with the abstract_launcher framework. |
| It does the following when it is asked to run an app: |
| 1. Uninstall existing appx. |
| 2. Upload and install new application. |
| 3. Wait for the install to finish |
| 4. Run the application |
| 5. Wait for the application to finish running |
| 6. Fetch the content and write it to the appropriate place. |
| |
| When combined with starboard/tools/example/app_launcher_client.py, this module |
| enables changing target names (e.g. nplb), the Xbox used for testing, and |
| |
| Examples: |
| python starboard/tools/example/app_launcher_client.py --p xb1 -t nplb |
| -c debug -d 172.31.164.179 --target_params="--gtest_filter=*SbSocket*ait*" |
| |
| python starboard/tools/example/app_launcher_client.py --p xb1 -t nplb |
| -c debug -d 172.31.164.179 |
| --target_params="--gtest_filter=*Sb*Win*" |
| |
| Caveats: |
| Since we cannot launch write to a location any of the user directories like |
| LocalCache before the application is run for the first time, we have to bake |
| in the commandline arguments at install time. Thus, we have to reinstall |
| the package each time we want to run the binary with different arguments. |
| |
| Sample output (truncated for brevity): |
| Uninstalling App (if it exists). |
| Uninstall successfully started |
| Installing on NAVREETXB1 |
| Uploading 6 files in appx/...Done. |
| . |
| . |
| Uploading 80 files in appx/content/data/icu/icudt56l/zone...Done. |
| Skipping appx/microsoft.system.package.metadata since it has 0 files. |
| Done uploading. |
| Uploading args. [ --gtest_filter=*Sb*Win*; ] |
| Uploading 1 files in appx/content/data/arguments...Done. |
| Done uploading. |
| Done registering. |
| Installation is running.. |
| Installation complete. |
| Starting GoogleInc.YouTube::nplb |
| App started successfully. |
| nplb is running. |
| nplb is running. |
| Fetching log.. |
| Found 12 installed apps. |
| [1800:61781033300:INFO:application_uwp.cc(465)] Starting nplb.exe |
| Note: Google Test filter = *Sb*Win* |
| [==========] Running 7 tests from 4 test cases. |
| [----------] Global test environment set-up. |
| [----------] 2 tests from SbWindowCreateTest |
| [ RUN ] SbWindowCreateTest.SunnyDayDefault |
| [ OK ] SbWindowCreateTest.SunnyDayDefault (1 ms) |
| . |
| . |
| . |
| [----------] Global test environment tear-down |
| [==========] 7 tests from 4 test cases ran. (5644 ms total) |
| [ PASSED ] 7 tests. |
| Killing Process (if running). |
| Found 12 installed apps. |
| Log file found. Deleting. |
| |
| [1] |
| https://docs.microsoft.com/en-us/windows/uwp/debug-test-perf/device-portal-api-core |
| """ |
| |
| from __future__ import print_function |
| |
| import os |
| import subprocess |
| import sys |
| import time |
| |
| from starboard.shared.win32 import mini_dump_printer |
| from starboard.tools import abstract_launcher |
| from starboard.tools.abstract_launcher import TargetStatus |
| from starboard.tools import net_args |
| from starboard.tools import net_log |
| from starboard.xb1.tools import packager |
| from starboard.xb1.tools import xb1_network_api |
| |
| _ARGS_DIRECTORY = 'content/data/arguments' |
| _STARBOARD_ARGUMENTS_FILE = 'starboard_arguments.txt' |
| _DEFAULT_PACKAGE_NAME = 'GoogleInc.YouTube' |
| _STUB_PACKAGE_NAME = 'Microsoft.Title.StubApp' |
| _DEBUG_VC_LIBS_PACKAGE_NAME = 'Microsoft.VCLibs.140.00.Debug' |
| _DEFAULT_APPX_NAME = 'cobalt.appx' |
| _DEFAULT_STAGING_APP_NAME = 'appx' |
| _EXTENSION_SDK_DIR = os.path.realpath( |
| os.path.expandvars('%ProgramFiles(x86)%\\Microsoft SDKs' |
| '\\Windows Kits\\10\\ExtensionSDKs')) |
| _DEBUG_VC_LIBS_PATH = os.path.join(_EXTENSION_SDK_DIR, 'Microsoft.VCLibs', |
| '14.0', 'Appx', 'Debug', 'x64', |
| 'Microsoft.VCLibs.x64.Debug.14.00.appx') |
| _XB1_LOG_FILE_PARAM = 'xb1_log_file' |
| _XB1_PORT = 11443 |
| _XB1_NET_LOG_PORT = 49353 |
| _XB1_NET_ARG_PORT = 49355 |
| |
| # Number of times a test will try or retry. |
| _TEST_MAX_TRIES = 4 |
| # Seconds to wait between retries (scales with backoff factor). |
| _TEST_RETRY_WAIT = 8 |
| # Amount to multiply retry time with each failed attempt (i.e. 2 doubles the |
| # amount of time to wait between retries). |
| _TEST_RETRY_BACKOFF_FACTOR = 2 |
| |
| _PROCESS_TIMEOUT = 60 * 5.0 |
| _PROCESS_KILL_TIMEOUT_SECONDS = 5.0 |
| |
| _RESTART_TIMEOUT = 180 |
| |
| |
| def ToAppxFriendlyName(app_name): |
| if app_name == 'cobalt': |
| app_name = 'App' |
| else: |
| app_name = app_name.replace('_', '') + 'App' |
| return app_name |
| |
| |
| def GlobalVars(): |
| return GlobalVars.__dict__ |
| |
| |
| # First call returns True, otherwise return false. |
| def FirstRun(): |
| v = GlobalVars() |
| if 'first_run' in v: |
| return False |
| v['first_run'] = False |
| return True |
| |
| |
| def GetXb1NetworkApiFor(device_id, port): |
| v = GlobalVars() |
| network_key = str(device_id) + str(port) |
| if network_key not in v: |
| v[network_key] = xb1_network_api.Xb1NetworkApi(device_id, port) |
| return v[network_key] |
| |
| |
| def TryGetAppxDirectory(target_dir): |
| appx_path = os.path.join(target_dir, _DEFAULT_STAGING_APP_NAME) |
| if os.path.exists(os.path.join(appx_path, 'AppxManifest.xml')): |
| return appx_path |
| return None |
| |
| |
| def TryGetTargetPath(appx_path, target_name): |
| if not appx_path: |
| return None |
| target_path = os.path.join(appx_path, target_name + '.exe') |
| if os.path.exists(target_path): |
| return target_path |
| return None |
| |
| |
| def TryGetPackagedBinary(target_path): |
| path = os.path.join(target_path, 'package', _DEFAULT_APPX_NAME) |
| if os.path.exists(path): |
| return path |
| return None |
| |
| |
| def CheckEnvironment(): |
| v = GlobalVars() |
| if 'os_checked' not in v: |
| v['os_checked'] = True |
| path = os.path.abspath(__file__) |
| if path.startswith('/cygdrive'): |
| # os.path.abspath() does not work in cygwin, maybe other |
| # path operations don't work either. |
| raise OSError('\n **** cygwin not supported ****') |
| |
| |
| # Input: "--url=https://www.youtube.com/tv?v=1FmJdQiKy_w,--install=True" |
| # Output: {'url': 'https://www.youtube.com/tv?v=1FmJdQiKy_w', 'install': 'True'} |
| def CommandArgsToDict(s): |
| if not s: |
| return {} |
| |
| array = s.split(',') |
| out = {} |
| for v in array: |
| elems = v.split('=', 1) |
| if len(elems) < 1: |
| continue |
| key = elems[0] |
| |
| if key.startswith('--'): |
| key = key[2:] |
| val = None |
| if len(elems) > 1: |
| val = elems[1] |
| out[key] = val |
| return out |
| |
| |
| def Str2Bool(s): |
| if not s: |
| return False |
| s = s.lower() |
| return s in ('true', '1') |
| |
| |
| class Launcher(abstract_launcher.AbstractLauncher): |
| """Class for launching Cobalt/tools on Xbox.""" |
| |
| def __init__(self, platform, target_name, config, device_id, **kwargs): |
| """Launcher constructor. |
| |
| Args: |
| platform: Platform name (e.g. xb1, or xb1-future) |
| target_name: Target name (e.g. "nplb") |
| config: Config name (e.g. debug, devel, qa, gold) |
| device_id: IP address of the device (e.g. 172.2.3.4) |
| """ |
| super().__init__(platform, target_name, config, device_id, **kwargs) |
| |
| CheckEnvironment() |
| self._network_api = GetXb1NetworkApiFor(device_id, _XB1_PORT) |
| self._network_api.SetLoggingFunction(self._Log) |
| self._logfile_name = target_name + '_stdout.txt' |
| # Global argument will be visible to every launched application. |
| # --net_args_wait_for_connection will instruct the app to wait |
| # until net_args can connect via a socket and receive the command |
| # arguments. |
| # --net_log_wait_for_connection will instruct the app to wait until |
| # the netlog attaches. |
| self._global_args = (';--net_args_wait_for_connection' |
| ';--net_log_wait_for_connection') |
| # Once net_args is connected, the following _target_args will be sent |
| # over the wire. |
| self._target_args = ['--' + _XB1_LOG_FILE_PARAM + '=' + self._logfile_name] |
| self._do_deploy = True |
| self._do_restart = True |
| self._do_run = True |
| self._device_id = device_id |
| self._platform = platform |
| self._config = config |
| |
| # Note that target_params is an array of strings, not a string. |
| user_params = kwargs.get('target_params', '') |
| if user_params and len(user_params) > 0: |
| self.InitTargetParams(','.join(user_params)) |
| |
| self._package_root_dir = os.path.split(self.GetTargetPath())[0] |
| |
| # Ensure that target is built and ready to be packaged. |
| self._source_appx_path = TryGetAppxDirectory(self._package_root_dir) |
| self._source_exe_path = TryGetTargetPath(self._source_appx_path, |
| target_name) |
| if not self._source_exe_path: |
| raise IOError(f'Target {target_name} not found in appx directory.') |
| |
| def InitTargetParams(self, user_params_str): |
| args_dict = CommandArgsToDict(user_params_str) |
| # If --restart is present, consume the field |
| self._do_restart = Str2Bool(args_dict.pop('restart', 'True')) |
| |
| # If --deploy is present, then consume the value and |
| # remove from the rest of the arguments. |
| if 'deploy' in args_dict: |
| val = args_dict['deploy'] |
| del args_dict['deploy'] |
| if val is not None: |
| self._do_deploy = Str2Bool(val) |
| |
| # If --run is present, then consume the value and |
| # remove from the rest of the arguments. |
| if 'run' in args_dict: |
| val = args_dict['run'] |
| del args_dict['run'] |
| if val is not None: |
| self._do_run = Str2Bool(val) |
| # If there are still arguments then add them to the target |
| # args. |
| if len(args_dict) > 0: |
| args = [] |
| for key, val in args_dict.items(): |
| entry = '--' + key |
| if val is not None: |
| entry += '=' + val |
| args.append(entry) |
| self._target_args.extend(args) |
| |
| def _Log(self, s): |
| try: |
| self.output_file.write(s) |
| except UnicodeEncodeError: |
| s = s.encode('ascii', 'replace').decode() |
| self.output_file.write(s) |
| self.output_file.flush() |
| |
| def _LogLn(self, s): |
| try: |
| # Attempt to convert to a string if the type is bytes. |
| s = s.decode() |
| except (UnicodeDecodeError, AttributeError): |
| pass |
| self._Log(s + '\n') |
| |
| def InstallStarboardArgument(self, global_args): |
| args_path = [self._source_appx_path] |
| args_path.extend(_ARGS_DIRECTORY.split('/')) |
| args_path.append(_STARBOARD_ARGUMENTS_FILE) |
| args_path = os.path.join(*args_path) # splat args_path into flat list. |
| |
| args_dir_path = os.path.dirname(args_path) |
| if not os.path.exists(args_dir_path): |
| os.makedirs(args_dir_path) |
| |
| if not global_args: |
| if os.path.exists(args_path): |
| os.remove(args_path) |
| self._LogLn('Removed global command argument file: ' + args_path) |
| else: |
| with open(args_path, 'w', encoding='utf-8') as fd: |
| fd.write(global_args) |
| self._LogLn('Installed global command arguments: \"' + global_args + |
| '\" to:\n' + os.path.realpath(args_path)) |
| |
| def GetSerial(self): |
| try: |
| poll = self._network_api.GetDeviceInfo() |
| return poll['SerialNumber'] |
| except IOError: |
| return '' |
| |
| def RestartDevkit(self): |
| dev_info = self._network_api.GetDeviceInfo() |
| serial = dev_info['SerialNumber'] |
| console_type = dev_info['ConsoleType'] |
| self._Log('Connected to: ') |
| self._LogLn(self._network_api.ComputerName() + ' Type:' + console_type + |
| ' Serial:' + serial + ', triggering restart') |
| start = time.time() |
| self._network_api.Restart() |
| |
| while self.GetSerial(): # Wait for at least one failed poll |
| time.sleep(1) |
| |
| while self.GetSerial() != serial: |
| elapsed = int(time.time() - start) |
| if elapsed > _RESTART_TIMEOUT: |
| raise RuntimeError('Timed out waiting for restart, ' |
| f'timeout:{_RESTART_TIMEOUT}') |
| self._LogLn(f'Waiting for restart, elapsed seconds:{elapsed} ' |
| f'( timeout: {_RESTART_TIMEOUT} )') |
| time.sleep(1) |
| |
| time.sleep(10) |
| self._Log('Restart complete: ') |
| self._LogLn(self._network_api.ComputerName()) |
| |
| def SignIn(self): |
| self._Log('Connected to: ') |
| self._LogLn(self._network_api.ComputerName()) |
| |
| users = self._network_api.GetXboxLiveUserInfos().get('Users', []) |
| |
| if len(users) == 0: |
| raise IOError('Please add at least one user to the test accounts on\n' + |
| 'the Xbox developer homescreen.') |
| |
| # If we find an existing user that is signed in, then use that. |
| for user in users: |
| if user.get('SignedIn', False): |
| return |
| |
| # Else, sign in the first user. |
| self._network_api.SetXboxLiveSignedInUserState(users[0]['EmailAddress'], |
| True) |
| |
| def WinAppDeployCmd(self, command: str): |
| try: |
| exe_path = os.path.join(packager.GetWinToolsPath(), 'WinAppDeployCmd.exe') |
| command_str = f'{exe_path} {command} -ip {self.GetDeviceIp()}' |
| self._LogLn('Running: ' + command_str) |
| out = subprocess.check_output(command_str).decode() |
| except subprocess.CalledProcessError as e: |
| self._LogLn(e.output) |
| raise e |
| |
| return out |
| |
| def PackageCobalt(self): |
| # Package all targets into an appx. |
| package_parameters = { |
| 'source_dir': self.out_directory, |
| 'output_dir': os.path.join(self.out_directory, 'package'), |
| 'publisher': None, |
| 'product': 'youtube', |
| } |
| if not os.path.exists(package_parameters['output_dir']): |
| os.makedirs(package_parameters['output_dir']) |
| packager.Package(**package_parameters) |
| |
| def UninstallSubPackages(self): |
| # Check for sub-packages left over. Force uninstall any that exist. |
| uninstalled_packages = [] |
| packages = self._network_api.GetInstalledPackages() |
| for package in packages: |
| try: |
| package_full_name = package['PackageFullName'] |
| if package_full_name.find( |
| _DEFAULT_PACKAGE_NAME) != -1 or package_full_name.find( |
| _STUB_PACKAGE_NAME) != -1: |
| if package_full_name not in uninstalled_packages: |
| self._LogLn('Existing YouTube app found on device. Uninstalling: ' + |
| package_full_name) |
| uninstalled_packages.append(package_full_name) |
| self.WinAppDeployCmd('uninstall -package ' + package_full_name) |
| except KeyError: |
| # Some packages don't include all fields. Ignore those. |
| pass |
| except subprocess.CalledProcessError as err: |
| self._LogLn(err.output) |
| |
| def DeleteLooseApps(self): |
| self._network_api.ClearLooseAppFiles() |
| |
| def Deploy(self): |
| # starboard_arguments.txt is packaged with the appx. It instructs the app |
| # to wait for the NetArgs thread to send command-line args via the socket. |
| self.InstallStarboardArgument(self._global_args) |
| |
| self.PackageCobalt() |
| appx_package_file = TryGetPackagedBinary(self._package_root_dir) |
| if not appx_package_file: |
| raise IOError('Packaged appx not found in package directory. Perhaps ' |
| 'package_cobalt script did not complete successfully.') |
| |
| existing_package = self.CheckPackageIsDeployed(_DEFAULT_PACKAGE_NAME) |
| if existing_package: |
| self._LogLn('Existing YouTube app found on device. Uninstalling.') |
| self.WinAppDeployCmd('uninstall -package ' + existing_package) |
| |
| if not self.CheckPackageIsDeployed(_DEBUG_VC_LIBS_PACKAGE_NAME): |
| self._LogLn('Required dependency missing. Attempting to install.') |
| self.WinAppDeployCmd(f'install -file "{_DEBUG_VC_LIBS_PATH}"') |
| |
| self._LogLn('Deleting temporary files') |
| self._network_api.ClearTempFiles() |
| |
| try: |
| self.WinAppDeployCmd(f'install -file {appx_package_file}') |
| except subprocess.CalledProcessError: |
| # Install exited with non-zero status code, clear everything out, restart, |
| # and attempt another install. |
| self._LogLn('Error installing appx. Attempting a clean install...') |
| self.UninstallSubPackages() |
| self.DeleteLooseApps() |
| self.RestartDevkit() |
| self.WinAppDeployCmd(f'install -file {appx_package_file}') |
| |
| # Cleanup starboard arguments file. |
| self.InstallStarboardArgument(None) |
| |
| # Validate that app was installed correctly by checking to make sure |
| # that the full package name can now be found. |
| def CheckPackageIsDeployed(self, package_name=_DEFAULT_PACKAGE_NAME): |
| package_list = self.WinAppDeployCmd('list') |
| package_index = package_list.find(package_name) |
| if package_index == -1: |
| return False |
| return package_list[package_index:].split('\n')[0].strip() |
| |
| def RunTest(self, appx_name: str): |
| self.net_args_thread = None |
| attempt_num = 0 |
| retry_wait_s = _TEST_RETRY_WAIT |
| while attempt_num < _TEST_MAX_TRIES: |
| if not self.net_args_thread or not self.net_args_thread.is_alive(): |
| # This thread must start before the app executes or else it is possible |
| # the app will hang at _network_api.ExecuteBinary() |
| self.net_args_thread = net_args.NetArgsThread(self.device_id, |
| _XB1_NET_ARG_PORT, |
| self._target_args) |
| if self._network_api.ExecuteBinary(_DEFAULT_PACKAGE_NAME, appx_name): |
| break |
| |
| if not self.net_args_thread.ArgsSent(): |
| self._LogLn( |
| 'Net Args were not sent to the test! This will likely cause ' |
| 'the test to fail!') |
| attempt_num += 1 |
| self._LogLn(f'Retry attempt {attempt_num}.') |
| time.sleep(retry_wait_s) |
| retry_wait_s *= _TEST_RETRY_BACKOFF_FACTOR |
| if hasattr(self, 'net_args_thread'): |
| self.net_args_thread.join() |
| |
| def InitDevice(self): |
| if not self._network_api.IsInDevMode(): |
| raise IOError('\n\n**** Please set the XBOX at ' + self._device_id + |
| ' to dev mode!!!! ****\n') |
| self.SignIn() |
| |
| def Run(self): |
| # Only upload and install Appx on the first run. |
| if FirstRun(): |
| if self._do_restart: |
| self.RestartDevkit() |
| |
| self.InitDevice() |
| if self._do_deploy: |
| self.Deploy() |
| else: |
| self._LogLn('Skipping deploy step.') |
| |
| if not self.CheckPackageIsDeployed(_DEFAULT_PACKAGE_NAME): |
| raise IOError('Could not resolve ' + _DEFAULT_PACKAGE_NAME + ' to\n' + |
| 'it\'s full package name after install! This means that' + |
| '\n the package is not deployed correctly!\n\n') |
| |
| if not self._do_run: |
| self._LogLn('Skipping running step.') |
| return 0 |
| |
| status, _ = self.Run2() |
| if status == TargetStatus.CRASH: |
| return 1 |
| else: |
| return 0 |
| |
| def Run2(self): |
| try: |
| self.Kill() # Kill existing running app. |
| # While binary is running, extract the net log and stream it to |
| # the output. |
| self.net_log_thread = net_log.NetLogThread(self.device_id, |
| _XB1_NET_LOG_PORT) |
| |
| appx_name = ToAppxFriendlyName(self.target_name) |
| |
| self.RunTest(appx_name) |
| |
| while self._network_api.IsBinaryRunning(self.target_name): |
| self._Log(self.net_log_thread.GetLog()) |
| |
| self._Log(self.net_log_thread.GetLog()) |
| self._LogLn('Program Exited...') |
| |
| except KeyboardInterrupt: |
| self._LogLn('User cancelled...') |
| except IOError as io_err: |
| _, _, exc_tb = sys.exc_info() |
| fname = os.path.split(exc_tb.tb_frame.f_code.co_filename)[1] |
| msg = (f'Exception happened at {fname}({exc_tb.tb_lineno}) during ' |
| f'to run cycle of {self.target_name} this ' |
| 'is common if no one is signed in to the xbox.\n' |
| f' Error: {repr(io_err)}') |
| raise IOError(msg) from io_err |
| finally: |
| if hasattr(self, 'net_args_thread'): |
| self.net_args_thread.join() |
| if hasattr(self, 'net_log_thread'): |
| self.net_log_thread.join() |
| |
| self.Kill() |
| self._LogLn('Finished running...') |
| if self._DetectAndHandleAnyCrashes(self.target_name): |
| return TargetStatus.CRASH, 0 |
| else: |
| return TargetStatus.NA, 0 |
| |
| def _DetectAndHandleAnyCrashes(self, target_name): |
| crashes_detected = False |
| (files, |
| _) = self._network_api.ListPackageFilesAndDirs(_DEFAULT_PACKAGE_NAME, 'AC') |
| for f in files: |
| if f.startswith(target_name) and f.endswith('dmp'): |
| self._LogLn('\n***** Application ' + target_name + |
| ' crashed! *****\nDump file: ' + f) |
| crashes_detected = True |
| data = self._network_api.FetchPackageFile( |
| _DEFAULT_PACKAGE_NAME, |
| 'AC/' + f, |
| raise_on_failure=True, |
| is_text=False) |
| # Ensure that directory exists. |
| dump_dir_path = os.path.join(os.environ['LOCALAPPDATA'], 'CrashDumps', |
| 'xbox_remote') |
| if not os.path.exists(dump_dir_path): |
| os.makedirs(dump_dir_path) |
| |
| dump_file_path = os.path.join(dump_dir_path, f) |
| # Clean previous dump file, if it exists. |
| if os.path.exists(dump_file_path): |
| try: |
| os.remove(dump_file_path) |
| except IOError as io_err: |
| self._LogLn('Error, could not remove old file ' + str(io_err)) |
| |
| try: |
| with open(dump_file_path, 'wb') as fp: |
| fp.write(data) |
| self._LogLn('Copied dump to ' + dump_file_path) |
| mini_dump_printer.PrintMiniDump(dump_file_path, self.output_file) |
| except Exception as err: # pylint: disable=broad-except |
| self._LogLn('Failed to dump file ' + dump_file_path + ' because: ' + |
| str(err)) |
| return crashes_detected |
| |
| def Kill(self): |
| self._Log('Killing Process (if running).\n') |
| running_processes = self._network_api.GetRunningProcesses() |
| exe_name = self.target_name + '.exe' |
| filtered_processes = list( |
| filter(lambda p: p['ImageName'] == exe_name, running_processes)) |
| if len(filtered_processes) == 0: |
| return |
| process_under_test = filtered_processes[0] |
| if not process_under_test['IsRunning']: |
| return |
| pid = process_under_test['ProcessId'] |
| self._Log('Going to kill pid ' + str(pid) + '\n') |
| |
| self._network_api.KillProcess(pid) |
| self._Log('Kill signal sent. Waiting for the process to die.\n') |
| self._network_api.WaitForBinaryToFinishRunning( |
| self.target_name, _PROCESS_KILL_TIMEOUT_SECONDS) |
| self._Log('Done killing.\n') |
| |
| def GetDeviceIp(self): |
| """Gets the device IP.""" |
| return self.device_id |
| |
| def GetDeviceOutputPath(self): |
| # TODO: Implement |
| return None |
