# Copyright 2014 Mirantis, Inc.
#
# 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 mock

from oslotest import base as test_base

from fuel_agent import errors
from fuel_agent.objects import partition


class TestMD(test_base.BaseTestCase):
    def setUp(self):
        super(TestMD, self).setUp()
        self.md = partition.Md('name', 'level')

    def test_add_device_ok(self):
        self.assertEqual(0, len(self.md.devices))
        self.md.add_device('device')
        self.assertEqual(1, len(self.md.devices))
        self.assertEqual('device', self.md.devices[0])

    def test_add_device_in_spares_fail(self):
        self.assertEqual(0, len(self.md.devices))
        self.assertEqual(0, len(self.md.spares))
        self.md.add_spare('device')
        self.assertRaises(errors.MDDeviceDuplicationError, self.md.add_device,
                          'device')

    def test_add_device_in_devices_fail(self):
        self.assertEqual(0, len(self.md.devices))
        self.assertEqual(0, len(self.md.spares))
        self.md.add_device('device')
        self.assertRaises(errors.MDDeviceDuplicationError, self.md.add_device,
                          'device')

    def test_add_spare_in_spares_fail(self):
        self.assertEqual(0, len(self.md.devices))
        self.assertEqual(0, len(self.md.spares))
        self.md.add_spare('device')
        self.assertRaises(errors.MDDeviceDuplicationError, self.md.add_spare,
                          'device')

    def test_add_spare_in_devices_fail(self):
        self.assertEqual(0, len(self.md.devices))
        self.assertEqual(0, len(self.md.spares))
        self.md.add_device('device')
        self.assertRaises(errors.MDDeviceDuplicationError, self.md.add_spare,
                          'device')


class TestPartition(test_base.BaseTestCase):
    def setUp(self):
        super(TestPartition, self).setUp()
        self.pt = partition.Partition('name', 'count', 'device', 'begin',
                                      'end', 'partition_type')

    def test_set_flag(self):
        self.assertEqual(0, len(self.pt.flags))
        self.pt.set_flag('fake_flag')
        self.assertEqual(1, len(self.pt.flags))
        self.assertIn('fake_flag', self.pt.flags)


class TestPartitionScheme(test_base.BaseTestCase):
    def setUp(self):
        super(TestPartitionScheme, self).setUp()
        self.p_scheme = partition.PartitionScheme()

    def test_root_device_not_found(self):
        self.assertRaises(errors.WrongPartitionSchemeError,
                          self.p_scheme.root_device)

    def test_fs_by_device(self):
        expected_fs = partition.Fs('device')
        self.p_scheme.fss.append(expected_fs)
        self.p_scheme.fss.append(partition.Fs('wrong_device'))
        actual_fs = self.p_scheme.fs_by_device('device')
        self.assertEqual(expected_fs, actual_fs)

    def test_fs_by_mount(self):
        expected_fs = partition.Fs('d', mount='mount')
        self.p_scheme.fss.append(expected_fs)
        self.p_scheme.fss.append(partition.Fs('w_d', mount='wrong_mount'))
        actual_fs = self.p_scheme.fs_by_mount('mount')
        self.assertEqual(expected_fs, actual_fs)

    def test_pv_by_name(self):
        expected_pv = partition.Pv('pv')
        self.p_scheme.pvs.append(expected_pv)
        self.p_scheme.pvs.append(partition.Pv('wrong_pv'))
        actual_pv = self.p_scheme.pv_by_name('pv')
        self.assertEqual(expected_pv, actual_pv)

    def test_vg_by_name(self):
        expected_vg = partition.Vg('vg')
        self.p_scheme.vgs.append(expected_vg)
        self.p_scheme.vgs.append(partition.Vg('wrong_vg'))
        actual_vg = self.p_scheme.vg_by_name('vg')
        self.assertEqual(expected_vg, actual_vg)

    def test_vg_attach_by_name(self):
        self.p_scheme.vg_attach_by_name('pvname', 'vgname')
        self.assertEqual(1, len(self.p_scheme.pvs))
        self.assertEqual(1, len(self.p_scheme.vgs))
        self.assertIn('pvname', self.p_scheme.vgs[0].pvnames)
        self.assertIn('vgname', self.p_scheme.vgs[0].name)

    def test_md_next_name_ok(self):
        expected_name = '/dev/md0'
        self.assertEqual(expected_name, self.p_scheme.md_next_name())

    def test_md_next_name_fail(self):
        self.p_scheme.mds = [
            partition.Md('/dev/md%s' % x, 'level') for x in range(0, 128)]
        self.assertRaises(errors.MDAlreadyExistsError,
                          self.p_scheme.md_next_name)

    def test_md_by_name(self):
        self.assertEqual(0, len(self.p_scheme.mds))
        expected_md = partition.Md('name', 'level')
        self.p_scheme.mds.append(expected_md)
        self.p_scheme.mds.append(partition.Md('wrong_name', 'level'))
        self.assertEqual(expected_md, self.p_scheme.md_by_name('name'))

    def test_md_by_mount(self):
        self.assertEqual(0, len(self.p_scheme.mds))
        self.assertEqual(0, len(self.p_scheme.fss))
        expected_md = partition.Md('name', 'level')
        expected_fs = partition.Fs('name', mount='mount')
        self.p_scheme.mds.append(expected_md)
        self.p_scheme.fss.append(expected_fs)
        self.p_scheme.fss.append(partition.Fs('wrong_name',
                                 mount='wrong_mount'))
        self.assertEqual(expected_md, self.p_scheme.md_by_mount('mount'))

    def test_md_attach_by_mount_md_exists(self):
        self.assertEqual(0, len(self.p_scheme.mds))
        self.assertEqual(0, len(self.p_scheme.fss))
        expected_md = partition.Md('name', 'level')
        expected_fs = partition.Fs('name', mount='mount')
        self.p_scheme.mds.append(expected_md)
        self.p_scheme.fss.append(expected_fs)
        actual_md = self.p_scheme.md_attach_by_mount('device', 'mount')
        self.assertIn('device', actual_md.devices)
        self.assertEqual(expected_md, actual_md)

    def test_md_attach_by_mount_no_md(self):
        self.assertEqual(0, len(self.p_scheme.mds))
        self.assertEqual(0, len(self.p_scheme.fss))
        actual_md = self.p_scheme.md_attach_by_mount(
            'device', 'mount', fs_type='fs_type', fs_options='-F',
            fs_label='fs_label', name='name', level='level')
        self.assertIn('device', actual_md.devices)
        self.assertEqual(1, len(self.p_scheme.fss))
        self.assertEqual('name', self.p_scheme.fss[0].device)
        self.assertEqual('mount', self.p_scheme.fss[0].mount)
        self.assertEqual('fs_type', self.p_scheme.fss[0].type)
        self.assertEqual('fs_label', self.p_scheme.fss[0].label)
        self.assertEqual('-F', self.p_scheme.fss[0].options)


