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
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
|
"""Tests for the VM test framework's vm-utils helpers.
Two robustness bugs under test:
1. init_vm_paths suffixed DISK_PATH and OVMF_VARS by FS_PROFILE but left
PID_FILE, MONITOR_SOCK, and SERIAL_LOG unsuffixed, so a concurrent btrfs
and zfs run shared them -- the second run's stop/is-running logic read the
first run's PID and could kill the other VM. All runtime paths now carry
the profile suffix.
2. kill_qemu sent kill -9 and deleted the PID file without waiting for the
process to die, so a force-kill fallback could run qemu-img snapshot
against a qcow2 the dying qemu still held locked. kill_qemu now waits for
the process to be dead (or reaped) before returning.
Method: sed-extract init_vm_paths / kill_qemu / _cleanup_qemu_files from
lib/vm-utils.sh and drive them with temp dirs and real background processes.
Run from repo root:
python3 -m unittest tests.vm-framework.test_vm_utils
"""
import os
import subprocess
import tempfile
import textwrap
import unittest
REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
VM_UTILS = os.path.join(REPO_ROOT, "scripts", "testing", "lib", "vm-utils.sh")
EXTRACT = (
"source <(sed -n '/^init_vm_paths() {{/,/^}}/p;"
"/^kill_qemu() {{/,/^}}/p;"
"/^_cleanup_qemu_files() {{/,/^}}/p' \"{lib}\")"
).format(lib=VM_UTILS)
def run(body):
script = f"fatal() {{ echo \"FATAL: $*\"; exit 1; }}\nwarn() {{ :; }}\n{EXTRACT}\n{body}\n"
return subprocess.run(
["bash", "-c", script], capture_output=True, text=True, timeout=30,
)
class InitVmPaths(unittest.TestCase):
def paths_for(self, profile):
with tempfile.TemporaryDirectory() as d:
body = textwrap.dedent(f"""\
FS_PROFILE={profile}
init_vm_paths "{d}"
echo "DISK=$DISK_PATH"
echo "VARS=$OVMF_VARS"
echo "PID=$PID_FILE"
echo "MON=$MONITOR_SOCK"
echo "SER=$SERIAL_LOG"
""")
return run(body).stdout
def test_zfs_profile_suffixes_all_runtime_paths(self):
out = self.paths_for("zfs")
self.assertIn("archsetup-base-zfs.qcow2", out)
self.assertIn("OVMF_VARS-zfs.fd", out)
# The bug: these three shared one name across profiles.
self.assertIn("PID=", out)
self.assertRegex(out, r"PID=.*-zfs",
"PID_FILE must carry the profile suffix")
self.assertRegex(out, r"MON=.*-zfs",
"MONITOR_SOCK must carry the profile suffix")
self.assertRegex(out, r"SER=.*-zfs",
"SERIAL_LOG must carry the profile suffix")
def test_btrfs_profile_keeps_legacy_unsuffixed_names(self):
out = self.paths_for("btrfs")
self.assertIn("DISK=", out)
self.assertIn("archsetup-base.qcow2", out)
self.assertNotIn("-btrfs", out)
def test_invalid_profile_fatals(self):
out = self.paths_for("ext4")
self.assertIn("FATAL", out)
class KillQemu(unittest.TestCase):
def test_process_is_dead_before_return(self):
# Contract pin for the snapshot-restore race: when kill_qemu returns,
# the process must be dead or reaped -- state Z or /proc entry gone --
# so a following qemu-img call can't hit a still-held qcow2 lock.
with tempfile.TemporaryDirectory() as d:
body = textwrap.dedent(f"""\
PID_FILE="{d}/qemu.pid"
MONITOR_SOCK="{d}/qemu-monitor.sock"
sleep 60 & victim=$!
echo "$victim" > "$PID_FILE"
kill_qemu
state=$(awk '{{print $3}}' "/proc/$victim/stat" 2>/dev/null || echo GONE)
echo "STATE=$state"
[ -f "$PID_FILE" ] && echo "PIDFILE=present" || echo "PIDFILE=removed"
""")
r = run(body)
self.assertRegex(r.stdout, r"STATE=(Z|GONE)",
f"process still live after kill_qemu: {r.stdout}")
self.assertIn("PIDFILE=removed", r.stdout)
def test_stale_pid_file_is_cleaned_quietly(self):
with tempfile.TemporaryDirectory() as d:
body = textwrap.dedent(f"""\
PID_FILE="{d}/qemu.pid"
MONITOR_SOCK="{d}/qemu-monitor.sock"
echo 99999999 > "$PID_FILE"
kill_qemu
echo "RC=$?"
[ -f "$PID_FILE" ] && echo "PIDFILE=present" || echo "PIDFILE=removed"
""")
r = run(body)
self.assertIn("RC=0", r.stdout)
self.assertIn("PIDFILE=removed", r.stdout)
def test_no_pid_file_is_a_noop(self):
with tempfile.TemporaryDirectory() as d:
body = textwrap.dedent(f"""\
PID_FILE="{d}/qemu.pid"
MONITOR_SOCK="{d}/qemu-monitor.sock"
kill_qemu
echo "RC=$?"
""")
self.assertIn("RC=0", run(body).stdout)
if __name__ == "__main__":
unittest.main()
|
