Commit 5fc7a66f authored by chrissi^'s avatar chrissi^
Browse files

parker: Add config for systemd-nspawn-container

Also adapt testsuite to handle multiple Shell-drivers.
parent fe572369
......@@ -31,8 +31,22 @@ def in_good_config(strategy):
def in_good_running(strategy):
strategy.transition("good_running")
def ipv6_link_local(target):
command = target.get_driver("ShellDriver")
@pytest.fixture(scope="function")
def command(target):
return target.get_driver("ShellDriver", name="dut_shell")
@pytest.fixture(scope="function")
def host_container(target):
container = target.get_driver("NSpawnDriver")
shell = target.get_driver("ShellDriver", name="container_shell")
target.activate(shell)
yield shell
target.deactivate(shell)
target.deactivate(container)
def ipv6_link_local(command):
stdout, _, code = command.run('ip address show br-client')
if code != 0:
raise Exception('Failed to get address from br-client:\n{}'.format(''.join(stdout)))
......
import logging
import warnings
import subprocess
import time
import attr
from pexpect import TIMEOUT
import serial
import serial.rfc2217
from labgrid.factory import target_factory
from labgrid.protocol import ConsoleProtocol
from labgrid.driver.common import Driver
from labgrid.driver.consoleexpectmixin import ConsoleExpectMixin
from labgrid.util.proxy import proxymanager
from labgrid.resource import SerialPort
@target_factory.reg_driver
@attr.s(eq=False)
class NSpawnDriver(ConsoleExpectMixin, Driver, ConsoleProtocol):
"""
Driver implementing the ConsoleProtocol interface for a systemd-nspawn container
"""
container_name = attr.ib(validator=attr.validators.instance_of(str))
command = attr.ib(default="systemd-nspawn", validator=attr.validators.instance_of(str))
command_prefix = attr.ib(default="", validator=attr.validators.instance_of(str))
network_interface = attr.ib(default="", validator=attr.validators.instance_of(str))
ephemeral_mode = attr.ib(default=True, validator=attr.validators.instance_of(bool))
txdelay = 0.0
def __attrs_post_init__(self):
super().__attrs_post_init__()
self.logger = logging.getLogger("{}({})".format(self, self.container_name))
self.process = None
def on_activate(self):
# spawn container in subprocess
if self.process:
raise Exception("Nspawn-Driver is already activated.")
args = []
if len(self.command_prefix) > 0:
args.append(self.command_prefix)
args.append(self.command)
args.append("-b")
args.append("--console=interactive")
if self.ephemeral_mode:
args.append("-x")
args.append("--machine={}".format(self.container_name))
if len(self.network_interface) > 0:
args.append("--network-macvlan={}".format(self.network_interface))
self.logger.info("Starting container " +str(args))
self.process = subprocess.Popen(
args,
bufsize=0,
stdin=subprocess.PIPE,
stderr=subprocess.STDOUT,
stdout=subprocess.PIPE,
)
self._poll()
def on_deactivate(self):
self._poll()
if self.process:
subprocess.run(["sudo", "kill", str(self.process.pid)])
for _ in range(5):
if not self.process.poll():
time.sleep(1)
else:
print("systemd-nspawn container exited in time. thanks! =)")
break
else:
print("systemd-nspawn failed to exit in time. killing {}".format(self.process.pid))
subprocess.run(["sudo", "kill", "-9", str(self.process.pid)])
self.process = None
def _read(self, size: int = 1, timeout: float = 0.0):
"""
Reads 'size' or more bytes from the subprocess's stdout
Keyword Arguments:
size -- amount of bytes to read, defaults to 1
"""
self._poll()
return self.process.stdout.read(size)
def _write(self, data: bytes):
"""
Writes 'data' to the serialport
Arguments:
data -- data to write, must be bytes
"""
self._poll()
self.process.stdin.write(data)
self.process.stdin.flush()
def _poll(self):
if self.process.poll():
self.logger.error("systemd-nspawn container {} terminated unexpected!".format(self.container_name))
raise Exception("systemd-nspan container {} terminated unexpected!".format(self.container_name))
......@@ -8,18 +8,37 @@ targets:
drivers:
- SerialDriver:
txdelay: 0.01
name: dut_serial
- NetworkPowerDriver: {}
- SmallUBootDriver:
prompt: 'ap143-2\.0> '
boot_expression: 'Autobooting in 1 seconds'
boot_secret: "tpl"
bindings: { console: "dut_serial" }
- ShellDriver:
prompt: 'root@[\w\(\)-]+:[^ ]+'
login_prompt: ' login: '
username: 'root'
console_ready: 'Please press Enter to activate this console.'
bindings: { console: "dut_serial" }
name: dut_shell
- SmallUBootStrategy: {}
- GpioDigitalOutputDriver: {}
- NSpawnDriver:
container_name: debian
command_prefix: sudo
network_interface: dut
name: container
ephemeral_mode: False
- ShellDriver:
prompt: 'root@[\w\(\)-]+:[^ ]+'
login_prompt: ' login: '
username: 'root'
password: 'root'
bindings: { console: "container" }
name: container_shell
imports:
- "parker.py"
- "nspawndriver.py"
......@@ -83,18 +83,16 @@ def test_good_config_autoupdater_enabled(target, in_good_config):
assert_web("http://192.168.1.1/cgi-bin/config/admin/autoupdater", 'value="1" id="id.1.1.enabled" name="id.1.1.enabled" checked="checked"')
@pytest.mark.config_mode
def test_generated_wg_key(target, in_good_config):
command = target.get_driver("ShellDriver")
def test_generated_wg_key(target, in_good_config, command):
stdout, _, code = command.run('cat /etc/parker/wg-privkey | wg pubkey')
assert code == 0, "Verification of WG private key failed: {}".format(''.join(stdout))
@pytest.mark.config_mode
def test_ssh_authorized_keys(target, in_good_config):
def test_ssh_authorized_keys(target, in_good_config, command):
r = requests.get('http://192.168.1.1/cgi-bin/config/admin/remote')
match = re.search('<textarea[^>]+>(.*)</textarea>', r.text, re.DOTALL)
assert match, 'Could not fetch authorized keys'
keys = match[1].strip()
command = target.get_driver("ShellDriver")
stdout, _, code = command.run('cat /lib/gluon/release')
assert code == 0, 'Error getting firmware version'
version = ''.join(stdout).strip()
......@@ -111,8 +109,8 @@ def test_transition_to_running_mode(in_good_running):
@pytest.mark.running_mode
@pytest.mark.flaky(reruns=10, reruns_delay=10)
def test_good_running_ssh(target, in_good_running, config):
ip = ipv6_link_local(target)
def test_good_running_ssh(target, in_good_running, config, command):
ip = ipv6_link_local(command)
r = subprocess.run(["ssh", "-o", "UserKnownHostsFile=/dev/null", "-o", "StrictHostKeyChecking=no", "-o", "PreferredAuthentications=publickey", "-i", config["ssh_key_file"], 'root@{}%dut'.format(ip), "echo 1"])
assert r.returncode == 0, "SSH connection during run mode failed"
......@@ -131,8 +129,7 @@ def test_good_running_ssh(target, in_good_running, config):
("wireless.radio0.basic_rate", "'1000'", False),
("wireless.radio0.basic_rate", "'2000'", False)
] )
def test_good_running_uci_config(target, in_good_running, uci_key, uci_value, should_be_set):
command = target.get_driver("ShellDriver")
def test_good_running_uci_config(target, in_good_running, uci_key, uci_value, should_be_set, command):
stdout, _, _ = command.run("uci show {}".format(uci_key))
......@@ -155,21 +152,18 @@ def test_good_running_uci_config(target, in_good_running, uci_key, uci_value, sh
assert test_value != uci_value, "UCI field {} is set to value {}".format(uci_key, uci_value)
@pytest.mark.running_mode
def test_good_running_private_wifi_ssid(target, in_good_running, config):
command = target.get_driver("ShellDriver")
def test_good_running_private_wifi_ssid(target, in_good_running, config, command):
stdout, _, _ = command.run("iw dev")
assert config["private_wifi_ssid"] in " ".join(stdout), "Did not find any wifi-interface with SSID {}".format(config["private_wifi_ssid"])
@pytest.mark.running_mode
def test_good_running_private_wifi_psk(target, in_good_running, config):
command = target.get_driver("ShellDriver")
def test_good_running_private_wifi_psk(target, in_good_running, config, command):
stdout, _, _ = command.run("uci show wireless")
assert config["private_wifi_psk"] in " ".join(stdout), "Did not find any wifi-interface with PSK {}".format(config["private_wifi_psk"])
@pytest.mark.running_mode
@pytest.mark.flaky(reruns=20, reruns_delay=10)
def test_good_running_wan_dns(target, in_good_running, config):
command = target.get_driver("ShellDriver")
def test_good_running_wan_dns(target, in_good_running, config, command):
stdout, _, _ = command.run(
"LIBPACKETMARK_MARK=1 LD_PRELOAD=libpacketmark.so ping -4 -c 1 gandolf.stratum0.org"
)
......@@ -177,8 +171,7 @@ def test_good_running_wan_dns(target, in_good_running, config):
@pytest.mark.running_mode
@pytest.mark.flaky(reruns=20, reruns_delay=10)
def test_good_running_wan_ping_confighost(target, in_good_running, config):
command = target.get_driver("ShellDriver")
def test_good_running_wan_ping_confighost(target, in_good_running, config, command):
stdout, _, _ = command.run(
"LIBPACKETMARK_MARK=1 LD_PRELOAD=libpacketmark.so ping -4 -c 10 config.freifunk-bs.de"
)
......@@ -193,8 +186,7 @@ def test_good_running_wan_ping_confighost(target, in_good_running, config):
@pytest.mark.running_mode
@pytest.mark.flaky(reruns=20, reruns_delay=10)
def test_good_running_wan_curl_confighost(target, in_good_running, config):
command = target.get_driver("ShellDriver")
def test_good_running_wan_curl_confighost(target, in_good_running, config, command):
stdout, _, rc = command.run(
"LIBPACKETMARK_MARK=1 LD_PRELOAD=libpacketmark.so wget http://config.freifunk-bs.de/config"
)
......@@ -205,8 +197,7 @@ def test_good_running_wan_curl_confighost(target, in_good_running, config):
@pytest.mark.running_mode
@pytest.mark.flaky(reruns=20, reruns_delay=10)
def test_good_running_count_wg_ifs(target, in_good_running, config):
command = target.get_driver("ShellDriver")
def test_good_running_count_wg_ifs(target, in_good_running, config, command):
stdout, _, rc = command.run(
"ip link"
)
......@@ -219,8 +210,7 @@ def test_good_running_count_wg_ifs(target, in_good_running, config):
@pytest.mark.running_mode
@pytest.mark.flaky(reruns=20, reruns_delay=10)
def test_good_running_ipv4_default(target, in_good_running, config):
command = target.get_driver("ShellDriver")
def test_good_running_ipv4_default(target, in_good_running, config, command):
stdout, _, rc = command.run(
"ip -4 route"
)
......@@ -233,8 +223,7 @@ def test_good_running_ipv4_default(target, in_good_running, config):
@pytest.mark.running_mode
@pytest.mark.flaky(reruns=20, reruns_delay=10)
def test_good_running_ipv6_default(target, in_good_running, config):
command = target.get_driver("ShellDriver")
def test_good_running_ipv6_default(target, in_good_running, config, command):
stdout, _, rc = command.run(
"ip -6 route"
)
......@@ -248,8 +237,7 @@ def test_good_running_ipv6_default(target, in_good_running, config):
@pytest.mark.running_mode
@pytest.mark.flaky(reruns=20, reruns_delay=10)
def test_good_running_ping_v4_parker(target, in_good_running, config):
command = target.get_driver("ShellDriver")
def test_good_running_ping_v4_parker(target, in_good_running, config, command):
stdout, _, rc = command.run(
"ping -c 10 -4 gandolf.stratum0.org"
)
......@@ -264,8 +252,7 @@ def test_good_running_ping_v4_parker(target, in_good_running, config):
@pytest.mark.running_mode
@pytest.mark.flaky(reruns=20, reruns_delay=10)
def test_good_running_ping_v6_parker(target, in_good_running, config):
command = target.get_driver("ShellDriver")
def test_good_running_ping_v6_parker(target, in_good_running, config, command):
stdout, _, rc = command.run(
"ping -c 10 -6 gandolf.stratum0.org"
)
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment