xref: /aosp_15_r20/external/autotest/client/cros/network.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7)
1# Lint as: python2, python3
2# Copyright (c) 2019 The Chromium OS Authors. All rights reserved.
3# Use of this source code is governed by a BSD-style license that can be
4# found in the LICENSE file.
5
6import logging
7from six.moves import urllib
8import socket
9import time
10
11from autotest_lib.client.bin import utils
12from autotest_lib.client.common_lib import error
13
14
15def CheckThatInterfaceCanAccessDestination(host,
16                                           interface,
17                                           families=[socket.AF_UNSPEC]):
18    """
19    Checks that we can access a host using a specific interface.
20
21    @param host: Destination host
22    @param interface: Name of the network interface to be used
23    @raises: error.TestFail if the interface cannot access the specified host.
24
25    """
26    logging.debug('Check connection to %s', host)
27    # addrinfo records: (family, type, proto, canonname, (addr, port))
28    server_addresses = []
29    for family in families:
30        try:
31            records = socket.getaddrinfo(host, 80, family)
32        except:
33            # Just ignore this family.
34            continue
35        server_addresses.extend(record[4][0] for record in records)
36
37    found_route = False
38    failing_addresses = []
39    for address in set(server_addresses):
40        # Routes may not always be up by this point. Note that routes for v4 or
41        # v6 may come up before the other, so we simply do this poll for all
42        # addresses.
43        try:
44            utils.poll_for_condition(condition=lambda: utils.ping(
45                    address, interface=interface, tries=2, timeout=3) == 0,
46                                     exception=Exception('No route to %s' %
47                                                         address),
48                                     timeout=2)
49        except Exception as e:
50            logging.info(e)
51            failing_addresses.append(address)
52        else:
53            found_route = True
54
55    if not found_route:
56        raise error.TestFail('Interface %s cannot connect to %s' % (interface,
57                             failing_addresses))
58
59
60FETCH_URL_PATTERN_FOR_TEST = \
61    'http://testing-chargen.appspot.com/download?size=%d'
62
63def FetchUrl(url_pattern, bytes_to_fetch=10, fetch_timeout=10):
64    """
65    Fetches a specified number of bytes from a URL.
66
67    @param url_pattern: URL pattern for fetching a specified number of bytes.
68            %d in the pattern is to be filled in with the number of bytes to
69            fetch.
70    @param bytes_to_fetch: Number of bytes to fetch.
71    @param fetch_timeout: Number of seconds to wait for the fetch to complete
72            before it times out.
73    @return: The time in seconds spent for fetching the specified number of
74            bytes.
75    @raises: error.TestError if one of the following happens:
76            - The fetch takes no time.
77            - The number of bytes fetched differs from the specified
78              number.
79
80    """
81    # Limit the amount of bytes to read at a time.
82    _MAX_FETCH_READ_BYTES = 1024 * 1024
83
84    url = url_pattern % bytes_to_fetch
85    logging.info('FetchUrl %s', url)
86    start_time = time.time()
87    result = urllib.request.urlopen(url, timeout=fetch_timeout)
88    bytes_fetched = 0
89    while bytes_fetched < bytes_to_fetch:
90        bytes_left = bytes_to_fetch - bytes_fetched
91        bytes_to_read = min(bytes_left, _MAX_FETCH_READ_BYTES)
92        bytes_read = len(result.read(bytes_to_read))
93        bytes_fetched += bytes_read
94        if bytes_read != bytes_to_read:
95            raise error.TestError('FetchUrl tried to read %d bytes, but got '
96                                  '%d bytes instead.' %
97                                  (bytes_to_read, bytes_read))
98        fetch_time = time.time() - start_time
99        if fetch_time > fetch_timeout:
100            raise error.TestError('FetchUrl exceeded timeout.')
101
102    return fetch_time
103