class TestParted(test_base.BaseTestCase):
    def setUp(self):
        super(TestParted, self).setUp()
        self.prtd = partition.Parted('name', 'label')

    @mock.patch.object(partition.Parted, 'next_count')
    @mock.patch.object(partition.Parted, 'next_type')
    def test_next_name_none(self, nt_mock, nc_mock):
        nc_mock.return_value = 1
        nt_mock.return_value = 'extended'
        self.assertEqual(None, self.prtd.next_name())

    @mock.patch.object(partition.Parted, 'next_count')
    @mock.patch.object(partition.Parted, 'next_type')
    def test_next_name_no_separator(self, nt_mock, nc_mock):
        nc_mock.return_value = 1
        nt_mock.return_value = 'not_extended'
        expected_name = '%s%s' % (self.prtd.name, 1)
        self.assertEqual(expected_name, self.prtd.next_name())

    @mock.patch.object(partition.Parted, 'next_count')
    @mock.patch.object(partition.Parted, 'next_type')
    def test_next_name_with_separator(self, nt_mock, nc_mock):
        nc_mock.return_value = 1
        nt_mock.return_value = 'not_extended'
        self.prtd.name = '/dev/cciss/c0d0'
        expected_name = '%sp%s' % (self.prtd.name, 1)
        self.assertEqual(expected_name, self.prtd.next_name())
        self.prtd.name = '/dev/loop123'
        expected_name = '%sp%s' % (self.prtd.name, 1)
        self.assertEqual(expected_name, self.prtd.next_name())
        self.prtd.name = '/dev/nvme0n1'
        expected_name = '%sp%s' % (self.prtd.name, 1)
        self.assertEqual(expected_name, self.prtd.next_name())

    def test_next_begin_empty_partitions(self):
        self.assertEqual(1, self.prtd.next_begin())

    def test_next_begin_last_extended_partition(self):
        self.prtd.partitions.append(
            partition.Partition('name', 'count', 'device', 'begin', 'end',
                                'extended'))
        self.assertEqual('begin', self.prtd.next_begin())

    def test_next_begin_no_last_extended_partition(self):
        self.prtd.partitions.append(
            partition.Partition('name', 'count', 'device', 'begin', 'end',
                                'primary'))
        self.assertEqual('end', self.prtd.next_begin())

    def test_next_count_no_logical(self):
        self.assertEqual(1, self.prtd.next_count('primary'))

    def test_next_count_has_logical(self):
        self.prtd.partitions.append(
            partition.Partition('name', 'count', 'device', 'begin', 'end',
                                'logical'))
        self.assertEqual(6, self.prtd.next_count('logical'))

    def test_next_type_gpt(self):
        self.prtd.label = 'gpt'
        self.assertEqual('primary', self.prtd.next_type())

    def test_next_type_no_extended(self):
        self.prtd.label = 'msdos'
        self.assertEqual('primary', self.prtd.next_type())
        self.prtd.partitions.extend(
            3 * [partition.Partition('name', 'count', 'device', 'begin',
                                     'end', 'primary')])
        self.assertEqual('extended', self.prtd.next_type())

    def test_next_type_has_extended(self):
        self.prtd.label = 'msdos'
        self.prtd.partitions.append(
            partition.Partition('name', 'count', 'device', 'begin', 'end',
                                'extended'))
        self.assertEqual('logical', self.prtd.next_type())

    def test_primary(self):
        expected_partitions = [partition.Partition('name', 'count', 'device',
                                                   'begin', 'end', 'primary')]
        self.prtd.partitions.extend(expected_partitions)
        self.assertEqual(expected_partitions, self.prtd.primary)

18035302f820e5a44a820dec29a476ae6c6eeb78
<mask token>


class TestPartitionScheme(test_base.BaseTestCase):

    def setUp(self):
        super(TestPartitionScheme, self).setUp()
        self.p_scheme = partition.PartitionScheme()

    def test_root_device_not_found(self):
        self.assertRaises(errors.WrongPartitionSchemeError, self.p_scheme.
            root_device)
    <mask token>

    def test_fs_by_mount(self):
        expected_fs = partition.Fs('d', mount='mount')
        self.p_scheme.fss.append(expected_fs)
        self.p_scheme.fss.append(partition.Fs('w_d', mount='wrong_mount'))
        actual_fs = self.p_scheme.fs_by_mount('mount')
        self.assertEqual(expected_fs, actual_fs)
    <mask token>

    def test_vg_by_name(self):
        expected_vg = partition.Vg('vg')
        self.p_scheme.vgs.append(expected_vg)
        self.p_scheme.vgs.append(partition.Vg('wrong_vg'))
        actual_vg = self.p_scheme.vg_by_name('vg')
        self.assertEqual(expected_vg, actual_vg)

    def test_vg_attach_by_name(self):
        self.p_scheme.vg_attach_by_name('pvname', 'vgname')
        self.assertEqual(1, len(self.p_scheme.pvs))
        self.assertEqual(1, len(self.p_scheme.vgs))
        self.assertIn('pvname', self.p_scheme.vgs[0].pvnames)
        self.assertIn('vgname', self.p_scheme.vgs[0].name)

    def test_md_next_name_ok(self):
        expected_name = '/dev/md0'
        self.assertEqual(expected_name, self.p_scheme.md_next_name())

    def test_md_next_name_fail(self):
        self.p_scheme.mds = [partition.Md('/dev/md%s' % x, 'level') for x in
            range(0, 128)]
        self.assertRaises(errors.MDAlreadyExistsError, self.p_scheme.
            md_next_name)

    def test_md_by_name(self):
        self.assertEqual(0, len(self.p_scheme.mds))
        expected_md = partition.Md('name', 'level')
        self.p_scheme.mds.append(expected_md)
        self.p_scheme.mds.append(partition.Md('wrong_name', 'level'))
        self.assertEqual(expected_md, self.p_scheme.md_by_name('name'))

    def test_md_by_mount(self):
        self.assertEqual(0, len(self.p_scheme.mds))
        self.assertEqual(0, len(self.p_scheme.fss))
        expected_md = partition.Md('name', 'level')
        expected_fs = partition.Fs('name', mount='mount')
        self.p_scheme.mds.append(expected_md)
        self.p_scheme.fss.append(expected_fs)
        self.p_scheme.fss.append(partition.Fs('wrong_name', mount=
            'wrong_mount'))
        self.assertEqual(expected_md, self.p_scheme.md_by_mount('mount'))
    <mask token>
    <mask token>


class TestParted(test_base.BaseTestCase):

    def setUp(self):
        super(TestParted, self).setUp()
        self.prtd = partition.Parted('name', 'label')

    @mock.patch.object(partition.Parted, 'next_count')
    @mock.patch.object(partition.Parted, 'next_type')
    def test_next_name_none(self, nt_mock, nc_mock):
        nc_mock.return_value = 1
        nt_mock.return_value = 'extended'
        self.assertEqual(None, self.prtd.next_name())

    @mock.patch.object(partition.Parted, 'next_count')
    @mock.patch.object(partition.Parted, 'next_type')
    def test_next_name_no_separator(self, nt_mock, nc_mock):
        nc_mock.return_value = 1
        nt_mock.return_value = 'not_extended'
        expected_name = '%s%s' % (self.prtd.name, 1)
        self.assertEqual(expected_name, self.prtd.next_name())

    @mock.patch.object(partition.Parted, 'next_count')
    @mock.patch.object(partition.Parted, 'next_type')
    def test_next_name_with_separator(self, nt_mock, nc_mock):
        nc_mock.return_value = 1
        nt_mock.return_value = 'not_extended'
        self.prtd.name = '/dev/cciss/c0d0'
        expected_name = '%sp%s' % (self.prtd.name, 1)
        self.assertEqual(expected_name, self.prtd.next_name())
        self.prtd.name = '/dev/loop123'
        expected_name = '%sp%s' % (self.prtd.name, 1)
        self.assertEqual(expected_name, self.prtd.next_name())
        self.prtd.name = '/dev/nvme0n1'
        expected_name = '%sp%s' % (self.prtd.name, 1)
        self.assertEqual(expected_name, self.prtd.next_name())

    def test_next_begin_empty_partitions(self):
        self.assertEqual(1, self.prtd.next_begin())

    def test_next_begin_last_extended_partition(self):
        self.prtd.partitions.append(partition.Partition('name', 'count',
            'device', 'begin', 'end', 'extended'))
        self.assertEqual('begin', self.prtd.next_begin())

    def test_next_begin_no_last_extended_partition(self):
        self.prtd.partitions.append(partition.Partition('name', 'count',
            'device', 'begin', 'end', 'primary'))
        self.assertEqual('end', self.prtd.next_begin())

    def test_next_count_no_logical(self):
        self.assertEqual(1, self.prtd.next_count('primary'))

    def test_next_count_has_logical(self):
        self.prtd.partitions.append(partition.Partition('name', 'count',
            'device', 'begin', 'end', 'logical'))
        self.assertEqual(6, self.prtd.next_count('logical'))

    def test_next_type_gpt(self):
        self.prtd.label = 'gpt'
        self.assertEqual('primary', self.prtd.next_type())

    def test_next_type_no_extended(self):
        self.prtd.label = 'msdos'
        self.assertEqual('primary', self.prtd.next_type())
        self.prtd.partitions.extend(3 * [partition.Partition('name',
            'count', 'device', 'begin', 'end', 'primary')])
        self.assertEqual('extended', self.prtd.next_type())

    def test_next_type_has_extended(self):
        self.prtd.label = 'msdos'
        self.prtd.partitions.append(partition.Partition('name', 'count',
            'device', 'begin', 'end', 'extended'))
        self.assertEqual('logical', self.prtd.next_type())

    def test_primary(self):
        expected_partitions = [partition.Partition('name', 'count',
            'device', 'begin', 'end', 'primary')]
        self.prtd.partitions.extend(expected_partitions)
        self.assertEqual(expected_partitions, self.prtd.primary)

<mask token>


class TestPartitionScheme(test_base.BaseTestCase):

    def setUp(self):
        super(TestPartitionScheme, self).setUp()
        self.p_scheme = partition.PartitionScheme()

    def test_root_device_not_found(self):
        self.assertRaises(errors.WrongPartitionSchemeError, self.p_scheme.
            root_device)

    def test_fs_by_device(self):
        expected_fs = partition.Fs('device')
        self.p_scheme.fss.append(expected_fs)
        self.p_scheme.fss.append(partition.Fs('wrong_device'))
        actual_fs = self.p_scheme.fs_by_device('device')
        self.assertEqual(expected_fs, actual_fs)

    def test_fs_by_mount(self):
        expected_fs = partition.Fs('d', mount='mount')
        self.p_scheme.fss.append(expected_fs)
        self.p_scheme.fss.append(partition.Fs('w_d', mount='wrong_mount'))
        actual_fs = self.p_scheme.fs_by_mount('mount')
        self.assertEqual(expected_fs, actual_fs)

    def test_pv_by_name(self):
        expected_pv = partition.Pv('pv')
        self.p_scheme.pvs.append(expected_pv)
        self.p_scheme.pvs.append(partition.Pv('wrong_pv'))
        actual_pv = self.p_scheme.pv_by_name('pv')
        self.assertEqual(expected_pv, actual_pv)

    def test_vg_by_name(self):
        expected_vg = partition.Vg('vg')
        self.p_scheme.vgs.append(expected_vg)
        self.p_scheme.vgs.append(partition.Vg('wrong_vg'))
        actual_vg = self.p_scheme.vg_by_name('vg')
        self.assertEqual(expected_vg, actual_vg)

    def test_vg_attach_by_name(self):
        self.p_scheme.vg_attach_by_name('pvname', 'vgname')
        self.assertEqual(1, len(self.p_scheme.pvs))
        self.assertEqual(1, len(self.p_scheme.vgs))
        self.assertIn('pvname', self.p_scheme.vgs[0].pvnames)
        self.assertIn('vgname', self.p_scheme.vgs[0].name)

    def test_md_next_name_ok(self):
        expected_name = '/dev/md0'
        self.assertEqual(expected_name, self.p_scheme.md_next_name())

    def test_md_next_name_fail(self):
        self.p_scheme.mds = [partition.Md('/dev/md%s' % x, 'level') for x in
            range(0, 128)]
        self.assertRaises(errors.MDAlreadyExistsError, self.p_scheme.
            md_next_name)

    def test_md_by_name(self):
        self.assertEqual(0, len(self.p_scheme.mds))
        expected_md = partition.Md('name', 'level')
        self.p_scheme.mds.append(expected_md)
        self.p_scheme.mds.append(partition.Md('wrong_name', 'level'))
        self.assertEqual(expected_md, self.p_scheme.md_by_name('name'))

    def test_md_by_mount(self):
        self.assertEqual(0, len(self.p_scheme.mds))
        self.assertEqual(0, len(self.p_scheme.fss))
        expected_md = partition.Md('name', 'level')
        expected_fs = partition.Fs('name', mount='mount')
        self.p_scheme.mds.append(expected_md)
        self.p_scheme.fss.append(expected_fs)
        self.p_scheme.fss.append(partition.Fs('wrong_name', mount=
            'wrong_mount'))
        self.assertEqual(expected_md, self.p_scheme.md_by_mount('mount'))

    def test_md_attach_by_mount_md_exists(self):
        self.assertEqual(0, len(self.p_scheme.mds))
        self.assertEqual(0, len(self.p_scheme.fss))
        expected_md = partition.Md('name', 'level')
        expected_fs = partition.Fs('name', mount='mount')
        self.p_scheme.mds.append(expected_md)
        self.p_scheme.fss.append(expected_fs)
        actual_md = self.p_scheme.md_attach_by_mount('device', 'mount')
        self.assertIn('device', actual_md.devices)
        self.assertEqual(expected_md, actual_md)

    def test_md_attach_by_mount_no_md(self):
        self.assertEqual(0, len(self.p_scheme.mds))
        self.assertEqual(0, len(self.p_scheme.fss))
        actual_md = self.p_scheme.md_attach_by_mount('device', 'mount',
            fs_type='fs_type', fs_options='-F', fs_label='fs_label', name=
            'name', level='level')
        self.assertIn('device', actual_md.devices)
        self.assertEqual(1, len(self.p_scheme.fss))
        self.assertEqual('name', self.p_scheme.fss[0].device)
        self.assertEqual('mount', self.p_scheme.fss[0].mount)
        self.assertEqual('fs_type', self.p_scheme.fss[0].type)
        self.assertEqual('fs_label', self.p_scheme.fss[0].label)
        self.assertEqual('-F', self.p_scheme.fss[0].options)


class TestParted(test_base.BaseTestCase):

    def setUp(self):
        super(TestParted, self).setUp()
        self.prtd = partition.Parted('name', 'label')

    @mock.patch.object(partition.Parted, 'next_count')
    @mock.patch.object(partition.Parted, 'next_type')
    def test_next_name_none(self, nt_mock, nc_mock):
        nc_mock.return_value = 1
        nt_mock.return_value = 'extended'
        self.assertEqual(None, self.prtd.next_name())

    @mock.patch.object(partition.Parted, 'next_count')
    @mock.patch.object(partition.Parted, 'next_type')
    def test_next_name_no_separator(self, nt_mock, nc_mock):
        nc_mock.return_value = 1
        nt_mock.return_value = 'not_extended'
        expected_name = '%s%s' % (self.prtd.name, 1)
        self.assertEqual(expected_name, self.prtd.next_name())

    @mock.patch.object(partition.Parted, 'next_count')
    @mock.patch.object(partition.Parted, 'next_type')
    def test_next_name_with_separator(self, nt_mock, nc_mock):
        nc_mock.return_value = 1
        nt_mock.return_value = 'not_extended'
        self.prtd.name = '/dev/cciss/c0d0'
        expected_name = '%sp%s' % (self.prtd.name, 1)
        self.assertEqual(expected_name, self.prtd.next_name())
        self.prtd.name = '/dev/loop123'
        expected_name = '%sp%s' % (self.prtd.name, 1)
        self.assertEqual(expected_name, self.prtd.next_name())
        self.prtd.name = '/dev/nvme0n1'
        expected_name = '%sp%s' % (self.prtd.name, 1)
        self.assertEqual(expected_name, self.prtd.next_name())

    def test_next_begin_empty_partitions(self):
        self.assertEqual(1, self.prtd.next_begin())

    def test_next_begin_last_extended_partition(self):
        self.prtd.partitions.append(partition.Partition('name', 'count',
            'device', 'begin', 'end', 'extended'))
        self.assertEqual('begin', self.prtd.next_begin())

    def test_next_begin_no_last_extended_partition(self):
        self.prtd.partitions.append(partition.Partition('name', 'count',
            'device', 'begin', 'end', 'primary'))
        self.assertEqual('end', self.prtd.next_begin())

    def test_next_count_no_logical(self):
        self.assertEqual(1, self.prtd.next_count('primary'))

    def test_next_count_has_logical(self):
        self.prtd.partitions.append(partition.Partition('name', 'count',
            'device', 'begin', 'end', 'logical'))
        self.assertEqual(6, self.prtd.next_count('logical'))

    def test_next_type_gpt(self):
        self.prtd.label = 'gpt'
        self.assertEqual('primary', self.prtd.next_type())

    def test_next_type_no_extended(self):
        self.prtd.label = 'msdos'
        self.assertEqual('primary', self.prtd.next_type())
        self.prtd.partitions.extend(3 * [partition.Partition('name',
            'count', 'device', 'begin', 'end', 'primary')])
        self.assertEqual('extended', self.prtd.next_type())

    def test_next_type_has_extended(self):
        self.prtd.label = 'msdos'
        self.prtd.partitions.append(partition.Partition('name', 'count',
            'device', 'begin', 'end', 'extended'))
        self.assertEqual('logical', self.prtd.next_type())

    def test_primary(self):
        expected_partitions = [partition.Partition('name', 'count',
            'device', 'begin', 'end', 'primary')]
        self.prtd.partitions.extend(expected_partitions)
        self.assertEqual(expected_partitions, self.prtd.primary)

<mask token>


class TestMD(test_base.BaseTestCase):
    <mask token>
    <mask token>

    def test_add_device_in_spares_fail(self):
        self.assertEqual(0, len(self.md.devices))
        self.assertEqual(0, len(self.md.spares))
        self.md.add_spare('device')
        self.assertRaises(errors.MDDeviceDuplicationError, self.md.
            add_device, 'device')

    def test_add_device_in_devices_fail(self):
        self.assertEqual(0, len(self.md.devices))
        self.assertEqual(0, len(self.md.spares))
        self.md.add_device('device')
        self.assertRaises(errors.MDDeviceDuplicationError, self.md.
            add_device, 'device')

    def test_add_spare_in_spares_fail(self):
        self.assertEqual(0, len(self.md.devices))
        self.assertEqual(0, len(self.md.spares))
        self.md.add_spare('device')
        self.assertRaises(errors.MDDeviceDuplicationError, self.md.
            add_spare, 'device')
    <mask token>


class TestPartition(test_base.BaseTestCase):

    def setUp(self):
        super(TestPartition, self).setUp()
        self.pt = partition.Partition('name', 'count', 'device', 'begin',
            'end', 'partition_type')

    def test_set_flag(self):
        self.assertEqual(0, len(self.pt.flags))
        self.pt.set_flag('fake_flag')
        self.assertEqual(1, len(self.pt.flags))
        self.assertIn('fake_flag', self.pt.flags)


