• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2#
3#   Copyright 2017 - Google, Inc.
4#
5#   Licensed under the Apache License, Version 2.0 (the "License");
6#   you may not use this file except in compliance with the License.
7#   You may obtain a copy of the License at
8#
9#       http://www.apache.org/licenses/LICENSE-2.0
10#
11#   Unless required by applicable law or agreed to in writing, software
12#   distributed under the License is distributed on an "AS IS" BASIS,
13#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14#   See the License for the specific language governing permissions and
15#   limitations under the License.
16
17import logging
18from acts.libs.proc import job
19
20GET_ALL_INTERFACE = 'ls /sys/class/net'
21GET_VIRTUAL_INTERFACE = 'ls /sys/devices/virtual/net'
22BRCTL_SHOW = 'brctl show'
23
24
25class ApInterfacesError(Exception):
26    """Error related to AP interfaces."""
27
28
29class ApInterfaces(object):
30    """Class to get network interface information for the device.
31
32    """
33    def __init__(self, ap, wan_interface_override=None):
34        """Initialize the ApInterface class.
35
36        Args:
37            ap: the ap object within ACTS
38            wan_interface_override: wan interface to use if specified by config
39        """
40        self.ssh = ap.ssh
41        self.wan_interface_override = wan_interface_override
42
43    def get_all_interface(self):
44        """Get all network interfaces on the device.
45
46        Returns:
47            interfaces_all: list of all the network interfaces on device
48        """
49        output = self.ssh.run(GET_ALL_INTERFACE)
50        interfaces_all = output.stdout.split('\n')
51
52        return interfaces_all
53
54    def get_virtual_interface(self):
55        """Get all virtual interfaces on the device.
56
57        Returns:
58            interfaces_virtual: list of all the virtual interfaces on device
59        """
60        output = self.ssh.run(GET_VIRTUAL_INTERFACE)
61        interfaces_virtual = output.stdout.split('\n')
62
63        return interfaces_virtual
64
65    def get_physical_interface(self):
66        """Get all the physical interfaces of the device.
67
68        Get all physical interfaces such as eth ports and wlan ports
69        Returns:
70            interfaces_phy: list of all the physical interfaces
71        """
72        interfaces_all = self.get_all_interface()
73        interfaces_virtual = self.get_virtual_interface()
74        interfaces_phy = list(set(interfaces_all) - set(interfaces_virtual))
75
76        return interfaces_phy
77
78    def get_bridge_interface(self):
79        """Get all the bridge interfaces of the device.
80
81        Returns:
82            interfaces_bridge: the list of bridge interfaces, return None if
83                bridge utility is not available on the device
84        """
85        interfaces_bridge = []
86        try:
87            output = self.ssh.run(BRCTL_SHOW)
88            lines = output.stdout.split('\n')
89            for line in lines:
90                interfaces_bridge.append(line.split('\t')[0])
91            interfaces_bridge.pop(0)
92            interfaces_bridge = [x for x in interfaces_bridge if x != '']
93            return interfaces_bridge
94        except job.Error:
95            logging.info('No brctl utility is available')
96            return None
97
98    def get_wlan_interface(self):
99        """Get all WLAN interfaces and specify 2.4 GHz and 5 GHz interfaces.
100
101        Returns:
102            interfaces_wlan: all wlan interfaces
103        Raises:
104            ApInterfacesError: Missing at least one WLAN interface
105        """
106        wlan_2g = None
107        wlan_5g = None
108        interfaces_phy = self.get_physical_interface()
109        for iface in interfaces_phy:
110            IW_LIST_FREQ = 'iwlist %s freq' % iface
111            output = self.ssh.run(IW_LIST_FREQ)
112            if 'Channel 06' in output.stdout and 'Channel 36' not in output.stdout:
113                wlan_2g = iface
114            elif 'Channel 36' in output.stdout and 'Channel 06' not in output.stdout:
115                wlan_5g = iface
116
117        interfaces_wlan = [wlan_2g, wlan_5g]
118
119        if None not in interfaces_wlan:
120            return interfaces_wlan
121
122        raise ApInterfacesError('Missing at least one WLAN interface')
123
124    def get_wan_interface(self):
125        """Get the WAN interface which has internet connectivity. If a wan
126        interface is already specified return that instead.
127
128        Returns:
129            wan: the only one WAN interface
130        Raises:
131            ApInterfacesError: no running WAN can be found
132        """
133        if self.wan_interface_override:
134            return self.wan_interface_override
135
136        wan = None
137        interfaces_phy = self.get_physical_interface()
138        interfaces_wlan = self.get_wlan_interface()
139        interfaces_eth = list(set(interfaces_phy) - set(interfaces_wlan))
140        for iface in interfaces_eth:
141            network_status = self.check_ping(iface)
142            if network_status == 1:
143                wan = iface
144                break
145        if wan:
146            return wan
147
148        output = self.ssh.run('ifconfig')
149        interfaces_all = output.stdout.split('\n')
150        logging.info("IFCONFIG output = %s" % interfaces_all)
151
152        raise ApInterfacesError('No WAN interface available')
153
154    def get_lan_interface(self):
155        """Get the LAN interface connecting to local devices.
156
157        Returns:
158            lan: the only one running LAN interface of the devices
159            None, if nothing was found.
160        """
161        lan = None
162        interfaces_phy = self.get_physical_interface()
163        interfaces_wlan = self.get_wlan_interface()
164        interfaces_eth = list(set(interfaces_phy) - set(interfaces_wlan))
165        interface_wan = self.get_wan_interface()
166        interfaces_eth.remove(interface_wan)
167        for iface in interfaces_eth:
168            LAN_CHECK = 'ifconfig %s' % iface
169            output = self.ssh.run(LAN_CHECK)
170            if 'RUNNING' in output.stdout:
171                lan = iface
172                break
173        return lan
174
175    def check_ping(self, iface):
176        """Check the ping status on specific interface to determine the WAN.
177
178        Args:
179            iface: the specific interface to check
180        Returns:
181            network_status: the connectivity status of the interface
182        """
183        PING = 'ping -c 3 -I %s 8.8.8.8' % iface
184        try:
185            self.ssh.run(PING)
186            return 1
187        except job.Error:
188            return 0
189