Move "functional" unit tests under block-device
This is code motion with some small changes to make follow-on's easier. test_blockdevice_mbr.py is moved alongside the other tests. It is modified slightly to use the standard base class and remove a lot of repeated test setup; a fixture is used for the tempdir (so it doesn't have to be torn-down, and is removed properly on error) and the partx args are moved into the setUp() so each test doesn't have to create it. No functional change. renamed test_mbr.py for shortness. test_blockdevice_utils.py is merged with existing test_utils.py. No change to the tests. test_blockdevice.py is removed. It isn't doing anything currently; to work it will need to take an approach based more on mocking of calls that require elevated permissions. It's in history if we need it. Change-Id: I87b1ea94afaaa0b44e6a57b9d073f95a63a04cf0
This commit is contained in:
Ian Wienand
parent
cae44c7eea
commit
cdb1a95be1
@ -10,14 +10,12 @@
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
import copy
|
||||
import fixtures
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import tempfile
|
||||
import testtools
|
||||
|
||||
import diskimage_builder.block_device.tests.test_base as tb
|
||||
|
||||
from diskimage_builder.block_device.level0.localloop \
|
||||
import LocalLoopNode as LocalLoop
|
||||
@ -27,14 +25,11 @@ from diskimage_builder.block_device.level1.mbr import MBR
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class TestMBR(testtools.TestCase):
|
||||
class TestMBR(tb.TestBase):
|
||||
|
||||
disk_size_10M = 10 * 1024 * 1024
|
||||
disk_size_1G = 1024 * 1024 * 1024
|
||||
|
||||
pargs = ["--raw", "--output",
|
||||
"NR,START,END,TYPE,FLAGS,SCHEME", "-g", "-b", "-"]
|
||||
|
||||
def _get_path_for_partx(self):
|
||||
"""Searches and sets the path for partx
|
||||
|
||||
@ -53,34 +48,27 @@ class TestMBR(testtools.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(TestMBR, self).setUp()
|
||||
fs = '%(asctime)s %(levelname)s [%(name)s] %(message)s'
|
||||
self.log_fixture = self.useFixture(
|
||||
fixtures.FakeLogger(level=logging.DEBUG,
|
||||
format=fs))
|
||||
|
||||
def _create_image(self):
|
||||
tmp_dir = tempfile.mkdtemp(prefix="dib-bd-mbr-")
|
||||
image_path = os.path.join(tmp_dir, "image.raw")
|
||||
LocalLoop.image_create(image_path, TestMBR.disk_size_1G)
|
||||
return tmp_dir, image_path
|
||||
self.tmp_dir = fixtures.TempDir()
|
||||
self.useFixture(self.tmp_dir)
|
||||
self.image_path = os.path.join(self.tmp_dir.path, "image.raw")
|
||||
LocalLoop.image_create(self.image_path, TestMBR.disk_size_1G)
|
||||
logger.debug("Temp image is %s", self.image_path)
|
||||
|
||||
self.partx_args = [self._get_path_for_partx(), "--raw",
|
||||
"--output", "NR,START,END,TYPE,FLAGS,SCHEME",
|
||||
"-g", "-b", "-", self.image_path]
|
||||
|
||||
def _run_partx(self, image_path):
|
||||
largs = copy.copy(TestMBR.pargs)
|
||||
partx_path = self._get_path_for_partx()
|
||||
largs.insert(0, partx_path)
|
||||
largs.append(image_path)
|
||||
logger.info("Running command [%s]", largs)
|
||||
return subprocess.check_output(largs).decode("ascii")
|
||||
logger.info("Running command: %s", self.partx_args)
|
||||
return subprocess.check_output(self.partx_args).decode("ascii")
|
||||
|
||||
def test_one_ext_partition(self):
|
||||
"""Creates one partition and check correctness with partx."""
|
||||
|
||||
tmp_dir, image_path = self._create_image()
|
||||
with MBR(image_path, TestMBR.disk_size_1G, 1024 * 1024) as mbr:
|
||||
with MBR(self.image_path, TestMBR.disk_size_1G, 1024 * 1024) as mbr:
|
||||
mbr.add_partition(False, False, TestMBR.disk_size_10M, 0x83)
|
||||
|
||||
output = self._run_partx(image_path)
|
||||
shutil.rmtree(tmp_dir)
|
||||
output = self._run_partx(self.image_path)
|
||||
self.assertEqual(
|
||||
"1 2048 2097151 0xf 0x0 dos\n"
|
||||
"5 4096 24575 0x83 0x0 dos\n", output)
|
||||
@ -88,25 +76,19 @@ class TestMBR(testtools.TestCase):
|
||||
def test_zero_partitions(self):
|
||||
"""Creates no partition and check correctness with partx."""
|
||||
|
||||
tmp_dir, image_path = self._create_image()
|
||||
with MBR(image_path, TestMBR.disk_size_1G, 1024 * 1024):
|
||||
with MBR(self.image_path, TestMBR.disk_size_1G, 1024 * 1024):
|
||||
pass
|
||||
|
||||
output = self._run_partx(image_path)
|
||||
shutil.rmtree(tmp_dir)
|
||||
output = self._run_partx(self.image_path)
|
||||
self.assertEqual("", output)
|
||||
|
||||
def test_many_ext_partitions(self):
|
||||
"""Creates many partition and check correctness with partx."""
|
||||
|
||||
tmp_dir, image_path = self._create_image()
|
||||
with MBR(image_path, TestMBR.disk_size_1G, 1024 * 1024) as mbr:
|
||||
with MBR(self.image_path, TestMBR.disk_size_1G, 1024 * 1024) as mbr:
|
||||
for nr in range(0, 64):
|
||||
mbr.add_partition(False, False, TestMBR.disk_size_10M, 0x83)
|
||||
|
||||
output = self._run_partx(image_path)
|
||||
|
||||
shutil.rmtree(tmp_dir)
|
||||
output = self._run_partx(self.image_path)
|
||||
|
||||
lines = output.split("\n")
|
||||
self.assertEqual(66, len(lines))
|
||||
@ -131,25 +113,21 @@ class TestMBR(testtools.TestCase):
|
||||
def test_one_pri_partition(self):
|
||||
"""Creates one primary partition and check correctness with partx."""
|
||||
|
||||
tmp_dir, image_path = self._create_image()
|
||||
with MBR(image_path, TestMBR.disk_size_1G, 1024 * 1024) as mbr:
|
||||
with MBR(self.image_path, TestMBR.disk_size_1G, 1024 * 1024) as mbr:
|
||||
mbr.add_partition(True, False, TestMBR.disk_size_10M, 0x83)
|
||||
|
||||
output = self._run_partx(image_path)
|
||||
shutil.rmtree(tmp_dir)
|
||||
output = self._run_partx(self.image_path)
|
||||
self.assertEqual(
|
||||
"1 2048 22527 0x83 0x0 dos\n", output)
|
||||
|
||||
def test_three_pri_partition(self):
|
||||
"""Creates three primary partition and check correctness with partx."""
|
||||
|
||||
tmp_dir, image_path = self._create_image()
|
||||
with MBR(image_path, TestMBR.disk_size_1G, 1024 * 1024) as mbr:
|
||||
with MBR(self.image_path, TestMBR.disk_size_1G, 1024 * 1024) as mbr:
|
||||
for _ in range(3):
|
||||
mbr.add_partition(True, False, TestMBR.disk_size_10M, 0x83)
|
||||
|
||||
output = self._run_partx(image_path)
|
||||
shutil.rmtree(tmp_dir)
|
||||
output = self._run_partx(self.image_path)
|
||||
self.assertEqual(
|
||||
"1 2048 22527 0x83 0x0 dos\n"
|
||||
"2 22528 43007 0x83 0x0 dos\n"
|
||||
@ -158,16 +136,14 @@ class TestMBR(testtools.TestCase):
|
||||
def test_many_pri_and_ext_partition(self):
|
||||
"""Creates many primary and extended partitions."""
|
||||
|
||||
tmp_dir, image_path = self._create_image()
|
||||
with MBR(image_path, TestMBR.disk_size_1G, 1024 * 1024) as mbr:
|
||||
with MBR(self.image_path, TestMBR.disk_size_1G, 1024 * 1024) as mbr:
|
||||
# Create three primary partitions
|
||||
for _ in range(3):
|
||||
mbr.add_partition(True, False, TestMBR.disk_size_10M, 0x83)
|
||||
for _ in range(7):
|
||||
mbr.add_partition(False, False, TestMBR.disk_size_10M, 0x83)
|
||||
|
||||
output = self._run_partx(image_path)
|
||||
shutil.rmtree(tmp_dir)
|
||||
output = self._run_partx(self.image_path)
|
||||
self.assertEqual(
|
||||
"1 2048 22527 0x83 0x0 dos\n" # Primary 1
|
||||
"2 22528 43007 0x83 0x0 dos\n" # Primary 2
|
||||
@ -12,12 +12,46 @@
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
import testtools
|
||||
import logging
|
||||
|
||||
import diskimage_builder.block_device.tests.test_base as tb
|
||||
|
||||
from diskimage_builder.block_device.utils import parse_abs_size_spec
|
||||
from diskimage_builder.block_device.utils import parse_rel_size_spec
|
||||
|
||||
|
||||
class TestLoggingConfig(testtools.TestCase):
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class TestBlockDeviceUtils(tb.TestBase):
|
||||
"""Tests for the utils.py
|
||||
|
||||
This tests mostly the error and failure cases - because the good
|
||||
cases are tested implicitly with the higher level unit tests.
|
||||
"""
|
||||
|
||||
def test_parse_rel_size_with_abs(self):
|
||||
"""Calls parse_rel_size_spec with an absolute number"""
|
||||
|
||||
is_rel, size = parse_rel_size_spec("154MiB", 0)
|
||||
self.assertFalse(is_rel)
|
||||
self.assertEqual(154 * 1024 * 1024, size)
|
||||
|
||||
def test_parse_abs_size_without_spec(self):
|
||||
"""Call parse_abs_size_spec without spec"""
|
||||
|
||||
size = parse_abs_size_spec("198")
|
||||
self.assertEqual(198, size)
|
||||
|
||||
def test_invalid_unit_spec(self):
|
||||
"""Call parse_abs_size_spec with invalid unit spec"""
|
||||
|
||||
self.assertRaises(RuntimeError, parse_abs_size_spec, "747InVaLiDUnIt")
|
||||
|
||||
def test_broken_unit_spec(self):
|
||||
"""Call parse_abs_size_spec with a completely broken unit spec"""
|
||||
|
||||
self.assertRaises(RuntimeError, parse_abs_size_spec, "_+!HuHi+-=")
|
||||
|
||||
def test_parse_size_spec(self):
|
||||
map(lambda tspec:
|
||||
|
||||
@ -1,86 +0,0 @@
|
||||
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
||||
# not use this file except in compliance with the License. You may obtain
|
||||
# a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
import fixtures
|
||||
import logging
|
||||
import subprocess
|
||||
import testtools
|
||||
|
||||
from diskimage_builder.block_device.blockdevice import BlockDevice
|
||||
from diskimage_builder.block_device.level0 import localloop
|
||||
from diskimage_builder.logging_config import setup
|
||||
|
||||
|
||||
# Setup Logging
|
||||
setup()
|
||||
|
||||
|
||||
class StateSavingBlockDevice(BlockDevice):
|
||||
def cmd_create(self):
|
||||
logging.info("StateSavingBlockDevice cmd_create()")
|
||||
super(StateSavingBlockDevice, self).cmd_create()
|
||||
_, _, self.state = self.load_state()
|
||||
|
||||
|
||||
class BlockDeviceFixture(fixtures.Fixture):
|
||||
def __init__(self, *args, **kwargs):
|
||||
logging.info("BlockDeviceFixture constructor")
|
||||
self.args = args
|
||||
self.kwargs = kwargs
|
||||
self.bd = None
|
||||
|
||||
def _setUp(self):
|
||||
logging.info("BlockDeviceFixture _setUp()")
|
||||
self.bd = StateSavingBlockDevice(*self.args, **self.kwargs)
|
||||
self.addCleanup(self.cleanup_loopbacks)
|
||||
|
||||
def _assert_loopback_detatched(self, loopback):
|
||||
if localloop.LocalLoop.loopdev_is_attached(loopback):
|
||||
localloop.LocalLoop.loopdev_detach(loopback)
|
||||
|
||||
def cleanup_loopbacks(self):
|
||||
for lb_dev in self.bd.state.get('loopback_devices', []):
|
||||
self._assert_loopback_detatched(lb_dev)
|
||||
|
||||
|
||||
class TestBlockDevice(testtools.TestCase):
|
||||
def _assert_loopbacks_cleaned(self, blockdevice):
|
||||
for lb_dev in blockdevice.state.get('loopback_devices', []):
|
||||
self.assertEqual(False,
|
||||
localloop.LocalLoop.loopdev_is_attached(lb_dev))
|
||||
|
||||
# ToDo: This calls sudo to setup the loop device - which is not allowed.
|
||||
# Currently no idea how to continue here...
|
||||
def _DONT_test_create_default_config(self):
|
||||
logging.info("test_create_default_config called")
|
||||
builddir = self.useFixture(fixtures.TempDir()).path
|
||||
imagedir = self.useFixture(fixtures.TempDir()).path
|
||||
logging.info("builddir [%s]", builddir)
|
||||
logging.info("imagedir [%s]", imagedir)
|
||||
|
||||
logging.info("Calling BlockDevice constructor")
|
||||
bd = self.useFixture(BlockDeviceFixture(
|
||||
None, builddir, '%dKiB' % (1024 * 1024), imagedir
|
||||
)).bd
|
||||
logging.info("Calling BlockDevice cmd_create()")
|
||||
bd.cmd_create()
|
||||
|
||||
logging.info("Check result")
|
||||
logging.info("State [%s]", bd.state)
|
||||
self.assertTrue('device' in bd.state['image0'])
|
||||
lb_dev = bd.state['image0']['device']
|
||||
# partprobe loopback so we can get partition info
|
||||
args = ['sudo', 'partprobe', lb_dev]
|
||||
logging.info("Call: %s", args)
|
||||
subprocess.check_call(args)
|
||||
bd.cmd_cleanup()
|
||||
self._assert_loopbacks_cleaned(bd)
|
||||
@ -1,49 +0,0 @@
|
||||
# Copyright 2016 Andreas Florath (andreas@florath.net)
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
||||
# not use this file except in compliance with the License. You may obtain
|
||||
# a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
|
||||
from diskimage_builder.block_device.utils import parse_abs_size_spec
|
||||
from diskimage_builder.block_device.utils import parse_rel_size_spec
|
||||
import testtools
|
||||
|
||||
|
||||
class TestBlockDeviceUtils(testtools.TestCase):
|
||||
"""Tests for the utils.py in the block_device dir.
|
||||
|
||||
This tests mostly the error and failure cases - because the good
|
||||
cases are tested implicitly with the higher level unit tests.
|
||||
"""
|
||||
|
||||
def test_parse_rel_size_with_abs(self):
|
||||
"""Calls parse_rel_size_spec with an absolute number"""
|
||||
|
||||
is_rel, size = parse_rel_size_spec("154MiB", 0)
|
||||
self.assertFalse(is_rel)
|
||||
self.assertEqual(154 * 1024 * 1024, size)
|
||||
|
||||
def test_parse_abs_size_without_spec(self):
|
||||
"""Call parse_abs_size_spec without spec"""
|
||||
|
||||
size = parse_abs_size_spec("198")
|
||||
self.assertEqual(198, size)
|
||||
|
||||
def test_invalid_unit_spec(self):
|
||||
"""Call parse_abs_size_spec with invalid unit spec"""
|
||||
|
||||
self.assertRaises(RuntimeError, parse_abs_size_spec, "747InVaLiDUnIt")
|
||||
|
||||
def test_broken_unit_spec(self):
|
||||
"""Call parse_abs_size_spec with a completely broken unit spec"""
|
||||
|
||||
self.assertRaises(RuntimeError, parse_abs_size_spec, "_+!HuHi+-=")
|
||||
Loading…
Reference in New Issue
Block a user