class TestPartitionScheme(test_base.BaseTestCase):

    def setUp(self):
        super(TestPartitionScheme, self).setUp()
        self.p_scheme = partition.PartitionScheme()

    def test_root_device_not_found(self):
        self.assertRaises(errors.WrongPartitionSchemeError, self.p_scheme.
            root_device)

    def test_fs_by_device(self):
        expected_fs = partition.Fs('device')
        self.p_scheme.fss.append(expected_fs)
        self.p_scheme.fss.append(partition.Fs('wrong_device'))
        actual_fs = self.p_scheme.fs_by_device('device')
        self.assertEqual(expected_fs, actual_fs)

    def test_fs_by_mount(self):
        expected_fs = partition.Fs('d', mount='mount')
        self.p_scheme.fss.append(expected_fs)
        self.p_scheme.fss.append(partition.Fs('w_d', mount='wrong_mount'))
        actual_fs = self.p_scheme.fs_by_mount('mount')
        self.assertEqual(expected_fs, actual_fs)

    def test_pv_by_name(self):
        expected_pv = partition.Pv('pv')
        self.p_scheme.pvs.append(expected_pv)
        self.p_scheme.pvs.append(partition.Pv('wrong_pv'))
        actual_pv = self.p_scheme.pv_by_name('pv')
        self.assertEqual(expected_pv, actual_pv)

    def test_vg_by_name(self):
        expected_vg = partition.Vg('vg')
        self.p_scheme.vgs.append(expected_vg)
        self.p_scheme.vgs.append(partition.Vg('wrong_vg'))
        actual_vg = self.p_scheme.vg_by_name('vg')
        self.assertEqual(expected_vg, actual_vg)

    def test_vg_attach_by_name(self):
        self.p_scheme.vg_attach_by_name('pvname', 'vgname')
        self.assertEqual(1, len(self.p_scheme.pvs))
        self.assertEqual(1, len(self.p_scheme.vgs))
        self.assertIn('pvname', self.p_scheme.vgs[0].pvnames)
        self.assertIn('vgname', self.p_scheme.vgs[0].name)

    def test_md_next_name_ok(self):
        expected_name = '/dev/md0'
        self.assertEqual(expected_name, self.p_scheme.md_next_name())

    def test_md_next_name_fail(self):
        self.p_scheme.mds = [partition.Md('/dev/md%s' % x, 'level') for x in
            range(0, 128)]
        self.assertRaises(errors.MDAlreadyExistsError, self.p_scheme.
            md_next_name)

    def test_md_by_name(self):
        self.assertEqual(0, len(self.p_scheme.mds))
        expected_md = partition.Md('name', 'level')
        self.p_scheme.mds.append(expected_md)
        self.p_scheme.mds.append(partition.Md('wrong_name', 'level'))
        self.assertEqual(expected_md, self.p_scheme.md_by_name('name'))

    def test_md_by_mount(self):
        self.assertEqual(0, len(self.p_scheme.mds))
        self.assertEqual(0, len(self.p_scheme.fss))
        expected_md = partition.Md('name', 'level')
        expected_fs = partition.Fs('name', mount='mount')
        self.p_scheme.mds.append(expected_md)
        self.p_scheme.fss.append(expected_fs)
        self.p_scheme.fss.append(partition.Fs('wrong_name', mount=
            'wrong_mount'))
        self.assertEqual(expected_md, self.p_scheme.md_by_mount('mount'))

    def test_md_attach_by_mount_md_exists(self):
        self.assertEqual(0, len(self.p_scheme.mds))
        self.assertEqual(0, len(self.p_scheme.fss))
        expected_md = partition.Md('name', 'level')
        expected_fs = partition.Fs('name', mount='mount')
        self.p_scheme.mds.append(expected_md)
        self.p_scheme.fss.append(expected_fs)
        actual_md = self.p_scheme.md_attach_by_mount('device', 'mount')
        self.assertIn('device', actual_md.devices)
        self.assertEqual(expected_md, actual_md)

    def test_md_attach_by_mount_no_md(self):
        self.assertEqual(0, len(self.p_scheme.mds))
        self.assertEqual(0, len(self.p_scheme.fss))
        actual_md = self.p_scheme.md_attach_by_mount('device', 'mount',
            fs_type='fs_type', fs_options='-F', fs_label='fs_label', name=
            'name', level='level')
        self.assertIn('device', actual_md.devices)
        self.assertEqual(1, len(self.p_scheme.fss))
        self.assertEqual('name', self.p_scheme.fss[0].device)
        self.assertEqual('mount', self.p_scheme.fss[0].mount)
        self.assertEqual('fs_type', self.p_scheme.fss[0].type)
        self.assertEqual('fs_label', self.p_scheme.fss[0].label)
        self.assertEqual('-F', self.p_scheme.fss[0].options)


class TestParted(test_base.BaseTestCase):

    def setUp(self):
        super(TestParted, self).setUp()
        self.prtd = partition.Parted('name', 'label')

    @mock.patch.object(partition.Parted, 'next_count')
    @mock.patch.object(partition.Parted, 'next_type')
    def test_next_name_none(self, nt_mock, nc_mock):
        nc_mock.return_value = 1
        nt_mock.return_value = 'extended'
        self.assertEqual(None, self.prtd.next_name())

    @mock.patch.object(partition.Parted, 'next_count')
    @mock.patch.object(partition.Parted, 'next_type')
    def test_next_name_no_separator(self, nt_mock, nc_mock):
        nc_mock.return_value = 1
        nt_mock.return_value = 'not_extended'
        expected_name = '%s%s' % (self.prtd.name, 1)
        self.assertEqual(expected_name, self.prtd.next_name())

    @mock.patch.object(partition.Parted, 'next_count')
    @mock.patch.object(partition.Parted, 'next_type')
    def test_next_name_with_separator(self, nt_mock, nc_mock):
        nc_mock.return_value = 1
        nt_mock.return_value = 'not_extended'
        self.prtd.name = '/dev/cciss/c0d0'
        expected_name = '%sp%s' % (self.prtd.name, 1)
        self.assertEqual(expected_name, self.prtd.next_name())
        self.prtd.name = '/dev/loop123'
        expected_name = '%sp%s' % (self.prtd.name, 1)
        self.assertEqual(expected_name, self.prtd.next_name())
        self.prtd.name = '/dev/nvme0n1'
        expected_name = '%sp%s' % (self.prtd.name, 1)
        self.assertEqual(expected_name, self.prtd.next_name())

    def test_next_begin_empty_partitions(self):
        self.assertEqual(1, self.prtd.next_begin())

    def test_next_begin_last_extended_partition(self):
        self.prtd.partitions.append(partition.Partition('name', 'count',
            'device', 'begin', 'end', 'extended'))
        self.assertEqual('begin', self.prtd.next_begin())

    def test_next_begin_no_last_extended_partition(self):
        self.prtd.partitions.append(partition.Partition('name', 'count',
            'device', 'begin', 'end', 'primary'))
        self.assertEqual('end', self.prtd.next_begin())

    def test_next_count_no_logical(self):
        self.assertEqual(1, self.prtd.next_count('primary'))

    def test_next_count_has_logical(self):
        self.prtd.partitions.append(partition.Partition('name', 'count',
            'device', 'begin', 'end', 'logical'))
        self.assertEqual(6, self.prtd.next_count('logical'))

    def test_next_type_gpt(self):
        self.prtd.label = 'gpt'
        self.assertEqual('primary', self.prtd.next_type())

    def test_next_type_no_extended(self):
        self.prtd.label = 'msdos'
        self.assertEqual('primary', self.prtd.next_type())
        self.prtd.partitions.extend(3 * [partition.Partition('name',
            'count', 'device', 'begin', 'end', 'primary')])
        self.assertEqual('extended', self.prtd.next_type())

    def test_next_type_has_extended(self):
        self.prtd.label = 'msdos'
        self.prtd.partitions.append(partition.Partition('name', 'count',
            'device', 'begin', 'end', 'extended'))
        self.assertEqual('logical', self.prtd.next_type())

    def test_primary(self):
        expected_partitions = [partition.Partition('name', 'count',
            'device', 'begin', 'end', 'primary')]
        self.prtd.partitions.extend(expected_partitions)
        self.assertEqual(expected_partitions, self.prtd.primary)

