+ (".bat" if os.name == "nt" else "")
)
-logcat_started = False
+# Whether we've seen any output from Python yet.
+python_started = False
+
+# Buffer for verbose output which will be displayed only if a test fails and
+# there has been no output from Python.
+hidden_output = []
+
+
+def log_verbose(context, line, stream=sys.stdout):
+ if context.verbose:
+ stream.write(line)
+ else:
+ hidden_output.append((stream, line))
def delete_glob(pattern):
env_script = ANDROID_DIR / "android-env.sh"
env_output = subprocess.run(
f"set -eu; "
- f"export HOST={host}; "
+ f"HOST={host}; "
f"PREFIX={prefix}; "
f". {env_script}; "
f"export",
# `--pid` requires API level 24 or higher.
args = [adb, "-s", serial, "logcat", "--pid", pid, "--format", "tag"]
- hidden_output = []
+ logcat_started = False
async with async_process(
*args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
) as process:
while line := (await process.stdout.readline()).decode(*DECODE_ARGS):
if match := re.fullmatch(r"([A-Z])/(.*)", line, re.DOTALL):
+ logcat_started = True
level, message = match.groups()
else:
- # If the regex doesn't match, this is probably the second or
- # subsequent line of a multi-line message. Python won't produce
- # such messages, but other components might.
+ # If the regex doesn't match, this is either a logcat startup
+ # error, or the second or subsequent line of a multi-line
+ # message. Python won't produce multi-line messages, but other
+ # components might.
level, message = None, line
# Exclude high-volume messages which are rarely useful.
# tag indicators from Python's stdout and stderr.
for prefix in ["python.stdout: ", "python.stderr: "]:
if message.startswith(prefix):
- global logcat_started
- logcat_started = True
+ global python_started
+ python_started = True
stream.write(message.removeprefix(prefix))
break
else:
- if context.verbose:
- # Non-Python messages add a lot of noise, but they may
- # sometimes help explain a failure.
- stream.write(line)
- else:
- hidden_output.append(line)
+ # Non-Python messages add a lot of noise, but they may
+ # sometimes help explain a failure.
+ log_verbose(context, line, stream)
# If the device disconnects while logcat is running, which always
# happens in --managed mode, some versions of adb return non-zero.
# Distinguish this from a logcat startup error by checking whether we've
- # received a message from Python yet.
+ # received any logcat messages yet.
status = await wait_for(process.wait(), timeout=1)
if status != 0 and not logcat_started:
- raise CalledProcessError(status, args, "".join(hidden_output))
+ raise CalledProcessError(status, args)
def stop_app(serial):
task_prefix = "connected"
env["ANDROID_SERIAL"] = context.connected
- hidden_output = []
-
- def log(line):
- # Gradle may take several minutes to install SDK packages, so it's worth
- # showing those messages even in non-verbose mode.
- if context.verbose or line.startswith('Preparing "Install'):
- sys.stdout.write(line)
- else:
- hidden_output.append(line)
-
if context.command:
mode = "-c"
module = context.command
]
if context.verbose >= 2:
args.append("--info")
- log("> " + join_command(args))
+ log_verbose(context, f"> {join_command(args)}\n")
try:
async with async_process(
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
) as process:
while line := (await process.stdout.readline()).decode(*DECODE_ARGS):
- log(line)
+ # Gradle may take several minutes to install SDK packages, so
+ # it's worth showing those messages even in non-verbose mode.
+ if line.startswith('Preparing "Install'):
+ sys.stdout.write(line)
+ else:
+ log_verbose(context, line)
status = await wait_for(process.wait(), timeout=1)
if status == 0:
else:
raise CalledProcessError(status, args)
finally:
- # If logcat never started, then something has gone badly wrong, so the
- # user probably wants to see the Gradle output even in non-verbose mode.
- if hidden_output and not logcat_started:
- sys.stdout.write("".join(hidden_output))
-
# Gradle does not stop the tests when interrupted.
if context.connected:
stop_app(context.connected)
except* MySystemExit as e:
raise SystemExit(*e.exceptions[0].args) from None
except* CalledProcessError as e:
+ # If Python produced no output, then the user probably wants to see the
+ # verbose output to explain why the test failed.
+ if not python_started:
+ for stream, line in hidden_output:
+ stream.write(line)
+
# Extract it from the ExceptionGroup so it can be handled by `main`.
raise e.exceptions[0]