xref: /aosp_15_r20/external/autotest/client/cros/cellular/labconfig.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7)
1# Lint as: python2, python3
2# Copyright (c) 2011 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
6
7from __future__ import absolute_import
8from __future__ import division
9from __future__ import print_function
10
11import optparse
12import pickle
13import re
14import six
15import subprocess
16
17import common
18
19from autotest_lib.client.cros.cellular import cellular
20from autotest_lib.client.cros.cellular import cellular_logging
21from autotest_lib.client.cros.cellular import labconfig_data
22
23log = cellular_logging.SetupCellularLogging('labconfig')
24
25
26class LabConfigError(Exception):
27    """Exception thrown on bad lab configuration"""
28    pass
29
30
31def get_interface_ip(interface='eth0'):
32    """Returns the IP address for an interface, or None if not found.
33    @param interface: the interface to request IP address for.
34    """
35
36    # We'd like to use
37    #  utils.system_output('ifconfig eth0 2>&1', retain_output=True)
38    # but that gives us a dependency on the rest of autotest, which
39    # means that running the unit test requires pythonpath manipulation
40    stdout = subprocess.Popen(['ip', '-4', 'addr', 'show', 'dev', interface],
41                              stdout=subprocess.PIPE).communicate()[0]
42
43    if six.PY2:
44        # stdout is a string in py2, but we need it to match a byte pattern.
45        stdout = stdout.encode('ascii')
46    match = re.search(b'inet ([0-9.]+)[/ ]', stdout)
47    if not match:
48        return None
49    return match.group(1).decode()
50
51
52class Configuration(object):
53    """Configuration for a cellular test.
54
55    This includes things like the address of the cell emulator device
56    and details of the RF switching between the emulated basestation
57    and the DUT."""
58
59    def __init__(self, args):
60        # For server tests, this constructor runs as part of the
61        # server control file, on whatever machine the test was
62        # started on.
63        parser = optparse.OptionParser()
64
65        # Record our args so we can serialize ourself.
66        self.args = args
67
68        self.ip = None
69
70        parser.add_option('--cell', dest='cell', default=None,
71                          help='Cellular test cell to use')
72        parser.add_option(
73            '--technology', dest='technology', default='all',
74            help='Radio access technologies to use (e.g. "WCDMA")')
75        (self.options, _) = parser.parse_args(args)
76
77        self.cell = self._get_cell(self.options.cell)
78
79    def _get_cell(self, name):
80        """Extracts the named cell from labconfig_data.CELLS."""
81        if not name:
82            raise LabConfigError(
83                'Could not find --cell argument.  ' +
84                'To specify a cell, pass --args=--cell=foo to test_that')
85
86        if name not in labconfig_data.CELLS:
87            raise LabConfigError(
88                'Could not find cell %s, valid cells are %s' % (
89                    name, list(labconfig_data.CELLS.keys())))
90
91        return labconfig_data.CELLS[name]
92
93    def _get_dut(self, machine=None):
94        """Returns the DUT record for machine from cell["duts"]
95        Args:
96            machine:  name or IP of machine.  None: for "the current machine".
97
98        Right now, we use the interface of eth0 to figure out which
99        machine we're running on.  The important thing is that this
100        matches the IP address in the cell duts configuration.  We'll
101        have to come up with a better way if this proves brittle."""
102
103        # TODO(byronk) : crosbug.com/235911:
104        # autotest: Getting IP address from eth0 by name is brittle
105        if self.ip and not machine:
106            machine = self.ip
107        ifconfig = ''
108        if not machine:
109            log.debug('self.ip is : %s ', self.ip)
110            # TODO(byronk): use sysfs to find network interface
111            possible_interfaces = ['eth0', 'eth1', 'eth_test']
112            log.debug('Looking for an up network interface in : %s',
113                      possible_interfaces)
114            for interface in possible_interfaces:
115                machine = get_interface_ip(interface)
116                if machine:
117                    log.debug('Got an IP address: %s Stopping the search.. ',
118                              machine)
119                    self.ip = machine
120                    break
121            else:
122                ifconfig = subprocess.Popen(['ip', 'addr', 'show'],
123                        stdout=subprocess.PIPE).communicate()[0]
124        if not machine:
125            raise LabConfigError(
126                'Could not determine which machine we are.\n'
127                '  Cell =  %s \n' % self.options.cell +
128                'Tried these interface names: %s \n' % possible_interfaces +
129                '`ip addr show` output:\n%s' % ifconfig
130            )
131
132        for dut in self.cell["duts"]:
133            if machine == dut["address"] or machine == dut["name"]:
134                return dut
135
136        raise LabConfigError(
137            'This machine %s not matching: (%s,%s) in config. Cell = %s: %s' %
138            (machine, dut['address'],
139             dut['name'], self.options.cell, self.cell['duts']))
140
141    def get_technologies(self, machine=None):
142        """Gets technologies to use for machine; defaults to all available.
143        @param machine: Machine to get technologies for.
144        """
145        technologies_list = self.options.technology.split(',')
146
147        if 'all' in technologies_list:
148            m = self._get_dut(machine)
149            technologies_list = m["technologies"]
150
151        enums = [getattr(cellular.Technology, t, None)
152                 for t in technologies_list]
153
154        if None in enums:
155            raise LabConfigError(
156                'Could not understand a technology in %s' % technologies_list)
157
158        return enums
159
160    def get_rf_switch_port(self, machine=None):
161        """Get the RF Switch Port for the specified machine.
162        @param machine: machine to get rf switch port for
163        """
164        dut = self._get_dut(machine)
165        print(dut)
166        return dut['rf_switch_port']
167
168    def get_pickle(self):
169        """Get pickled object."""
170        return pickle.dumps(self)
171