public inbox for passt-dev@passt.top
 help / color / mirror / code / Atom feed
From: David Gibson <david@gibson.dropbear.id.au>
To: passt-dev@passt.top, Stefano Brivio <sbrivio@redhat.com>
Cc: David Gibson <david@gibson.dropbear.id.au>
Subject: [PATCH 5/7] avocado: Helper functions and classes to run commands via nstool
Date: Thu, 20 Apr 2023 11:11:06 +1000	[thread overview]
Message-ID: <20230420011108.494181-6-david@gibson.dropbear.id.au> (raw)
In-Reply-To: <20230420011108.494181-1-david@gibson.dropbear.id.au>

These are roughly equivalent to the 'context' subsystem in the existing
shell test framework.

Signed-off-by: David Gibson <david@gibson.dropbear.id.au>
---
 avocado/common.py | 117 +++++++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 116 insertions(+), 1 deletion(-)

diff --git a/avocado/common.py b/avocado/common.py
index 036ef3d..94308b8 100644
--- a/avocado/common.py
+++ b/avocado/common.py
@@ -7,8 +7,15 @@
 # Copyright Red Hat
 # Author: David Gibson <david@gibson.dropbear.id.au>
 
+import json
+import os.path
+import sys
+
 import avocado
-from avocado.utils.process import system_output, CmdError
+from avocado.utils.process import system_output, SubProcess, CmdError
+
+
+NSTOOL = './test/nstool'
 
 
 class BaseTest(avocado.Test):
@@ -19,6 +26,51 @@ class BaseTest(avocado.Test):
         return system_output(cmd, **kwargs)
 
 
+class NsTool:
+    def __init__(self, sockpath):
+        self.sockpath = sockpath
+        pid = system_output(
+            '{} info -wp {}'.format(NSTOOL, sockpath), timeout=1)
+        self.pid = int(pid)
+        print('NsTool object: sockpath={} PID={}'.format(sockpath, self.pid),
+              file=sys.stderr)
+
+    def pid(self):
+        return self.pid
+
+    def _xcmd(self, cmd, sudo=False):
+        if sudo:
+            opts = '--keep-caps'
+        else:
+            opts = ''
+        return '{} exec {} {} -- {}'.format(NSTOOL, opts, self.sockpath, cmd)
+
+    def subprocess(self, cmd, sudo=False, **kwargs):
+        return SubProcess(self._xcmd(cmd, sudo), **kwargs)
+
+    def system_output(self, cmd, sudo=False, **kwargs):
+        return system_output(self._xcmd(cmd, sudo), **kwargs)
+
+
+class NsToolUnshare(NsTool):
+    def __init__(self, workdir, sockname, unshare_opts, parent=None):
+        sockpath = os.path.join(workdir, sockname)
+        holdcmd = 'unshare {} -- {} hold {}'.format(
+            unshare_opts, NSTOOL, sockpath)
+        if parent is None:
+            self.holder = SubProcess(holdcmd)
+        else:
+            self.holder = parent.subprocess(holdcmd, sudo=True)
+
+        self.holder.start()
+        super().__init__(sockpath)
+
+    def __del__(self):
+        cmd = '{} stop {}'.format(NSTOOL, self.sockpath)
+        system_output(cmd)
+        self.holder.stop()
+
+
 #
 # Tests for the test infrastructure itself
 #
@@ -29,3 +81,66 @@ class HostExecTests(BaseTest):
 
     def test_false(self):
         self.assertRaises(CmdError, self.hostx, 'false')
