diff --git a/Android/README.md b/Android/README.md
index f5f463ca116..bae9150ef05 100644
--- a/Android/README.md
+++ b/Android/README.md
@@ -25,7 +25,7 @@ you don't already have the SDK, here's how to install it:
 The `android.py` script also requires the following commands to be on the `PATH`:
 
 * `curl`
-* `java`
+* `java` (or set the `JAVA_HOME` environment variable)
 * `tar`
 * `unzip`
 
@@ -80,18 +80,54 @@ call. For example, if you want a pydebug build that also caches the results from
 
 ## Testing
 
-To run the Python test suite on Android:
+The tests can be run on Linux, macOS, or Windows, although on Windows you'll
+have to build the `cross-build/HOST` subdirectory on one of the other platforms
+and copy it over.
 
-* Install Android Studio, if you don't already have it.
-* Follow the instructions in the previous section to build all supported
-  architectures.
-* Run `./android.py setup-testbed` to download the Gradle wrapper.
-* Open the `testbed` directory in Android Studio.
-* In the *Device Manager* dock, connect a device or start an emulator.
-  Then select it from the drop-down list in the toolbar.
-* Click the "Run" button in the toolbar.
-* The testbed app displays nothing on screen while running. To see its output,
-  open the [Logcat window](https://developer.android.com/studio/debug/logcat).
+The test suite can usually be run on a device with 2 GB of RAM, though for some
+configurations or test orders you may need to increase this. As of Android
+Studio Koala, 2 GB is the default for all emulators, although the user interface
+may indicate otherwise. The effective setting is `hw.ramSize` in
+~/.android/avd/*.avd/hardware-qemu.ini, whereas Android Studio displays the
+value from config.ini. Changing the value in Android Studio will update both of
+these files.
 
-To run specific tests, or pass any other arguments to the test suite, edit the
-command line in testbed/app/src/main/python/main.py.
+Before running the test suite, follow the instructions in the previous section
+to build the architecture you want to test. Then run the test script in one of
+the following modes:
+
+* In `--connected` mode, it runs on a device or emulator you have already
+  connected to the build machine. List the available devices with
+  `$ANDROID_HOME/platform-tools/adb devices -l`, then pass a device ID to the
+  script like this:
+
+  ```sh
+  ./android.py test --connected emulator-5554
+  ```
+
+* In `--managed` mode, it uses a temporary headless emulator defined in the
+  `managedDevices` section of testbed/app/build.gradle.kts. This mode is slower,
+  but more reproducible.
+
+  We currently define two devices: `minVersion` and `maxVersion`, corresponding
+  to our minimum and maximum supported Android versions. For example:
+
+  ```sh
+  ./android.py test --managed maxVersion
+  ```
+
+By default, the only messages the script will show are Python's own stdout and
+stderr. Add the `-v` option to also show Gradle output, and non-Python logcat
+messages.
+
+Any other arguments on the `android.py test` command line will be passed through
+to `python -m test` – use `--` to separate them from android.py's own options.
+See the [Python Developer's
+Guide](https://devguide.python.org/testing/run-write-tests/) for common options
+– most of them will work on Android, except for those that involve subprocesses,
+such as `-j`.
+
+Every time you run `android.py test`, changes in pure-Python files in the
+repository's `Lib` directory will be picked up immediately. Changes in C files,
+and architecture-specific files such as sysconfigdata, will not take effect
+until you re-run `android.py make-host` or `build`.
diff --git a/Android/android-env.sh b/Android/android-env.sh
index 545d559d93a..93372e3fe1c 100644
--- a/Android/android-env.sh
+++ b/Android/android-env.sh
@@ -28,7 +28,7 @@ ndk_version=26.2.11394342
 
 ndk=$ANDROID_HOME/ndk/$ndk_version
 if ! [ -e $ndk ]; then
-    log "Installing NDK: this may take several minutes"
+    log "Installing NDK - this may take several minutes"
     yes | $ANDROID_HOME/cmdline-tools/latest/bin/sdkmanager "ndk;$ndk_version"
 fi
 
diff --git a/Android/android.py b/Android/android.py
index a78b15c9c4e..b5403b5d2a4 100755
--- a/Android/android.py
+++ b/Android/android.py
@@ -1,21 +1,51 @@
 #!/usr/bin/env python3
 
+import asyncio
 import argparse
 from glob import glob
 import os
 import re
+import shlex
 import shutil
+import signal
 import subprocess
 import sys
 import sysconfig
+from asyncio import wait_for
+from contextlib import asynccontextmanager
 from os.path import basename, relpath
 from pathlib import Path
+from subprocess import CalledProcessError
 from tempfile import TemporaryDirectory
 
+
 SCRIPT_NAME = Path(__file__).name
 CHECKOUT = Path(__file__).resolve().parent.parent
+ANDROID_DIR = CHECKOUT / "Android"
+TESTBED_DIR = ANDROID_DIR / "testbed"
 CROSS_BUILD_DIR = CHECKOUT / "cross-build"
 
+APP_ID = "org.python.testbed"
+DECODE_ARGS = ("UTF-8", "backslashreplace")
+
+
+try:
+    android_home = Path(os.environ['ANDROID_HOME'])
+except KeyError:
+    sys.exit("The ANDROID_HOME environment variable is required.")
+
+adb = Path(
+    f"{android_home}/platform-tools/adb"
+    + (".exe" if os.name == "nt" else "")
+)
+
+gradlew = Path(
+    f"{TESTBED_DIR}/gradlew"
+    + (".bat" if os.name == "nt" else "")
+)
+
+logcat_started = False
+
 
 def delete_glob(pattern):
     # Path.glob doesn't accept non-relative patterns.
@@ -42,10 +72,14 @@ def subdir(name, *, clean=None):
     return path
 
 
-def run(command, *, host=None, **kwargs):
-    env = os.environ.copy()
+def run(command, *, host=None, env=None, log=True, **kwargs):
+    kwargs.setdefault("check", True)
+    if env is None:
+        env = os.environ.copy()
+    original_env = env.copy()
+
     if host:
-        env_script = CHECKOUT / "Android/android-env.sh"
+        env_script = ANDROID_DIR / "android-env.sh"
         env_output = subprocess.run(
             f"set -eu; "
             f"HOST={host}; "
@@ -66,15 +100,13 @@ def run(command, *, host=None, **kwargs):
                     print(line)
                     env[key] = value
 
-        if env == os.environ:
+        if env == original_env:
             raise ValueError(f"Found no variables in {env_script.name} output:\n"
                              + env_output)
 
-    print(">", " ".join(map(str, command)))
-    try:
-        subprocess.run(command, check=True, env=env, **kwargs)
-    except subprocess.CalledProcessError as e:
-        sys.exit(e)
+    if log:
+        print(">", " ".join(map(str, command)))
+    return subprocess.run(command, env=env, **kwargs)
 
 
 def build_python_path():
@@ -180,31 +212,334 @@ def clean_all(context):
     delete_glob(CROSS_BUILD_DIR)
 
 
+def setup_sdk():
+    sdkmanager = android_home / (
+        "cmdline-tools/latest/bin/sdkmanager"
+        + (".bat" if os.name == "nt" else "")
+    )
+
+    # Gradle will fail if it needs to install an SDK package whose license
+    # hasn't been accepted, so pre-accept all licenses.
+    if not all((android_home / "licenses" / path).exists() for path in [
+        "android-sdk-arm-dbt-license", "android-sdk-license"
+    ]):
+        run([sdkmanager, "--licenses"], text=True, input="y\n" * 100)
+
+    # Gradle may install this automatically, but we can't rely on that because
+    # we need to run adb within the logcat task.
+    if not adb.exists():
+        run([sdkmanager, "platform-tools"])
+
+
 # To avoid distributing compiled artifacts without corresponding source code,
 # the Gradle wrapper is not included in the CPython repository. Instead, we
 # extract it from the Gradle release.
-def setup_testbed(context):
+def setup_testbed():
+    if all((TESTBED_DIR / path).exists() for path in [
+        "gradlew", "gradlew.bat", "gradle/wrapper/gradle-wrapper.jar",
+    ]):
+        return
+
     ver_long = "8.7.0"
     ver_short = ver_long.removesuffix(".0")
-    testbed_dir = CHECKOUT / "Android/testbed"
 
     for filename in ["gradlew", "gradlew.bat"]:
         out_path = download(
             f"https://raw.githubusercontent.com/gradle/gradle/v{ver_long}/{filename}",
-            testbed_dir)
+            TESTBED_DIR)
         os.chmod(out_path, 0o755)
 
     with TemporaryDirectory(prefix=SCRIPT_NAME) as temp_dir:
-        os.chdir(temp_dir)
         bin_zip = download(
-            f"https://services.gradle.org/distributions/gradle-{ver_short}-bin.zip")
+            f"https://services.gradle.org/distributions/gradle-{ver_short}-bin.zip",
+            temp_dir)
         outer_jar = f"gradle-{ver_short}/lib/plugins/gradle-wrapper-{ver_short}.jar"
-        run(["unzip", bin_zip, outer_jar])
-        run(["unzip", "-o", "-d", f"{testbed_dir}/gradle/wrapper", outer_jar,
-             "gradle-wrapper.jar"])
+        run(["unzip", "-d", temp_dir, bin_zip, outer_jar])
+        run(["unzip", "-o", "-d", f"{TESTBED_DIR}/gradle/wrapper",
+             f"{temp_dir}/{outer_jar}", "gradle-wrapper.jar"])
 
 
-def main():
+# run_testbed will build the app automatically, but it hides the Gradle output
+# by default, so it's useful to have this as a separate command for the buildbot.
+def build_testbed(context):
+    setup_sdk()
+    setup_testbed()
+    run(
+        [gradlew, "--console", "plain", "packageDebug", "packageDebugAndroidTest"],
+        cwd=TESTBED_DIR,
+    )
+
+
+# Work around a bug involving sys.exit and TaskGroups
+# (https://github.com/python/cpython/issues/101515).
+def exit(*args):
+    raise MySystemExit(*args)
+
+
+class MySystemExit(Exception):
+    pass
+
+
+# The `test` subcommand runs all subprocesses through this context manager so
+# that no matter what happens, they can always be cancelled from another task,
+# and they will always be cleaned up on exit.
+@asynccontextmanager
+async def async_process(*args, **kwargs):
+    process = await asyncio.create_subprocess_exec(*args, **kwargs)
+    try:
+        yield process
+    finally:
+        if process.returncode is None:
+            # Allow a reasonably long time for Gradle to clean itself up,
+            # because we don't want stale emulators left behind.
+            timeout = 10
+            process.terminate()
+            try:
+                await wait_for(process.wait(), timeout)
+            except TimeoutError:
+                print(
+                    f"Command {args} did not terminate after {timeout} seconds "
+                    f" - sending SIGKILL"
+                )
+                process.kill()
+
+                # Even after killing the process we must still wait for it,
+                # otherwise we'll get the warning "Exception ignored in __del__".
+                await wait_for(process.wait(), timeout=1)
+
+
+async def async_check_output(*args, **kwargs):
+    async with async_process(
+        *args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, **kwargs
+    ) as process:
+        stdout, stderr = await process.communicate()
+        if process.returncode == 0:
+            return stdout.decode(*DECODE_ARGS)
+        else:
+            raise CalledProcessError(
+                process.returncode, args,
+                stdout.decode(*DECODE_ARGS), stderr.decode(*DECODE_ARGS)
+            )
+
+
+# Return a list of the serial numbers of connected devices. Emulators will have
+# serials of the form "emulator-5678".
+async def list_devices():
+    serials = []
+    header_found = False
+
+    lines = (await async_check_output(adb, "devices")).splitlines()
+    for line in lines:
+        # Ignore blank lines, and all lines before the header.
+        line = line.strip()
+        if line == "List of devices attached":
+            header_found = True
+        elif header_found and line:
+            try:
+                serial, status = line.split()
+            except ValueError:
+                raise ValueError(f"failed to parse {line!r}")
+            if status == "device":
+                serials.append(serial)
+
+    if not header_found:
+        raise ValueError(f"failed to parse {lines}")
+    return serials
+
+
+async def find_device(context, initial_devices):
+    if context.managed:
+        print("Waiting for managed device - this may take several minutes")
+        while True:
+            new_devices = set(await list_devices()).difference(initial_devices)
+            if len(new_devices) == 0:
+                await asyncio.sleep(1)
+            elif len(new_devices) == 1:
+                serial = new_devices.pop()
+                print(f"Serial: {serial}")
+                return serial
+            else:
+                exit(f"Found more than one new device: {new_devices}")
+    else:
+        return context.connected
+
+
+# An older version of this script in #121595 filtered the logs by UID instead.
+# But logcat can't filter by UID until API level 31. If we ever switch back to
+# filtering by UID, we'll also have to filter by time so we only show messages
+# produced after the initial call to `stop_app`.
+#
+# We're more likely to miss the PID because it's shorter-lived, so there's a
+# workaround in PythonSuite.kt to stop it being *too* short-lived.
+async def find_pid(serial):
+    print("Waiting for app to start - this may take several minutes")
+    shown_error = False
+    while True:
+        try:
+            pid = (await async_check_output(
+                adb, "-s", serial, "shell", "pidof", "-s", APP_ID
+            )).strip()
+        except CalledProcessError as e:
+            # If the app isn't running yet, pidof gives no output. So if there
+            # is output, there must have been some other error. However, this
+            # sometimes happens transiently, especially when running a managed
+            # emulator for the first time, so don't make it fatal.
+            if (e.stdout or e.stderr) and not shown_error:
+                print_called_process_error(e)
+                print("This may be transient, so continuing to wait")
+                shown_error = True
+        else:
+            # Some older devices (e.g. Nexus 4) return zero even when no process
+            # was found, so check whether we actually got any output.
+            if pid:
+                print(f"PID: {pid}")
+                return pid
+
+        # Loop fairly rapidly to avoid missing a short-lived process.
+        await asyncio.sleep(0.2)
+
+
+async def logcat_task(context, initial_devices):
+    # Gradle may need to do some large downloads of libraries and emulator
+    # images. This will happen during find_device in --managed mode, or find_pid
+    # in --connected mode.
+    startup_timeout = 600
+    serial = await wait_for(find_device(context, initial_devices), startup_timeout)
+    pid = await wait_for(find_pid(serial), startup_timeout)
+
+    args = [adb, "-s", serial, "logcat", "--pid", pid,  "--format", "tag"]
+    hidden_output = []
+    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):
+                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.
+                level, message = None, line
+
+            # Put high-level messages on stderr so they're highlighted in the
+            # buildbot logs. This will include Python's own stderr.
+            stream = (
+                sys.stderr
+                if level in ["E", "F"]  # ERROR and FATAL (aka ASSERT)
+                else sys.stdout
+            )
+
+            # To simplify automated processing of the output, e.g. a buildbot
+            # posting a failure notice on a GitHub PR, we strip the level and
+            # 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
+                    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)
+
+        # 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.
+        status = await wait_for(process.wait(), timeout=1)
+        if status != 0 and not logcat_started:
+            raise CalledProcessError(status, args, "".join(hidden_output))
+
+
+def stop_app(serial):
+    run([adb, "-s", serial, "shell", "am", "force-stop", APP_ID], log=False)
+
+
+async def gradle_task(context):
+    env = os.environ.copy()
+    if context.managed:
+        task_prefix = context.managed
+    else:
+        task_prefix = "connected"
+        env["ANDROID_SERIAL"] = context.connected
+
+    args = [
+        gradlew, "--console", "plain", f"{task_prefix}DebugAndroidTest",
+        "-Pandroid.testInstrumentationRunnerArguments.pythonArgs="
+        + shlex.join(context.args),
+    ]
+    hidden_output = []
+    try:
+        async with async_process(
+            *args, cwd=TESTBED_DIR, env=env,
+            stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+        ) as process:
+            while line := (await process.stdout.readline()).decode(*DECODE_ARGS):
+                # 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)
+
+            status = await wait_for(process.wait(), timeout=1)
+            if status == 0:
+                exit(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)
+
+
+async def run_testbed(context):
+    setup_sdk()
+    setup_testbed()
+
+    if context.managed:
+        # In this mode, Gradle will create a device with an unpredictable name.
+        # So we save a list of the running devices before starting Gradle, and
+        # find_device then waits for a new device to appear.
+        initial_devices = await list_devices()
+    else:
+        # In case the previous shutdown was unclean, make sure the app isn't
+        # running, otherwise we might show logs from a previous run. This is
+        # unnecessary in --managed mode, because Gradle creates a new emulator
+        # every time.
+        stop_app(context.connected)
+        initial_devices = None
+
+    try:
+        async with asyncio.TaskGroup() as tg:
+            tg.create_task(logcat_task(context, initial_devices))
+            tg.create_task(gradle_task(context))
+    except* MySystemExit as e:
+        raise SystemExit(*e.exceptions[0].args) from None
+    except* CalledProcessError as e:
+        # Extract it from the ExceptionGroup so it can be handled by `main`.
+        raise e.exceptions[0]
+
+
+# Handle SIGTERM the same way as SIGINT. This ensures that if we're terminated
+# by the buildbot worker, we'll make an attempt to clean up our subprocesses.
+def install_signal_handler():
+    def signal_handler(*args):
+        os.kill(os.getpid(), signal.SIGINT)
+
+    signal.signal(signal.SIGTERM, signal_handler)
+
+
+def parse_args():
     parser = argparse.ArgumentParser()
     subcommands = parser.add_subparsers(dest="subcommand")
     build = subcommands.add_parser("build", help="Build everything")
@@ -219,8 +554,6 @@ def main():
                                        help="Run `make` for Android")
     subcommands.add_parser(
         "clean", help="Delete the cross-build directory")
-    subcommands.add_parser(
-        "setup-testbed", help="Download the testbed Gradle wrapper")
 
     for subcommand in build, configure_build, configure_host:
         subcommand.add_argument(
@@ -235,15 +568,66 @@ def main():
         subcommand.add_argument("args", nargs="*",
                                 help="Extra arguments to pass to `configure`")
 
-    context = parser.parse_args()
+    subcommands.add_parser(
+        "build-testbed", help="Build the testbed app")
+    test = subcommands.add_parser(
+        "test", help="Run the test suite")
+    test.add_argument(
+        "-v", "--verbose", action="store_true",
+        help="Show Gradle output, and non-Python logcat messages")
+    device_group = test.add_mutually_exclusive_group(required=True)
+    device_group.add_argument(
+        "--connected", metavar="SERIAL", help="Run on a connected device. "
+        "Connect it yourself, then get its serial from `adb devices`.")
+    device_group.add_argument(
+        "--managed", metavar="NAME", help="Run on a Gradle-managed device. "
+        "These are defined in `managedDevices` in testbed/app/build.gradle.kts.")
+    test.add_argument(
+        "args", nargs="*", help=f"Arguments for `python -m test`. "
+        f"Separate them from {SCRIPT_NAME}'s own arguments with `--`.")
+
+    return parser.parse_args()
+
+
+def main():
+    install_signal_handler()
+    context = parse_args()
     dispatch = {"configure-build": configure_build_python,
                 "make-build": make_build_python,
                 "configure-host": configure_host_python,
                 "make-host": make_host_python,
                 "build": build_all,
                 "clean": clean_all,
-                "setup-testbed": setup_testbed}
-    dispatch[context.subcommand](context)
+                "build-testbed": build_testbed,
+                "test": run_testbed}
+
+    try:
+        result = dispatch[context.subcommand](context)
+        if asyncio.iscoroutine(result):
+            asyncio.run(result)
+    except CalledProcessError as e:
+        print_called_process_error(e)
+        sys.exit(1)
+
+
+def print_called_process_error(e):
+    for stream_name in ["stdout", "stderr"]:
+        content = getattr(e, stream_name)
+        stream = getattr(sys, stream_name)
+        if content:
+            stream.write(content)
+            if not content.endswith("\n"):
+                stream.write("\n")
+
+    # Format the command so it can be copied into a shell. shlex uses single
+    # quotes, so we surround the whole command with double quotes.
+    args_joined = (
+        e.cmd if isinstance(e.cmd, str)
+        else " ".join(shlex.quote(str(arg)) for arg in e.cmd)
+    )
+    print(
+        f'Command "{args_joined}" returned exit status {e.returncode}'
+    )
 
 
 if __name__ == "__main__":
diff --git a/Android/testbed/app/build.gradle.kts b/Android/testbed/app/build.gradle.kts
index 7320b21e98b..7e0bef58ed8 100644
--- a/Android/testbed/app/build.gradle.kts
+++ b/Android/testbed/app/build.gradle.kts
@@ -1,17 +1,18 @@
 import com.android.build.api.variant.*
+import kotlin.math.max
 
 plugins {
     id("com.android.application")
     id("org.jetbrains.kotlin.android")
 }
 
-val PYTHON_DIR = File(projectDir, "../../..").canonicalPath
+val PYTHON_DIR = file("../../..").canonicalPath
 val PYTHON_CROSS_DIR = "$PYTHON_DIR/cross-build"
 
 val ABIS = mapOf(
     "arm64-v8a" to "aarch64-linux-android",
     "x86_64" to "x86_64-linux-android",
-).filter { File("$PYTHON_CROSS_DIR/${it.value}").exists() }
+).filter { file("$PYTHON_CROSS_DIR/${it.value}").exists() }
 if (ABIS.isEmpty()) {
     throw GradleException(
         "No Android ABIs found in $PYTHON_CROSS_DIR: see Android/README.md " +
@@ -19,7 +20,7 @@ if (ABIS.isEmpty()) {
     )
 }
 
-val PYTHON_VERSION = File("$PYTHON_DIR/Include/patchlevel.h").useLines {
+val PYTHON_VERSION = file("$PYTHON_DIR/Include/patchlevel.h").useLines {
     for (line in it) {
         val match = """#define PY_VERSION\s+"(\d+\.\d+)""".toRegex().find(line)
         if (match != null) {
@@ -29,6 +30,16 @@ val PYTHON_VERSION = File("$PYTHON_DIR/Include/patchlevel.h").useLines {
     throw GradleException("Failed to find Python version")
 }
 
+android.ndkVersion = file("../../android-env.sh").useLines {
+    for (line in it) {
+        val match = """ndk_version=(\S+)""".toRegex().find(line)
+        if (match != null) {
+            return@useLines match.groupValues[1]
+        }
+    }
+    throw GradleException("Failed to find NDK version")
+}
+
 
 android {
     namespace = "org.python.testbed"
@@ -45,6 +56,8 @@ android {
         externalNativeBuild.cmake.arguments(
             "-DPYTHON_CROSS_DIR=$PYTHON_CROSS_DIR",
             "-DPYTHON_VERSION=$PYTHON_VERSION")
+
+        testInstrumentationRunner = "androidx.test.runner.AndroidJUnitRunner"
     }
 
     externalNativeBuild.cmake {
@@ -62,41 +75,81 @@ android {
     kotlinOptions {
         jvmTarget = "1.8"
     }
+
+    testOptions {
+        managedDevices {
+            localDevices {
+                create("minVersion") {
+                    device = "Small Phone"
+
+                    // Managed devices have a minimum API level of 27.
+                    apiLevel = max(27, defaultConfig.minSdk!!)
+
+                    // ATD devices are smaller and faster, but have a minimum
+                    // API level of 30.
+                    systemImageSource = if (apiLevel >= 30) "aosp-atd" else "aosp"
+                }
+
+                create("maxVersion") {
+                    device = "Small Phone"
+                    apiLevel = defaultConfig.targetSdk!!
+                    systemImageSource = "aosp-atd"
+                }
+            }
+
+            // If the previous test run succeeded and nothing has changed,
+            // Gradle thinks there's no need to run it again. Override that.
+            afterEvaluate {
+                (localDevices.names + listOf("connected")).forEach {
+                    tasks.named("${it}DebugAndroidTest") {
+                        outputs.upToDateWhen { false }
+                    }
+                }
+            }
+        }
+    }
 }
 
 dependencies {
     implementation("androidx.appcompat:appcompat:1.6.1")
     implementation("com.google.android.material:material:1.11.0")
     implementation("androidx.constraintlayout:constraintlayout:2.1.4")
+    androidTestImplementation("androidx.test.ext:junit:1.1.5")
+    androidTestImplementation("androidx.test:rules:1.5.0")
 }
 
 
 // Create some custom tasks to copy Python and its standard library from
 // elsewhere in the repository.
 androidComponents.onVariants { variant ->
+    val pyPlusVer = "python$PYTHON_VERSION"
     generateTask(variant, variant.sources.assets!!) {
         into("python") {
-            for (triplet in ABIS.values) {
-                for (subDir in listOf("include", "lib")) {
-                    into(subDir) {
-                        from("$PYTHON_CROSS_DIR/$triplet/prefix/$subDir")
-                        include("python$PYTHON_VERSION/**")
-                        duplicatesStrategy = DuplicatesStrategy.EXCLUDE
-                    }
+            into("include/$pyPlusVer") {
+                for (triplet in ABIS.values) {
+                    from("$PYTHON_CROSS_DIR/$triplet/prefix/include/$pyPlusVer")
                 }
+                duplicatesStrategy = DuplicatesStrategy.EXCLUDE
             }
-            into("lib/python$PYTHON_VERSION") {
-                // Uncomment this to pick up edits from the source directory
-                // without having to rerun `make install`.
-                // from("$PYTHON_DIR/Lib")
-                // duplicatesStrategy = DuplicatesStrategy.INCLUDE
+
+            into("lib/$pyPlusVer") {
+                // To aid debugging, the source directory takes priority.
+                from("$PYTHON_DIR/Lib")
+
+                // The cross-build directory provides ABI-specific files such as
+                // sysconfigdata.
+                for (triplet in ABIS.values) {
+                    from("$PYTHON_CROSS_DIR/$triplet/prefix/lib/$pyPlusVer")
+                }
 
                 into("site-packages") {
                     from("$projectDir/src/main/python")
                 }
+
+                duplicatesStrategy = DuplicatesStrategy.EXCLUDE
+                exclude("**/__pycache__")
             }
         }
