#!/usr/bin/python

#

# CDDL HEADER START

#

# The contents of this file are subject to the terms of the

# Common Development and Distribution License (the "License").

# You may not use this file except in compliance with the License.

#

# You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE

# or http://www.opensolaris.org/os/licensing.

# See the License for the specific language governing permissions

# and limitations under the License.

#

# When distributing Covered Code, include this CDDL HEADER in each

# file and include the License file at usr/src/OPENSOLARIS.LICENSE.

# If applicable, add the following below this CDDL HEADER, with the

# fields enclosed by brackets "[]" replaced with your own identifying

# information: Portions Copyright [yyyy] [name of copyright owner]

#

# CDDL HEADER END

#

#

# Copyright (c) 2011, Oracle and/or its affiliates. All rights reserved.

#

""" test_shadow_list.py - collection of unittests for testing target validation

through shadow list manipulation

"""

import copy

import platform

import unittest

from nose.plugins.skip import SkipTest

import osol_install.errsvc as errsvc

from osol_install.liberrsvc import ES_DATA_EXCEPTION

from solaris_install import Popen

from solaris_install.target import Target

from solaris_install.target.physical import Disk, DiskGeometry, DiskProp, \

    Slice, Partition

from solaris_install.target.libadm.const import MAX_EXT_PARTS, V_NUMPAR

from solaris_install.target.logical import BE, Logical, Vdev, Zpool

from solaris_install.target.shadow.physical import LOGICAL_ADJUSTMENT, \

    ShadowPhysical

from solaris_install.target.shadow.logical import ShadowLogical

from solaris_install.target.shadow.zpool import ShadowZpool

from solaris_install.target.size import Size

from solaris_install.target.vdevs import _get_vdev_mapping

GBSECTOR = long(1024 * 1024 * 1024 / 512)  # 1 GB of sectors

BLOCKSIZE = 512

CYLSIZE = 16065

class TestZpool(unittest.TestCase):

    """ test case to test manipulation of Zpool objects via the API and shadow

    list interfaces

    """

    def setUp(self):

        # create DOC objects

        self.target = Target("target")

        self.logical = Logical("logical")

        self.target.insert_children(self.logical)

        # reset the errsvc

        errsvc.clear_error_list()

    def tearDown(self):

        self.target.delete_children()

        self.target.delete()

        # reset the errsvc

        errsvc.clear_error_list()

    def test_add_simple_zpool(self):

        # create a DOC tree

        self.logical.add_zpool("test_zpool")

        self.assertFalse(errsvc._ERRORS)

    def test_add_duplicate_zpool_names(self):

        self.logical.add_zpool("test_zpool")

        self.logical.add_zpool("test_zpool")

        # verify there is only one error in the errsvc list and that it is the

        # proper error

        self.assertEqual(len(errsvc._ERRORS), 1)

        error = errsvc._ERRORS[0]

        self.assertTrue(isinstance(error.error_data[ES_DATA_EXCEPTION],

                                   ShadowZpool.DuplicateZpoolNameError))

    def test_add_two_root_pools(self):

        self.logical.add_zpool("test_zpool", is_root=True)

        self.logical.add_zpool("test_zpool2", is_root=True)

        # verify there is only one error in the errsvc list and that it is the

        # proper error

        self.assertEqual(len(errsvc._ERRORS), 1)

        error = errsvc._ERRORS[0]

        self.assertTrue(isinstance(error.error_data[ES_DATA_EXCEPTION],

                                   ShadowZpool.TooManyRootPoolsError))

    def test_add_overlapping_mountpoints(self):

        self.logical.add_zpool("test_zpool", mountpoint="/foo")

        self.logical.add_zpool("test_zpool2", mountpoint="/foo/bar")

        # verify there is only one error in the errsvc list and that it is the

        # proper error

        self.assertEqual(len(errsvc._ERRORS), 1)

        error = errsvc._ERRORS[0]

        self.assertTrue(isinstance(error.error_data[ES_DATA_EXCEPTION],

                                   ShadowZpool.DuplicateMountpointError))

    def test_delete_zpool(self):

        zpool1 = self.logical.add_zpool("test_zpool")

        zpool2 = self.logical.add_zpool("test_zpool2")

        # verify there are 2 entries in the _children list

        self.assertEqual(len(self.logical._children), 2)

        # delete zpool2

        zpool2.delete()

        # verify there is only one entry in the _children list and the proper

        # zpool was removed

        self.assertEqual(len(self.logical._children), 1)

        self.assertEqual(self.logical._children[0].name, "test_zpool")

        # delete zpool1

        zpool1.delete()

        # verify there are no entries in the _children list

        self.assertFalse(self.logical._children)

    def test_delete_zpool_twice(self):

        zpool = self.logical.add_zpool("test_zpool")

        self.logical.delete_zpool(zpool)

        self.assertFalse(self.logical._children)

        # try to delete the zpool again.

        self.logical.delete_zpool(zpool)

        self.assertFalse(self.logical._children)

    def test_delete_zpool_from_logical(self):

        zpool1 = self.logical.add_zpool("test_zpool")

        zpool2 = self.logical.add_zpool("test_zpool2")

        # verify there are 2 entries in the _children list

        self.assertEqual(len(self.logical._children), 2)

        # delete zpool2

        self.logical.delete_zpool(zpool2)

        # verify there is only one entry in the _children list and the proper

        # zpool was removed

        self.assertEqual(len(self.logical._children), 1)

        self.assertEqual(self.logical._children[0].name, "test_zpool")

    def test_add_vdev(self):

        # create a zpool and a vdev

        zpool = self.logical.add_zpool("test_zpool")

        zpool.add_vdev(label="test_vdev", redundancy="mirror")

        self.assertTrue(zpool.get_descendants(class_type=Vdev))

    def test_delete_vdev(self):

        # create a zpool and two vdevs

        zpool = self.logical.add_zpool("test_zpool")

        vdev1 = zpool.add_vdev(label="test_vdev1", redundancy="mirror")

        vdev2 = zpool.add_vdev(label="test_vdev2", redundancy="raidz")

        # verify there are two vdevs

        self.assertEqual(len(zpool.get_descendants(class_type=Vdev)), 2)

        # delete vdev2

        zpool.delete_vdev(vdev2)

        # verify there is only one vdev

        self.assertEqual(len(zpool.get_descendants(class_type=Vdev)), 1)

        # delete vdev1

        zpool.delete_vdev(vdev1)

        # verify there are no vdevs

        self.assertFalse(zpool.get_descendants(class_type=Vdev))

