Skip to content

Commit

Permalink
tests: some more for qvm-template
Browse files Browse the repository at this point in the history
QubesOS/qubes-issues#
  • Loading branch information
marmarek committed Feb 19, 2021
1 parent f4e826e commit e00f35b
Showing 1 changed file with 192 additions and 9 deletions.
201 changes: 192 additions & 9 deletions qubesadmin/tests/tools/qvm_template.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,11 +46,12 @@ def test_000_verify_rpm_success(self, mock_proc, mock_call, mock_ts):
hdr = {
rpm.RPMTAG_SIGPGP: 'xxx', # non-empty
rpm.RPMTAG_SIGGPG: 'xxx', # non-empty
rpm.RPMTAG_NAME: 'qubes-template-test-vm',
}
mock_ts.return_value.hdrFromFdno.return_value = hdr
mock_proc.return_value = b'dummy.rpm: digests signatures OK\n'
ret = qubesadmin.tools.qvm_template.verify_rpm('/dev/null',
['/path/to/key'])
'/path/to/key', template_name='test-vm')
mock_call.assert_called_once()
mock_proc.assert_called_once()
self.assertEqual(hdr, ret)
Expand All @@ -69,7 +70,7 @@ def test_001_verify_rpm_nosig_fail(self, mock_proc, mock_call, mock_ts):
mock_proc.return_value = b'dummy.rpm: digests OK\n'
with self.assertRaises(Exception) as e:
qubesadmin.tools.qvm_template.verify_rpm('/dev/null',
['/path/to/key'])
'/path/to/key')
mock_call.assert_called_once()
mock_proc.assert_called_once()
self.assertIn('Signature verification failed', e.exception.args[0])
Expand All @@ -88,7 +89,7 @@ def test_002_verify_rpm_nosig_success(self, mock_proc, mock_call, mock_ts):
mock_ts.return_value.hdrFromFdno.return_value = hdr
mock_proc.return_value = b'dummy.rpm: digests OK\n'
ret = qubesadmin.tools.qvm_template.verify_rpm('/dev/null',
['/path/to/key'], True)
'/path/to/key', True)
mock_proc.assert_not_called()
mock_call.assert_not_called()
self.assertEqual(ret, hdr)
Expand All @@ -102,7 +103,28 @@ def test_003_verify_rpm_badsig_fail(self, mock_proc, mock_call, mock_ts):
['rpmkeys', '--checksig'], b'/dev/null: digests SIGNATURES NOT OK\n')
with self.assertRaises(Exception) as e:
qubesadmin.tools.qvm_template.verify_rpm('/dev/null',
['/path/to/key'])
'/path/to/key')
mock_call.assert_called_once()
mock_proc.assert_called_once()
self.assertIn('Signature verification failed', e.exception.args[0])
mock_ts.assert_not_called()
self.assertAllCalled()

