mirror of
https://github.com/clearlinux/swupd-client.git
synced 2026-09-08 06:31:50 +00:00
William Douglas
7544162243
Add support in update to properly support bundles that include other bundles inside them. This feature is intended to work with bundle additions by adding included bundles (from bundles already subscribed) to the subscription list prior to downloading packs and full files. As part of this change the add_subscriptions function from bundle.c was modified to handle retrying manifest downloads.
580 lines
24 KiB
Python
Executable File
580 lines
24 KiB
Python
Executable File
#!/usr/bin/python3
|
|
|
|
# swupd functional test driver
|
|
#
|
|
# Things to test:
|
|
# update
|
|
# update --status
|
|
# update --download
|
|
# verify
|
|
# verify --fix
|
|
# verify --install
|
|
# bundle add
|
|
# bundle add --list
|
|
# bundle remove
|
|
# check-update
|
|
#
|
|
# Out of scope (for now at least):
|
|
# Files going into /boot (integration test)
|
|
#
|
|
# Current issues with not running as root:
|
|
# Use of system /var
|
|
# chown of files to root for hash comparison
|
|
|
|
import functools
|
|
import inspect
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import unittest
|
|
|
|
from tap import TAPTestRunner
|
|
|
|
COMMAND_MAP = {'bundleadd': 'bundle-add',
|
|
'bundleremove': 'bundle-remove',
|
|
'checkupdate': 'check-update',
|
|
'hashdump': 'hashdump',
|
|
'update': 'update',
|
|
'search': 'search',
|
|
'verify': 'verify'}
|
|
PATH_PREFIX = 'test/functional'
|
|
PORT = '8089'
|
|
|
|
|
|
def path_from_name(name, path_type):
|
|
"""Based on a test class name and path_type, create a path"""
|
|
# Basic conversion looks like:
|
|
# class name == 'subfunction' + '_' + 'test_name'
|
|
# Paths will use '-' instead of '_'
|
|
# So conversion will split on '_' and the first element will be the
|
|
# subfunction test directory and the rest will be a specific test
|
|
# folder within that subfunction
|
|
# So the checkupdate_version_match class would use the
|
|
# test/functional/checkupdate/version-match/ path
|
|
# with path_type determining the base folder
|
|
type_map = {'web': 'web-dir', 'target': 'target-dir'}
|
|
folders = name.split('_')
|
|
subfunction_path = folders[0]
|
|
test_dir = ('-').join(folders[1:])
|
|
if path_type == 'plain':
|
|
return os.path.join(PATH_PREFIX, subfunction_path, test_dir)
|
|
else:
|
|
return os.path.join(PATH_PREFIX, subfunction_path, test_dir,
|
|
type_map[path_type])
|
|
|
|
|
|
def http_server(func):
|
|
"""Create an http server for the lifetime of a test function"""
|
|
@functools.wraps(func)
|
|
def wrapper(*args):
|
|
"""http_server wrapper"""
|
|
web_path = path_from_name(args[0].__class__.__name__, 'web')
|
|
httpd = subprocess.Popen(['python3', '-m', 'http.server', PORT],
|
|
cwd=web_path,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL)
|
|
# Stupid but give the web server a chance to run
|
|
time.sleep(.1)
|
|
try:
|
|
func(*args)
|
|
finally:
|
|
httpd.kill()
|
|
return wrapper
|
|
|
|
|
|
def with_command(func):
|
|
"""Add the command variable to the lexical environment of a function"""
|
|
@functools.wraps(func)
|
|
def wrapper(*args):
|
|
"""with_command wrapper"""
|
|
target_path = path_from_name(args[0].__class__.__name__, 'target')
|
|
subcommand = COMMAND_MAP[args[0].__class__.__name__.split('_')[0]]
|
|
command = ['sudo', './swupd', subcommand, '-u', 'http://localhost/',
|
|
'-P', PORT, '-p', target_path]
|
|
func(*args, command)
|
|
return wrapper
|
|
|
|
|
|
def http_command(**kwargs):
|
|
"""Make the standard test class method"""
|
|
def class_wrapper(original_class):
|
|
"""Update class to add test function"""
|
|
@http_server
|
|
@with_command
|
|
def runTest(*args):
|
|
subprocess.run('sudo rm -fr /var/lib/swupd'.split())
|
|
"""Basic test wrapper function"""
|
|
# Options are for subcommand so insert after the subcommand index.
|
|
# Currently defaults to the -u option
|
|
if kwargs.get('option') != "":
|
|
i = args[1].index('-u')
|
|
args[1][i:i] = kwargs['option'].split(' ')
|
|
process = subprocess.run(args[1], stdout=subprocess.PIPE,
|
|
universal_newlines=True)
|
|
args[0].validate(process.stdout.splitlines())
|
|
msg = kwargs.get('skip')
|
|
if msg:
|
|
original_class.runTest = unittest.skip(msg)(runTest)
|
|
else:
|
|
original_class.runTest = runTest
|
|
return original_class
|
|
return class_wrapper
|
|
|
|
|
|
@http_command(option="test-bundle")
|
|
class bundleadd_add_directory(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn('Downloading test-bundle pack for version 10',
|
|
test_output)
|
|
self.assertIn('Installing bundle(s) files...', test_output)
|
|
self.assertIn('Tracking test-bundle bundle on the system', test_output)
|
|
|
|
|
|
@http_command(option="test-bundle1 test-bundle2")
|
|
class bundleadd_add_multiple(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
newfile = os.path.join(os.getcwd(),
|
|
path_from_name(__class__.__name__, 'target'),
|
|
'usr/foo')
|
|
newdir = os.path.join(os.getcwd(),
|
|
path_from_name(__class__.__name__, 'target'),
|
|
'usr/bin')
|
|
self.assertTrue(os.path.isfile(newfile))
|
|
self.assertTrue(os.path.isdir(newdir))
|
|
|
|
|
|
@http_command(option="test-bundle")
|
|
class bundleadd_boot_file(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn('Downloading test-bundle pack for version 10',
|
|
test_output)
|
|
self.assertIn('Installing bundle(s) files...', test_output)
|
|
self.assertIn('Tracking test-bundle bundle on the system', test_output)
|
|
# bundle-add output does not list files it did or did not install,
|
|
# so check for the file existence in the target directory.
|
|
newfile = os.path.join(os.getcwd(),
|
|
path_from_name(__class__.__name__, 'target'),
|
|
'usr/lib/kernel/testfile')
|
|
self.assertTrue(os.path.isfile(newfile))
|
|
|
|
|
|
@http_command(option="test-bundle")
|
|
class bundleadd_include(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn('Added bundle test-bundle for installation', test_output)
|
|
self.assertIn('Added bundle os-core for installation', test_output)
|
|
newfile = os.path.join(os.getcwd(),
|
|
path_from_name(__class__.__name__, 'target'),
|
|
'usr/foo')
|
|
newdir = os.path.join(os.getcwd(),
|
|
path_from_name(__class__.__name__, 'target'),
|
|
'usr/bin')
|
|
self.assertTrue(os.path.isfile(newfile))
|
|
self.assertTrue(os.path.isdir(newdir))
|
|
|
|
|
|
@http_command(option="--list")
|
|
class bundleadd_list(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn('os-core', test_output)
|
|
|
|
|
|
@http_command(option="test-bundle")
|
|
class bundleremove_boot_file(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn('Deleting bundle files...', test_output)
|
|
self.assertIn('Total deleted files: 1', test_output)
|
|
self.assertIn('Untracking bundle from system...', test_output)
|
|
self.assertIn('Success: Bundle removed', test_output)
|
|
# bundle-remove output does not list files it removed,
|
|
# so check for the file absence in the target directory.
|
|
newfile = os.path.join(os.getcwd(),
|
|
path_from_name(__class__.__name__, 'target'),
|
|
'usr/lib/kernel/testfile')
|
|
self.assertFalse(os.path.isfile(newfile))
|
|
|
|
|
|
@http_command(option="test-bundle")
|
|
class bundleremove_remove_file(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn('Deleting bundle files...', test_output)
|
|
self.assertIn('Total deleted files: 1', test_output)
|
|
self.assertIn('Untracking bundle from system...', test_output)
|
|
self.assertIn('Success: Bundle removed', test_output)
|
|
|
|
# Positive test for a library in lib32/
|
|
@http_command(option="-l -a test-lib32")
|
|
class search_content_check_poslib32(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn('\'test-bundle\' : \'/usr/lib/test-lib32\'', test_output)
|
|
self.assertNotIn('/libtest-nohit', test_output)
|
|
|
|
# Positive test for a binary
|
|
@http_command(option="-b test-bin")
|
|
class search_content_check_posbin(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn('\'test-bundle\' : \'/usr/bin/test-bin\'', test_output)
|
|
self.assertNotIn('/libtest-nohit', test_output)
|
|
|
|
# Posititve test for a binary using the -e everywhere flag
|
|
@http_command(option="-e test-bin")
|
|
class search_content_check_posebin(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn('\'test-bundle\' : \'/usr/bin/test-bin\'', test_output)
|
|
self.assertNotIn('/libtest-nohit', test_output)
|
|
|
|
# Negative test for a binary using the -l library flag
|
|
@http_command(option="-l test-bin")
|
|
class search_content_check_poslbin(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertNotIn('\'test-bundle\' : \'/usr/bin/test-bin\'', test_output)
|
|
self.assertNotIn('/libtest-nohit', test_output)
|
|
|
|
# Positive test for a library in lib64/
|
|
@http_command(option="-l -a test-lib64")
|
|
class search_content_check_poslib64(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn('\'test-bundle\' : \'/usr/lib64/test-lib64\'', test_output)
|
|
self.assertNotIn('/libtest-nohit', test_output)
|
|
|
|
#Negative test for a non-existent library
|
|
@http_command(option="-a -l test-lib36")
|
|
class search_content_check_neglib32(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertNotIn('\'test-bundle\' : \'test-lib36\'', test_output)
|
|
self.assertNotIn('/libtest-nohit', test_output)
|
|
|
|
# Negative test for a file which exists but is not a library
|
|
@http_command(option="-l -a libtest-nohit")
|
|
class search_content_check_neglibtest(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertNotIn('\'test-bundle\' : \'libtest-nohit\'', test_output)
|
|
self.assertNotIn('/libtest-nohit', test_output)
|
|
|
|
# Positive test for a file, specifying full path
|
|
@http_command(option="/usr/lib64/test-lib64")
|
|
class search_content_check_posfull_path(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn('\'test-bundle\' : \'/usr/lib64/test-lib64\'', test_output)
|
|
self.assertNotIn('/libtest-nohit', test_output)
|
|
|
|
# Nagative test for a non-existent file, specifying full path
|
|
@http_command(option="/usr/lib64/test-lib100")
|
|
class search_content_check_negfull_path(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertNotIn('\'test-bundle\' : \'/usr/lib64/test-lib100\'', test_output)
|
|
self.assertNotIn('\'test-bundle\' : \'/usr/lib64/test-lib64\'', test_output)
|
|
self.assertNotIn('/libtest-nohit', test_output)
|
|
|
|
# Negative test for a file which exists, but providing incorrect path
|
|
@http_command(option="/usr/lib/test-lib64")
|
|
class search_content_check_negwrongpath(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertNotIn('\'test-bundle\' : \'/usr/lib/test-lib64\'', test_output)
|
|
self.assertNotIn('/libtest-nohit', test_output)
|
|
|
|
@http_command(option="")
|
|
class checkupdate_version_match(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn('There are no updates available', test_output)
|
|
|
|
|
|
@http_command(option="")
|
|
class checkupdate_new_version(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn('There is a new OS version available: 100', test_output)
|
|
|
|
|
|
@http_command(option="")
|
|
class checkupdate_no_server_content(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn('Cannot reach update server', test_output)
|
|
|
|
|
|
@http_command(option="")
|
|
class checkupdate_no_target_content(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn('Unable to determine current OS version', test_output)
|
|
|
|
|
|
class hashdump_file_hash(unittest.TestCase):
|
|
def runTest(self):
|
|
target_path = path_from_name(self.__class__.__name__, 'target')
|
|
subcommand = COMMAND_MAP[self.__class__.__name__.split('_')[0]]
|
|
command = ['./swupd', subcommand, "--basepath={0}/{1}"
|
|
.format(os.getcwd(), target_path), "/test-hash"]
|
|
process = subprocess.run(command, stdout=subprocess.PIPE,
|
|
universal_newlines=True)
|
|
test_output = process.stdout.splitlines()
|
|
self.assertIn('b03e11e3307de4d4f3f545c8a955c2208507dbc1927e9434d1da42917712c15b', test_output)
|
|
|
|
|
|
@http_command(option="")
|
|
class update_boot_file(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn(' new files : 1', test_output)
|
|
# update must always add a new boot file, so check for the file
|
|
# existence in the target directory.
|
|
newfile = os.path.join(os.getcwd(),
|
|
path_from_name(__class__.__name__, 'target'),
|
|
'usr/lib/kernel/testfile')
|
|
self.assertTrue(os.path.isfile(newfile))
|
|
|
|
|
|
@http_command(option="--download")
|
|
class update_download(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn(' changed files : 1', test_output)
|
|
self.assertIn(' changed manifests : 1', test_output)
|
|
|
|
|
|
@http_command(option="")
|
|
class update_include(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
core = os.path.join(os.getcwd(),
|
|
path_from_name(__class__.__name__, 'target'),
|
|
'usr/core')
|
|
self.assertTrue(os.path.isfile(core))
|
|
test1 = os.path.join(os.getcwd(),
|
|
path_from_name(__class__.__name__, 'target'),
|
|
'usr/bin')
|
|
self.assertTrue(os.path.isdir(test1))
|
|
test2 = os.path.join(os.getcwd(),
|
|
path_from_name(__class__.__name__, 'target'),
|
|
'usr/foo2')
|
|
self.assertTrue(os.path.isfile(test2))
|
|
test3 = os.path.join(os.getcwd(),
|
|
path_from_name(__class__.__name__, 'target'),
|
|
'usr/foo3')
|
|
self.assertTrue(os.path.isfile(test3))
|
|
test4 = os.path.join(os.getcwd(),
|
|
path_from_name(__class__.__name__, 'target'),
|
|
'usr/foo4')
|
|
self.assertTrue(os.path.isfile(test4))
|
|
test4 = os.path.join(os.getcwd(),
|
|
path_from_name(__class__.__name__, 'target'),
|
|
'usr/foo4')
|
|
self.assertTrue(os.path.isfile(test4))
|
|
test5 = os.path.join(os.getcwd(),
|
|
path_from_name(__class__.__name__, 'target'),
|
|
'usr/foo5')
|
|
self.assertTrue(os.path.isfile(test5))
|
|
test6 = os.path.join(os.getcwd(),
|
|
path_from_name(__class__.__name__, 'target'),
|
|
'usr/foo6')
|
|
self.assertTrue(os.path.isfile(test6))
|
|
test7 = os.path.join(os.getcwd(),
|
|
path_from_name(__class__.__name__, 'target'),
|
|
'usr/foo7')
|
|
self.assertTrue(os.path.isfile(test7))
|
|
self.assertIn('os-core bundle already installed, skipping it', test_output)
|
|
self.assertIn('Added bundle test-bundle2 for installation', test_output)
|
|
self.assertIn('Added bundle test-bundle3 for installation', test_output)
|
|
self.assertIn('Added bundle test-bundle4 for installation', test_output)
|
|
self.assertIn('Added bundle test-bundle5 for installation', test_output)
|
|
self.assertIn('Added bundle test-bundle6 for installation', test_output)
|
|
self.assertIn('Added bundle test-bundle7 for installation', test_output)
|
|
|
|
|
|
@http_command(option="")
|
|
class update_missing_os_core(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
os_core_dir = os.path.join(os.getcwd(),
|
|
path_from_name(__class__.__name__, 'target'),
|
|
'usr/bin')
|
|
self.assertTrue(os.path.isdir(os_core_dir))
|
|
|
|
|
|
@http_command(option="--status")
|
|
class update_status(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn('Current OS version: 10', test_output)
|
|
self.assertIn('Latest server version: 100', test_output)
|
|
|
|
|
|
@http_command(option="--status")
|
|
class update_status_no_server_content(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn('Current OS version: 10', test_output)
|
|
self.assertIn('Cannot get latest the server version.Could not reach server', test_output)
|
|
|
|
|
|
@http_command(option="--status")
|
|
class update_status_no_target_content(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn('Cannot determine current OS version', test_output)
|
|
self.assertIn('Latest server version: 100', test_output)
|
|
|
|
|
|
@http_command(option="--status")
|
|
class update_status_version_single_quote(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn('Current OS version: 10', test_output)
|
|
self.assertIn('Latest server version: 100', test_output)
|
|
|
|
|
|
@http_command(option="--status")
|
|
class update_status_version_double_quote(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn('Current OS version: 10', test_output)
|
|
self.assertIn('Latest server version: 100', test_output)
|
|
|
|
|
|
@http_command(option="")
|
|
class update_use_full_file(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn(' changed files : 1', test_output)
|
|
self.assertIn(' changed manifests : 1', test_output)
|
|
self.assertIn('Staging file content', test_output)
|
|
self.assertIn('Update was applied.', test_output)
|
|
self.assertIn('Update successful. System updated from version 10 to version 100', test_output)
|
|
|
|
@http_command(option="")
|
|
class update_verify_fix_path_missing_dir(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn(' changed files : 1', test_output)
|
|
self.assertIn(' changed manifests : 1', test_output)
|
|
self.assertIn('Staging file content', test_output)
|
|
target_path = os.path.join(os.getcwd(), path_from_name(__class__.__name__, 'target') , 'usr/bin')
|
|
self.assertIn('Update target directory does not exist: ' + target_path + '. Trying to fix it', test_output)
|
|
self.assertIn('Update was applied.', test_output)
|
|
self.assertIn('Update successful. System updated from version 10 to version 100', test_output)
|
|
|
|
|
|
@http_command(option="")
|
|
class update_verify_fix_path_hash_mismatch(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn(' changed files : 1', test_output)
|
|
self.assertIn(' changed manifests : 1', test_output)
|
|
self.assertIn('Staging file content', test_output)
|
|
target_path = os.path.join(os.getcwd(), path_from_name(__class__.__name__, 'target') , 'usr/bin')
|
|
self.assertIn('Update target directory does not exist: ' + target_path + '. Trying to fix it', test_output)
|
|
self.assertIn('Hash did not match for path : /usr', test_output)
|
|
self.assertIn('Path /usr/bin is missing on the file system', test_output)
|
|
self.assertIn('Update was applied.', test_output)
|
|
self.assertIn('Update successful. System updated from version 10 to version 100', test_output)
|
|
|
|
|
|
@http_command(option="")
|
|
class update_use_pack(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn(' changed files : 1', test_output)
|
|
self.assertIn(' changed manifests : 1', test_output)
|
|
self.assertIn('Staging file content', test_output)
|
|
self.assertIn('Update was applied.', test_output)
|
|
self.assertIn('Update successful. System updated from version 10 to version 100', test_output)
|
|
|
|
|
|
@http_command(option="--fix")
|
|
class verify_add_missing_directory(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn('Verifying version 10', test_output)
|
|
self.assertIn(' 1 files were missing', test_output)
|
|
self.assertIn(' 1 of 1 missing files were replaced', test_output)
|
|
self.assertIn('Fix successful', test_output)
|
|
|
|
|
|
@http_command(option="")
|
|
class verify_boot_file_mismatch(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn('Verifying version 10', test_output)
|
|
corruptfile = os.path.join(os.getcwd(),
|
|
path_from_name(__class__.__name__, 'target'),
|
|
'usr/lib/kernel/testfile')
|
|
self.assertIn('Hash mismatch for file: ' + corruptfile, test_output)
|
|
self.assertIn(' 1 files did not match', test_output)
|
|
self.assertIn('Verify successful', test_output)
|
|
|
|
|
|
@http_command(option="")
|
|
class verify_check_missing_directory(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn('Verifying version 10', test_output)
|
|
self.assertIn(' 1 files did not match', test_output)
|
|
self.assertIn('Verify successful', test_output)
|
|
|
|
|
|
@http_command(option="--fix")
|
|
class verify_boot_file_deleted(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn('Verifying version 10', test_output)
|
|
# Deleted boot files should be ignored completely
|
|
self.assertNotIn(' 1 files found which should be deleted', test_output)
|
|
bootfile = os.path.join(os.getcwd(),
|
|
path_from_name(__class__.__name__, 'target'),
|
|
'usr/lib/kernel/testfile')
|
|
self.assertTrue(os.path.isfile(bootfile))
|
|
self.assertIn('Fix successful', test_output)
|
|
|
|
|
|
@http_command(option="--fix")
|
|
class verify_boot_file_mismatch_fix(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn('Verifying version 10', test_output)
|
|
corruptfile = os.path.join(os.getcwd(),
|
|
path_from_name(__class__.__name__, 'target'),
|
|
'usr/lib/kernel/testfile')
|
|
self.assertIn('Hash mismatch for file: ' + corruptfile, test_output)
|
|
self.assertIn(' 1 of 1 files were fixed', test_output)
|
|
self.assertIn('Fix successful', test_output)
|
|
|
|
|
|
@http_command(option="--install -m 10")
|
|
class verify_install_directory(unittest.TestCase):
|
|
def validate(self, test_output):
|
|
self.assertIn('Verifying version 10', test_output)
|
|
self.assertIn(' 1 files were missing', test_output)
|
|
self.assertIn(' 1 of 1 missing files were replaced', test_output)
|
|
self.assertIn('Fix successful', test_output)
|
|
|
|
|
|
def get_test_cases():
|
|
tests = ['bundleadd', 'bundleremove', 'checkupdate', 'hashdump', 'update',
|
|
'verify', 'search']
|
|
test_cases = []
|
|
class_members = inspect.getmembers(sys.modules[__name__], inspect.isclass)
|
|
for (name, cls) in class_members:
|
|
if name.split('_')[0] in tests:
|
|
test_cases.append((name, cls))
|
|
return test_cases
|
|
|
|
|
|
def run_setups(tests):
|
|
setups = []
|
|
for test in tests:
|
|
test_dir = path_from_name(test[0], 'plain')
|
|
if os.path.isfile(os.path.join(test_dir, 'setup.sh')):
|
|
setups.append(subprocess.Popen('./setup.sh', cwd=test_dir,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL))
|
|
for setup in setups:
|
|
setup.wait()
|
|
if setup.returncode != 0:
|
|
raise Exception('Setup failed')
|
|
|
|
|
|
def run_teardowns(tests):
|
|
teardowns = []
|
|
for test in tests:
|
|
test_dir = path_from_name(test[0], 'plain')
|
|
if os.path.isfile(os.path.join(test_dir, 'teardown.sh')):
|
|
teardowns.append(subprocess.Popen('./teardown.sh', cwd=test_dir,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL))
|
|
for teardown in teardowns:
|
|
teardown.wait()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
tests = get_test_cases()
|
|
run_setups(tests)
|
|
suite = unittest.TestSuite()
|
|
for test in tests:
|
|
suite.addTest(test[1]())
|
|
runner = TAPTestRunner()
|
|
runner.set_format('{method_name}')
|
|
runner.set_stream(True)
|
|
runner.run(suite)
|
|
run_teardowns(tests)
|