xref: /aosp_15_r20/external/cronet/build/fuchsia/test/test_connection.py (revision 6777b5387eb2ff775bb5750e3f5d96f37fb7352b)
1#!/usr/bin/env vpython3
2
3# Copyright 2023 The Chromium Authors
4# Use of this source code is governed by a BSD-style license that can be
5# found in the LICENSE file.
6"""Tests the connection of a target."""
7
8# Note, this is a temporary tool and should be removed in favor of a better way
9# to expose the functionality or merge with other use cases of get_ssh_address.
10
11import logging
12import sys
13import time
14
15from typing import Optional
16
17from boot_device import boot_device, BootMode
18from common import run_ffx_command
19
20
21def test_connection(target_id: Optional[str], wait_sec: int = 60) -> None:
22    """Runs echo tests to verify that the device can be connected to.
23
24    Devices may not be connectable right after being discovered by ffx, e.g.
25    after a `ffx target wait`, so this function retries up to |wait_sec| before
26    throwing an exception.
27    """
28    start_sec = time.time()
29    while time.time() - start_sec < wait_sec:
30        if run_ffx_command(cmd=('target', 'echo'),
31                           target_id=target_id,
32                           check=False).returncode == 0:
33            return
34        time.sleep(10)
35
36    run_ffx_command(cmd=('target', 'echo'), target_id=target_id)
37
38
39def test_device_connection(target_id: Optional[str]) -> None:
40    """Runs test_connection against the target_id and restarts the device if
41    it cannot be connected."""
42    start_sec = time.time()
43    while time.time() - start_sec < 1800:
44        # pylint: disable=bare-except
45        # First, test_connection with ffx target echo.
46        try:
47            test_connection(target_id=target_id, wait_sec=600)
48            return
49        except:
50            # If anything wrong, reboot the device and try again.
51            try:
52                boot_device(target_id, BootMode.REGULAR, must_boot=True)
53            except:
54                # If unfortunately, the reboot failed, it's still worth
55                # continuing the test rather than failing here.
56                pass
57            logging.warning(
58                run_ffx_command(cmd=('target', 'wait'),
59                                target_id=target_id,
60                                check=False,
61                                capture_output=True).stdout)
62
63def main():
64    """Test a connection against a fuchsia target via ffx."""
65    if len(sys.argv) < 2:
66        raise ValueError('test_connection.py target')
67    test_connection(sys.argv[1])
68
69
70if __name__ == '__main__':
71    sys.exit(main())
72