forked from chromium/chromium
-
Notifications
You must be signed in to change notification settings - Fork 0
/
run_test.py
executable file
·165 lines (142 loc) · 7.12 KB
/
run_test.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
#!/usr/bin/env vpython3
# Copyright 2022 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Implements commands for running tests E2E on a Fuchsia device."""
import argparse
import logging
import os
import sys
import tempfile
from contextlib import ExitStack
from typing import List
import monitors
from common import has_ffx_isolate_dir, is_daemon_running, \
register_common_args, register_device_args, \
register_log_args, resolve_packages
from compatible_utils import running_unattended
from ffx_integration import ScopedFfxConfig
from flash_device import register_update_args, update
from isolate_daemon import IsolateDaemon
from log_manager import LogManager, start_system_log
from publish_package import publish_packages, register_package_args
from run_blink_test import BlinkTestRunner
from run_executable_test import create_executable_test_runner, \
register_executable_test_args
from run_telemetry_test import TelemetryTestRunner
from run_webpage_test import WebpageTestRunner
from serve_repo import register_serve_args, serve_repository
from start_emulator import create_emulator_from_args, register_emulator_args
from test_connection import test_connection, test_device_connection
from test_runner import TestRunner
def _get_test_runner(runner_args: argparse.Namespace,
test_args: List[str]) -> TestRunner:
"""Initialize a suitable TestRunner class."""
if not runner_args.out_dir:
raise ValueError('--out-dir must be specified.')
if runner_args.test_type == 'blink':
return BlinkTestRunner(runner_args.out_dir, test_args,
runner_args.target_id)
if runner_args.test_type in ['gpu', 'perf']:
return TelemetryTestRunner(runner_args.test_type, runner_args.out_dir,
test_args, runner_args.target_id)
if runner_args.test_type == 'webpage':
return WebpageTestRunner(runner_args.out_dir, test_args,
runner_args.target_id, runner_args.logs_dir)
return create_executable_test_runner(runner_args, test_args)
# pylint: disable=too-many-statements
def main():
"""E2E method for installing packages and running a test."""
# Always add time stamps to the logs.
logging.basicConfig(format='%(levelname)s %(asctime)s %(message)s')
parser = argparse.ArgumentParser()
parser.add_argument(
'test_type',
help='The type of test to run. Options include \'blink\', \'gpu\', '
'or in the case of executable tests, the test name.')
parser.add_argument('--device',
'-d',
action='store_true',
default=False,
help='Use an existing device.')
parser.add_argument('--extra-path',
action='append',
help='Extra paths to append to the PATH environment')
# Register arguments
register_common_args(parser)
register_device_args(parser)
register_emulator_args(parser)
register_executable_test_args(parser)
register_update_args(parser, default_os_check='ignore')
register_log_args(parser)
register_package_args(parser, allow_temp_repo=True)
register_serve_args(parser)
# Treat unrecognized arguments as test specific arguments.
runner_args, test_args = parser.parse_known_args()
if runner_args.target_id:
runner_args.device = True
with ExitStack() as stack:
if runner_args.logs_dir:
# TODO(crbug.com/343242386): Find a way to upload metric output when
# logs_dir is not defined.
stack.push(lambda *_: monitors.dump(
os.path.join(runner_args.logs_dir, 'invocations')))
if runner_args.extra_path:
os.environ['PATH'] += os.pathsep + os.pathsep.join(
runner_args.extra_path)
if running_unattended():
# Only restart the daemon if 1) daemon will be run in a new isolate
# dir, or 2) if there isn't a daemon running in the predefined
# isolate dir.
if not has_ffx_isolate_dir() or not is_daemon_running():
stack.enter_context(IsolateDaemon(runner_args.logs_dir))
if runner_args.everlasting:
# Setting the emu.instance_dir to match the named cache, so
# we can keep these files across multiple runs.
# The configuration attaches to the daemon isolate-dir, so it
# needs to go after the IsolateDaemon.
# There isn't a point of enabling the feature on devbox, it
# won't use isolate-dir and the emu.instance_dir always goes to
# the HOME directory.
stack.enter_context(
ScopedFfxConfig(
'emu.instance_dir',
os.path.join(os.environ['HOME'],
'.fuchsia_emulator/')))
elif runner_args.logs_dir:
# Never restart daemon if not in the unattended mode.
logging.warning('You are using a --logs-dir, ensure the ffx '
'daemon is started with the logs.dir config '
'updated. We won\'t restart the daemon randomly'
' anymore.')
log_manager = LogManager(runner_args.logs_dir)
stack.enter_context(log_manager)
if runner_args.device:
update(runner_args.system_image_dir, runner_args.os_check,
runner_args.target_id, runner_args.serial_num)
# Try to reboot the device if necessary since the ffx may ignore the
# device state after the flash. See
# https://cs.opensource.google/fuchsia/fuchsia/+/main:src/developer/ffx/lib/fastboot/src/common/fastboot.rs;drc=cfba0bdd4f8857adb6409f8ae9e35af52c0da93e;l=454
test_device_connection(runner_args.target_id)
else:
runner_args.target_id = stack.enter_context(
create_emulator_from_args(runner_args))
test_connection(runner_args.target_id)
test_runner = _get_test_runner(runner_args, test_args)
package_deps = test_runner.package_deps
# Start system logging, after all possible restarts of the ffx daemon
# so that logging will not be interrupted.
start_system_log(log_manager, False, package_deps.values(),
('--since', 'now'), runner_args.target_id)
if package_deps:
if not runner_args.repo:
# Create a directory that serves as a temporary repository.
runner_args.repo = stack.enter_context(
tempfile.TemporaryDirectory())
publish_packages(package_deps.values(), runner_args.repo,
not runner_args.no_repo_init)
stack.enter_context(serve_repository(runner_args))
resolve_packages(package_deps.keys(), runner_args.target_id)
return test_runner.run_test().returncode
if __name__ == '__main__':
sys.exit(main())