42c7c0691c
Beakerlib upstream can't do this yet, but might at some point: https://github.com/beakerlib/beakerlib/issues/42 This is only enabled in combination with the `--sit` option of the `test/check-*` scripts. It leaves the system in exacly the state it was in when an assertion failed. Finishing the test run would run cleanup as well (such as deleting created images). It also takes longer.
125 lines
4.2 KiB
Python
125 lines
4.2 KiB
Python
#!/usr/bin/python3
|
|
|
|
import argparse
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
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
|
|
|
|
|
|
def print_exception(etype, value, tb):
|
|
import traceback
|
|
|
|
# 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 ComposerTestCase(unittest.TestCase):
|
|
image = testvm.DEFAULT_IMAGE
|
|
sit = False
|
|
|
|
def setUp(self):
|
|
self.network = testvm.VirtNetwork(0)
|
|
self.machine = testvm.VirtMachine(self.image, networking=self.network.host(), memory_mb=2048)
|
|
|
|
print(f"Starting virtual machine '{self.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()
|
|
|
|
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):
|
|
# 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 = filter(None, [ e[1] for e in self._outcome.errors ])
|
|
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)
|
|
|
|
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 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(f"Error: '{name}' does not match a test", 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 = unittest.TextTestRunner(verbosity=2, failfast=args.sit)
|
|
result = runner.run(tests)
|
|
|
|
sys.exit(not result.wasSuccessful())
|