-# Copyright (c) 2016 Cisco and/or its affiliates.
+# Copyright (c) 2021 Cisco and/or its affiliates.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at:
"""Linux namespace utilities library."""
+from copy import deepcopy
+
from resources.libraries.python.ssh import exec_cmd_no_error, exec_cmd
-class Namespaces(object):
+class Namespaces:
"""Linux namespace utilities."""
- def __init__(self):
- self._namespaces = []
+ __namespaces = []
- def create_namespace(self, node, namespace_name):
+ @staticmethod
+ def create_namespace(node, namespace, delete_before_create=True):
"""Create namespace and add the name to the list for later clean-up.
:param node: Where to create namespace.
- :param namespace_name: Name for namespace.
+ :param namespace: Name for namespace.
+ :param delete_before_create: Delete namespace prior to create
:type node: dict
- :type namespace_name: str
+ :type namespace: str
+ :type delete_before_create: bool
"""
- cmd = ('ip netns add {0}'.format(namespace_name))
+ if delete_before_create:
+ Namespaces.delete_namespace(node, namespace)
+
+ cmd = f"ip netns add {namespace}"
exec_cmd_no_error(node, cmd, sudo=True)
- self._namespaces.append(namespace_name)
+ Namespaces.__namespaces.append(namespace)
+
+ @staticmethod
+ def delete_namespace(node, namespace):
+ """Delete namespace from the node and list.
+
+ :param node: Where to delete namespace.
+ :param namespace: Name for namespace.
+ :param delete_before_create: Delete namespace prior to create
+ :type node: dict
+ :type namespace: str
+ :type delete_before_create: bool
+ """
+ cmd_timeout = 5
+ cmd = f"ip netns delete {namespace}"
+ (ret_code, _, delete_errmsg) = \
+ exec_cmd(node, cmd, timeout=cmd_timeout, sudo=True)
+ if ret_code != 0:
+ cmd = f"ip netns list {namespace}"
+ (stdout, _) = \
+ exec_cmd_no_error(node, cmd, timeout=cmd_timeout, sudo=True)
+ if stdout == namespace:
+ raise RuntimeError(f"Could not delete namespace "
+ f"({namespace}): {delete_errmsg}")
+ try:
+ Namespaces.__namespaces.remove(namespace)
+ except ValueError:
+ pass
@staticmethod
def attach_interface_to_namespace(node, namespace, interface):
:type interface: str
:raises RuntimeError: Interface could not be attached.
"""
- cmd = 'ip link set {0} netns {1}'.format(interface, namespace)
- (ret_code, _, stderr) = exec_cmd(node, cmd, timeout=5, sudo=True)
+ cmd = f"ip link set {interface} netns {namespace}"
+
+ ret_code, _, stderr = exec_cmd(node, cmd, timeout=5, sudo=True)
if ret_code != 0:
- raise RuntimeError(
- 'Could not attach interface, reason:{}'.format(stderr))
- cmd = 'ip netns exec {} ip link set {} up'.format(
- namespace, interface)
- (ret_code, _, stderr) = exec_cmd(node, cmd, timeout=5, sudo=True)
+ raise RuntimeError(f"Could not attach interface, reason:\n{stderr}")
+
+ cmd = f"ip netns exec {namespace} ip link set {interface} up"
+
+ ret_code, _, stderr = exec_cmd(node, cmd, timeout=5, sudo=True)
if ret_code != 0:
raise RuntimeError(
- 'Could not set interface state, reason:{}'.format(stderr))
+ f"Could not set interface state, reason:\n{stderr}"
+ )
+
+ @staticmethod
+ def add_default_route_to_namespace(node, namespace, default_route):
+ """Add IPv4 default route to interface in namespace.
+
+ :param node: Node where to execute command.
+ :param namespace: Namespace to execute command on.
+ :param default_route: Default route address.
+ :type node: dict
+ :type namespace: str
+ :type default_route: str
+ """
+ cmd = f"ip netns exec {namespace} ip route add default " \
+ f"via {default_route}"
+ exec_cmd_no_error(node, cmd, sudo=True)
@staticmethod
def create_bridge_for_int_in_namespace(
:type bridge_name: str
:type interfaces: list
"""
- cmd = 'ip netns exec {} brctl addbr {}'.format(namespace, bridge_name)
+ cmd = f"ip netns exec {namespace} brctl addbr {bridge_name}"
exec_cmd_no_error(node, cmd, sudo=True)
+
for interface in interfaces:
- cmd = 'ip netns exec {} brctl addif {} {}'.format(
- namespace, bridge_name, interface)
+ cmd = f"ip netns exec {namespace} brctl addif {bridge_name} " \
+ f"{interface}"
exec_cmd_no_error(node, cmd, sudo=True)
- cmd = 'ip netns exec {} ip link set dev {} up'.format(
- namespace, bridge_name)
+
+ cmd = f"ip netns exec {namespace} ip link set dev {bridge_name} up"
exec_cmd_no_error(node, cmd, sudo=True)
- def clean_up_namespaces(self, node):
- """Remove all old namespaces.
+ @staticmethod
+ def clean_up_namespaces(node, namespace=None):
+ """Delete all old namespaces.
:param node: Node where to execute command.
+ :param namespace: Namespace to delete, if None delete all namespaces
:type node: dict
+ :type namespace: str
:raises RuntimeError: Namespaces could not be cleaned properly.
"""
- for namespace in self._namespaces:
- print "Cleaning namespace {}".format(namespace)
- cmd = 'ip netns delete {}'.format(namespace)
- (ret_code, _, _) = exec_cmd(node, cmd, timeout=5, sudo=True)
- if ret_code != 0:
- raise RuntimeError('Could not delete namespace')
+ if namespace is not None:
+ Namespaces.delete_namespace(node, namespace)
+ return
+
+ namespace_copy = deepcopy(Namespaces.__namespaces)
+ for namespace_name in namespace_copy:
+ Namespaces.delete_namespace(node, namespace_name)
Code Review / csit.git / blobdiff