+
+
+class UserNsTests(BaseTest):
+    def setUp(self):
+        super().setUp()
+
+        self.ns = NsToolUnshare(self.workdir, 'userns', '-Uc')
+
+    def tearDown(self):
+        del(self.ns)
+
+        super().tearDown()
+
+    def test(self):
+        capcmd = 'capsh --has-p=CAP_SETUID'
+        self.assertRaises(CmdError, self.hostx, capcmd)
+        self.ns.system_output(capcmd, sudo=True)
+
+
+class NestedNsTests(BaseTest):
+    def setUp(self):
+        super().setUp()
+
+        self.userns = NsToolUnshare(self.workdir, 'userns', '-Uc')
+        self.netns = NsToolUnshare(
+            self.workdir, 'netns', '-n', parent=self.userns)
+
+    def tearDown(self):
+        del(self.netns)
+        del(self.userns)
+        super().tearDown()
+
+    def test_unnested(self):
+        # Shouldn't have permission to create a netns without nesting
+        # it in the userns
+        self.assertRaises(CmdError, NsToolUnshare,
+                          self.workdir, 'netns2', '-n')
+
+    def test_nested(self):
+        self.netns.system_output('true')
+        output = self.netns.system_output('ip -j link show')
+        ifs = json.loads(output)
+        self.assertEquals(len(ifs), 1)
+        self.assertEquals(ifs[0]['ifname'], 'lo')
+
+
+class NsConnectTests(BaseTest):
+    def setUp(self):
+        super().setUp()
+
+        self.sockpath = os.path.join(self.workdir, 'hostns')
+        holdcmd = '{} hold {}'.format(NSTOOL, self.sockpath)
+        self.holder = SubProcess(holdcmd)
+        self.holder.start()
+
+    def tearDown(self):
+        self.holder.stop()
+
+        super().tearDown()
+
+    def test_connect(self):
+        hostns = NsTool(self.sockpath)
+        hostns.system_output('true')
-- 
@@ -7,8 +7,15 @@
 # Copyright Red Hat
 # Author: David Gibson <david@gibson.dropbear.id.au>
 
+import json
+import os.path
+import sys
+
 import avocado
-from avocado.utils.process import system_output, CmdError
+from avocado.utils.process import system_output, SubProcess, CmdError
+
+
+NSTOOL = './test/nstool'
 
 
 class BaseTest(avocado.Test):
@@ -19,6 +26,51 @@ class BaseTest(avocado.Test):
         return system_output(cmd, **kwargs)
 
 
+class NsTool:
+    def __init__(self, sockpath):
+        self.sockpath = sockpath
+        pid = system_output(
+            '{} info -wp {}'.format(NSTOOL, sockpath), timeout=1)
+        self.pid = int(pid)
+        print('NsTool object: sockpath={} PID={}'.format(sockpath, self.pid),
+              file=sys.stderr)
+
+    def pid(self):
+        return self.pid
+
+    def _xcmd(self, cmd, sudo=False):
+        if sudo:
+            opts = '--keep-caps'
+        else:
+            opts = ''
+        return '{} exec {} {} -- {}'.format(NSTOOL, opts, self.sockpath, cmd)
+
+    def subprocess(self, cmd, sudo=False, **kwargs):
+        return SubProcess(self._xcmd(cmd, sudo), **kwargs)
+
+    def system_output(self, cmd, sudo=False, **kwargs):
+        return system_output(self._xcmd(cmd, sudo), **kwargs)
+
+
+class NsToolUnshare(NsTool):
+    def __init__(self, workdir, sockname, unshare_opts, parent=None):
+        sockpath = os.path.join(workdir, sockname)
+        holdcmd = 'unshare {} -- {} hold {}'.format(
+            unshare_opts, NSTOOL, sockpath)
+        if parent is None:
+            self.holder = SubProcess(holdcmd)
+        else:
+            self.holder = parent.subprocess(holdcmd, sudo=True)
+
+        self.holder.start()
+        super().__init__(sockpath)
+
+    def __del__(self):
+        cmd = '{} stop {}'.format(NSTOOL, self.sockpath)
+        system_output(cmd)
+        self.holder.stop()
+
+
 #
 # Tests for the test infrastructure itself
 #
@@ -29,3 +81,66 @@ class HostExecTests(BaseTest):
 
     def test_false(self):
         self.assertRaises(CmdError, self.hostx, 'false')