<mask token>


class TestMD(test_base.BaseTestCase):

    def setUp(self):
        super(TestMD, self).setUp()
        self.md = partition.Md('name', 'level')

    def test_add_device_ok(self):
        self.assertEqual(0, len(self.md.devices))
        self.md.add_device('device')
        self.assertEqual(1, len(self.md.devices))
        self.assertEqual('device', self.md.devices[0])

    def test_add_device_in_spares_fail(self):
        self.assertEqual(0, len(self.md.devices))
        self.assertEqual(0, len(self.md.spares))
        self.md.add_spare('device')
        self.assertRaises(errors.MDDeviceDuplicationError, self.md.
            add_device, 'device')

    def test_add_device_in_devices_fail(self):
        self.assertEqual(0, len(self.md.devices))
        self.assertEqual(0, len(self.md.spares))
        self.md.add_device('device')
        self.assertRaises(errors.MDDeviceDuplicationError, self.md.
            add_device, 'device')

    def test_add_spare_in_spares_fail(self):
        self.assertEqual(0, len(self.md.devices))
        self.assertEqual(0, len(self.md.spares))
        self.md.add_spare('device')
        self.assertRaises(errors.MDDeviceDuplicationError, self.md.
            add_spare, 'device')

    def test_add_spare_in_devices_fail(self):
        self.assertEqual(0, len(self.md.devices))
        self.assertEqual(0, len(self.md.spares))
        self.md.add_device('device')
        self.assertRaises(errors.MDDeviceDuplicationError, self.md.
            add_spare, 'device')


class TestPartition(test_base.BaseTestCase):

    def setUp(self):
        super(TestPartition, self).setUp()
        self.pt = partition.Partition('name', 'count', 'device', 'begin',
            'end', 'partition_type')

    def test_set_flag(self):
        self.assertEqual(0, len(self.pt.flags))
        self.pt.set_flag('fake_flag')
        self.assertEqual(1, len(self.pt.flags))
        self.assertIn('fake_flag', self.pt.flags)


class TestPartitionScheme(test_base.BaseTestCase):

    def setUp(self):
        super(TestPartitionScheme, self).setUp()
        self.p_scheme = partition.PartitionScheme()

    def test_root_device_not_found(self):
        self.assertRaises(errors.WrongPartitionSchemeError, self.p_scheme.
            root_device)

    def test_fs_by_device(self):
        expected_fs = partition.Fs('device')
        self.p_scheme.fss.append(expected_fs)
        self.p_scheme.fss.append(partition.Fs('wrong_device'))
        actual_fs = self.p_scheme.fs_by_device('device')
        self.assertEqual(expected_fs, actual_fs)

    def test_fs_by_mount(self):
        expected_fs = partition.Fs('d', mount='mount')
        self.p_scheme.fss.append(expected_fs)
        self.p_scheme.fss.append(partition.Fs('w_d', mount='wrong_mount'))
        actual_fs = self.p_scheme.fs_by_mount('mount')
        self.assertEqual(expected_fs, actual_fs)

    def test_pv_by_name(self):
        expected_pv = partition.Pv('pv')
        self.p_scheme.pvs.append(expected_pv)
        self.p_scheme.pvs.append(partition.Pv('wrong_pv'))
        actual_pv = self.p_scheme.pv_by_name('pv')
        self.assertEqual(expected_pv, actual_pv)

    def test_vg_by_name(self):
        expected_vg = partition.Vg('vg')
        self.p_scheme.vgs.append(expected_vg)
        self.p_scheme.vgs.append(partition.Vg('wrong_vg'))
        actual_vg = self.p_scheme.vg_by_name('vg')
        self.assertEqual(expected_vg, actual_vg)

    def test_vg_attach_by_name(self):
        self.p_scheme.vg_attach_by_name('pvname', 'vgname')
        self.assertEqual(1, len(self.p_scheme.pvs))
        self.assertEqual(1, len(self.p_scheme.vgs))
        self.assertIn('pvname', self.p_scheme.vgs[0].pvnames)
        self.assertIn('vgname', self.p_scheme.vgs[0].name)

    def test_md_next_name_ok(self):
        expected_name = '/dev/md0'
        self.assertEqual(expected_name, self.p_scheme.md_next_name())

    def test_md_next_name_fail(self):
        self.p_scheme.mds = [partition.Md('/dev/md%s' % x, 'level') for x in
            range(0, 128)]
        self.assertRaises(errors.MDAlreadyExistsError, self.p_scheme.
            md_next_name)

    def test_md_by_name(self):
        self.assertEqual(0, len(self.p_scheme.mds))
        expected_md = partition.Md('name', 'level')
        self.p_scheme.mds.append(expected_md)
        self.p_scheme.mds.append(partition.Md('wrong_name', 'level'))
        self.assertEqual(expected_md, self.p_scheme.md_by_name('name'))

    def test_md_by_mount(self):
        self.assertEqual(0, len(self.p_scheme.mds))
        self.assertEqual(0, len(self.p_scheme.fss))
        expected_md = partition.Md('name', 'level')
        expected_fs = partition.Fs('name', mount='mount')
        self.p_scheme.mds.append(expected_md)
        self.p_scheme.fss.append(expected_fs)
        self.p_scheme.fss.append(partition.Fs('wrong_name', mount=
            'wrong_mount'))
        self.assertEqual(expected_md, self.p_scheme.md_by_mount('mount'))

    def test_md_attach_by_mount_md_exists(self):
        self.assertEqual(0, len(self.p_scheme.mds))
        self.assertEqual(0, len(self.p_scheme.fss))
        expected_md = partition.Md('name', 'level')
        expected_fs = partition.Fs('name', mount='mount')
        self.p_scheme.mds.append(expected_md)
        self.p_scheme.fss.append(expected_fs)
        actual_md = self.p_scheme.md_attach_by_mount('device', 'mount')
        self.assertIn('device', actual_md.devices)
        self.assertEqual(expected_md, actual_md)

    def test_md_attach_by_mount_no_md(self):
        self.assertEqual(0, len(self.p_scheme.mds))
        self.assertEqual(0, len(self.p_scheme.fss))
        actual_md = self.p_scheme.md_attach_by_mount('device', 'mount',
            fs_type='fs_type', fs_options='-F', fs_label='fs_label', name=
            'name', level='level')
        self.assertIn('device', actual_md.devices)
        self.assertEqual(1, len(self.p_scheme.fss))
        self.assertEqual('name', self.p_scheme.fss[0].device)
        self.assertEqual('mount', self.p_scheme.fss[0].mount)
        self.assertEqual('fs_type', self.p_scheme.fss[0].type)
        self.assertEqual('fs_label', self.p_scheme.fss[0].label)
        self.assertEqual('-F', self.p_scheme.fss[0].options)


