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
|
import io
import json
import logging
import os
import shutil
import yaml
def install_vmadm(ctx):
runcmd_prepend_to_path = globals()["runcmd_prepend_to_path"]
srcdir = globals()["srcdir"]
# Add the directory with built Rust binaries to the path.
default_target = os.path.join(srcdir, "target")
target = os.environ.get("CARGO_TARGET_DIR", default_target)
runcmd_prepend_to_path(ctx, dirname=os.path.join(target, "debug"))
# Set permissions on the datadir and its parent. They are 0o700 by default,
# which means that the libvirt daemon can't access the virtual machine
# image we create.
os.chmod(".", 0o711)
os.chmod("..", 0o711)
# Create .ssh directory, so that the scenario can put files there later.
# This can be removed once the Subplot lib/files library creates
# directories.
os.mkdir(".ssh")
os.makedirs("expected/init-test")
os.mkdir("images")
def ensure_base_image(ctx):
assert_eq = globals()["assert_eq"]
assert_ne = globals()["assert_ne"]
base = os.environ.get("BASE_IMAGE", "")
_, ext = os.path.splitext(base)
assert_ne(base, "")
assert_eq(ext, ".qcow2")
shutil.copy(base, "base.qcow2")
def directories_match(ctx, actual=None, expected=None):
assert_eq = globals()["assert_eq"]
efiles = list(sorted(os.listdir(expected)))
afiles = list(sorted(os.listdir(actual)))
assert_eq(efiles, afiles)
for filename in efiles:
with open(os.path.join(expected, filename)) as f:
edata = yaml.safe_load(f)
with open(os.path.join(actual, filename)) as f:
adata = yaml.safe_load(f)
if "runcmd" in adata:
del adata["runcmd"]
logging.debug(f"directories_match: filename={filename}")
logging.debug(f"directories_match: edata={edata!r}")
logging.debug(f"directories_match: adata={adata!r}")
assert_eq(edata, adata)
def stdout_json_matches(ctx, filename=None):
runcmd_get_stdout = globals()["runcmd_get_stdout"]
stdout = io.StringIO(runcmd_get_stdout(ctx))
actual = yaml.safe_load(stdout)
with open(filename) as f:
expected = json.load(f)
expected = _expand_tilde(expected)
logging.debug(f"actual: {actual}")
logging.debug(f"expect: {expected}")
_assert_equal_objects(actual, expected)
def _expand_tilde(o):
if isinstance(o, str):
return os.path.expanduser(o)
elif isinstance(o, dict):
return {key: _expand_tilde(o[key]) for key in o}
elif isinstance(o, list):
return [_expand_tilde(value) for value in o]
else:
return o
def _assert_equal_objects(a, b):
logging.debug(f"comparing {a!r} and {b!r}")
assert type(a) == type(b)
if isinstance(a, dict):
logging.debug("both a and b are dicts")
for key in a:
assert key in b, f"wanted b to have key {key!r}"
logging.debug(f"comparing values for key {key!r}")
_assert_equal_objects(a[key], b[key])
for key in b:
assert key in a, f"wanted a to have key {key!r}"
elif isinstance(a, list):
assert len(a) == len(
b
), f"wanted a and b to be of same length ({len(a)} vs {len(b)})"
for i in range(len(a)):
_assert_equal_objects(a[i], b[i])
else:
assert a == b, f"wanted {a!r} and {b!r} to be equal"
|
