diff --git a/test/check-api b/test/check-api index d47393cf..ffa5c889 100755 --- a/test/check-api +++ b/test/check-api @@ -1,41 +1,37 @@ #!/usr/bin/python3 -from time import sleep - import composertest import requests import subprocess - class TestApi(composertest.ComposerTestCase): """Test Composer HTTP API""" def setUp(self): - super(TestApi, self).setUp() + super().setUp() # Forward /run/weldr/api.socket to a port on the host # Set ExitOnForwardFailure so that ssh blocks until the forward is set # up before going to the background (-f), which it closes stdout. We # wait for that by calling read() on it. self.composer_port = self.network._lock(8080) - forwarder_command = self.ssh_command[:] - forwarder_command.extend(["-fNT", - "-o", "ExitOnForwardFailure=yes", - "-L", "localhost:%d:/run/weldr/api.socket" % self.composer_port]) + forwarder_command = [*self.ssh_command, "-fNT", + "-o", "ExitOnForwardFailure=yes", + "-L", f"localhost:{self.composer_port}:/run/weldr/api.socket"] self.forwarder_proc = subprocess.Popen(forwarder_command, stdout=subprocess.PIPE) self.forwarder_proc.stdout.read() def tearDown(self): self.forwarder_proc.terminate() - sleep(1) # wait and check for timeout - if self.forwarder_proc.poll() is None: + try: + self.forwarder_proc.wait(timeout=1) + except TimeoutError: self.forwarder_proc.kill() - super(TestApi, self).tearDown() + super().tearDown() - def request(self, method, path, check=True): + def request(self, method, path, json=None, check=True): self.assertEqual(path[0], "/") - url = "http://localhost:%d%s" % (self.composer_port, path) - r = requests.request(method, url, timeout=30) + r = requests.request(method, f"http://localhost:{self.composer_port}{path}", json=json, timeout=30) if check: r.raise_for_status() return r @@ -70,6 +66,23 @@ class TestApi(composertest.ComposerTestCase): "errors": [{ "id": "HTTPError", "code": 405, "msg": "Method Not Allowed" }] }) + # + # API create blueprint with InvalidChars + # + invalid_blueprint = { + "name": "Name,With,Commas", + "description": "", + "version": "0.0.1", + "modules": [], + "groups": [] + } + r = self.request("POST", "/api/v0/blueprints/new", json=invalid_blueprint, check=False) + self.assertEqual(r.status_code, 400) + self.assertEqual(r.json(), { + "status": False, + "errors": [{ "id": "InvalidChars", "msg": "Invalid characters in API path" }] + }) + if __name__ == '__main__': composertest.main() diff --git a/test/composertest.py b/test/composertest.py index 739e9750..f8a49dae 100755 --- a/test/composertest.py +++ b/test/composertest.py @@ -1,11 +1,10 @@ #!/usr/bin/python3 -from __future__ import print_function - import argparse import os import subprocess import sys +import traceback import unittest # import Cockpit's machinery for test VMs and its browser test API @@ -14,8 +13,6 @@ 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: @@ -29,22 +26,11 @@ class ComposerTestCase(unittest.TestCase): image = testvm.DEFAULT_IMAGE sit = False - def __init__(self, methodName='runTest'): - super(ComposerTestCase, self).__init__(methodName=methodName) - # by default run() does this and defaultTestResult() - # always creates new object which is local for the .run() method - self.ci_result = self.defaultTestResult() - - def run(self, result=None): - # so we override run() and use an object attribute which we can - # reference later in tearDown() and extract the errors from - super(ComposerTestCase, self).run(result=self.ci_result) - def setUp(self): self.network = testvm.VirtNetwork(0) self.machine = testvm.VirtMachine(self.image, networking=self.network.host(), memory_mb=2048) - print("Starting virtual machine '%s'" % self.image) + print("Starting virtual machine '{}'".format(self.image)) self.machine.start() self.machine.wait_boot() @@ -64,12 +50,13 @@ class ComposerTestCase(unittest.TestCase): "--silent", "--unix-socket", "/run/weldr/api.socket", "http://localhost/api/status"] - r = subprocess.call(self.ssh_command + curl_command, stdout=open(os.devnull, 'w')) + r = subprocess.run(self.ssh_command + curl_command, stdout=subprocess.DEVNULL) self.assertEqual(r.returncode, 0) def tearDown(self): - # `errors` is a list of tuples (method, error) - errors = list(e[1] for e in self.ci_result.errors if e[1]) + # 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: @@ -81,31 +68,98 @@ class ComposerTestCase(unittest.TestCase): self.machine.stop() - def execute(self, command): - """Execute a command on the test machine.""" - return subprocess.call(self.ssh_command + command) + 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): - execute_params = ["CLI=/usr/bin/composer-cli", + 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", - "/tests/test_cli.sh", script] - if self.sit: - execute_params.insert(0, "COMPOSER_TEST_FAIL_FAST=1") - - r = self.execute(execute_params) + *extra_env, + "/tests/test_cli.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) - # I don't know how this is used when running the tests - # (maybe not used from what it looks like) so not sure how to refactor it - # elif isinstance(test, unittest.loader._FailedTest): - # name = test.id().replace("unittest.loader._FailedTest.", "") - # print("Error: '%s' does not match a test" % name, file=sys.stderr) + 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__.", "")) @@ -129,7 +183,7 @@ def main(): print_tests(tests) return 0 - runner = unittest.TextTestRunner(verbosity=2, failfast=args.sit) + runner = ComposerTestRunner(failfast=args.sit) result = runner.run(tests) if tests.countTestCases() != result.testsRun: diff --git a/tests/pylint/runpylint.py b/tests/pylint/runpylint.py index 04cc1500..7d7223bc 100755 --- a/tests/pylint/runpylint.py +++ b/tests/pylint/runpylint.py @@ -32,6 +32,10 @@ class LoraxLintConfig(PocketLintConfig): FalsePositive(r"Module 'composer' has no 'version' member"), ] + @property + def ignoreNames(self): + return { 'test' } + @property def pylintPlugins(self): retval = super(LoraxLintConfig, self).pylintPlugins