-        exclude("**/__pycache__")
     }
 
     generateTask(variant, variant.sources.jniLibs!!) {
diff --git a/Android/testbed/app/src/androidTest/java/org/python/testbed/PythonSuite.kt b/Android/testbed/app/src/androidTest/java/org/python/testbed/PythonSuite.kt
new file mode 100644
index 00000000000..0e888ab71d8
--- /dev/null
+++ b/Android/testbed/app/src/androidTest/java/org/python/testbed/PythonSuite.kt
@@ -0,0 +1,35 @@
+package org.python.testbed
+
+import androidx.test.annotation.UiThreadTest
+import androidx.test.platform.app.InstrumentationRegistry
+import androidx.test.ext.junit.runners.AndroidJUnit4
+
+import org.junit.Test
+import org.junit.runner.RunWith
+
+import org.junit.Assert.*
+
+
+@RunWith(AndroidJUnit4::class)
+class PythonSuite {
+    @Test
+    @UiThreadTest
+    fun testPython() {
+        val start = System.currentTimeMillis()
+        try {
+            val context =
+                InstrumentationRegistry.getInstrumentation().targetContext
+            val args =
+                InstrumentationRegistry.getArguments().getString("pythonArgs", "")
+            val status = PythonTestRunner(context).run(args)
+            assertEquals(0, status)
+        } finally {
+            // Make sure the process lives long enough for the test script to
+            // detect it (see `find_pid` in android.py).
+            val delay = 2000 - (System.currentTimeMillis() - start)
+            if (delay > 0) {
+                Thread.sleep(delay)
+            }
+        }
+    }
+}
diff --git a/Android/testbed/app/src/main/c/main_activity.c b/Android/testbed/app/src/main/c/main_activity.c
index 73aba4164d0..53470904899 100644
--- a/Android/testbed/app/src/main/c/main_activity.c
+++ b/Android/testbed/app/src/main/c/main_activity.c
@@ -84,7 +84,7 @@ static char *redirect_stream(StreamInfo *si) {
     return 0;
 }
 
-JNIEXPORT void JNICALL Java_org_python_testbed_MainActivity_redirectStdioToLogcat(
+JNIEXPORT void JNICALL Java_org_python_testbed_PythonTestRunner_redirectStdioToLogcat(
     JNIEnv *env, jobject obj
 ) {
     for (StreamInfo *si = STREAMS; si->file; si++) {
@@ -115,7 +115,7 @@ static void throw_status(JNIEnv *env, PyStatus status) {
     throw_runtime_exception(env, status.err_msg ? status.err_msg : "");
 }
 
-JNIEXPORT void JNICALL Java_org_python_testbed_MainActivity_runPython(
+JNIEXPORT int JNICALL Java_org_python_testbed_PythonTestRunner_runPython(
     JNIEnv *env, jobject obj, jstring home, jstring runModule
 ) {
     PyConfig config;
@@ -125,13 +125,13 @@ JNIEXPORT void JNICALL Java_org_python_testbed_MainActivity_runPython(
     status = set_config_string(env, &config, &config.home, home);
     if (PyStatus_Exception(status)) {
         throw_status(env, status);
-        return;
+        return 1;
     }
 
     status = set_config_string(env, &config, &config.run_module, runModule);
     if (PyStatus_Exception(status)) {
         throw_status(env, status);
-        return;
+        return 1;
     }
 
     // Some tests generate SIGPIPE and SIGXFSZ, which should be ignored.
@@ -140,8 +140,8 @@ JNIEXPORT void JNICALL Java_org_python_testbed_MainActivity_runPython(
     status = Py_InitializeFromConfig(&config);
     if (PyStatus_Exception(status)) {
         throw_status(env, status);
-        return;
+        return 1;
     }
 
-    Py_RunMain();
+    return Py_RunMain();
 }
diff --git a/Android/testbed/app/src/main/java/org/python/testbed/MainActivity.kt b/Android/testbed/app/src/main/java/org/python/testbed/MainActivity.kt
index 5a590d5d04e..c4bf6cbe83d 100644
--- a/Android/testbed/app/src/main/java/org/python/testbed/MainActivity.kt
+++ b/Android/testbed/app/src/main/java/org/python/testbed/MainActivity.kt
@@ -1,38 +1,56 @@
 package org.python.testbed
 
+import android.content.Context
 import android.os.*
 import android.system.Os
 import android.widget.TextView
 import androidx.appcompat.app.*
 import java.io.*
 
+
+// Launching the tests from an activity is OK for a quick check, but for
+// anything more complicated it'll be more convenient to use `android.py test`
+// to launch the tests via PythonSuite.
 class MainActivity : AppCompatActivity() {
     override fun onCreate(savedInstanceState: Bundle?) {
         super.onCreate(savedInstanceState)
         setContentView(R.layout.activity_main)
+        val status = PythonTestRunner(this).run("-W -uall")
+        findViewById<TextView>(R.id.tvHello).text = "Exit status $status"
+    }
+}
+
+
+class PythonTestRunner(val context: Context) {
+    /** @param args Extra arguments for `python -m test`.
+     * @return The Python exit status: zero if the tests passed, nonzero if
+     * they failed. */
+    fun run(args: String = "") : Int {
+        Os.setenv("PYTHON_ARGS", args, true)
 
         // Python needs this variable to help it find the temporary directory,
         // but Android only sets it on API level 33 and later.
-        Os.setenv("TMPDIR", cacheDir.toString(), false)
+        Os.setenv("TMPDIR", context.cacheDir.toString(), false)
 
         val pythonHome = extractAssets()
         System.loadLibrary("main_activity")
         redirectStdioToLogcat()
-        runPython(pythonHome.toString(), "main")
-        findViewById<TextView>(R.id.tvHello).text = "Python complete"
+
+        // The main module is in src/main/python/main.py.
+        return runPython(pythonHome.toString(), "main")
     }
 
     private fun extractAssets() : File {
-        val pythonHome = File(filesDir, "python")
+        val pythonHome = File(context.filesDir, "python")
         if (pythonHome.exists() && !pythonHome.deleteRecursively()) {
             throw RuntimeException("Failed to delete $pythonHome")
         }
-        extractAssetDir("python", filesDir)
+        extractAssetDir("python", context.filesDir)
         return pythonHome
     }
 
     private fun extractAssetDir(path: String, targetDir: File) {
-        val names = assets.list(path)
+        val names = context.assets.list(path)
             ?: throw RuntimeException("Failed to list $path")
         val targetSubdir = File(targetDir, path)
         if (!targetSubdir.mkdirs()) {
@@ -43,7 +61,7 @@ class MainActivity : AppCompatActivity() {
             val subPath = "$path/$name"
             val input: InputStream
             try {
-                input = assets.open(subPath)
+                input = context.assets.open(subPath)
             } catch (e: FileNotFoundException) {
                 extractAssetDir(subPath, targetDir)
                 continue
@@ -57,5 +75,5 @@ class MainActivity : AppCompatActivity() {
     }
 
     private external fun redirectStdioToLogcat()
-    private external fun runPython(home: String, runModule: String)
-}
\ No newline at end of file
+    private external fun runPython(home: String, runModule: String) : Int
+}
diff --git a/Android/testbed/app/src/main/python/main.py b/Android/testbed/app/src/main/python/main.py
index a1b6def34ed..c7314b500bf 100644
--- a/Android/testbed/app/src/main/python/main.py
+++ b/Android/testbed/app/src/main/python/main.py
@@ -1,4 +1,6 @@
+import os
 import runpy
+import shlex
 import signal
 import sys
 
@@ -8,10 +10,7 @@ import sys
 # profile save"), so disabling it should not weaken the tests.
 signal.pthread_sigmask(signal.SIG_UNBLOCK, [signal.SIGUSR1])
 
-# To run specific tests, or pass any other arguments to the test suite, edit
-# this command line.
-sys.argv[1:] = [
-    "--use", "all,-cpu",
-    "--verbose3",
-]
+sys.argv[1:] = shlex.split(os.environ["PYTHON_ARGS"])
+
+# The test module will call sys.exit to indicate whether the tests passed.
 runpy.run_module("test")
diff --git a/Android/testbed/build.gradle.kts b/Android/testbed/build.gradle.kts
index 53f4a67287f..2dad1501c24 100644
--- a/Android/testbed/build.gradle.kts
+++ b/Android/testbed/build.gradle.kts
@@ -1,5 +1,5 @@
 // Top-level build file where you can add configuration options common to all sub-projects/modules.
 plugins {
-    id("com.android.application") version "8.2.2" apply false
+    id("com.android.application") version "8.4.2" apply false
     id("org.jetbrains.kotlin.android") version "1.9.22" apply false
-}
\ No newline at end of file
+}
diff --git a/Android/testbed/gradle.properties b/Android/testbed/gradle.properties
index 3c5031eb7d6..e9f345c8c26 100644
--- a/Android/testbed/gradle.properties
+++ b/Android/testbed/gradle.properties
@@ -20,4 +20,9 @@ kotlin.code.style=official
 # Enables namespacing of each library's R class so that its R class includes only the
 # resources declared in the library itself and none from the library's dependencies,
 # thereby reducing the size of the R class for that library
-android.nonTransitiveRClass=true
\ No newline at end of file
+android.nonTransitiveRClass=true
+
+# By default, the app will be uninstalled after the tests finish (apparently
+# after 10 seconds in case of an unclean shutdown). We disable this, because
+# when using android.py it can conflict with the installation of the next run.
+android.injected.androidTest.leaveApksInstalledAfterRun=true
diff --git a/Android/testbed/gradle/wrapper/gradle-wrapper.properties b/Android/testbed/gradle/wrapper/gradle-wrapper.properties
index 2dc3339a3ef..57b2f57cc86 100644
--- a/Android/testbed/gradle/wrapper/gradle-wrapper.properties
+++ b/Android/testbed/gradle/wrapper/gradle-wrapper.properties
@@ -1,6 +1,6 @@
 #Mon Feb 19 20:29:06 GMT 2024
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-8.2-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.6-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
diff --git a/Lib/_android_support.py b/Lib/_android_support.py
index d5d13ec6a48..066e54708fb 100644
--- a/Lib/_android_support.py
+++ b/Lib/_android_support.py
@@ -31,15 +31,17 @@ def init_streams(android_log_write, stdout_prio, stderr_prio):
     logcat = Logcat(android_log_write)
 
     sys.stdout = TextLogStream(
-        stdout_prio, "python.stdout", errors=sys.stdout.errors)
+        stdout_prio, "python.stdout", sys.stdout.fileno(),
+        errors=sys.stdout.errors)
     sys.stderr = TextLogStream(
-        stderr_prio, "python.stderr", errors=sys.stderr.errors)
+        stderr_prio, "python.stderr", sys.stderr.fileno(),
+        errors=sys.stderr.errors)
 
 
 class TextLogStream(io.TextIOWrapper):
-    def __init__(self, prio, tag, **kwargs):
+    def __init__(self, prio, tag, fileno=None, **kwargs):
         kwargs.setdefault("encoding", "UTF-8")
-        super().__init__(BinaryLogStream(prio, tag), **kwargs)
+        super().__init__(BinaryLogStream(prio, tag, fileno), **kwargs)
         self._lock = RLock()
         self._pending_bytes = []
         self._pending_bytes_count = 0
@@ -98,9 +100,10 @@ class TextLogStream(io.TextIOWrapper):
 
 
 class BinaryLogStream(io.RawIOBase):
-    def __init__(self, prio, tag):
+    def __init__(self, prio, tag, fileno=None):
         self.prio = prio
         self.tag = tag
+        self._fileno = fileno
 
     def __repr__(self):
         return f"<BinaryLogStream {self.tag!r}>"
@@ -122,6 +125,12 @@ class BinaryLogStream(io.RawIOBase):
             logcat.write(self.prio, self.tag, b)
         return len(b)
 
+    # This is needed by the test suite --timeout option, which uses faulthandler.
+    def fileno(self):
+        if self._fileno is None:
+            raise io.UnsupportedOperation("fileno")
+        return self._fileno
+
 
 # When a large volume of data is written to logcat at once, e.g. when a test
 # module fails in --verbose3 mode, there's a risk of overflowing logcat's own
diff --git a/Lib/test/libregrtest/cmdline.py b/Lib/test/libregrtest/cmdline.py
index 2ff4715e82a..8bef04cba81 100644
--- a/Lib/test/libregrtest/cmdline.py
+++ b/Lib/test/libregrtest/cmdline.py
@@ -428,9 +428,7 @@ def _parse_args(args, **kwargs):
     # Continuous Integration (CI): common options for fast/slow CI modes
     if ns.slow_ci or ns.fast_ci:
         # Similar to options:
-        #
-        #     -j0 --randomize --fail-env-changed --fail-rerun --rerun
-        #     --slowest --verbose3
+        #   -j0 --randomize --fail-env-changed --rerun --slowest --verbose3
         if ns.use_mp is None:
             ns.use_mp = 0
         ns.randomize = True
diff --git a/Lib/test/test_android.py b/Lib/test/test_android.py
index 82035061bb6..09fa21cea87 100644
--- a/Lib/test/test_android.py
+++ b/Lib/test/test_android.py
@@ -19,6 +19,9 @@ if sys.platform != "android":
 
 api_level = platform.android_ver().api_level
 
+# (name, level, fileno)
+STREAM_INFO = [("stdout", "I", 1), ("stderr", "W", 2)]
+
 
 # Test redirection of stdout and stderr to the Android log.
 @unittest.skipIf(
@@ -94,19 +97,21 @@ class TestAndroidOutput(unittest.TestCase):
         stack = ExitStack()
         stack.enter_context(self.subTest(stream_name))
         stream = getattr(sys, stream_name)
+        native_stream = getattr(sys, f"__{stream_name}__")
         if isinstance(stream, io.StringIO):
             stack.enter_context(
                 patch(
                     f"sys.{stream_name}",
                     TextLogStream(
-                        prio, f"python.{stream_name}", errors="backslashreplace"
+                        prio, f"python.{stream_name}", native_stream.fileno(),
+                        errors="backslashreplace"
                     ),
                 )
             )
         return stack
 
     def test_str(self):
-        for stream_name, level in [("stdout", "I"), ("stderr", "W")]:
+        for stream_name, level, fileno in STREAM_INFO:
             with self.stream_context(stream_name, level):
                 stream = getattr(sys, stream_name)
                 tag = f"python.{stream_name}"
@@ -114,6 +119,7 @@ class TestAndroidOutput(unittest.TestCase):
 
                 self.assertIs(stream.writable(), True)
                 self.assertIs(stream.readable(), False)
+                self.assertEqual(stream.fileno(), fileno)
                 self.assertEqual("UTF-8", stream.encoding)
                 self.assertIs(stream.line_buffering, True)
                 self.assertIs(stream.write_through, False)
@@ -257,13 +263,14 @@ class TestAndroidOutput(unittest.TestCase):
                 write("\n", [s * 51])  # 0 bytes in, 510 bytes out
 
     def test_bytes(self):
-        for stream_name, level in [("stdout", "I"), ("stderr", "W")]:
+        for stream_name, level, fileno in STREAM_INFO:
             with self.stream_context(stream_name, level):
                 stream = getattr(sys, stream_name).buffer
                 tag = f"python.{stream_name}"
                 self.assertEqual(f"<BinaryLogStream '{tag}'>", repr(stream))
                 self.assertIs(stream.writable(), True)
                 self.assertIs(stream.readable(), False)
+                self.assertEqual(stream.fileno(), fileno)
 
                 def write(b, lines=None, *, write_len=None):
                     if write_len is None: