How to use _get_bondinterface_details method in avocado

Best Python code snippet using avocado_python

interfaces.py

Source:interfaces.py

...61            LOG.error(msg)62    @property63    def slave_config_filename(self):64        try:65            slave_dict = self._get_bondinterface_details()66            return ["{}/ifcfg-{}".format(self.config_file_path,67                    slave) for slave in slave_dict['slaves']]68        except Exception:69            msg = "Slave config filename not available"70            LOG.debug(msg)71            return72    def _get_interface_details(self, version=None):73        cmd = "ip -j link show {}".format(self.name)74        if version:75            cmd = "ip -{} -j address show {}".format(version, self.name)76        output = run_command(cmd, self.host)77        try:78            result = json.loads(output)79            for item in result:80                if item.get('ifname') == self.name:81                    return item82            raise NWException("Interface not found")83        except (NWException, json.JSONDecodeError):84            msg = "Unable to get the details of interface {}".format(self.name)85            LOG.error(msg)86            raise NWException(msg)87    def _get_bondinterface_details(self):88        cmd = "cat /sys/class/net/{}/bonding/mode \89               /sys/class/net/{}/bonding/slaves".format(self.name, self.name)90        try:91            mode, slaves = run_command(cmd, self.host).splitlines()92            return {'mode': mode.split(),93                    'slaves': slaves.split()}94        except Exception:95            raise NWException("Slave interface not found for the bond {}".format(self.name))96    def _move_file_to_backup(self, filename, ignore_missing=True):97        destination = "{}.backup".format(filename)98        if os.path.exists(filename):99            shutil.move(filename, destination)100        else:101            if not ignore_missing:102                raise NWException("%s interface not available" % self.name)103    def _write_to_file(self, filename, values):104        self._move_file_to_backup(filename)105        with open(filename, 'w+') as fp:  # pylint: disable=W1514106            for key, value in values.items():107                fp.write("{}={}\n".format(key, value))108    def set_hwaddr(self, hwaddr):109        """Sets a Hardware Address (MAC Address) to the interface.110        This method will try to set a new hwaddr to this interface, if111        fails it will raise a NWException.112        You must have sudo permissions to run this method on a host.113        :param hwaddr: Hardware Address (Mac Address)114        """115        cmd = "ip link set dev {} address {}".format(self.name, hwaddr)116        try:117            run_command(cmd, self.self.host, sudo=True)118        except Exception as ex:119            raise NWException("Adding hw address fails: %s" % ex)120    def add_ipaddr(self, ipaddr, netmask):121        """Add an IP Address (with netmask) to the interface.122        This method will try to add a new ipaddr/netmask this interface, if123        fails it will raise a NWException.124        You must have sudo permissions to run this method on a host.125        :param ipaddr: IP Address126        :param netmask: Network mask127        """128        ip = ip_interface("{}/{}".format(ipaddr, netmask))129        cmd = 'ip addr add {} dev {}'.format(ip.compressed,130                                             self.name)131        try:132            run_command(cmd, self.host, sudo=True)133        except Exception as ex:134            raise NWException("Failed to add address {}".format(ex))135    @property136    def vlans(self):137        """Return all interface's VLAN.138        This is a dict were key is the VLAN number and the value is the name of139        the VLAN interface.140        rtype: dict141        """142        vlans = {}143        if not os.path.exists('/proc/net/vlan/config'):144            return vlans145        with open('/proc/net/vlan/config', encoding="utf-8") as vlan_config_file:146            for line in vlan_config_file:147                # entry is formatted as "vlan_name | vlan_id | parent_device"148                line = "".join(line.split())149                if line.endswith(self.name):150                    line = line.split('|')151                    vlans[line[1]] = line[0]152        return vlans153    def add_vlan_tag(self, vlan_num, vlan_name=None):154        """Configure 802.1Q VLAN tagging to the interface.155        This method will attempt to add a VLAN tag to this interface. If it156        fails, the method will raise a NWException.157        :param vlan_num: VLAN ID158        :param vlan_name: option to name VLAN interface, by default it is named159                          <interface_name>.<vlan_num>160        """161        vlan_name = vlan_name or "{}.{}".format(self.name, vlan_num)162        cmd = "ip link add link {} name {} type vlan id {}".format(self.name,163                                                                   vlan_name,164                                                                   vlan_num)165        try:166            run_command(cmd, self.host, sudo=True)167        except Exception as ex:168            raise NWException("Failed to add VLAN tag: {}".format(ex))169    def remove_vlan_by_tag(self, vlan_num):170        """Remove the VLAN of the interface by tag number.171        This method will try to remove the VLAN tag of this interface. If it fails,172        the method will raise a NWException.173        :param vlan_num: VLAN ID174        :return: True or False, True if it found the VLAN interface and removed175                 it successfully, otherwise it will return False.176        """177        if str(vlan_num) in self.vlans:178            vlan_name = self.vlans[str(vlan_num)]179        else:180            return False181        cmd = "ip link delete {}".format(vlan_name)182        try:183            run_command(cmd, self.host, sudo=True)184            return True185        except Exception as ex:186            raise NWException("Failed to remove VLAN interface: {}".format(ex))187    def remove_all_vlans(self):188        """Remove all VLANs of this interface.189        This method will remove all the VLAN interfaces associated by the190        interface. If it fails, the method will raise a NWException.191        """192        try:193            for v in self.vlans.values():194                cmd = "ip link delete {}".format(v)195                run_command(cmd, self.host, sudo=True)196        except Exception as ex:197            raise NWException("Failed to remove VLAN interface: {}".format(ex))198    def bring_down(self):199        """Shutdown the interface.200        This will shutdown the interface link. Be careful, you might lost201        connection to the host.202        You must have sudo permissions to run this method on a host.203        """204        cmd = "ip link set {} down".format(self.name)205        try:206            run_command(cmd, self.host, sudo=True)207        except Exception as ex:208            raise NWException("Failed to bring down: %s" % ex)209    def bring_up(self):210        """"Wake-up the interface.211        This will wake-up the interface link.212        You must have sudo permissions to run this method on a host.213        """214        cmd = "ip link set {} up".format(self.name)215        try:216            run_command(cmd, self.host, sudo=True)217        except Exception as ex:218            raise NWException("Failed to bring up: %s" % ex)219    def is_admin_link_up(self):220        """Check the admin link state is up or not.221        :return: True or False, True if network interface state is 'UP'222                 otherwise will return False.223        """224        try:225            if 'UP' in self._get_interface_details().get('flags'):226                return True227        except (NWException, IndexError):228            raise NWException("Could not get Administrative link state.")229        return False230    def is_operational_link_up(self):231        """Check Operational link state is up or not.232        :return: True or False. True if operational link state is LOWER_UP,233                 otherwise will return False.234        """235        try:236            if 'LOWER_UP' in self._get_interface_details().get('flags'):237                return True238        except (NWException, IndexError):239            raise NWException("Could not get operational link state.")240        return False241    def is_link_up(self):242        """Check if the interface is up or not.243        :return: True or False. True if admin link state and operational244                 link state is up otherwise will return False.245        """246        return self.is_admin_link_up() and self.is_operational_link_up()247    def get_ipaddrs(self, version=4):248        """Get the IP addresses from a network interface.249        Interfaces can hold multiple IP addresses. This method will return a250        list with all addresses on this interface.251        :param version: Address Family Version (4 or 6). This must be a integer252                        and default is 4.253        :return: IP address as string.254        """255        if version not in [4, 6]:256            raise NWException("Version {} not supported".format(version))257        try:258            details = self._get_interface_details(version)259            addr_info = details.get('addr_info')260            if addr_info:261                return [x.get('local') for x in addr_info]262        except (NWException, IndexError):263            msg = "Could not get ip addresses for {}".format(self.name)264            LOG.debug(msg)265            return []266    def get_hwaddr(self):267        """Get the Hardware Address (MAC) of this interface.268        This method will try to get the address and if fails it will raise a269        NWException.270        """271        cmd = "cat /sys/class/net/{}/address".format(self.name)272        try:273            return run_command(cmd, self.host)274        except Exception as ex:275            raise NWException("Failed to get hw address: {}".format(ex))276    def get_mtu(self):277        """Return the current MTU value of this interface.278        This method will try to get the current MTU value, if fails will279        raise a NWException.280        """281        try:282            return self._get_interface_details().get('mtu')283        except (NWException, IndexError):284            raise NWException("Could not get MUT value.")285    def ping_check(self, peer_ip, count=2, options=None):286        """This method will try to ping a peer address (IPv4 or IPv6).287        You should provide a IPv4 or IPV6 that would like to ping. This288        method will try to ping the peer and if fails it will raise a289        NWException.290        :param peer_ip: Peer IP address (IPv4 or IPv6)291        :param count: How many packets to send. Default is 2292        :param options: ping command options. Default is None293        """294        cmd = "ping -I {} {} -c {}".format(self.name, peer_ip, count)295        if options is not None:296            cmd = "{} {}".format(cmd, options)297        try:298            run_command(cmd, self.host)299        except Exception as ex:300            raise NWException("Failed to ping: {}".format(ex))301    def save(self, ipaddr, netmask):302        """Save current interface IP Address to the system configuration file.303        If the ipaddr is valid (currently being used by the interface)304        this will try to save the current settings into /etc/. This305        check is necessary to avoid inconsistency. Before save, you306        should add_ipaddr, first.307        Currently, only RHEL, Fedora and SuSE are supported. And this308        will create a backup file of your current configuration if309        found.310        :param ipaddr : IP Address which need to configure for interface311        :param netmask: Network mask which is associated to the provided IP312        """313        if ipaddr not in self.get_ipaddrs():314            msg = ('ipaddr not configured on interface. To avoid '315                   'inconsistency, please add the ipaddr first.')316            raise NWException(msg)317        current_distro = distro_detect()318        filename = "ifcfg-{}".format(self.name)319        if current_distro.name in ['rhel', 'fedora']:320            path = "/etc/sysconfig/network-scripts"321        elif current_distro.name == 'SuSE':322            path = "/etc/sysconfig/network"323        else:324            msg = 'Distro not supported by API. Could not save ipaddr.'325            raise NWException(msg)326        ifcfg_dict = {'TYPE': self.if_type,327                      'BOOTPROTO': 'static',328                      'NAME': self.name,329                      'DEVICE': self.name,330                      'ONBOOT': 'yes',331                      'IPADDR': ipaddr,332                      'NETMASK': netmask,333                      'IPV6INIT': 'yes',334                      'IPV6_AUTOCONF': 'yes',335                      'IPV6_DEFROUTE': 'yes'}336        if current_distro.name == 'SuSE':337            ifcfg_dict.pop('BOOTPROTO')338        if self.if_type == 'Bond':339            ifcfg_dict['BONDING_MASTER'] = 'yes'340            bond_dict = self._get_bondinterface_details()341            ifcfg_slave_dict = {'SLAVE': 'yes',342                                'ONBOOT': 'yes',343                                'MASTER': self.name}344            if current_distro.name == 'SuSE':345                ifcfg_dict['BONDING_MODULE_OPTS'] = 'mode=' \346                           + bond_dict['mode'][0]347                for index, slave in enumerate(bond_dict['slaves']):348                    bonding_slave = 'BONDING_SLAVE{}'.format(index)349                    ifcfg_dict[bonding_slave] = slave350                    ifcfg_slave_dict.update({'NAME': slave,351                                             'DEVICE': slave})352                    self._write_to_file("{}/ifcfg-{}".format(path, slave),353                                        ifcfg_slave_dict)354            elif current_distro.name in ['rhel', 'fedora']:...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.