df7a018ee2
This splits out the lorax-composer specific execution so that the built image can be downloaded from the build vm and booted by the host instead of using nested virt to try and boot it inside the build vm. Also adds copying the ssh key from the build vm so that it can log into the image and run the test_boot_* scripts.
234 lines
7.8 KiB
Python
234 lines
7.8 KiB
Python
#!/usr/bin/python3
|
|
|
|
import argparse
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import traceback
|
|
import unittest
|
|
|
|
# import Cockpit's machinery for test VMs and its browser test API
|
|
sys.path.append(os.path.join(os.path.dirname(__file__), "../bots/machine"))
|
|
import testvm # pylint: disable=import-error
|
|
|
|
#pylint: disable=subprocess-run-check
|
|
|
|
def print_exception(etype, value, tb):
|
|
# only include relevant lines
|
|
limit = 0
|
|
while tb and '__unittest' in tb.tb_frame.f_globals:
|
|
limit += 1
|
|
tb = tb.tb_next
|
|
|
|
traceback.print_exception(etype, value, tb, limit=limit)
|
|
|
|
|
|
class VirtMachineTestCase(unittest.TestCase):
|
|
sit = False
|
|
network = None
|
|
machine = None
|
|
ssh_command = None
|
|
|
|
def setUpTestMachine(self, image, identity_file=None):
|
|
self.network = testvm.VirtNetwork(0)
|
|
if identity_file:
|
|
self.machine = testvm.VirtMachine(image, networking=self.network.host(), cpus=2, memory_mb=2048, identity_file=identity_file)
|
|
else:
|
|
self.machine = testvm.VirtMachine(image, networking=self.network.host(), cpus=2, memory_mb=2048)
|
|
|
|
print("Starting virtual machine '{}'".format(image))
|
|
self.machine.start()
|
|
self.machine.wait_boot()
|
|
|
|
# run a command to force starting the SSH master
|
|
self.machine.execute("uptime")
|
|
|
|
self.ssh_command = ["ssh", "-o", "ControlPath=" + self.machine.ssh_master,
|
|
"-p", self.machine.ssh_port,
|
|
self.machine.ssh_user + "@" + self.machine.ssh_address]
|
|
|
|
print("Machine is up. Connect to it via:")
|
|
print(" ".join(self.ssh_command))
|
|
print()
|
|
|
|
def tearDownTestMachine(self):
|
|
if os.environ.get('TEST_ATTACHMENTS'):
|
|
self.machine.download_dir('/var/log/tests', os.environ.get('TEST_ATTACHMENTS'))
|
|
|
|
# Peek into internal data structure, because there's no way to get the
|
|
# TestResult at this point. `errors` is a list of tuples (method, error)
|
|
errors = list(e[1] for e in self._outcome.errors if e[1])
|
|
|
|
if errors and self.sit:
|
|
for e in errors:
|
|
print_exception(*e)
|
|
|
|
print()
|
|
print(" ".join(self.ssh_command))
|
|
input("Press RETURN to continue...")
|
|
|
|
self.machine.stop()
|
|
|
|
def execute(self, command, **args):
|
|
"""Execute a command on the test machine.
|
|
|
|
**args and return value are the same as those for subprocess.run().
|
|
"""
|
|
return subprocess.run(self.ssh_command + command, **args)
|
|
|
|
|
|
class ComposerTestCase(VirtMachineTestCase):
|
|
def setUp(self):
|
|
self.setUpTestMachine(testvm.DEFAULT_IMAGE)
|
|
|
|
# Upload the contents of the ./tests/ directory to the machine (it must have beakerlib already installed)
|
|
self.machine.upload(["../tests"], "/")
|
|
|
|
print("Waiting for lorax-composer to become ready...")
|
|
curl_command = ["curl", "--max-time", "360",
|
|
"--silent",
|
|
"--unix-socket", "/run/weldr/api.socket",
|
|
"http://localhost/api/status"]
|
|
r = subprocess.run(self.ssh_command + curl_command, stdout=subprocess.DEVNULL)
|
|
self.assertEqual(r.returncode, 0)
|
|
|
|
def tearDown(self):
|
|
self.tearDownVirt()
|
|
|
|
def tearDownVirt(self, virt_dir=None, local_dir=None):
|
|
if os.environ.get('TEST_ATTACHMENTS'):
|
|
self.machine.download_dir('/var/log/tests', os.environ.get('TEST_ATTACHMENTS'))
|
|
|
|
if virt_dir and local_dir:
|
|
self.machine.download_dir(virt_dir, local_dir)
|
|
|
|
self.tearDownTestMachine()
|
|
return local_dir
|
|
|
|
def runCliTest(self, script):
|
|
extra_env = []
|
|
if self.sit:
|
|
extra_env.append("COMPOSER_TEST_FAIL_FAST=1")
|
|
|
|
r = self.execute(["CLI=/usr/bin/composer-cli",
|
|
"TEST=" + self.id(),
|
|
"PACKAGE=composer-cli",
|
|
*extra_env,
|
|
"/tests/test_cli.sh", script])
|
|
self.assertEqual(r.returncode, 0)
|
|
|
|
def runImageTest(self, script):
|
|
extra_env = []
|
|
if self.sit:
|
|
extra_env.append("COMPOSER_TEST_FAIL_FAST=1")
|
|
|
|
r = self.execute(["TEST=" + self.id(),
|
|
*extra_env,
|
|
"/tests/test_image.sh", script])
|
|
self.assertEqual(r.returncode, 0)
|
|
|
|
|
|
class ComposerTestResult(unittest.TestResult):
|
|
def name(self, test):
|
|
name = test.id().replace("__main__.", "")
|
|
if test.shortDescription():
|
|
name += ": " + test.shortDescription()
|
|
return name
|
|
|
|
def startTest(self, test):
|
|
super().startTest(test)
|
|
|
|
print("# ----------------------------------------------------------------------")
|
|
print("# ", self.name(test))
|
|
print("", flush=True)
|
|
|
|
def stopTest(self, test):
|
|
print(flush=True)
|
|
|
|
def addSuccess(self, test):
|
|
super().addSuccess(test)
|
|
print("ok {} {}".format(self.testsRun, self.name(test)))
|
|
|
|
def addError(self, test, err):
|
|
super().addError(test, err)
|
|
traceback.print_exception(*err, file=sys.stdout)
|
|
print("not ok {} {}".format(self.testsRun, self.name(test)))
|
|
|
|
def addFailure(self, test, err):
|
|
super().addError(test, err)
|
|
traceback.print_exception(*err, file=sys.stdout)
|
|
print("not ok {} {}".format(self.testsRun, self.name(test)))
|
|
|
|
def addSkip(self, test, reason):
|
|
super().addSkip(test, reason)
|
|
print("ok {} {} # SKIP {}".format(self.testsRun, self.name(test), reason))
|
|
|
|
def addExpectedFailure(self, test, err):
|
|
super().addExpectedFailure(test, err)
|
|
print("ok {} {}".format(self.testsRun, self.name(test)))
|
|
|
|
def addUnexpectedSuccess(self, test):
|
|
super().addUnexpectedSuccess(test)
|
|
print("not ok {} {}".format(self.testsRun, self.name(test)))
|
|
|
|
|
|
class ComposerTestRunner(object):
|
|
"""A test runner that (in combination with ComposerTestResult) outputs
|
|
results in a way that cockpit's log.html can read and format them.
|
|
"""
|
|
|
|
def __init__(self, failfast=False):
|
|
self.failfast = failfast
|
|
|
|
def run(self, testable):
|
|
result = ComposerTestResult()
|
|
result.failfast = self.failfast
|
|
result.startTestRun()
|
|
count = testable.countTestCases()
|
|
print("1.." + str(count))
|
|
try:
|
|
testable(result)
|
|
finally:
|
|
result.stopTestRun()
|
|
return result
|
|
|
|
|
|
def print_tests(tests):
|
|
for test in tests:
|
|
if isinstance(test, unittest.TestSuite):
|
|
print_tests(test)
|
|
elif isinstance(test, unittest.loader._FailedTest):
|
|
name = test.id().replace("unittest.loader._FailedTest.", "")
|
|
print("Error: '{}' does not match a test".format(name), file=sys.stderr)
|
|
else:
|
|
print(test.id().replace("__main__.", ""))
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("tests", nargs="*", help="List of tests modules, classes, and methods")
|
|
parser.add_argument("-l", "--list", action="store_true", help="Print the list of tests that would be executed")
|
|
parser.add_argument("-s", "--sit", action="store_true", help="Halt test execution (but keep VM running) when a test fails")
|
|
args = parser.parse_args()
|
|
|
|
ComposerTestCase.sit = args.sit
|
|
|
|
module = __import__("__main__")
|
|
if args.tests:
|
|
tests = unittest.defaultTestLoader.loadTestsFromNames(args.tests, module)
|
|
else:
|
|
tests = unittest.defaultTestLoader.loadTestsFromModule(module)
|
|
|
|
if args.list:
|
|
print_tests(tests)
|
|
return 0
|
|
|
|
runner = ComposerTestRunner(failfast=args.sit)
|
|
result = runner.run(tests)
|
|
|
|
if tests.countTestCases() != result.testsRun:
|
|
print("Error: unexpected number of tests were run", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
sys.exit(not result.wasSuccessful())
|