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
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
|
# mypy: allow-untyped-defs
import argparse
import json
import logging
import multiprocessing
import os
import sys
from tools import localpaths # noqa: F401
from . import virtualenv
here = os.path.dirname(__file__)
wpt_root = os.path.abspath(os.path.join(here, os.pardir, os.pardir))
def load_conditional_requirements(props, base_dir):
"""Load conditional requirements from commands.json."""
conditional_requirements = props.get("conditional_requirements")
if not conditional_requirements:
return {}
commandline_flag_requirements = {}
for key, value in conditional_requirements.items():
if key == "commandline_flag":
for flag_name, requirements_paths in value.items():
commandline_flag_requirements[flag_name] = [
os.path.join(base_dir, path) for path in requirements_paths]
else:
raise KeyError(
f'Unsupported conditional requirement key: {key}')
return {
"commandline_flag": commandline_flag_requirements,
}
def load_commands():
rv = {}
with open(os.path.join(here, "paths")) as f:
paths = [item.strip().replace("/", os.path.sep) for item in f if item.strip()]
for path in paths:
abs_path = os.path.join(wpt_root, path, "commands.json")
base_dir = os.path.dirname(abs_path)
with open(abs_path) as f:
data = json.load(f)
for command, props in data.items():
assert "path" in props
assert "script" in props
rv[command] = {
"path": os.path.join(base_dir, props["path"]),
"script": props["script"],
"parser": props.get("parser"),
"parse_known": props.get("parse_known", False),
"help": props.get("help"),
"virtualenv": props.get("virtualenv", True),
"requirements": [os.path.join(base_dir, item)
for item in props.get("requirements", [])]
}
rv[command]["conditional_requirements"] = load_conditional_requirements(
props, base_dir)
if rv[command]["requirements"] or rv[command]["conditional_requirements"]:
assert rv[command]["virtualenv"]
return rv
def parse_args(argv, commands=load_commands()):
parser = argparse.ArgumentParser()
parser.add_argument("--venv", action="store", help="Path to an existing virtualenv to use")
parser.add_argument("--skip-venv-setup", action="store_true",
dest="skip_venv_setup",
help="Whether to use the virtualenv as-is. Must set --venv as well")
parser.add_argument("--debug", action="store_true", help="Run the debugger in case of an exception")
subparsers = parser.add_subparsers(dest="command")
for command, props in commands.items():
subparsers.add_parser(command, help=props["help"], add_help=False)
if not argv:
parser.print_help()
return None, None
args, extra = parser.parse_known_args(argv)
return args, extra
def import_command(prog, command, props):
# This currently requires the path to be a module,
# which probably isn't ideal but it means that relative
# imports inside the script work
rel_path = os.path.relpath(props["path"], wpt_root)
parts = os.path.splitext(rel_path)[0].split(os.path.sep)
mod_name = ".".join(parts)
mod = __import__(mod_name)
for part in parts[1:]:
mod = getattr(mod, part)
script = getattr(mod, props["script"])
if props["parser"] is not None:
parser = getattr(mod, props["parser"])()
parser.prog = f"{os.path.basename(prog)} {command}"
else:
parser = None
return script, parser
def create_complete_parser():
"""Eagerly load all subparsers. This involves more work than is required
for typical command-line usage. It is maintained for the purposes of
documentation generation as implemented in WPT's top-level `/docs`
directory."""
commands = load_commands()
parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers()
# We should already be in a virtual environment from the top-level
# `wpt build-docs` command but we need to look up the environment to
# find out where it's located.
venv_path = os.environ["VIRTUAL_ENV"]
venv = virtualenv.Virtualenv(venv_path, True)
for command in commands:
props = commands[command]
try:
venv.install_requirements(*props.get("requirements", []))
except Exception:
logging.warning(
f"Unable to install requirements ({props['requirements']!r}) for command {command}"
)
continue
subparser = import_command('wpt', command, props)[1]
if not subparser:
continue
subparsers.add_parser(command,
help=props["help"],
add_help=False,
parents=[subparser])
return parser
def venv_dir():
return f"_venv{sys.version_info[0]}"
def setup_virtualenv(path, skip_venv_setup, props):
if skip_venv_setup and path is None:
raise ValueError("Must set --venv when --skip-venv-setup is used")
should_skip_setup = path is not None and skip_venv_setup
if path is None:
path = os.path.join(wpt_root, venv_dir())
venv = virtualenv.Virtualenv(path, should_skip_setup)
if not should_skip_setup:
venv.start()
venv.install_requirements(*props.get("requirements", []))
return venv
def install_command_flag_requirements(venv, props, kwargs):
requirements = props["conditional_requirements"].get("commandline_flag", {})
install_paths = []
for command_flag_name, requirement_paths in requirements.items():
if command_flag_name in kwargs:
install_paths.extend(requirement_paths)
venv.install_requirements(*install_paths)
def main(prog=None, argv=None):
logging.basicConfig(level=logging.INFO)
# Ensure we use the spawn start method for all multiprocessing
try:
multiprocessing.set_start_method('spawn')
except RuntimeError as e:
# This can happen if we call back into wpt having already set the context
start_method = multiprocessing.get_start_method()
if start_method != "spawn":
logging.critical("The multiprocessing start method was set to %s by a caller", start_method)
raise e
if prog is None:
prog = sys.argv[0]
if argv is None:
argv = sys.argv[1:]
commands = load_commands()
main_args, command_args = parse_args(argv, commands)
if not main_args:
return
command = main_args.command
props = commands[command]
venv = None
if props["virtualenv"]:
venv = setup_virtualenv(main_args.venv, main_args.skip_venv_setup, props)
script, parser = import_command(prog, command, props)
if parser:
if props["parse_known"]:
kwargs, extras = parser.parse_known_args(command_args)
extras = (extras,)
kwargs = vars(kwargs)
else:
extras = ()
kwargs = vars(parser.parse_args(command_args))
else:
extras = ()
kwargs = {}
if venv is not None:
if not main_args.skip_venv_setup:
install_command_flag_requirements(venv, props, kwargs)
args = (venv,) + extras
else:
args = extras
if script:
try:
rv = script(*args, **kwargs)
if rv is not None:
sys.exit(int(rv))
except Exception:
if main_args.debug:
import pdb
pdb.post_mortem()
else:
raise
sys.exit(0)
if __name__ == "__main__":
main() # type: ignore
|
