1a20a9a0dd
Define the remote endpoint "model". To execute most meaningful device
driver tests we need to be able to communicate with a remote system,
and have it send traffic to the device under test.
Various test environments will have different requirements.
0) "Local" netdevsim-based testing can simply use net namespaces.
netdevsim supports connecting two devices now, to form a veth-like
construct.
1) Similarly on hosts with multiple NICs, the NICs may be connected
together with a loopback cable or internal device loopback.
One interface may be placed into separate netns, and tests
would proceed much like in the netdevsim case. Note that
the loopback config or the moving of one interface
into a netns is not expected to be part of selftest code.
2) Some systems may need to communicate with the remote endpoint
via SSH.
3) Last but not least environment may have its own custom communication
method.
Fundamentally we only need two operations:
- run a command remotely
- deploy a binary (if some tool we need is built as part of kselftests)
Wrap these two in a class. Use dynamic loading to load the Remote
class. This will allow very easy definition of other communication
methods without bothering upstream code base.
Stick to the "simple" / "no unnecessary abstractions" model for
referring to the remote endpoints. The host / remote object are
passed as an argument to the usual cmd() or ip() invocation.
For example:
ip("link show", json=True, host=remote)
Reviewed-by: Willem de Bruijn <willemb@google.com>
Link: https://lore.kernel.org/r/20240420025237.3309296-2-kuba@kernel.org
Signed-off-by: Jakub Kicinski <kuba@kernel.org>
49 lines
1.5 KiB
Python
49 lines
1.5 KiB
Python
# SPDX-License-Identifier: GPL-2.0
|
|
|
|
import json as _json
|
|
import subprocess
|
|
|
|
class cmd:
|
|
def __init__(self, comm, shell=True, fail=True, ns=None, background=False, host=None):
|
|
if ns:
|
|
comm = f'ip netns exec {ns} ' + comm
|
|
|
|
self.stdout = None
|
|
self.stderr = None
|
|
self.ret = None
|
|
|
|
self.comm = comm
|
|
if host:
|
|
self.proc = host.cmd(comm)
|
|
else:
|
|
self.proc = subprocess.Popen(comm, shell=shell, stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE)
|
|
if not background:
|
|
self.process(terminate=False, fail=fail)
|
|
|
|
def process(self, terminate=True, fail=None):
|
|
if terminate:
|
|
self.proc.terminate()
|
|
stdout, stderr = self.proc.communicate(timeout=5)
|
|
self.stdout = stdout.decode("utf-8")
|
|
self.stderr = stderr.decode("utf-8")
|
|
self.proc.stdout.close()
|
|
self.proc.stderr.close()
|
|
self.ret = self.proc.returncode
|
|
|
|
if self.proc.returncode != 0 and fail:
|
|
if len(stderr) > 0 and stderr[-1] == "\n":
|
|
stderr = stderr[:-1]
|
|
raise Exception("Command failed: %s\nSTDOUT: %s\nSTDERR: %s" %
|
|
(self.proc.args, stdout, stderr))
|
|
|
|
|
|
def ip(args, json=None, ns=None, host=None):
|
|
cmd_str = "ip "
|
|
if json:
|
|
cmd_str += '-j '
|
|
cmd_str += args
|
|
cmd_obj = cmd(cmd_str, ns=ns, host=host)
|
|
if json:
|
|
return _json.loads(cmd_obj.stdout)
|
|
return cmd_obj
|