pungi/tests/test_iso_wrapper.py
Haibo Lin c0193c9fca Fix flake8 complaints - E501
E501 line too long (92 > 88 characters)
E501 line too long (103 > 88 characters)
...

JIRA: COMPOSE-4108
Signed-off-by: Haibo Lin <hlin@redhat.com>
2020-02-07 14:36:46 +08:00

215 lines
8.7 KiB
Python

# -*- coding: utf-8 -*-
import itertools
import mock
import os
import six
try:
import unittest2 as unittest
except ImportError:
import unittest
from pungi.wrappers import iso
CORRECT_OUTPUT = """dummy.iso: 31ff3e405e26ad01c63b62f6b11d30f6
Fragment sums: 6eb92e7bda221d7fe5f19b4d21468c9bf261d84c96d145d96c76444b9cbc
Fragment count: 20
Supported ISO: no
"""
INCORRECT_OUTPUT = """This should never happen: File not found"""
# Cached to use in tests that mock os.listdir
orig_listdir = os.listdir
def fake_listdir(pattern, result=None, exc=None):
"""Create a function that mocks os.listdir. If the path contains pattern,
result will be returned or exc raised. Otherwise it's normal os.listdir
"""
# The point of this is to avoid issues on Python 2, where apparently
# isdir() is using listdir(), so the mocking is breaking it.
def worker(path):
if isinstance(path, six.string_types) and pattern in path:
if exc:
raise exc
return result
return orig_listdir(path)
return worker
class TestIsoUtils(unittest.TestCase):
@mock.patch("pungi.wrappers.iso.run")
def test_get_implanted_md5_correct(self, mock_run):
mock_run.return_value = (0, CORRECT_OUTPUT)
logger = mock.Mock()
self.assertEqual(
iso.get_implanted_md5("dummy.iso", logger=logger),
"31ff3e405e26ad01c63b62f6b11d30f6",
)
self.assertEqual(
mock_run.call_args_list,
[
mock.call(
["/usr/bin/checkisomd5", "--md5sumonly", "dummy.iso"],
universal_newlines=True,
)
],
)
self.assertEqual(logger.mock_calls, [])
@mock.patch("pungi.wrappers.iso.run")
def test_get_implanted_md5_incorrect(self, mock_run):
mock_run.return_value = (0, INCORRECT_OUTPUT)
logger = mock.Mock()
self.assertEqual(iso.get_implanted_md5("dummy.iso", logger=logger), None)
self.assertEqual(
mock_run.call_args_list,
[
mock.call(
["/usr/bin/checkisomd5", "--md5sumonly", "dummy.iso"],
universal_newlines=True,
)
],
)
self.assertTrue(len(logger.mock_calls) > 0)
@mock.patch("pungi.util.run_unmount_cmd")
@mock.patch("pungi.wrappers.iso.run")
def test_mount_iso(self, mock_run, mock_unmount):
# first tuple is return value for command 'which guestmount'
# value determines type of the mount/unmount
# command ('1' - guestmount is not available)
# for approach as a root, pair commands mount-umount are used
mock_run.side_effect = [(1, ""), (0, "")]
with iso.mount("dummy") as temp_dir:
self.assertTrue(os.path.isdir(temp_dir))
self.assertEqual(len(mock_run.call_args_list), 2)
mount_call_str = str(mock_run.call_args_list[1])
self.assertTrue(mount_call_str.startswith("call(['mount'"))
self.assertEqual(len(mock_unmount.call_args_list), 1)
unmount_call_str = str(mock_unmount.call_args_list[0])
self.assertTrue(unmount_call_str.startswith("call(['umount'"))
self.assertFalse(os.path.isdir(temp_dir))
@mock.patch("pungi.util.rmtree")
@mock.patch("os.listdir", new=fake_listdir("guestfs", ["root"]))
@mock.patch("pungi.util.run_unmount_cmd")
@mock.patch("pungi.wrappers.iso.run")
def test_guestmount(self, mock_run, mock_unmount, mock_rmtree):
# first tuple is return value for command 'which guestmount'
# value determines type of the mount/unmount
# command ('0' - guestmount is available)
# for approach as a non-root, pair commands guestmount-fusermount are used
mock_run.side_effect = [(0, ""), (0, "")]
with iso.mount("dummy") as temp_dir:
self.assertTrue(os.path.isdir(temp_dir))
self.assertEqual(len(mock_run.call_args_list), 2)
mount_call_str = str(mock_run.call_args_list[1])
self.assertTrue(mount_call_str.startswith("call(['guestmount'"))
self.assertEqual(len(mock_unmount.call_args_list), 1)
unmount_call_str = str(mock_unmount.call_args_list[0])
self.assertTrue(unmount_call_str.startswith("call(['fusermount'"))
self.assertFalse(os.path.isdir(temp_dir))
@mock.patch("pungi.util.rmtree")
@mock.patch("os.listdir", new=fake_listdir("guestfs", []))
@mock.patch("pungi.util.run_unmount_cmd")
@mock.patch("pungi.wrappers.iso.run")
def test_guestmount_cleans_up_cache(self, mock_run, mock_unmount, mock_rmtree):
# first tuple is return value for command 'which guestmount'
# value determines type of the mount/unmount
# command ('0' - guestmount is available)
# for approach as a non-root, pair commands guestmount-fusermount are used
mock_run.side_effect = [(0, ""), (0, "")]
with iso.mount("dummy") as temp_dir:
self.assertTrue(os.path.isdir(temp_dir))
self.assertEqual(len(mock_run.call_args_list), 2)
mount_call_str = str(mock_run.call_args_list[1])
self.assertTrue(mount_call_str.startswith("call(['guestmount'"))
self.assertEqual(len(mock_unmount.call_args_list), 1)
unmount_call_str = str(mock_unmount.call_args_list[0])
self.assertTrue(unmount_call_str.startswith("call(['fusermount'"))
self.assertFalse(os.path.isdir(temp_dir))
@mock.patch("pungi.util.rmtree")
@mock.patch("os.listdir", new=fake_listdir("guestfs", OSError("No such file")))
@mock.patch("pungi.util.run_unmount_cmd")
@mock.patch("pungi.wrappers.iso.run")
def test_guestmount_handles_missing_cache(
self, mock_run, mock_unmount, mock_rmtree
):
# first tuple is return value for command 'which guestmount'
# value determines type of the mount/unmount
# command ('0' - guestmount is available)
# for approach as a non-root, pair commands guestmount-fusermount are used
mock_run.side_effect = [(0, ""), (0, "")]
with iso.mount("dummy") as temp_dir:
self.assertTrue(os.path.isdir(temp_dir))
self.assertEqual(len(mock_run.call_args_list), 2)
mount_call_str = str(mock_run.call_args_list[1])
self.assertTrue(mount_call_str.startswith("call(['guestmount'"))
self.assertEqual(len(mock_unmount.call_args_list), 1)
unmount_call_str = str(mock_unmount.call_args_list[0])
self.assertTrue(unmount_call_str.startswith("call(['fusermount'"))
self.assertFalse(os.path.isdir(temp_dir))
@mock.patch("pungi.util.run_unmount_cmd")
@mock.patch("pungi.wrappers.iso.run")
def test_mount_iso_always_unmounts(self, mock_run, mock_unmount):
mock_run.side_effect = [(1, ""), (0, "")]
try:
with iso.mount("dummy") as temp_dir:
self.assertTrue(os.path.isdir(temp_dir))
raise RuntimeError()
except RuntimeError:
pass
self.assertEqual(len(mock_run.call_args_list), 2)
self.assertEqual(len(mock_unmount.call_args_list), 1)
self.assertFalse(os.path.isdir(temp_dir))
@mock.patch("pungi.util.run_unmount_cmd")
@mock.patch("pungi.wrappers.iso.run")
def test_mount_iso_raises_on_error(self, mock_run, mock_unmount):
log = mock.Mock()
mock_run.side_effect = [(1, ""), (1, "Boom")]
with self.assertRaises(RuntimeError):
with iso.mount("dummy", logger=log) as temp_dir:
self.assertTrue(os.path.isdir(temp_dir))
self.assertEqual(len(mock_run.call_args_list), 2)
self.assertEqual(len(mock_unmount.call_args_list), 0)
self.assertEqual(len(log.mock_calls), 1)
class TestCmpGraftPoints(unittest.TestCase):
def assertSorted(self, *args):
"""Tests that all permutations of arguments yield the same sorted results."""
for perm in itertools.permutations(args):
self.assertEqual(sorted(perm, key=iso.graft_point_sort_key), list(args))
def test_eq(self):
self.assertSorted("pkgs/foo.rpm", "pkgs/foo.rpm")
def test_rpms_sorted_alphabetically(self):
self.assertSorted("pkgs/bar.rpm", "pkgs/foo.rpm")
def test_images_sorted_alphabetically(self):
self.assertSorted("aaa.img", "images/foo", "isolinux/foo")
def test_other_files_sorted_alphabetically(self):
self.assertSorted("bar.txt", "foo.txt")
def test_rpms_after_images(self):
self.assertSorted("foo.ins", "bar.rpm")
def test_other_after_images(self):
self.assertSorted("EFI/anything", "zzz.txt")
def test_rpm_after_other(self):
self.assertSorted("bbb.txt", "aaa.rpm")
def test_all_kinds(self):
self.assertSorted("etc/file", "ppc/file", "c.txt", "d.txt", "a.rpm", "b.rpm")