class TestZFSFilesystem(unittest.TestCase):

    """ test case to test the manipulation of ZFS filesystems via the API and

    shadow list interface.

    """

    def setUp(self):

        self.zpool = Zpool("zpool")

        # reset the errsvc

        errsvc.clear_error_list()

    def tearDown(self):

        self.zpool.delete_children()

        # reset the errsvc

        errsvc.clear_error_list()

    def test_add_simple_filesystem(self):

        self.zpool.add_filesystem("test_filesystem")

        self.assertFalse(errsvc._ERRORS)

    def test_add_simple_filesystem_with_mountpoint(self):

        self.zpool.add_filesystem("test_filesystem", "/mountpoint")

        self.assertFalse(errsvc._ERRORS)

    def test_add_duplicate_filesystem_names(self):

        self.zpool.add_filesystem("test_filesystem")

        self.assertFalse(errsvc._ERRORS)

        self.zpool.add_filesystem("test_filesystem")

        # verify there is only one error in the errsvc list and that it is the

        # proper error

        self.assertEqual(len(errsvc._ERRORS), 1)

        error = errsvc._ERRORS[0]

        self.assertTrue(isinstance(error.error_data[ES_DATA_EXCEPTION],

            ShadowLogical.DuplicateDatasetNameError))

    def test_add_duplicate_mountpoints(self):

        self.zpool.add_filesystem("test_filesystem", "/a")

        self.assertFalse(errsvc._ERRORS)

        self.zpool.add_filesystem("test_filesystem_2", "/a")

        # verify there is only one error in the errsvc list and that it is the

        # proper error

        self.assertEqual(len(errsvc._ERRORS), 1)

        error = errsvc._ERRORS[0]

        self.assertTrue(isinstance(error.error_data[ES_DATA_EXCEPTION],

            ShadowLogical.DuplicateMountpointError))

    def test_delete_filesystem(self):

        fs1 = self.zpool.add_filesystem("test_filesystem1")

        fs2 = self.zpool.add_filesystem("test_filesystem2")

        # verify there are 2 entries in _children

        self.assertEqual(len(self.zpool._children), 2)

        # delete fs2

        fs2.delete()

        # verify there is only one entry in the _children list and the proper

        # fs was removed

        self.assertEqual(len(self.zpool._children), 1)

        self.assertEqual(self.zpool._children[0].name, "test_filesystem1")

        # delete fs1

        fs1.delete()

        # verify there are no entries in _children

        self.assertFalse(self.zpool._children)

    def test_delete_filesystem_from_zpool(self):

        fs1 = self.zpool.add_filesystem("test_filesystem1")

        fs2 = self.zpool.add_filesystem("test_filesystem2")

        # verify there are 2 entries in the _children list

        self.assertEqual(len(self.zpool._children), 2)

        # delete zpool2

        self.zpool.delete_filesystem(fs2)

        # verify there is only one entry in the _children list and the proper

        # zpool was removed

        self.assertEqual(len(self.zpool._children), 1)

        self.assertEqual(self.zpool._children[0].name, "test_filesystem1")

        # delete zpool2 again

        self.zpool.delete_filesystem(fs2)

        # verify there is only one entry in the _children list and the proper

        # zpool was removed

        self.assertEqual(len(self.zpool._children), 1)

        self.assertEqual(self.zpool._children[0].name, "test_filesystem1")