@mock.patch('rpm.TransactionSet')
@mock.patch('subprocess.check_call')
@mock.patch('subprocess.check_output')
def test_004_verify_rpm_badname(self, mock_proc, mock_call, mock_ts):
mock_proc.side_effect = subprocess.CalledProcessError(1,
['rpmkeys', '--checksig'], b'/dev/null: digests signatures OK\n')
hdr = {
rpm.RPMTAG_SIGPGP: 'xxx', # non-empty
rpm.RPMTAG_SIGGPG: 'xxx', # non-empty
rpm.RPMTAG_NAME: 'qubes-template-unexpected',
}
mock_ts.return_value.hdrFromFdno.return_value = hdr
with self.assertRaises(Exception) as e:
qubesadmin.tools.qvm_template.verify_rpm('/dev/null',
'/path/to/key', template_name='test-vm')
mock_call.assert_called_once()
mock_proc.assert_called_once()
self.assertIn('Signature verification failed', e.exception.args[0])
Expand Down Expand Up @@ -3677,7 +3699,7 @@ def f(*args):
@mock.patch('qubesadmin.tools.qvm_template.verify_rpm')
@mock.patch('qubesadmin.tools.qvm_template.get_dl_list')
@mock.patch('qubesadmin.tools.qvm_template.qrexec_download')
def test_189_download_verify_fail(self, mock_qrexec, mock_dllist,
def test_190_download_fail_verify(self, mock_qrexec, mock_dllist,
mock_verify_rpm):
mock_qrexec.side_effect = self._mock_qrexec_download
mock_verify_rpm.side_effect = \
Expand All @@ -3689,7 +3711,7 @@ def test_189_download_verify_fail(self, mock_qrexec, mock_dllist,
repo_files=[],
keyring='/tmp/keyring.gpg',
releasever='4.1',
nogpgcheck=False,
nogpgcheck=True, # make sure it gets ignored
downloaddir=dir
)
with self.assertRaises(qubesadmin.tools.qvm_template.SignatureVerificationError):
Expand All @@ -3705,7 +3727,168 @@ def test_189_download_verify_fail(self, mock_qrexec, mock_dllist,
])
self.assertEqual(mock_dllist.mock_calls, [])
self.assertEqual(os.listdir(dir), [])
self.assertEqual(mock_verify_rpm.mock_calls, [
mock.call(re_str(dir + '/.*/qubes-template-fedora-31-1:2-3.rpm.UNTRUSTED'),
'/tmp/keyring.gpg', template_name='fedora-31'),
])

def _mock_qrexec_download_short(self, args, app, spec, path,
dlsize=None, refresh=False):
self.assertFalse(os.path.exists(path),
'{} should not exist before'.format(path))
# just create an empty file
with open(path, 'wb') as f:
if f is not None:
f.truncate(dlsize // 2)

@mock.patch('qubesadmin.tools.qvm_template.verify_rpm')
@mock.patch('qubesadmin.tools.qvm_template.get_dl_list')
@mock.patch('qubesadmin.tools.qvm_template.qrexec_download')
def test_191_download_fail_short(self, mock_qrexec, mock_dllist,
mock_verify_rpm):
mock_qrexec.side_effect = self._mock_qrexec_download_short
with tempfile.TemporaryDirectory() as dir, \
self.assertRaises(SystemExit):
args = argparse.Namespace(
repo_files=[],
keyring='/tmp/keyring.gpg',
releasever='4.1',
nogpgcheck=False,
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),
})
self.assertEqual(mock_qrexec.mock_calls, [
mock.call(args, self.app, 'qubes-template-fedora-31-1:2-3',
re_str(dir + '/.*/qubes-template-fedora-31-1:2-3.rpm.UNTRUSTED'),
1048576),
])
self.assertEqual(mock_dllist.mock_calls, [])
self.assertEqual(mock_verify_rpm.mock_calls, [])


