First implementation of network namespaces in addr module (#596)

Author: romainx

test/test_modules.py

@@ -590,3 +590,15 @@ def test_addr(host):
 
     for ip in google_addr.ip_addresses:
         assert isinstance(ip_address(ip), (IPv4Address, IPv6Address))
+
+
+@pytest.mark.testinfra_hosts("ansible://debian_buster")
+def test_addr_namespace(host):
+    namespace_lookup = host.addr("localhost", "ns1")
+    # ns1 network namespace does not exist so everything is false
+    assert not namespace_lookup.namespace_exists
+    assert not namespace_lookup.is_reachable
+    assert not namespace_lookup.is_resolvable
+    with pytest.raises(NotImplementedError):
+        # nc is not available so an error is raised
+        assert not namespace_lookup.port("443").is_reachable

testinfra/modules/addr.py

@@ -20,15 +20,20 @@ def __init__(self, addr, port):
 
     @property
     def is_reachable(self):
-        if not self._addr._host.exists('nc'):
+        """Return if port is reachable"""
+        if not self._addr._host.exists("nc"):
+            if self._addr.namespace:
+                # in this case cannot use namespace
+                raise NotImplementedError(
+                    "nc command not available, namespace cannot be used")
             # Fallback to bash if netcat is not available
             return self._addr.run_expect(
                 [0, 1, 124],
                 "timeout 1 bash -c 'cat < /dev/null > /dev/tcp/%s/%s'",
                 self._addr.name, self._port).rc == 0
 
-        return self._addr.run(
-            "nc -w 1 -z %s %s", self._addr.name, self._port).rc == 0
+        return self._addr.run("{}nc -w 1 -z {} {}".format(self._addr._prefix,
+                              self._addr.name, self._port)).rc == 0
 
 
 class Addr(Module):
@@ -47,17 +52,54 @@ class Addr(Module):
     True
     >>> google.port(666).is_reachable
     False
-    """
 
-    def __init__(self, name):
+    Can also be use within a network namespace_.
+
+    >>> localhost = host.addr("localhost", "ns1")
+    >>> localhost.is_resolvable
+    True
+
+    Network namespaces can only be used if ip_ command is available
+    because in this case, the module use ip-netns_ as command prefix.
+    In the other case, it will raise NotImplementedError.
+
+    .. _namespace: https://man7.org/linux/man-pages/man7/namespaces.7.html
+    .. _ip: https://man7.org/linux/man-pages/man8/ip.8.html
+    .. _ip-netns: https://man7.org/linux/man-pages/man8/ip-netns.8.html
+    """
+    def __init__(self, name, namespace=None):
         self._name = name
+        self._namespace = namespace
+        if self.namespace and not self._host.exists("ip"):
+            raise NotImplementedError(
+                "ip command not available, namespace cannot be used")
         super().__init__()
 
     @property
     def name(self):
         """Return host name"""
         return self._name
 
+    @property
+    def namespace(self):
+        """Return network namespace"""
+        return self._namespace
+
+    @property
+    def _prefix(self):
+        """Return the prefix to use for commands"""
+        prefix = ""
+        if self.namespace:
+            prefix = "ip netns exec %s " % self.namespace
+        return prefix
+
+    @property
+    def namespace_exists(self):
+        """Test if the network namespace exists"""
+        # could use ip netns list instead
+        return self.namespace and self.run_test("test -e /var/run/netns/%s",
+                                                self.namespace).rc == 0
+
     @property
     def is_resolvable(self):
         """Return if address is resolvable"""
@@ -66,8 +108,8 @@ def is_resolvable(self):
     @property
     def is_reachable(self):
         """Return if address is reachable"""
-        return self.run_expect([0, 1, 2],
-                               "ping -W 1 -c 1 %s", self.name).rc == 0
+        return self.run_expect([0, 1, 2], "{}ping -W 1 -c 1 {}".format(
+            self._prefix, self.name)).rc == 0
 
     @property
     def ip_addresses(self):
@@ -92,6 +134,7 @@ def __repr__(self):
         return "<addr %s>" % (self.name,)
 
     def _resolve(self, method):
-        result = self.run_expect([0, 2], "getent %s %s", method, self.name)
+        result = self.run_expect([0, 1, 2], "{}getent {} {}".format(
+            self._prefix, method, self.name))
         lines = result.stdout.splitlines()
         return list(set(line.split()[0] for line in lines))