diff options
Diffstat (limited to 'pytest/test_embedded/tests')
10 files changed, 310 insertions, 82 deletions
diff --git a/pytest/test_embedded/tests/conftest.py b/pytest/test_embedded/tests/conftest.py index defd9f53..35beffb8 100644 --- a/pytest/test_embedded/tests/conftest.py +++ b/pytest/test_embedded/tests/conftest.py @@ -50,6 +50,8 @@ AOSP_ROOT = Path(os.path.dirname(__file__)).absolute().parents[4] SDK_EMULATOR = ( AOSP_ROOT / "prebuilts" / "android-emulator-build" / "system-images" / OS_NAME ) +# Path to all the gRPC services +GRPC_SERVICES = AOSP_ROOT / "external" / "qemu" / "android" / "android-grpc" def pytest_addoption(parser): @@ -256,7 +258,11 @@ def pytest_sessionfinish( def get_crash_reporter(pytestconfig): exe = pytestconfig.getoption("emulator") emulator_directory = Path(exe).parent if exe else None - return CrashReporter(emulator_directory, pytestconfig.getoption("symbols")) + return CrashReporter( + emulator_directory, + pytestconfig.getoption("symbols"), + GRPC_SERVICES, + ) @pytest.fixture(autouse=True) @@ -292,10 +298,13 @@ async def crash_reporter(pytestconfig): if log_file and Path(log_file).exists(): log_dir = Path(log_file).parent await crash_report.write_reports_to_disk(log_dir) + else: + await crash_report.list_crashes() await crash_report.clear() logging.info("=== completed crash reporter") + @pytest.fixture(scope="module") @pytest.mark.async_timeout(200) async def emulator(request, pytestconfig) -> BaseEmulator: @@ -347,7 +356,10 @@ async def emulator(request, pytestconfig) -> BaseEmulator: """ avd_configs = json.loads(pytestconfig.getoption("avd_configs")) - return await manage_emulator(request, pytestconfig, avd_configs[0] if avd_configs else {}) + return await manage_emulator( + request, pytestconfig, avd_configs[0] if avd_configs else {} + ) + @pytest.fixture(scope="module") @pytest.mark.async_timeout(200) @@ -380,7 +392,7 @@ async def manage_emulator(request, pytestconfig, avd_param_config) -> BaseEmulat avd_pram_config: avd_config of the emulator specified by cfg files Returns: - BaseEmulator: A successfully booted emulator with the debug apk installed. + BaseEmulator: A successfully booted emulator with the debug apk installed. """ avd_config = { @@ -395,7 +407,7 @@ async def manage_emulator(request, pytestconfig, avd_param_config) -> BaseEmulat avd_config.update(avd_param_config) name = f"{avd_config['api']}_{avd_config['tag.id']}_{avd_config['cpu']}_{avd_config['device.name']}" - if 'AvdId' in avd_config: + if "AvdId" in avd_config: name += f"_{avd_config['AvdId']}" logging.info("--> Setting up emulator using avd config:%s", avd_config) @@ -419,7 +431,7 @@ async def manage_emulator(request, pytestconfig, avd_param_config) -> BaseEmulat ) emu.symbols = pytestconfig.getoption("symbols") - emu.launch_flags = avd_config['launch_flags'] + emu.launch_flags = avd_config.get("launch_flags", []) pytest.emulators[name] = emu emu = pytest.emulators[name] @@ -432,7 +444,7 @@ async def manage_emulator(request, pytestconfig, avd_param_config) -> BaseEmulat @pytest.fixture(scope="module") @pytest.mark.async_timeout(200) -async def avd(emulator: BaseEmulator) -> BaseEmulator: +async def avd_launcher(emulator: BaseEmulator) -> BaseEmulator: """Makes a booted emulator accessible and with the animation apk installed. Note that the following holds: @@ -451,6 +463,32 @@ async def avd(emulator: BaseEmulator) -> BaseEmulator: return await anext(manage_avd(emulator)) +@pytest.fixture(scope="function") +@pytest.mark.async_timeout(200) +async def avd(avd_launcher: BaseEmulator) -> BaseEmulator: + """Makes a booted emulator accessible and with the animation apk installed. + + This fixture has function scope, which will make sure the emulator will be + restarted if it has crashed. A new emulator will be brought up once for each + module. + + The emulator will be (re-)started if needed. + + Args: + avd_launcher (BaseEmulator): Test fixture that provides the configured emulator. + + Returns: + BaseEmulator: A successfully booted emulator with the debug apk installed. + """ + if not avd_launcher.is_alive(): + logging.info("--> Restarting emulator") + await avd_launcher.restart(avd_launcher.launch_flags) + assert await emulator.wait_for_boot() + else: + logging.info("--> Reusing emulator") + return avd_launcher + + @pytest.fixture(scope="module") @pytest.mark.async_timeout(200) async def avds(emulators: list[BaseEmulator]) -> list[BaseEmulator]: @@ -469,7 +507,10 @@ async def avds(emulators: list[BaseEmulator]) -> list[BaseEmulator]: Returns: List of BaseEmulator: Successfully booted emulators with the debug apk installed. """ - return await asyncio.gather(*[anext(manage_avd(emulator)) for emulator in emulators]) + return await asyncio.gather( + *[anext(manage_avd(emulator)) for emulator in emulators] + ) + async def manage_avd(emulator) -> BaseEmulator: """Helper to manage a booted emulator and make it available for avd and avds fixtures. @@ -749,16 +790,20 @@ def log_directory(pytestconfig): @pytest.fixture async def get_screenshot(emulator_controller, log_directory, request): - async def do_get_screenshot(image_format: ImageFormat): + async def do_get_screenshot( + image_format: ImageFormat = None, screenshot_dir: str = "" + ): """Get a screenshot from the emulator and save it to a file. Args: - image_format: The format of the screenshot image. + image_format: The format of the screenshot image. Defaults to 'ImageFormat()' + screenshot_dir: Screenshot directory. Defaults to '<log_directory> / screenshots' Returns: A tuple of the raw screenshot image and the Pillow image object. """ - screenshot_dir = Path(log_directory) / "screenshots" + image_format = image_format or ImageFormat() + screenshot_dir = Path(screenshot_dir) or Path(log_directory) / "screenshots" img = await emulator_controller.getScreenshot(image_format) test_name = request.node.nodeid.split("::")[-1] file_name = re.sub(r"[\\/\{\}:]", "_", test_name) diff --git a/pytest/test_embedded/tests/display/test_screenshot.py b/pytest/test_embedded/tests/display/test_screenshot.py index e2f62490..b7b54a20 100644 --- a/pytest/test_embedded/tests/display/test_screenshot.py +++ b/pytest/test_embedded/tests/display/test_screenshot.py @@ -249,12 +249,11 @@ async def test_screenshot_should_fail_if_does_not_exist( @pytest.mark.graphics -async def test_screenshot_saved_to_other_folder(emulator_controller, request): +async def test_screenshot_saved_to_other_folder(get_screenshot): """Ensure a screenshot can be saved to a non-default location. Args: - emulator_controller (EmulatorControllerStub): Emulator controller fixture. - request (FixtureRequest): information of the requesting test function. + get_screenshot: Screenshot fixture Test UUID: 2505c26f-0d87-4b2a-81c1-e504787cb5ac @@ -265,15 +264,6 @@ async def test_screenshot_saved_to_other_folder(emulator_controller, request): Verify: Screen capture file should appear in the specified location. """ - #Get the name of the current test being executed from the test collection tree. - test_name = re.sub(r"[\\/\{\}:]", "_", request.node.nodeid.split("::")[-1]) - with tempfile.TemporaryDirectory() as screenshot_dir: - received_image = await emulator_controller.getScreenshot(ImageFormat()) - epoch_time_ms = int(time.time() * 1000) - image_path = Path(screenshot_dir) / f"screenshot-{test_name}-{epoch_time_ms}.png" - - pillow_img = proto_to_pillow(received_image) - pillow_img.save(image_path, "PNG") - - assert image_path.exists(), "Temporary screnshot not created." + _, pillow_image = await get_screenshot(screenshot_dir=screenshot_dir) + assert pillow_image.filename.exists(), "Temporary screenshot not created." diff --git a/pytest/test_embedded/tests/display/test_stream_screenshot.py b/pytest/test_embedded/tests/display/test_stream_screenshot.py index f490f17d..74bdc4c4 100644 --- a/pytest/test_embedded/tests/display/test_stream_screenshot.py +++ b/pytest/test_embedded/tests/display/test_stream_screenshot.py @@ -206,44 +206,6 @@ def calculate_frame_rate(timestamp1, timestamp2): @pytest.mark.graphics @pytest.mark.embedded -async def test_stream_screenshot_has_min_fps( - at_home, - power_down, - emulator_controller, -): - """ - Verifies that the screenshot stream maintains a minimum frame rate (b/312136269). - - We explicity turn down the device to make sure we have a "black" screen with 0 changes. - """ - MIN_FPS = 1.5 - MAX_FRAMES_TO_PROCESS = 10 - - stream = emulator_controller.streamScreenshot( - ImageFormat( - width=18, - height=18, - format=ImageFormat.RGBA8888, - ), - ) - - # Track initial timestamp for comparison - previous_timestamp = 0 - - async for img in stream: - logging.info("Received frame: %s", img.seq) - if previous_timestamp != 0: - assert ( - calculate_frame_rate(previous_timestamp, img.timestampUs) > MIN_FPS - ), "Frame rate dropped below minimum FPS" - - if img.seq > 10: - break - previous_timestamp = img.timestampUs - - -@pytest.mark.graphics -@pytest.mark.embedded @pytest.mark.skipos("win", "reason: b/305258769 - error at setup.") async def test_stream_screenshot_should_fail_if_does_not_exist( at_home, diff --git a/pytest/test_embedded/tests/events/needs_debug_events/test_keys.py b/pytest/test_embedded/tests/events/needs_debug_events/test_keys.py index a0f63143..ab261514 100644 --- a/pytest/test_embedded/tests/events/needs_debug_events/test_keys.py +++ b/pytest/test_embedded/tests/events/needs_debug_events/test_keys.py @@ -14,6 +14,8 @@ import asyncio import logging import re +from xml.etree import ElementTree as ET +import signal import pytest from aemu.proto.emulator_controller_pb2 import KeyboardEvent @@ -121,11 +123,12 @@ async def test_unicode_no_deadlock(at_home, emulator_controller): @pytest.mark.sanity @pytest.mark.embedded @pytest.mark.async_timeout(50000) -async def test_emulator_controls_keys(avd): +async def test_emulator_controls_keys(avd, emulator_controller): """Ensure the emulator controls keys and events work. Args: avd (BaseEmulator): Fixture that gives access to the running emulator. + emulator_controller (EmulatorControllerStub): Emulator controller fixture. Test Steps: 1. Click on Power button (Verify 1). @@ -145,14 +148,13 @@ async def test_emulator_controls_keys(avd): 6. Back button, home and recents work as expected. 7. Extended Controls window is displayed. """ - controller = EmulatorControllerStub(avd.channel) - async def keypress(key, n_times=1): # Send the keypress 'key' event 'n_time' times. for i in range(n_times): logging.info("Sending %s key", key) - await controller.sendKey(KeyboardEvent(key=key, - eventType=KeyboardEvent.keypress)) + await emulator_controller.sendKey( + KeyboardEvent(key=key, eventType=KeyboardEvent.keypress) + ) if n_times > 1: # Delay between successive key events. await asyncio.sleep(1) @@ -167,10 +169,20 @@ async def test_emulator_controls_keys(avd): return "Awake" in await avd.adb.shell("dumpsys power | grep mWakefulness=") async def get_volume(stream_type='STREAM_MUSIC'): - # Return the current volume level of the stream 'stream_type'. - output = await avd.adb.shell("dumpsys audio") - match = re.search(f'{stream_type}.*streamVolume:(\d+)', output) - return int(match.groups()[0]) + # Wait until 'stream_type' appears in dumpsys and return the current volume level. + async def get_stream_volume_dump(output: list): + # Return 'True' if the stream is observed in the system dump. + # Store the volume level in the 'output' list. + dumpsys = await avd.adb.shell("dumpsys audio") + match = re.search(f'{stream_type}.*streamVolume:(\d+)', dumpsys) + if match is None: + return False + output.append(int(match.groups()[0])) + return True + volume = [] + assert await eventually(partial(get_stream_volume_dump, volume)), \ + f"Coudn't detect the stream {stream_type} in the system dump" + return volume[0] async def check_volume_raises(volume): current_volume = await get_volume() @@ -241,17 +253,15 @@ async def test_emulator_controls_keys(avd): volume = await get_volume() await keypress("AudioVolumeUp", 2) assert ( - await eventually(partial(check_volume_raises, volume)), - "Volume was not raised" - ) + await eventually(partial(check_volume_raises, volume)) + ), "Volume was not raised" # Click on Volume Down. volume = await get_volume() await keypress("AudioVolumeDown", 2) assert ( - await eventually(partial(check_volume_lowers, volume)), - "Volume was not lowered" - ) + await eventually(partial(check_volume_lowers, volume)) + ), "Volume was not lowered" ############ Step 4 - Rotation keys ## @@ -321,3 +331,96 @@ async def test_emulator_controls_keys(avd): controlStatus = await ui_controller.showExtendedControls(empty_pb2.Empty()) # Verify the extended controls window appeared. assert controlStatus.visibilityChanged + + +@pytest.mark.e2e +@pytest.mark.sanity +@pytest.mark.async_timeout(1080) +async def test_close_emulator(avd): + """Ensure the emulator windows closes cleanly. + + Args: + avd (BaseEmulator): Fixture that gives access to the running emulator. + + Test Steps: + 1. Launch an emulator AVD. + 2. Send the kill command from the emulator console (Verify 1). + 3. Repeat step 1. + 4. Click and hold Power plus Volume Up buttons for a couple of seconds. + 5. Tap on "Power off" on the emulator (Verify 2). + 6. Restart the emulator. + 7. Send the Control + C (SIGINT) event to the emulator process (Verify 3). + + Verification: + 1. Emulator window closes. + 2. Emulator shuts down and window closes. + 3. Emulator window closes. + """ + def emulator_is_off(): + return not avd.is_alive() + + async def get_window_dump(): + dump = await avd.adb.shell("uiautomator dump /sdcard/window_dump.xml") + assert "uiautomator: inaccessible or not found" not in dump, \ + "Uiautomator binary not found!" + return await avd.adb.shell("cat /sdcard/window_dump.xml") + + def get_center_coords(bounds: str) -> tuple: + # Return the center coordinates (x, y) from element bounds string '[x0y0][x1 y1]' + coords = list(map(int, bounds[1:-1].replace('][',',').split(','))) + return ((coords[0] + coords[2]) / 2, (coords[1] + coords[3]) / 2) + + async def open_power_menu(): + # Triger the Power Options menu and return 'True' when it is opened. + # Send Volume Up and Power keystrokes. + await avd.adb.shell("input keyevent KEYCODE_VOLUME_UP & \ + input keyevent KEYCODE_POWER") + window_dump = await get_window_dump() + if "text=\"Power off\"" not in window_dump: + await asyncio.sleep(5) + return False + return True + + async def click_button(text: str): + # Tap the center of the button containing the text <text> + # Return 'True' if the button is found and clicked. + window_dump = await get_window_dump() + if f"text=\"{text}\"" not in window_dump: + return False + xml = ET.fromstring(window_dump) + bounds = xml.find(f".//*[@text='{text}']/..").get('bounds') + center = get_center_coords(bounds) + await avd.adb.shell("input tap " + ' '.join([*map(str, center)])) + return True + + # Ensure the emulator goes off following a 'kill' event (emulator window closed) + console = await avd.console() + await console.send("kill") + assert await ( + eventually(emulator_is_off) + ), "The emulator was not shut down after the window was closed." + + await avd.restart(avd.launch_flags) + await avd.wait_for_boot() + + # Ensure the emulator shuts down after the Power off button is tapped. + assert await ( + eventually(open_power_menu) + ), "Couldn't open the Power options menu." + + assert await ( + eventually(partial(click_button, "Power off")) + ), "Couldn't click the Power off button." + + assert await ( + eventually(emulator_is_off) + ), "The emulator was not shut down after the Power off button was clicked." + + await avd.restart(avd.launch_flags) + await avd.wait_for_boot() + + # Ensure the emulator shuts down after the CTRL-C event is sent + avd.cmd.process.send_signal(signal.SIGINT) + assert await ( + eventually(emulator_is_off) + ), "The emulator was not shut down after the CTRL-C event was sent." diff --git a/pytest/test_embedded/tests/netsim/test_handle_multi_devices.py b/pytest/test_embedded/tests/netsim/test_handle_multi_devices.py index 1296be4a..e6650b82 100644 --- a/pytest/test_embedded/tests/netsim/test_handle_multi_devices.py +++ b/pytest/test_embedded/tests/netsim/test_handle_multi_devices.py @@ -6,7 +6,7 @@ from netsim_grpc import netsim_client # @pytest.mark.boot @pytest.mark.netsim @pytest.mark.multi -@pytest.mark.async_timeout(200) +@pytest.mark.async_timeout(1080) def test_two_devices_attach_to_netsimd(avds): """Test case to verify that a device is attached to netsimd.""" assert len(netsim_client.NetsimClient().get_devices()) == 2 diff --git a/pytest/test_embedded/tests/netsim/test_handle_single_device.py b/pytest/test_embedded/tests/netsim/test_handle_single_device.py index e4e32720..61a3f110 100644 --- a/pytest/test_embedded/tests/netsim/test_handle_single_device.py +++ b/pytest/test_embedded/tests/netsim/test_handle_single_device.py @@ -5,7 +5,7 @@ from netsim_grpc import netsim_client @pytest.mark.e2e @pytest.mark.boot @pytest.mark.netsim -@pytest.mark.async_timeout(200) +@pytest.mark.async_timeout(1080) def test_device_attaches_to_netsimd(avd): """Test case to verify that a device is attached to netsimd.""" assert len(netsim_client.NetsimClient().get_devices()) != 0 @@ -14,7 +14,7 @@ def test_device_attaches_to_netsimd(avd): @pytest.mark.e2e @pytest.mark.boot @pytest.mark.netsim -@pytest.mark.async_timeout(200) +@pytest.mark.async_timeout(1080) def test_netsim_patch_and_reset(avd): """Test case to verify that patch and reset device in netsim""" netsim = netsim_client.NetsimClient() @@ -46,7 +46,7 @@ def test_netsim_patch_and_reset(avd): @pytest.mark.e2e @pytest.mark.boot @pytest.mark.netsim -@pytest.mark.async_timeout(200) +@pytest.mark.async_timeout(1080) def test_netsim_radio_state_toggle(avd): """Test case to verify patch radio in netsim""" netsim = netsim_client.NetsimClient() diff --git a/pytest/test_embedded/tests/netsim/test_netsim.py b/pytest/test_embedded/tests/netsim/test_netsim.py index 68468194..86455762 100644 --- a/pytest/test_embedded/tests/netsim/test_netsim.py +++ b/pytest/test_embedded/tests/netsim/test_netsim.py @@ -7,7 +7,7 @@ from emu.timing import eventually @pytest.mark.e2e @pytest.mark.boot @pytest.mark.netsim -@pytest.mark.async_timeout(200) +@pytest.mark.async_timeout(1080) async def test_netsimd_is_launched(avd): """Test case to verify that the 'netsimd' process is launched.""" diff --git a/pytest/test_embedded/tests/recording/test_recording.py b/pytest/test_embedded/tests/recording/test_recording.py index 81284038..c69b013c 100644 --- a/pytest/test_embedded/tests/recording/test_recording.py +++ b/pytest/test_embedded/tests/recording/test_recording.py @@ -91,6 +91,8 @@ async def test_can_only_record_once(screen_service, tmp_path): @pytest.mark.flaky @pytest.mark.graphics @pytest.mark.sanity +@pytest.mark.wear +@pytest.mark.atv async def test_screen_records_video_in_webm(screen_service, animation_app, tmp_path): sample_file = tmp_path / "sample.webm" sample_file_header = b"\x1A\x45\xDF\xA3" diff --git a/pytest/test_embedded/tests/snapshot/test_snapshot.py b/pytest/test_embedded/tests/snapshot/test_snapshot.py index 8ec86e29..cc33f43f 100644 --- a/pytest/test_embedded/tests/snapshot/test_snapshot.py +++ b/pytest/test_embedded/tests/snapshot/test_snapshot.py @@ -19,6 +19,8 @@ import tarfile import pytest from aemu.proto.snapshot_service_pb2_grpc import SnapshotServiceStub from snaptool.snapshot import AsyncSnapshotService +from emu.timing import eventually +from functools import partial @pytest.fixture @@ -134,3 +136,71 @@ async def test_snapshot_can_save_and_list(telnet, snapshot_service): snapshots = await telnet.send("avd snapshot list") assert any("foo1" in sublist for sublist in snapshots) + +@pytest.mark.e2e +@pytest.mark.fast +@pytest.mark.async_timeout(1080) +async def test_avd_launch_after_wipe_data(avd, telnet): + """Verify AVD launch after data is wiped. + + Args: + avd (BaseEmulator): Fixture that gives access to a booted emulator. + telnet (EmulatorConnection): Fixture that gives access to thee mulator console. + + Test Steps: + 1. Launch the AVD. + 2. Modify the "Auto-rotate" and "Airplane mode" system settings. + 3. Restart the AVD (verify 1). + 4. Repeat Step 3 with the launch option '-wipe-data' (verify 2). + + Verification: + 1. The AVD is loaded and the previously saved settings are kept. + 2. The emulator loads and all settings are reverted to default. + """ + async def get_system_key(key: str): + # Return the value of the system {key} + value = await avd.adb.shell(f"settings get system {key}") + return (0 if value == 'null' else int(value)) + + async def toggle_system_key(key: str): + # Toggle the integer valued system {key} and return its original value + initial_value = await get_system_key(key) + await avd.adb.shell( + f"settings put system {key} {initial_value ^ 1}" + ) + return initial_value + + async def key_has_value(key: str, expected_value: str): + # Return 'True' if the system {key} has its value equal to {expected_value} + return expected_value == await get_system_key(key) + + # Toggle the "Auto-rotate" and "Airplane mode" settings. + initial_rotation_lock = await toggle_system_key("accelerometer_rotation") + initial_airplane_mode = await toggle_system_key("airplane_mode_on") + + # Make the changes persistent. + await telnet.send("avd snapshot save default_boot") + + # Restart the emulator and verify the changes persist. + await avd.restart(avd.launch_flags) + await avd.wait_for_boot() + rotation_lock = await get_system_key("accelerometer_rotation") + airplane_mode = await get_system_key("airplane_mode_on") + assert rotation_lock != initial_rotation_lock \ + and airplane_mode != initial_airplane_mode, \ + "System settings changed when launched from the saved snapshot." + + # Restart the emulator with the '-wipe-data' launch option. + await avd.stop() + myflags = ["-wipe-data"] + await avd.launch(myflags) + await avd.wait_for_boot() + + # Verify the "Auto-rotate" and "Airplane-mode" setttings reverted to the defaults. + assert await eventually( + partial(key_has_value, "accelerometer_rotation", initial_rotation_lock) + ), "The key 'accelerometer_rotation' didn't revert to the default value." + + assert await eventually( + partial(key_has_value, "airplane_mode_on", initial_airplane_mode) + ), "The key 'airplane_mode_on' didn't revert to the default value." diff --git a/pytest/test_embedded/tests/test_boot.py b/pytest/test_embedded/tests/test_boot.py index e60b17ae..0bff03dd 100644 --- a/pytest/test_embedded/tests/test_boot.py +++ b/pytest/test_embedded/tests/test_boot.py @@ -26,6 +26,8 @@ from google.protobuf import empty_pb2 from emu.apk import APP_DEBUG_APK from emu.timing import eventually +from emu.emulator import Emulator +import json # This will run the tests in this module using this # user configuration. This will fetch an image with api 33 and @@ -301,3 +303,57 @@ async def test_first_time_booted_old_api(emulator): if emulator.is_alive(): await emulator.stop() logging.info("emulator is shut down successfully") + + +@pytest.mark.e2e +@pytest.mark.fast +@pytest.mark.async_timeout(1080) +@pytest.mark.parametrize("core", [1, 2]) +async def test_multicore_startup(emulator, core): + """Verify emulator launches without issues on single and dual core CPUs.""" + + # Launch the emulator with the specificed multicore configuration. + myflags = ["-cores", core] + logging.info(f"Launching emulator with {core} core ...") + + await emulator.restart(emu_flags=myflags) + assert ( + await emulator.wait_for_boot(timeout=1080) + ), f"The emulator couldn't be launched with {core} core" + await emulator.stop() + + +@pytest.mark.e2e +@pytest.mark.fast +@pytest.mark.async_timeout(1080) +async def test_boot_without_internet(emulator): + """Verify emulator can boot with no internet. + + Args: + emulator (BaseEmulator): Fixture that gives access to a configured emulator. + + Notes: + To disable internet access at boot time, the 'restrict=on' option + is added to the wifi and radio (user mode) network settings. + This causes the emulator to be isolated, not being able to contact + the host. No IP packages should be routed over the host to the outside. + """ + my_flags = ["-wifi-user-mode-options", "restrict=on", + "-network-user-mode-options", "restrict=on"] + + # Launch the emulator. + await emulator.restart(emulator.launch_flags + my_flags) + assert ( + await emulator.wait_for_boot(timeout=180) + ), f"The emulator wasn't able to boot without internet." + + # Make sure the emulator launched without internet access. + async def emulator_has_no_internet_access(emulator): + result = await emulator.adb.shell("ping -c 3 www.google.com") + for line in result.rstrip().splitlines(): + if "64 bytes from" in line and "icmp_seq" in line and "ttl" in line: + return False + return True + + assert await emulator_has_no_internet_access(emulator), \ + "The emulator was launched with internet access." |