class TestZvol(unittest.TestCase):

    """ test case to test the manipulation of Zvols via the API and

    shadow list interface.

    """

    def setUp(self):

        self.target = Target("target")

        # create a disk object

        self.disk = Disk("test disk")

        self.disk.in_zpool = "zpool"

        self.disk.ctd = "c12345t0d0"

        self.disk.geometry = DiskGeometry(BLOCKSIZE, CYLSIZE)

        # 250GB disk

        self.disk.disk_prop = DiskProp()

        self.disk.disk_prop.dev_size = Size(

            str(GBSECTOR * 250) + Size.sector_units)

        self.disk.disk_prop.blocksize = BLOCKSIZE

        self.logical = Logical("logical")

        self.zpool = Zpool("zpool")

        self.logical.insert_children(self.zpool)

        self.target.insert_children([self.disk, self.logical])

        # reset the errsvc

        errsvc.clear_error_list()

    def tearDown(self):

        self.logical.delete_children()

        self.logical.delete()

        # reset the errsvc

        errsvc.clear_error_list()

    def test_add_simple_zvol(self):

        self.zpool.add_zvol("test_zvol", "10")

        self.assertFalse(errsvc._ERRORS)

    def test_add_duplicate_zvol_names(self):

        self.zpool.add_zvol("test_zvol", "10")

        self.assertFalse(errsvc._ERRORS)

        self.zpool.add_zvol("test_zvol", "10")

        # verify there is only one error in the errsvc list and that it is the

        # proper error

        self.assertEqual(len(errsvc._ERRORS), 1)

        error = errsvc._ERRORS[0]

        self.assertTrue(isinstance(error.error_data[ES_DATA_EXCEPTION],

            ShadowLogical.DuplicateDatasetNameError))

    def test_delete_zvol(self):

        zvol1 = self.zpool.add_zvol("zvol1", "10")

        zvol2 = self.zpool.add_zvol("zvol2", "10")

        # verify there are 2 entries in the _children list

        self.assertEqual(len(self.zpool._children), 2)

        # delete zvol2

        zvol2.delete()

        # verify there is only one entry in the _children list and the proper

        # zvol was removed

        self.assertEqual(len(self.zpool._children), 1)

        self.assertEqual(self.zpool._children[0].name, "zvol1")

    def test_delete_zvol_from_zpool(self):

        zvol1 = self.zpool.add_zvol("zvol1", "10")

        zvol2 = self.zpool.add_zvol("zvol2", "10")

        # verify there are 2 entries in the _children list

        self.assertEqual(len(self.zpool._children), 2)

        # delete zvol2

        self.zpool.delete_zvol(zvol2)

        # verify there is only one entry in the _children list and the proper

        # zvol was removed

        self.assertEqual(len(self.zpool._children), 1)

        self.assertEqual(self.zpool._children[0].name, "zvol1")

    def test_resize_zvol(self):

        zvol = self.zpool.add_zvol("zvol1", "10")

        # verify the size is 10gb

        self.assertEqual(self.zpool._children[0].size, "10g")

        # resize it to 20gb

        zvol.resize("20", Size.gb_units)

        # verify the size is 20gb

        self.assertEqual(self.zpool._children[0].size, "20g")

    def test_swap_zvol_with_noswap_set(self):

        self.logical.noswap = True

        zvol = self.zpool.add_zvol("swap_zvol", "10", Size.gb_units, "swap")

        # verify there is only one error in the errsvc list and that it is the

        # proper error

        self.assertEqual(len(errsvc._ERRORS), 1)

        error = errsvc._ERRORS[0]

        self.assertTrue(isinstance(error.error_data[ES_DATA_EXCEPTION],

            ShadowLogical.NoswapMismatchError))

    def test_dump_zvol_with_nodump_set(self):

        self.logical.nodump = True

        zvol = self.zpool.add_zvol("dump_zvol", "10", Size.gb_units, "dump")

        # verify there is only one error in the errsvc list and that it is the

        # proper error

        self.assertEqual(len(errsvc._ERRORS), 1)

        error = errsvc._ERRORS[0]

        self.assertTrue(isinstance(error.error_data[ES_DATA_EXCEPTION],

            ShadowLogical.NodumpMismatchError))