class TestParted(test_base.BaseTestCase):

    def setUp(self):
        super(TestParted, self).setUp()
        self.prtd = partition.Parted('name', 'label')

    @mock.patch.object(partition.Parted, 'next_count')
    @mock.patch.object(partition.Parted, 'next_type')
    def test_next_name_none(self, nt_mock, nc_mock):
        nc_mock.return_value = 1
        nt_mock.return_value = 'extended'
        self.assertEqual(None, self.prtd.next_name())

    @mock.patch.object(partition.Parted, 'next_count')
    @mock.patch.object(partition.Parted, 'next_type')
    def test_next_name_no_separator(self, nt_mock, nc_mock):
        nc_mock.return_value = 1
        nt_mock.return_value = 'not_extended'
        expected_name = '%s%s' % (self.prtd.name, 1)
        self.assertEqual(expected_name, self.prtd.next_name())

    @mock.patch.object(partition.Parted, 'next_count')
    @mock.patch.object(partition.Parted, 'next_type')
    def test_next_name_with_separator(self, nt_mock, nc_mock):
        nc_mock.return_value = 1
        nt_mock.return_value = 'not_extended'
        self.prtd.name = '/dev/cciss/c0d0'
        expected_name = '%sp%s' % (self.prtd.name, 1)
        self.assertEqual(expected_name, self.prtd.next_name())
        self.prtd.name = '/dev/loop123'
        expected_name = '%sp%s' % (self.prtd.name, 1)
        self.assertEqual(expected_name, self.prtd.next_name())
        self.prtd.name = '/dev/nvme0n1'
        expected_name = '%sp%s' % (self.prtd.name, 1)
        self.assertEqual(expected_name, self.prtd.next_name())

    def test_next_begin_empty_partitions(self):
        self.assertEqual(1, self.prtd.next_begin())

    def test_next_begin_last_extended_partition(self):
        self.prtd.partitions.append(partition.Partition('name', 'count',
            'device', 'begin', 'end', 'extended'))
        self.assertEqual('begin', self.prtd.next_begin())

    def test_next_begin_no_last_extended_partition(self):
        self.prtd.partitions.append(partition.Partition('name', 'count',
            'device', 'begin', 'end', 'primary'))
        self.assertEqual('end', self.prtd.next_begin())

    def test_next_count_no_logical(self):
        self.assertEqual(1, self.prtd.next_count('primary'))

    def test_next_count_has_logical(self):
        self.prtd.partitions.append(partition.Partition('name', 'count',
            'device', 'begin', 'end', 'logical'))
        self.assertEqual(6, self.prtd.next_count('logical'))

    def test_next_type_gpt(self):
        self.prtd.label = 'gpt'
        self.assertEqual('primary', self.prtd.next_type())

    def test_next_type_no_extended(self):
        self.prtd.label = 'msdos'
        self.assertEqual('primary', self.prtd.next_type())
        self.prtd.partitions.extend(3 * [partition.Partition('name',
            'count', 'device', 'begin', 'end', 'primary')])
        self.assertEqual('extended', self.prtd.next_type())

    def test_next_type_has_extended(self):
        self.prtd.label = 'msdos'
        self.prtd.partitions.append(partition.Partition('name', 'count',
            'device', 'begin', 'end', 'extended'))
        self.assertEqual('logical', self.prtd.next_type())

    def test_primary(self):
        expected_partitions = [partition.Partition('name', 'count',
            'device', 'begin', 'end', 'primary')]
        self.prtd.partitions.extend(expected_partitions)
        self.assertEqual(expected_partitions, self.prtd.primary)

# Copyright 2014 Mirantis, Inc.
#
# 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 mock

from oslotest import base as test_base

from fuel_agent import errors
from fuel_agent.objects import partition


class TestMD(test_base.BaseTestCase):
    def setUp(self):
        super(TestMD, self).setUp()
        self.md = partition.Md('name', 'level')

    def test_add_device_ok(self):
        self.assertEqual(0, len(self.md.devices))
        self.md.add_device('device')
        self.assertEqual(1, len(self.md.devices))
        self.assertEqual('device', self.md.devices[0])

    def test_add_device_in_spares_fail(self):
        self.assertEqual(0, len(self.md.devices))
        self.assertEqual(0, len(self.md.spares))
        self.md.add_spare('device')
        self.assertRaises(errors.MDDeviceDuplicationError, self.md.add_device,
                          'device')

    def test_add_device_in_devices_fail(self):
        self.assertEqual(0, len(self.md.devices))
        self.assertEqual(0, len(self.md.spares))
        self.md.add_device('device')
        self.assertRaises(errors.MDDeviceDuplicationError, self.md.add_device,
                          'device')

    def test_add_spare_in_spares_fail(self):
        self.assertEqual(0, len(self.md.devices))
        self.assertEqual(0, len(self.md.spares))
        self.md.add_spare('device')
        self.assertRaises(errors.MDDeviceDuplicationError, self.md.add_spare,
                          'device')

    def test_add_spare_in_devices_fail(self):
        self.assertEqual(0, len(self.md.devices))
        self.assertEqual(0, len(self.md.spares))
        self.md.add_device('device')
        self.assertRaises(errors.MDDeviceDuplicationError, self.md.add_spare,
                          'device')


class TestPartition(test_base.BaseTestCase):
    def setUp(self):
        super(TestPartition, self).setUp()
        self.pt = partition.Partition('name', 'count', 'device', 'begin',
                                      'end', 'partition_type')

    def test_set_flag(self):
        self.assertEqual(0, len(self.pt.flags))
        self.pt.set_flag('fake_flag')
        self.assertEqual(1, len(self.pt.flags))
        self.assertIn('fake_flag', self.pt.flags)