# TODO: test unexpected name downloaded
# TODO: test truncated download
# TODO: test no disk space
@mock.patch('os.rename')
@mock.patch('os.makedirs')
@mock.patch('subprocess.check_call')
@mock.patch('qubesadmin.tools.qvm_template.confirm_action')
@mock.patch('qubesadmin.tools.qvm_template.extract_rpm')
@mock.patch('qubesadmin.tools.qvm_template.download')
@mock.patch('qubesadmin.tools.qvm_template.get_dl_list')
@mock.patch('qubesadmin.tools.qvm_template.verify_rpm')
def test_200_reinstall_success(
self,
mock_verify,
mock_dl_list,
mock_dl,
mock_extract,
mock_confirm,
mock_call,
mock_mkdirs,
mock_rename):
self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
b'0\x00test-vm class=TemplateVM state=Halted\n'
build_time = '2020-09-01 14:30:00' # 1598970600
install_time = '2020-09-01 15:30:00'
for key, val in [
('name', 'test-vm'),
('epoch', '2'),
('version', '4.1'),
('release', '2020')]:
self.app.expected_calls[(
'test-vm',
'admin.vm.feature.Get',
f'template-{key}',
None)] = b'0\0' + val.encode()
for key, val in [
('name', 'test-vm'),
('epoch', '2'),
('version', '4.1'),
('release', '2020'),
('reponame', 'qubes-templates-itl'),
('buildtime', build_time),
('installtime', install_time),
('license', 'GPL'),
('url', 'https://qubes-os.org'),
('summary', 'Summary'),
('description', 'Desc|desc')]:
self.app.expected_calls[(
'test-vm',
'admin.vm.feature.Set',
f'template-{key}',
val.encode())] = b'0\0'
rpm_hdr = {
rpm.RPMTAG_NAME : 'qubes-template-test-vm',
rpm.RPMTAG_BUILDTIME : 1598970600,
rpm.RPMTAG_DESCRIPTION : 'Desc\ndesc',
rpm.RPMTAG_EPOCHNUM : 2,
rpm.RPMTAG_LICENSE : 'GPL',
rpm.RPMTAG_RELEASE : '2020',
rpm.RPMTAG_SUMMARY : 'Summary',
rpm.RPMTAG_URL : 'https://qubes-os.org',
rpm.RPMTAG_VERSION : '4.1'
}
mock_dl.return_value = {'test-vm': rpm_hdr}
dl_list = {
'test-vm': qubesadmin.tools.qvm_template.DlEntry(
('1', '4.1', '20200101'), 'qubes-templates-itl', 1048576)
}
mock_dl_list.return_value = dl_list
mock_call.side_effect = self.add_new_vm_side_effect
mock_time = mock.Mock(wraps=datetime.datetime)
mock_time.now.return_value = \
datetime.datetime(2020, 9, 1, 15, 30, tzinfo=datetime.timezone.utc)
selector = qubesadmin.tools.qvm_template.VersionSelector.REINSTALL
with mock.patch('qubesadmin.tools.qvm_template.LOCK_FILE', '/tmp/test.lock'), \
mock.patch('datetime.datetime', new=mock_time), \
mock.patch('tempfile.TemporaryDirectory') as mock_tmpdir:
args = argparse.Namespace(
templates=['test-vm'],
keyring='/tmp/keyring.gpg',
nogpgcheck=False,
cachedir='/var/cache/qvm-template',
repo_files=[],
releasever='4.1',
yes=False,
keep_cache=True,
allow_pv=False,
pool=None
)
mock_tmpdir.return_value.__enter__.return_value = \
'/var/tmp/qvm-template-tmpdir'
qubesadmin.tools.qvm_template.install(args, self.app,
version_selector=selector,
override_existing=True)
# Attempt to get download list
self.assertEqual(mock_dl_list.mock_calls, [
mock.call(args, self.app, version_selector=selector)
])
mock_dl.assert_called_with(args, self.app,
dl_list=dl_list, version_selector=selector)
# already verified by download()
self.assertEqual(mock_verify.mock_calls, [])
# Package is extracted
mock_extract.assert_called_with('test-vm',
'/var/cache/qvm-template/qubes-template-test-vm-1:4.1-20200101.rpm',
'/var/tmp/qvm-template-tmpdir')
# Expect override confirmation
self.assertEqual(mock_confirm.mock_calls,
[mock.call(re_str(r'.*override changes.*:'), ['test-vm'])])
# qvm-template-postprocess is called
self.assertEqual(mock_call.mock_calls, [
mock.call([
'qvm-template-postprocess',
'--really',
'--no-installed-by-rpm',
'post-install',
'test-vm',
'/var/tmp/qvm-template-tmpdir'
'/var/lib/qubes/vm-templates/test-vm'
])
])
# Cache directory created
self.assertEqual(mock_mkdirs.mock_calls, [
mock.call(args.cachedir, exist_ok=True)
])
self.assertAllCalled()

0 comments on commit e00f35b

Please sign in to comment.