Add to the tasst library a SimNetHost class used to represent a
simulated network host of some type (e.g. namespaces, VMs). For now
all it does is lets you execute commands, either foreground or
background in the context of the simulated host. We add some "meta"
exeter tests for it.
Signed-off-by: David Gibson <david(a)gibson.dropbear.id.au>
---
test/Makefile | 2 +-
test/tasst/__main__.py | 6 +-
test/tasst/snh.py | 187 +++++++++++++++++++++++++++++++++++++++++
3 files changed, 190 insertions(+), 5 deletions(-)
create mode 100644 test/tasst/snh.py
diff --git a/test/Makefile b/test/Makefile
index 81f94f70..8373ae77 100644
--- a/test/Makefile
+++ b/test/Makefile
@@ -70,7 +70,7 @@ EXETER_PY = build/build.py
EXETER_JOBS = $(EXETER_SH:%.sh=%.json) $(EXETER_PY:%.py=%.json)
AVOCADO_JOBS = $(EXETER_JOBS) avocado/static_checkers.json
-TASST_SRCS = __init__.py __main__.py
+TASST_SRCS = __init__.py __main__.py snh.py
EXETER_META = meta/lint.json meta/tasst.json
META_JOBS = $(EXETER_META)
diff --git a/test/tasst/__main__.py b/test/tasst/__main__.py
index c365b986..91499128 100644
--- a/test/tasst/__main__.py
+++ b/test/tasst/__main__.py
@@ -12,10 +12,8 @@ library of test helpers for passt & pasta
import exeter
-
-(a)exeter.test
-def placeholder():
- pass
+# We import just to get the exeter tests, which flake8 can't see
+from . import snh # noqa: F401
if __name__ == '__main__':
diff --git a/test/tasst/snh.py b/test/tasst/snh.py
new file mode 100644
index 00000000..dfbe2c84
--- /dev/null
+++ b/test/tasst/snh.py
@@ -0,0 +1,187 @@
+#! /usr/bin/env python3
+
+# SPDX-License-Identifier: GPL-2.0-or-later
+#
+# Copyright Red Hat
+# Author: David Gibson <david(a)gibson.dropbear.id.au>
+
+"""
+Test A Simple Socket Transport
+
+tasst/snh.py - Simulated network hosts for testing
+"""
+
+
+import contextlib
+import subprocess
+import sys
+
+import exeter
+
+
+STDOUT = 1
+
+
+class SnhProcess(contextlib.AbstractContextManager):
+ """
+ A background process running on a SimNetHost
+ """
+
+ def __init__(self, snh, *cmd, check=True, context_timeout=1.0, **kwargs):
+ self.snh = snh
+ self.cmd = cmd
+ self.check = check
+ self.context_timeout = float(context_timeout)
+
+ self.kwargs = kwargs
+
+ def __enter__(self):
+ self.popen = subprocess.Popen(self.cmd, **self.kwargs)
+ return self
+
+ def run(self, **kwargs):
+ stdout, stderr = self.popen.communicate(**kwargs)
+ cp = subprocess.CompletedProcess(self.popen.args,
+ self.popen.returncode,
+ stdout, stderr)
+ if self.check:
+ cp.check_returncode()
+ return cp
+
+ def terminate(self):
+ self.popen.terminate()
+
+ def kill(self):
+ self.popen.kill()
+
+ def __exit__(self, *exc_details):
+ try:
+ self.popen.wait(timeout=self.context_timeout)
+ except subprocess.TimeoutExpired as e:
+ self.terminate()
+ try:
+ self.popen.wait(timeout=self.context_timeout)
+ except subprocess.TimeoutExpired:
+ self.kill()
+ raise e
+
+
+class SimNetHost(contextlib.AbstractContextManager):
+ """
+ A (usually virtual or simulated) location where we can execute
+ commands and configure networks.
+
+ """
+
+ def __init__(self, name):
+ self.name = name # For debugging
+
+ def hostify(self, *cmd, **kwargs):
+ raise NotImplementedError
+
+ def __enter__(self):
+ return self
+
+ def __exit__(self, *exc_details):
+ pass
+
+ def output(self, *cmd, **kwargs):
+ proc = self.fg(*cmd, capture=STDOUT, **kwargs)
+ return proc.stdout
+
+ def fg(self, *cmd, timeout=None, **kwargs):
+ # We don't use subprocess.run() because it kills without
+ # attempting to terminate on timeout
+ with self.bg(*cmd, **kwargs) as proc:
+ res = proc.run(timeout=timeout)
+ return res
+
+ def bg(self, *cmd, capture=None, **kwargs):
+ if capture == STDOUT:
+ kwargs['stdout'] = subprocess.PIPE
+ hostcmd, kwargs = self.hostify(*cmd, **kwargs)
+ proc = SnhProcess(self, *hostcmd, **kwargs)
+ print(f"SimNetHost {self.name}: Started {cmd} => {proc}",
+ file=sys.stderr)
+ return proc
+
+ # Internal tests
+ def test_true(self):
+ with self as snh:
+ snh.fg('true')
+
+ def test_false(self):
+ with self as snh:
+ exeter.assert_raises(subprocess.CalledProcessError,
+ snh.fg, 'false')
+
+ def test_echo(self):
+ msg = 'Hello tasst'
+ with self as snh:
+ out = snh.output('echo', f'{msg}')
+ exeter.assert_eq(out, msg.encode('utf-8') + b'\n')
+
+ def test_timeout(self):
+ with self as snh:
+ exeter.assert_raises(subprocess.TimeoutExpired, snh.fg,
+ 'sleep', 'infinity', timeout=0.1,
check=False)
+
+ def test_bg_true(self):
+ with self as snh:
+ with snh.bg('true'):
+ pass
+
+ def test_bg_false(self):
+ with self as snh:
+ with snh.bg('false') as proc:
+ exeter.assert_raises(subprocess.CalledProcessError, proc.run)
+
+ def test_bg_echo(self):
+ msg = 'Hello tasst'
+ with self as snh:
+ with snh.bg('echo', f'{msg}', capture=STDOUT) as proc:
+ res = proc.run()
+ exeter.assert_eq(res.stdout, msg.encode('utf-8') +
b'\n')
+
+ def test_bg_timeout(self):
+ with self as snh:
+ with snh.bg('sleep', 'infinity') as proc:
+ exeter.assert_raises(subprocess.TimeoutExpired,
+ proc.run, timeout=0.1)
+ proc.terminate()
+
+ def test_bg_context_timeout(self):
+ with self as snh:
+ def run_timeout():
+ with snh.bg('sleep', 'infinity',
context_timeout=0.1):
+ pass
+ exeter.assert_raises(subprocess.TimeoutExpired, run_timeout)
+
+ SELFTESTS = [test_true, test_false, test_echo, test_timeout,
+ test_bg_true, test_bg_false, test_bg_echo, test_bg_timeout,
+ test_bg_context_timeout]
+
+ @classmethod
+ def selftest(cls, setup):
+ "Register standard snh tests for instance returned by setup"
+ for t in cls.SELFTESTS:
+ testid = f'{setup.__qualname__}|{t.__qualname__}'
+ exeter.register_pipe(testid, setup, t)
+
+
+class RealHost(SimNetHost):
+ """Represents the host on which the tests are running (as opposed
+ to some simulated host created by the tests)
+
+ """
+
+ def __init__(self):
+ super().__init__('REAL_HOST')
+
+ def hostify(self, *cmd, capable=False, **kwargs):
+ assert not capable, \
+ "BUG: Shouldn't run commands with capabilities on host"
+ return cmd, kwargs
+
+
+SimNetHost.selftest(RealHost)
--
2.45.2