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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
|
#!/usr/bin/python3 -cimport os, sys; os.execv(os.path.dirname(sys.argv[1]) + "/../common/pywrap", sys.argv)
#
# Copyright (C) 2018 Red Hat, Inc.
# SPDX-License-Identifier: LGPL-2.1-or-later
import json
import os
import shutil
import subprocess
import tempfile
import unittest
from glob import glob
import testlib
from machine import testvm
@unittest.skipUnless("TEST_OS" in os.environ, "TEST_OS not set")
class TestImageCustomize(unittest.TestCase):
def checkBoot(self, image):
with testvm.Timeout(seconds=300, error_message="Timed out waiting for image to run"):
network = testvm.VirtNetwork(0, image=image)
machine = testvm.VirtMachine(image=image, networking=network.host())
machine.boot()
out = machine.execute('cat /var/custom-test')
machine.stop()
self.assertEqual(out, "hello\n")
def testCustomDir(self):
dest = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, dest)
img = os.path.join(dest, os.environ["TEST_OS"])
with testvm.Timeout(seconds=300, error_message="Timed out waiting for image-customize"):
subprocess.check_call(["bots/image-customize", "--verbose", "--run-command",
"echo hello > /var/custom-test", img])
self.assertTrue(os.path.exists(img))
self.checkBoot(img)
def testBaseImage(self):
img = "custom-" + os.environ["TEST_OS"]
def cleanup():
for f in glob(f"test/images/{img}*"):
os.unlink(f)
self.addCleanup(cleanup)
with testvm.Timeout(seconds=300, error_message="Timed out waiting for image-customize"):
subprocess.check_call(["bots/image-customize", "--verbose", "--run-command",
"echo hello > /var/custom-test", "--base-image", os.environ["TEST_OS"], img])
self.assertTrue(os.path.exists(os.path.join("test/images", img)))
# notice, not giving directory here - test/images/ should be the default
self.checkBoot(img)
def testScriptRelativePath(self):
dest = tempfile.mkdtemp(dir=".")
self.addCleanup(shutil.rmtree, dest)
script = os.path.join(dest, "setup.sh")
with open(script, "w") as f:
f.write("#!/bin/sh -eu\necho hello > /var/custom-test\n")
img = os.path.join(dest, os.environ["TEST_OS"])
with testvm.Timeout(seconds=300, error_message="Timed out waiting for image-customize"):
subprocess.check_call(["bots/image-customize", "--verbose", "--script", script, img])
self.assertTrue(os.path.exists(img))
self.checkBoot(img)
def testUpload(self):
dest = tempfile.mkdtemp(dir=".")
self.addCleanup(shutil.rmtree, dest)
img = os.path.join(dest, os.environ["TEST_OS"])
with testvm.Timeout(seconds=300, error_message="Timed out waiting for image-customize"):
subprocess.check_call(["bots/image-customize", "--verbose",
"--upload", "/etc/passwd:/tmp/passwd",
"--run-command", "echo hello > /var/custom-test",
"--run-command", "grep ^root: /tmp/passwd", img])
self.checkBoot(img)
def testFailurePropagation(self):
dest = tempfile.mkdtemp(dir=".")
self.addCleanup(shutil.rmtree, dest)
img = os.path.join(dest, os.environ["TEST_OS"])
with testvm.Timeout(seconds=300, error_message="Timed out waiting for image-customize"):
subprocess.check_call(["bots/image-customize", "--verbose",
"--run-command", "true", img])
with self.assertRaises(subprocess.CalledProcessError):
subprocess.check_call(["bots/image-customize", "--verbose",
"--run-command", "false", img])
def testResize(self):
dest = tempfile.mkdtemp(dir=".")
self.addCleanup(shutil.rmtree, dest)
img = os.path.join(dest, os.environ["TEST_OS"])
with testvm.Timeout(seconds=300, error_message="Timed out waiting for image-customize"):
subprocess.check_call(["bots/image-customize", "--verbose",
"--resize", "30G", img])
output = subprocess.check_output(["qemu-img", "info", "--output=json", img], encoding="utf-8")
info = json.loads(output)
self.assertEqual(int(info['virtual-size']) // 1024 // 1024 // 1024, 30)
@unittest.skipUnless("TEST_OS" in os.environ, "TEST_OS not set")
class TestBotsVM(unittest.TestCase):
def testBasic(self):
dest = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, dest)
img = os.path.join(dest, os.environ["TEST_OS"])
with testvm.Timeout(seconds=300, error_message="Timed out waiting for image-customize"):
subprocess.check_call(["bots/image-customize", "--verbose", "--run-command",
"echo hello > /var/custom-test", img])
# boot it and wait for RUNNING marker, parse out ssh and cockpit addresses
with testvm.Timeout(seconds=300, error_message="Timed out waiting for testvm.py to boot VM"):
vm = subprocess.Popen(["bots/machine/testvm.py", img],
stdout=subprocess.PIPE, universal_newlines=True)
# first line should be the SSH command
ssh_command = vm.stdout.readline().split()
# second line is the redirected cockpit address
cockpit_address = vm.stdout.readline()
# third should be the "I am ready" flag
running = vm.stdout.readline()
self.assertEqual(running, "RUNNING\n")
self.assertTrue(cockpit_address.startswith("http://127.0.0.2:9"), cockpit_address)
# test SSH command and that we have the expected flag file
self.assertEqual(ssh_command[0], "ssh")
with testvm.Timeout(seconds=30, error_message="Timed out waiting for ssh command"):
out = subprocess.check_output([*ssh_command, "cat", "/var/custom-test"])
self.assertEqual(out, b"hello\n")
# should cleanly stop on SIGTERM
vm.terminate()
with testvm.Timeout(seconds=60, error_message="Timed out waiting for script to terminate"):
self.assertEqual(vm.wait(), 0)
if __name__ == '__main__':
testlib.test_main()
|
