From 1671b4216fe61e46dc0022a0a147c76fbdfd9d51 Mon Sep 17 00:00:00 2001 From: WillyPillow Date: Sun, 4 Oct 2020 03:05:14 +0800 Subject: [PATCH] qvm-template: Add tests for download function and fix minor bugs --- qubesadmin/tests/tools/qvm_template.py | 289 ++++++++++++++++++++++++- qubesadmin/tools/qvm_template.py | 26 ++- 2 files changed, 301 insertions(+), 14 deletions(-) diff --git a/qubesadmin/tests/tools/qvm_template.py b/qubesadmin/tests/tools/qvm_template.py index 6713c5f..8638f8e 100644 --- a/qubesadmin/tests/tools/qvm_template.py +++ b/qubesadmin/tests/tools/qvm_template.py @@ -191,7 +191,7 @@ class TC_00_qvm_template(qubesadmin.tests.QubesTestCase): mock_dl_list.return_value = {} mock_call.side_effect = self.add_new_vm_side_effect mock_time = mock.Mock(wraps=datetime.datetime) - mock_time.today.return_value = \ + mock_time.now.return_value = \ datetime.datetime(2020, 9, 1, 15, 30, tzinfo=datetime.timezone.utc) with mock.patch('builtins.open', mock.mock_open()) as mock_open, \ mock.patch('datetime.datetime', new=mock_time), \ @@ -322,7 +322,7 @@ class TC_00_qvm_template(qubesadmin.tests.QubesTestCase): mock_dl_list.return_value = {} mock_call.side_effect = self.add_new_vm_side_effect mock_time = mock.Mock(wraps=datetime.datetime) - mock_time.today.return_value = \ + mock_time.now.return_value = \ datetime.datetime(2020, 9, 1, 15, 30, tzinfo=datetime.timezone.utc) with mock.patch('builtins.open', mock.mock_open()) as mock_open, \ mock.patch('datetime.datetime', new=mock_time), \ @@ -3243,3 +3243,288 @@ test-vm : Qubes template for fedora-31 mock.call(args, self.app) ]) self.assertAllCalled() + + @mock.patch('qubesadmin.tools.qvm_template.get_dl_list') + @mock.patch('qubesadmin.tools.qvm_template.qrexec_download') + def test_180_download_success(self, mock_qrexec, mock_dllist): + with tempfile.TemporaryDirectory() as dir: + args = argparse.Namespace( + retries=1 + ) + qubesadmin.tools.qvm_template.download(args, self.app, dir, { + 'fedora-31': qubesadmin.tools.qvm_template.DlEntry( + ('1', '2', '3'), 'qubes-templates-itl', 1048576), + 'fedora-32': qubesadmin.tools.qvm_template.DlEntry( + ('0', '1', '2'), + 'qubes-templates-itl-testing', + 2048576) + }, '.unverified') + self.assertEqual(mock_qrexec.mock_calls, [ + mock.call(args, self.app, 'qubes-template-fedora-31-1:2-3', + dir + '/qubes-template-fedora-31-1:2-3.rpm.unverified', + 1048576), + mock.call(args, self.app, 'qubes-template-fedora-32-0:1-2', + dir + '/qubes-template-fedora-32-0:1-2.rpm.unverified', + 2048576) + ]) + self.assertEqual(mock_dllist.mock_calls, []) + self.assertTrue(all( + [x.endswith('.unverified') for x in os.listdir(dir)])) + + @mock.patch('qubesadmin.tools.qvm_template.get_dl_list') + @mock.patch('qubesadmin.tools.qvm_template.qrexec_download') + def test_181_download_success_nosuffix(self, mock_qrexec, mock_dllist): + with tempfile.TemporaryDirectory() as dir: + args = argparse.Namespace( + retries=1, + downloaddir=dir + ) + with mock.patch('sys.stderr', new=io.StringIO()) as mock_err: + qubesadmin.tools.qvm_template.download(args, self.app, None, { + 'fedora-31': qubesadmin.tools.qvm_template.DlEntry( + ('1', '2', '3'), 'qubes-templates-itl', 1048576) + }) + self.assertEqual(mock_qrexec.mock_calls, [ + mock.call(args, self.app, 'qubes-template-fedora-31-1:2-3', + dir + '/qubes-template-fedora-31-1:2-3.rpm', + 1048576) + ]) + self.assertEqual(mock_dllist.mock_calls, []) + + @mock.patch('qubesadmin.tools.qvm_template.get_dl_list') + @mock.patch('qubesadmin.tools.qvm_template.qrexec_download') + def test_182_download_success_getdllist(self, mock_qrexec, mock_dllist): + mock_dllist.return_value = { + 'fedora-31': qubesadmin.tools.qvm_template.DlEntry( + ('1', '2', '3'), 'qubes-templates-itl', 1048576) + } + with tempfile.TemporaryDirectory() as dir: + args = argparse.Namespace( + retries=1 + ) + qubesadmin.tools.qvm_template.download(args, self.app, + dir, None, '.unverified', + qubesadmin.tools.qvm_template.VersionSelector.LATEST_LOWER) + self.assertEqual(mock_qrexec.mock_calls, [ + mock.call(args, self.app, 'qubes-template-fedora-31-1:2-3', + dir + '/qubes-template-fedora-31-1:2-3.rpm.unverified', + 1048576) + ]) + self.assertEqual(mock_dllist.mock_calls, [ + mock.call(args, self.app, + version_selector=\ + qubesadmin.tools.qvm_template.\ + VersionSelector.LATEST_LOWER) + ]) + self.assertTrue(all( + [x.endswith('.unverified') for x in os.listdir(dir)])) + + @mock.patch('qubesadmin.tools.qvm_template.get_dl_list') + @mock.patch('qubesadmin.tools.qvm_template.qrexec_download') + def test_183_download_success_downloaddir(self, mock_qrexec, mock_dllist): + with tempfile.TemporaryDirectory() as dir: + args = argparse.Namespace( + retries=1, + downloaddir=dir + ) + qubesadmin.tools.qvm_template.download(args, self.app, None, { + 'fedora-31': qubesadmin.tools.qvm_template.DlEntry( + ('1', '2', '3'), 'qubes-templates-itl', 1048576) + }, '.unverified') + self.assertEqual(mock_qrexec.mock_calls, [ + mock.call(args, self.app, 'qubes-template-fedora-31-1:2-3', + dir + '/qubes-template-fedora-31-1:2-3.rpm.unverified', + 1048576) + ]) + self.assertEqual(mock_dllist.mock_calls, []) + self.assertTrue(all( + [x.endswith('.unverified') for x in os.listdir(dir)])) + + @mock.patch('qubesadmin.tools.qvm_template.get_dl_list') + @mock.patch('qubesadmin.tools.qvm_template.qrexec_download') + def test_184_download_success_exists(self, mock_qrexec, mock_dllist): + with tempfile.TemporaryDirectory() as dir: + with open(os.path.join( + dir, 'qubes-template-fedora-31-1:2-3.rpm.unverified'), + 'w') as _: + pass + args = argparse.Namespace( + retries=1, + downloaddir=dir + ) + with mock.patch('sys.stderr', new=io.StringIO()) as mock_err: + qubesadmin.tools.qvm_template.download(args, self.app, None, { + 'fedora-31': qubesadmin.tools.qvm_template.DlEntry( + ('1', '2', '3'), 'qubes-templates-itl', 1048576), + 'fedora-32': qubesadmin.tools.qvm_template.DlEntry( + ('0', '1', '2'), + 'qubes-templates-itl-testing', + 2048576) + }, '.unverified') + self.assertTrue('already exists, skipping' + in mock_err.getvalue()) + self.assertEqual(mock_qrexec.mock_calls, [ + mock.call(args, self.app, 'qubes-template-fedora-32-0:1-2', + dir + '/qubes-template-fedora-32-0:1-2.rpm.unverified', + 2048576) + ]) + self.assertEqual(mock_dllist.mock_calls, []) + self.assertTrue(all( + [x.endswith('.unverified') for x in os.listdir(dir)])) + + @mock.patch('qubesadmin.tools.qvm_template.get_dl_list') + @mock.patch('qubesadmin.tools.qvm_template.qrexec_download') + def test_185_download_success_existsmove(self, mock_qrexec, mock_dllist): + with tempfile.TemporaryDirectory() as dir: + with open(os.path.join( + dir, 'qubes-template-fedora-31-1:2-3.rpm'), + 'w') as _: + pass + args = argparse.Namespace( + retries=1, + downloaddir=dir + ) + with mock.patch('sys.stderr', new=io.StringIO()) as mock_err: + qubesadmin.tools.qvm_template.download(args, self.app, None, { + 'fedora-31': qubesadmin.tools.qvm_template.DlEntry( + ('1', '2', '3'), 'qubes-templates-itl', 1048576) + }, '.unverified') + self.assertTrue('already exists, skipping' + in mock_err.getvalue()) + self.assertEqual(mock_qrexec.mock_calls, []) + self.assertEqual(mock_dllist.mock_calls, []) + self.assertTrue(os.path.exists( + dir + '/qubes-template-fedora-31-1:2-3.rpm.unverified')) + self.assertTrue(all( + [x.endswith('.unverified') for x in os.listdir(dir)])) + + @mock.patch('qubesadmin.tools.qvm_template.get_dl_list') + @mock.patch('qubesadmin.tools.qvm_template.qrexec_download') + def test_186_download_success_existsnosuffix(self, mock_qrexec, mock_dllist): + with tempfile.TemporaryDirectory() as dir: + with open(os.path.join( + dir, 'qubes-template-fedora-31-1:2-3.rpm'), + 'w') as _: + pass + args = argparse.Namespace( + retries=1, + downloaddir=dir + ) + with mock.patch('sys.stderr', new=io.StringIO()) as mock_err: + qubesadmin.tools.qvm_template.download(args, self.app, None, { + 'fedora-31': qubesadmin.tools.qvm_template.DlEntry( + ('1', '2', '3'), 'qubes-templates-itl', 1048576) + }) + self.assertTrue('already exists, skipping' + in mock_err.getvalue()) + self.assertEqual(mock_qrexec.mock_calls, []) + self.assertEqual(mock_dllist.mock_calls, []) + self.assertTrue(os.path.exists( + dir + '/qubes-template-fedora-31-1:2-3.rpm')) + + @mock.patch('qubesadmin.tools.qvm_template.get_dl_list') + @mock.patch('qubesadmin.tools.qvm_template.qrexec_download') + def test_187_download_success_retry(self, mock_qrexec, mock_dllist): + counter = 0 + def f(*args): + nonlocal counter + counter += 1 + if counter == 1: + raise ConnectionError + mock_qrexec.side_effect = f + with tempfile.TemporaryDirectory() as dir: + args = argparse.Namespace( + retries=2, + downloaddir=dir + ) + with mock.patch('sys.stderr', new=io.StringIO()) as mock_err, \ + mock.patch('os.remove') as mock_rm: + qubesadmin.tools.qvm_template.download(args, self.app, None, { + 'fedora-31': qubesadmin.tools.qvm_template.DlEntry( + ('1', '2', '3'), 'qubes-templates-itl', 1048576) + }) + self.assertTrue('retrying...' in mock_err.getvalue()) + self.assertEqual(mock_rm.mock_calls, [ + mock.call(dir + '/qubes-template-fedora-31-1:2-3.rpm') + ]) + self.assertEqual(mock_qrexec.mock_calls, [ + mock.call(args, self.app, 'qubes-template-fedora-31-1:2-3', + dir + '/qubes-template-fedora-31-1:2-3.rpm', + 1048576), + mock.call(args, self.app, 'qubes-template-fedora-31-1:2-3', + dir + '/qubes-template-fedora-31-1:2-3.rpm', + 1048576) + ]) + self.assertEqual(mock_dllist.mock_calls, []) + + @mock.patch('qubesadmin.tools.qvm_template.get_dl_list') + @mock.patch('qubesadmin.tools.qvm_template.qrexec_download') + def test_188_download_fail_retry(self, mock_qrexec, mock_dllist): + counter = 0 + def f(*args): + nonlocal counter + counter += 1 + if counter <= 3: + raise ConnectionError + mock_qrexec.side_effect = f + with tempfile.TemporaryDirectory() as dir: + args = argparse.Namespace( + retries=3, + downloaddir=dir + ) + with mock.patch('sys.stderr', new=io.StringIO()) as mock_err, \ + mock.patch('os.remove') as mock_rm: + with self.assertRaises(SystemExit): + qubesadmin.tools.qvm_template.download( + args, self.app, None, { + 'fedora-31': qubesadmin.tools.qvm_template.DlEntry( + ('1', '2', '3'), 'qubes-templates-itl', 1048576) + }) + self.assertEqual(mock_err.getvalue().count('retrying...'), 2) + self.assertTrue('download failed' in mock_err.getvalue()) + self.assertEqual(mock_rm.mock_calls, [ + mock.call(dir + '/qubes-template-fedora-31-1:2-3.rpm'), + mock.call(dir + '/qubes-template-fedora-31-1:2-3.rpm'), + mock.call(dir + '/qubes-template-fedora-31-1:2-3.rpm') + ]) + self.assertEqual(mock_qrexec.mock_calls, [ + mock.call(args, self.app, 'qubes-template-fedora-31-1:2-3', + dir + '/qubes-template-fedora-31-1:2-3.rpm', + 1048576), + mock.call(args, self.app, 'qubes-template-fedora-31-1:2-3', + dir + '/qubes-template-fedora-31-1:2-3.rpm', + 1048576), + mock.call(args, self.app, 'qubes-template-fedora-31-1:2-3', + dir + '/qubes-template-fedora-31-1:2-3.rpm', + 1048576) + ]) + self.assertEqual(mock_dllist.mock_calls, []) + + @mock.patch('qubesadmin.tools.qvm_template.get_dl_list') + @mock.patch('qubesadmin.tools.qvm_template.qrexec_download') + def test_189_download_fail_interrupt(self, mock_qrexec, mock_dllist): + def f(*args): + raise RuntimeError + mock_qrexec.side_effect = f + with tempfile.TemporaryDirectory() as dir: + args = argparse.Namespace( + retries=3, + downloaddir=dir + ) + with mock.patch('sys.stderr', new=io.StringIO()) as mock_err, \ + mock.patch('os.remove') as mock_rm: + with self.assertRaises(RuntimeError): + qubesadmin.tools.qvm_template.download( + args, self.app, None, { + 'fedora-31': qubesadmin.tools.qvm_template.DlEntry( + ('1', '2', '3'), 'qubes-templates-itl', 1048576) + }) + self.assertEqual(mock_rm.mock_calls, [ + mock.call(dir + '/qubes-template-fedora-31-1:2-3.rpm') + ]) + self.assertEqual(mock_qrexec.mock_calls, [ + mock.call(args, self.app, 'qubes-template-fedora-31-1:2-3', + dir + '/qubes-template-fedora-31-1:2-3.rpm', + 1048576) + ]) + self.assertEqual(mock_dllist.mock_calls, []) diff --git a/qubesadmin/tools/qvm_template.py b/qubesadmin/tools/qvm_template.py index 4227975..dfc006b 100644 --- a/qubesadmin/tools/qvm_template.py +++ b/qubesadmin/tools/qvm_template.py @@ -53,6 +53,8 @@ def qubes_release() -> str: continue val = val.strip('\'"') # strip possible quotes return val + # Return default value instead of throwing so that it works on CI + return '4.1' def parser_gen() -> argparse.ArgumentParser: """Generate argument parser for the application.""" @@ -257,7 +259,7 @@ def is_match_spec(name: str, epoch: str, version: str, release: str, spec: str :return: A tuple. The first element indicates whether there is a match; the second element represents the priority of the match (lower is better) """ - if epoch != 0: + if epoch != '0': targets = [ f'{name}-{epoch}:{version}-{release}', f'{name}', @@ -713,14 +715,13 @@ def download( spec = PACKAGE_NAME_PREFIX + name + '-' + version_str target = os.path.join(path, '%s.rpm' % spec) target_suffix = target + suffix - if suffix != '' and os.path.exists(target_suffix): + if os.path.exists(target_suffix): print('\'%s\' already exists, skipping...' % target, file=sys.stderr) - if os.path.exists(target): + elif os.path.exists(target): print('\'%s\' already exists, skipping...' % target, file=sys.stderr) - if suffix != '': - os.rename(target, target_suffix) + os.rename(target, target_suffix) else: print('Downloading \'%s\'...' % spec, file=sys.stderr) done = False @@ -930,7 +931,7 @@ def install( tz=datetime.timezone.utc) \ .strftime(DATE_FMT) tpl.features['template-installtime'] = \ - datetime.datetime.today( + datetime.datetime.now( tz=datetime.timezone.utc).strftime(DATE_FMT) tpl.features['template-license'] = \ package_hdr[rpm.RPMTAG_LICENSE] @@ -1100,18 +1101,19 @@ def list_templates(args: argparse.Namespace, if args.machine_readable: if operation == 'info': - tpl_list = info_to_machine_output(tpl_list) + tpl_list_dict = info_to_machine_output(tpl_list) elif operation == 'list': - tpl_list = list_to_machine_output(tpl_list) - for status, grp in tpl_list.items(): + tpl_list_dict = list_to_machine_output(tpl_list) + for status, grp in tpl_list_dict.items(): for line in grp: print('|'.join([status] + list(line.values()))) elif args.machine_readable_json: if operation == 'info': - tpl_list = info_to_machine_output(tpl_list, replace_newline=False) + tpl_list_dict = \ + info_to_machine_output(tpl_list, replace_newline=False) elif operation == 'list': - tpl_list = list_to_machine_output(tpl_list) - print(json.dumps(tpl_list)) + tpl_list_dict = list_to_machine_output(tpl_list) + print(json.dumps(tpl_list_dict)) else: if operation == 'info': tpl_list = info_to_human_output(tpl_list)