1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
|
import json
class IncusHost(QemuMachine):
def __init__(self, base):
with subtest("Wait for startup"):
base.wait_for_unit("incus.service")
base.wait_for_unit("incus-preseed.service")
with subtest("Verify preseed resources created"):
base.succeed("incus profile show default")
base.succeed("incus network info incusbr0")
base.succeed("incus storage show default")
self._parent = base
# delegate attribute access to the parent
def __getattr__(self, name):
return getattr(self._parent, name)
def instance_exec(self, name: str, command: str, project: str = "default"):
return super().execute(
f"incus exec {name} --disable-stdin --force-interactive --project {project} -- {command}"
)
def instance_succeed(self, name: str, command: str, project: str = "default"):
return super().succeed(
f"incus exec {name} --disable-stdin --force-interactive --project {project} -- {command}"
)
def wait_for_instance(self, name: str, project: str = "default"):
self.wait_instance_exec_success(
name,
"/run/current-system/sw/bin/systemctl is-system-running",
project=project,
)
def wait_instance_exec_success(
self, name: str, command: str, timeout: int = 900, project: str = "default"
):
def check_command(_) -> bool:
status, _ = self.instance_exec(name, command, project)
return status == 0
with super().nested(
f"Waiting for successful instance exec, instance={name}, project={project}, command={command}"
):
retry(check_command, timeout)
def set_instance_config(
self, name: str, config: str, restart: bool = False, unset: bool = False
):
if restart:
super().succeed(f"incus stop {name}")
if unset:
super().succeed(f"incus config unset {name} {config}")
else:
super().succeed(f"incus config set {name} {config}")
if restart:
super().succeed(f"incus start {name}")
self.wait_for_instance(name)
else:
# give a moment to settle
super().sleep(1)
def cleanup(self):
# avoid conflict between preseed and cleanup operations
super().execute("systemctl kill incus-preseed.service")
instances = json.loads(
super().succeed("incus list --format json --all-projects")
)
for instance in [a for a in instances if a["status"] == "Running"]:
super().execute(
f"incus stop --force {instance['name']} --project {instance['project']}"
)
super().execute(
f"incus delete --force {instance['name']} --project {instance['project']}"
)
def check_instance_sysctl(self, name: str, project: str = "default"):
self.instance_succeed(name, "systemctl status systemd-sysctl", project)
sysctl = (
self.instance_succeed(name, "sysctl net.ipv4.ip_forward", project)
.strip()
.split(" ")[-1]
)
assert "1" == sysctl, (
f"systemd-sysctl configuration not correctly applied, {sysctl} != 1"
)
|