+
+
+class UserNsTests(BaseTest):
+    def setUp(self):
+        super().setUp()
+
+        self.ns = NsToolUnshare(self.workdir, 'userns', '-Uc')
+
+    def tearDown(self):
+        del(self.ns)
+
+        super().tearDown()
+
+    def test(self):
+        capcmd = 'capsh --has-p=CAP_SETUID'
+        self.assertRaises(CmdError, self.hostx, capcmd)
+        self.ns.system_output(capcmd, sudo=True)
+
+
+class NestedNsTests(BaseTest):
+    def setUp(self):
+        super().setUp()
+
+        self.userns = NsToolUnshare(self.workdir, 'userns', '-Uc')
+        self.netns = NsToolUnshare(
+            self.workdir, 'netns', '-n', parent=self.userns)
+
+    def tearDown(self):
+        del(self.netns)
+        del(self.userns)
+        super().tearDown()
+
+    def test_unnested(self):
+        # Shouldn't have permission to create a netns without nesting
+        # it in the userns
+        self.assertRaises(CmdError, NsToolUnshare,
+                          self.workdir, 'netns2', '-n')
+
+    def test_nested(self):
+        self.netns.system_output('true')
+        output = self.netns.system_output('ip -j link show')
+        ifs = json.loads(output)
+        self.assertEquals(len(ifs), 1)
+        self.assertEquals(ifs[0]['ifname'], 'lo')
+
+
+class NsConnectTests(BaseTest):
+    def setUp(self):
+        super().setUp()
+
+        self.sockpath = os.path.join(self.workdir, 'hostns')
+        holdcmd = '{} hold {}'.format(NSTOOL, self.sockpath)
+        self.holder = SubProcess(holdcmd)
+        self.holder.start()
+
+    def tearDown(self):
+        self.holder.stop()
+
+        super().tearDown()
+
+    def test_connect(self):
+        hostns = NsTool(self.sockpath)
+        hostns.system_output('true')
-- 
2.40.0


  parent reply	other threads:[~2023-04-20  1:11 UTC|newest]

Thread overview: 8+ messages / expand[flat|nested]  mbox.gz  Atom feed  top
2023-04-20  1:11 [PATCH 0/7] RFC: Proof-of-concept conversion of some tests to Avocado framework David Gibson
2023-04-20  1:11 ` [PATCH 1/7] avocado: Make a duplicate copy of testsuite for comparison purposes David Gibson
2023-04-20  1:11 ` [PATCH 2/7] avocado: Don't double download assets for test/ and oldtest/ David Gibson
2023-04-20  1:11 ` [PATCH 3/7] avocado: Move static checkers to avocado David Gibson
2023-04-20  1:11 ` [PATCH 4/7] avocado: Convert build tests " David Gibson
2023-04-20  1:11 ` David Gibson [this message]
2023-04-20  1:11 ` [PATCH 6/7] avocado: Helper to get link-local address and wait for SLAAC to complete David Gibson
2023-04-20  1:11 ` [PATCH 7/7] avocado: Convert pasta transfer tests to Avocado David Gibson

Reply instructions:

You may reply publicly to this message via plain-text email
using any one of the following methods:

* Save the following mbox file, import it into your mail client,
  and reply-to-all from there: mbox

  Avoid top-posting and favor interleaved quoting:
  https://en.wikipedia.org/wiki/Posting_style#Interleaved_style

* Reply using the --to, --cc, and --in-reply-to
  switches of git-send-email(1):

  git send-email \
    --in-reply-to=20230420011108.494181-6-david@gibson.dropbear.id.au \
    --to=david@gibson.dropbear.id.au \
    --cc=passt-dev@passt.top \
    --cc=sbrivio@redhat.com \
    /path/to/YOUR_REPLY

  https://kernel.org/pub/software/scm/git/docs/git-send-email.html

* If your mail client supports setting the In-Reply-To header
  via mailto: links, try the mailto: link
Be sure your reply has a Subject: header at the top and a blank line before the message body.
Code repositories for project(s) associated with this public inbox

	https://passt.top/passt

This is a public inbox, see mirroring instructions
for how to clone and mirror all data and code used for this inbox;
as well as URLs for IMAP folder(s).