class TestPartitionScheme(test_base.BaseTestCase):
    def setUp(self):
        super(TestPartitionScheme, self).setUp()
        self.p_scheme = partition.PartitionScheme()

    def test_root_device_not_found(self):
        self.assertRaises(errors.WrongPartitionSchemeError,
                          self.p_scheme.root_device)

    def test_fs_by_device(self):
        expected_fs = partition.Fs('device')
        self.p_scheme.fss.append(expected_fs)
        self.p_scheme.fss.append(partition.Fs('wrong_device'))
        actual_fs = self.p_scheme.fs_by_device('device')
        self.assertEqual(expected_fs, actual_fs)

    def test_fs_by_mount(self):
        expected_fs = partition.Fs('d', mount='mount')
        self.p_scheme.fss.append(expected_fs)
        self.p_scheme.fss.append(partition.Fs('w_d', mount='wrong_mount'))
        actual_fs = self.p_scheme.fs_by_mount('mount')
        self.assertEqual(expected_fs, actual_fs)

    def test_pv_by_name(self):
        expected_pv = partition.Pv('pv')
        self.p_scheme.pvs.append(expected_pv)
        self.p_scheme.pvs.append(partition.Pv('wrong_pv'))
        actual_pv = self.p_scheme.pv_by_name('pv')
        self.assertEqual(expected_pv, actual_pv)

    def test_vg_by_name(self):
        expected_vg = partition.Vg('vg')
        self.p_scheme.vgs.append(expected_vg)
        self.p_scheme.vgs.append(partition.Vg('wrong_vg'))
        actual_vg = self.p_scheme.vg_by_name('vg')
        self.assertEqual(expected_vg, actual_vg)

    def test_vg_attach_by_name(self):
        self.p_scheme.vg_attach_by_name('pvname', 'vgname')
        self.assertEqual(1, len(self.p_scheme.pvs))
        self.assertEqual(1, len(self.p_scheme.vgs))
        self.assertIn('pvname', self.p_scheme.vgs[0].pvnames)
        self.assertIn('vgname', self.p_scheme.vgs[0].name)

    def test_md_next_name_ok(self):
        expected_name = '/dev/md0'
        self.assertEqual(expected_name, self.p_scheme.md_next_name())

    def test_md_next_name_fail(self):
        self.p_scheme.mds = [
            partition.Md('/dev/md%s' % x, 'level') for x in range(0, 128)]
        self.assertRaises(errors.MDAlreadyExistsError,
                          self.p_scheme.md_next_name)

    def test_md_by_name(self):
        self.assertEqual(0, len(self.p_scheme.mds))
        expected_md = partition.Md('name', 'level')
        self.p_scheme.mds.append(expected_md)
        self.p_scheme.mds.append(partition.Md('wrong_name', 'level'))
        self.assertEqual(expected_md, self.p_scheme.md_by_name('name'))

    def test_md_by_mount(self):
        self.assertEqual(0, len(self.p_scheme.mds))
        self.assertEqual(0, len(self.p_scheme.fss))
        expected_md = partition.Md('name', 'level')
        expected_fs = partition.Fs('name', mount='mount')
        self.p_scheme.mds.append(expected_md)
        self.p_scheme.fss.append(expected_fs)
        self.p_scheme.fss.append(partition.Fs('wrong_name',
                                 mount='wrong_mount'))
        self.assertEqual(expected_md, self.p_scheme.md_by_mount('mount'))

    def test_md_attach_by_mount_md_exists(self):
        self.assertEqual(0, len(self.p_scheme.mds))
        self.assertEqual(0, len(self.p_scheme.fss))
        expected_md = partition.Md('name', 'level')
        expected_fs = partition.Fs('name', mount='mount')
        self.p_scheme.mds.append(expected_md)
        self.p_scheme.fss.append(expected_fs)
        actual_md = self.p_scheme.md_attach_by_mount('device', 'mount')
        self.assertIn('device', actual_md.devices)
        self.assertEqual(expected_md, actual_md)

    def test_md_attach_by_mount_no_md(self):
        self.assertEqual(0, len(self.p_scheme.mds))
        self.assertEqual(0, len(self.p_scheme.fss))
        actual_md = self.p_scheme.md_attach_by_mount(
            'device', 'mount', fs_type='fs_type', fs_options='-F',
            fs_label='fs_label', name='name', level='level')
        self.assertIn('device', actual_md.devices)
        self.assertEqual(1, len(self.p_scheme.fss))
        self.assertEqual('name', self.p_scheme.fss[0].device)
        self.assertEqual('mount', self.p_scheme.fss[0].mount)
        self.assertEqual('fs_type', self.p_scheme.fss[0].type)
        self.assertEqual('fs_label', self.p_scheme.fss[0].label)
        self.assertEqual('-F', self.p_scheme.fss[0].options)


class TestParted(test_base.BaseTestCase):
    def setUp(self):
        super(TestParted, self).setUp()
        self.prtd = partition.Parted('name', 'label')

    @mock.patch.object(partition.Parted, 'next_count')
    @mock.patch.object(partition.Parted, 'next_type')
    def test_next_name_none(self, nt_mock, nc_mock):
        nc_mock.return_value = 1
        nt_mock.return_value = 'extended'
        self.assertEqual(None, self.prtd.next_name())

    @mock.patch.object(partition.Parted, 'next_count')
    @mock.patch.object(partition.Parted, 'next_type')
    def test_next_name_no_separator(self, nt_mock, nc_mock):
        nc_mock.return_value = 1
        nt_mock.return_value = 'not_extended'
        expected_name = '%s%s' % (self.prtd.name, 1)
        self.assertEqual(expected_name, self.prtd.next_name())

    @mock.patch.object(partition.Parted, 'next_count')
    @mock.patch.object(partition.Parted, 'next_type')
    def test_next_name_with_separator(self, nt_mock, nc_mock):
        nc_mock.return_value = 1
        nt_mock.return_value = 'not_extended'
        self.prtd.name = '/dev/cciss/c0d0'
        expected_name = '%sp%s' % (self.prtd.name, 1)
        self.assertEqual(expected_name, self.prtd.next_name())
        self.prtd.name = '/dev/loop123'
        expected_name = '%sp%s' % (self.prtd.name, 1)
        self.assertEqual(expected_name, self.prtd.next_name())
        self.prtd.name = '/dev/nvme0n1'
        expected_name = '%sp%s' % (self.prtd.name, 1)
        self.assertEqual(expected_name, self.prtd.next_name())

    def test_next_begin_empty_partitions(self):
        self.assertEqual(1, self.prtd.next_begin())

    def test_next_begin_last_extended_partition(self):
        self.prtd.partitions.append(
            partition.Partition('name', 'count', 'device', 'begin', 'end',
                                'extended'))
        self.assertEqual('begin', self.prtd.next_begin())

    def test_next_begin_no_last_extended_partition(self):
        self.prtd.partitions.append(
            partition.Partition('name', 'count', 'device', 'begin', 'end',
                                'primary'))
        self.assertEqual('end', self.prtd.next_begin())

    def test_next_count_no_logical(self):
        self.assertEqual(1, self.prtd.next_count('primary'))

    def test_next_count_has_logical(self):
        self.prtd.partitions.append(
            partition.Partition('name', 'count', 'device', 'begin', 'end',
                                'logical'))
        self.assertEqual(6, self.prtd.next_count('logical'))

    def test_next_type_gpt(self):
        self.prtd.label = 'gpt'
        self.assertEqual('primary', self.prtd.next_type())

    def test_next_type_no_extended(self):
        self.prtd.label = 'msdos'
        self.assertEqual('primary', self.prtd.next_type())
        self.prtd.partitions.extend(
            3 * [partition.Partition('name', 'count', 'device', 'begin',
                                     'end', 'primary')])
        self.assertEqual('extended', self.prtd.next_type())

    def test_next_type_has_extended(self):
        self.prtd.label = 'msdos'
        self.prtd.partitions.append(
            partition.Partition('name', 'count', 'device', 'begin', 'end',
                                'extended'))
        self.assertEqual('logical', self.prtd.next_type())

    def test_primary(self):
        expected_partitions = [partition.Partition('name', 'count', 'device',
                                                   'begin', 'end', 'primary')]
        self.prtd.partitions.extend(expected_partitions)
        self.assertEqual(expected_partitions, self.prtd.primary)