class TestPartition(unittest.TestCase):

    def setUp(self):

        self.disk = Disk("test disk")

        self.disk.ctd = "c12345t0d0"

        self.disk.geometry = DiskGeometry(BLOCKSIZE, CYLSIZE)

        self.disk.label = "VTOC"

        # 100GB disk

        self.disk.disk_prop = DiskProp()

        self.disk.disk_prop.dev_size = Size(

            str(GBSECTOR * 100) + Size.sector_units)

        self.disk.disk_prop.blocksize = BLOCKSIZE

        # reset the errsvc

        errsvc.clear_error_list()

    def tearDown(self):

        self.disk.delete_children()

        self.disk.delete()

        # reset the errsvc

        errsvc.clear_error_list()

    def test_add_single_partition(self):

        self.disk.add_partition(1, 0, 1, Size.gb_units)

        self.assertFalse(errsvc._ERRORS)

    def test_duplicate_partition(self):

        # add 2 1GB partition with the same index but different starting sector

        self.disk.add_partition(1, 0, 1, Size.gb_units)

        # account for the first silce moving from a start_sector of 0 to

        # CYLSIZE

        self.disk.add_partition(1, CYLSIZE + GBSECTOR + 1, 1,

                                Size.gb_units)

        # verify there is only one error in the errsvc list and that it is the

        # proper error

        self.assertEqual(len(errsvc._ERRORS), 1)

        error = errsvc._ERRORS[0]

        self.assertTrue(isinstance(error.error_data[ES_DATA_EXCEPTION],

            ShadowPhysical.DuplicatePartitionNameError))

    def test_overlapping_starting_sector(self):

        # add a 10 GB disk

        self.disk.add_partition(0, CYLSIZE, 10, Size.gb_units)

        # add an 11 GB disk

        self.disk.add_partition(1, CYLSIZE, 11, Size.gb_units)

        # verify there is only one error in the errsvc list and that it is the

        # proper error

        self.assertEqual(len(errsvc._ERRORS), 1)

        error = errsvc._ERRORS[0]

        self.assertTrue(isinstance(error.error_data[ES_DATA_EXCEPTION],

                                   ShadowPhysical.OverlappingPartitionError))

    def test_overlapping_ending_sector(self):

        # add a 9 GB disk on the second cylinder boundary

        self.disk.add_partition(0, CYLSIZE * 2, 9, Size.gb_units)

        # add a 10 GB disk on the first cylinder boundary

        self.disk.add_partition(1, CYLSIZE, 10, Size.gb_units)

        # verify there is only one error in the errsvc list and that it is the

        # proper error

        self.assertEqual(len(errsvc._ERRORS), 1)

        error = errsvc._ERRORS[0]

        self.assertTrue(isinstance(error.error_data[ES_DATA_EXCEPTION],

                                   ShadowPhysical.OverlappingPartitionError))

    def test_add_two_extended_partitions(self):

        # add 2 1GB partitions

        self.disk.add_partition(1, 0, 1, Size.gb_units,

            partition_type=Partition.name_to_num("WIN95 Extended(LBA)"))

        # account for the first silce moving from a start_sector of 0 to

        # CYLSIZE

        self.disk.add_partition(2, CYLSIZE + GBSECTOR + 1, 1, Size.gb_units,

            partition_type=Partition.name_to_num("WIN95 Extended(LBA)"))

        # verify there is only one error in the errsvc list and that it is the

        # proper error

        self.assertEqual(len(errsvc._ERRORS), 1)

        error = errsvc._ERRORS[0]

        self.assertTrue(isinstance(error.error_data[ES_DATA_EXCEPTION],

            ShadowPhysical.TooManyExtPartitionsError))

    def test_delete_partition(self):

        p1 = self.disk.add_partition(1, 0, 1, Size.gb_units)

        p2 = self.disk.add_partition(2, GBSECTOR + 1, 1, Size.gb_units)

        # verify there are 2 entries in the _children

        self.assertEqual(len(self.disk._children), 2)

        # delete partition 2

        p2.delete()

        # verify there is only one entry in the _children and the proper

        # partition was removed.

        self.assertEqual(len(self.disk._children), 1)

        self.assertEqual(self.disk._children[0].name, 1)

    def test_delete_partition_with_slice_children(self):

        # add a 10gb partition

        p1 = self.disk.add_partition(1, 0, 10, Size.gb_units)

        # add two slices

        p1.add_slice(0, CYLSIZE, 1, Size.gb_units)

        p1.add_slice(1, GBSECTOR + CYLSIZE + 1, 1, Size.gb_units)

        # delete the partition

        p1.delete()

        # verify there are no errors

        self.assertFalse(errsvc._ERRORS)

        # verify the _children is empty

        self.assertFalse(self.disk._children)

    def test_delete_partition_from_disk(self):

        p1 = self.disk.add_partition(1, 0, 1, Size.gb_units)

        p2 = self.disk.add_partition(2, GBSECTOR + 1, 1, Size.gb_units)

        # verify there are 2 entries in the _children

        self.assertEqual(len(self.disk._children), 2)

        # delete partition 2

        self.disk.delete_partition(p2)