# Copyright 2015, 2018 IBM Corp.
#
# All Rights Reserved.
#
#    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.

"""Tests for pypowervm.tasks.partition."""

import mock
import testtools

import pypowervm.const as c
import pypowervm.entities as ent
import pypowervm.exceptions as ex
import pypowervm.tasks.partition as tpar
import pypowervm.tests.tasks.util as tju
import pypowervm.tests.test_fixtures as fx
import pypowervm.tests.test_utils.test_wrapper_abc as twrap
import pypowervm.wrappers.base_partition as bp
import pypowervm.wrappers.logical_partition as lpar
import pypowervm.wrappers.virtual_io_server as vios


LPAR_FEED_WITH_MGMT = 'lpar.txt'
VIO_FEED_WITH_MGMT = 'fake_vios_feed.txt'
LPAR_FEED_NO_MGMT = 'lpar_ibmi.txt'
VIO_FEED_NO_MGMT = 'fake_vios_feed2.txt'


def mock_vios(name, state, rmc_state, is_mgmt=False, uptime=3601):
    ret = mock.Mock()
    ret.configure_mock(name=name, state=state, rmc_state=rmc_state,
                       is_mgmt_partition=is_mgmt, uptime=uptime)
    return ret


class TestPartition(testtools.TestCase):

    def setUp(self):
        super(TestPartition, self).setUp()
        self.adpt = self.useFixture(
            fx.AdapterFx(traits=fx.RemotePVMTraits)).adpt
        self.mgmt_vio = tju.load_file(VIO_FEED_WITH_MGMT, self.adpt)
        self.mgmt_lpar = tju.load_file(LPAR_FEED_WITH_MGMT, self.adpt)
        self.nomgmt_vio = tju.load_file(VIO_FEED_NO_MGMT, self.adpt)
        self.nomgmt_lpar = tju.load_file(LPAR_FEED_NO_MGMT, self.adpt)

    def test_get_mgmt_lpar(self):
        "Happy path where the LPAR is the mgmt VM is a LPAR."
        self.adpt.read.side_effect = [self.nomgmt_vio, self.mgmt_lpar]

        mgmt_w = tpar.get_mgmt_partition(self.adpt)
        self.assertTrue(mgmt_w.is_mgmt_partition)
        self.assertEqual('089FFB20-5D19-4A8C-BB80-13650627D985', mgmt_w.uuid)
        self.assertIsInstance(mgmt_w, lpar.LPAR)
        self.assertEqual(2, self.adpt.read.call_count)

    def test_get_mgmt_vio(self):
        "Happy path where the LPAR is the mgmt VM is a VIOS."
        self.adpt.read.side_effect = [self.mgmt_vio, self.nomgmt_lpar]

        mgmt_w = tpar.get_mgmt_partition(self.adpt)
        self.assertTrue(mgmt_w.is_mgmt_partition)
        self.assertEqual('7DBBE705-E4C4-4458-8223-3EBE07015CA9', mgmt_w.uuid)
        self.assertIsInstance(mgmt_w, vios.VIOS)
        self.assertEqual(1, self.adpt.read.call_count)

    def test_get_mgmt_none(self):
        """Failure path with no mgmt VMs."""
        self.adpt.read.side_effect = [self.nomgmt_lpar, self.nomgmt_vio]

        self.assertRaises(ex.ManagementPartitionNotFoundException,
                          tpar.get_mgmt_partition, self.adpt)

    @mock.patch('pypowervm.wrappers.virtual_io_server.VIOS.search')
    @mock.patch('pypowervm.wrappers.logical_partition.LPAR.search')
    @mock.patch('pypowervm.util.my_partition_id')
    def test_get_me(self, mock_my_id, mock_lp_search, mock_vio_search):
        """Test get_this_partition()."""
        # Good path - one hit on LPAR
        mock_lp_search.return_value = [lpar.LPAR.wrap(self.mgmt_lpar)[0]]
        mock_vio_search.return_value = []
        mock_my_id.return_value = 9
        my_w = tpar.get_this_partition(self.adpt)
        self.assertEqual(9, my_w.id)
        self.assertEqual('089FFB20-5D19-4A8C-BB80-13650627D985', my_w.uuid)
        mock_lp_search.assert_called_with(self.adpt, id=9)
        mock_vio_search.assert_called_with(self.adpt, id=9)

        # Good path - one hit on VIOS
        mock_lp_search.reset_mock()
        mock_lp_search.return_value = []
        mock_vio_search.return_value = [vios.VIOS.wrap(self.mgmt_vio)[0]]
        mock_my_id.return_value = 2
        my_w = tpar.get_this_partition(self.adpt)
        self.assertEqual(2, my_w.id)
        self.assertEqual('1300C76F-9814-4A4D-B1F0-5B69352A7DEA', my_w.uuid)
        mock_lp_search.assert_not_called()
        mock_vio_search.assert_called_with(self.adpt, id=2)

        # Bad path - no hits
        mock_lp_search.return_value = []
        mock_vio_search.return_value = []
        self.assertRaises(ex.ThisPartitionNotFoundException,
                          tpar.get_this_partition, self.adpt)

    def test_has_physical_io(self):
        """test partition has physical io."""
        part_w = mock.Mock(io_config=mock.Mock(
            io_slots=[mock.Mock(description='1 Gigabit Ethernet (UTP) 4 '
                                'Port Adapter PCIE  Short')]))
        self.assertTrue(tpar.has_physical_io(part_w))

        part_w = mock.Mock(io_config=mock.Mock(
            io_slots=[mock.Mock(description='test Graphics 3.0 test')]))
        self.assertFalse(tpar.has_physical_io(part_w))

        part_w = mock.Mock(io_config=mock.Mock(
            io_slots=[mock.Mock(description='My 3D Controller test')]))
        self.assertFalse(tpar.has_physical_io(part_w))

        part_w = mock.Mock(io_config=mock.Mock(io_slots=[]))
        self.assertFalse(tpar.has_physical_io(part_w))


class TestVios(twrap.TestWrapper):
    file = 'fake_vios_feed2.txt'
    wrapper_class_to_test = vios.VIOS

    def setUp(self):
        super(TestVios, self).setUp()
        sleep_p = mock.patch('time.sleep')
        self.mock_sleep = sleep_p.start()
        self.addCleanup(sleep_p.stop)
        vioget_p = mock.patch('pypowervm.wrappers.virtual_io_server.VIOS.get')
        self.mock_vios_get = vioget_p.start()
        self.addCleanup(vioget_p.stop)

    def test_get_active_vioses(self):
        self.mock_vios_get.return_value = self.entries
        vioses = tpar.get_active_vioses(self.adpt)
        self.assertEqual(1, len(vioses))
        self.mock_vios_get.assert_called_once_with(self.adpt, xag=())
        vio = vioses[0]
        self.assertEqual(bp.LPARState.RUNNING, vio.state)
        self.assertEqual(bp.RMCState.ACTIVE, vio.rmc_state)
        self.mock_vios_get.assert_called_once_with(self.adpt, xag=())

        self.mock_vios_get.reset_mock()

        # Test with actual xag.  find_min equal to the number found - works.
        vioses = tpar.get_active_vioses(self.adpt, xag='xaglist', find_min=1)
        self.assertEqual(1, len(vioses))
        vio = vioses[0]
        self.assertEqual(bp.LPARState.RUNNING, vio.state)
        self.assertEqual(bp.RMCState.ACTIVE, vio.rmc_state)
        self.mock_vios_get.assert_called_once_with(self.adpt, xag='xaglist')

        # Violates find_min
        self.assertRaises(ex.NotEnoughActiveVioses, tpar.get_active_vioses,
                          self.adpt, find_min=2)

    def test_get_active_vioses_w_vios_wraps(self):
        mock_vios1 = mock_vios('vios1', 'running', 'active')
        mock_vios2 = mock_vios('vios2', 'running', 'inactive')
        mock_vios3 = mock_vios('mgmt', 'running', 'inactive', is_mgmt=True)
        vios_wraps = [mock_vios1, mock_vios2, mock_vios3]

        vioses = tpar.get_active_vioses(self.adpt, vios_wraps=vios_wraps)
        self.assertEqual(2, len(vioses))
        self.mock_vios_get.assert_not_called()

        # The first should be the mgmt partition
        vio = vioses[0]
        self.assertEqual(bp.LPARState.RUNNING, vio.state)
        self.assertEqual(bp.RMCState.INACTIVE, vio.rmc_state)

        # The second should be the active one
        vio = vioses[1]
        self.assertEqual(bp.LPARState.RUNNING, vio.state)
        self.assertEqual(bp.RMCState.ACTIVE, vio.rmc_state)
        self.mock_vios_get.assert_not_called()

    def test_get_physical_wwpns(self):
        self.mock_vios_get.return_value = self.entries
        expected = {'21000024FF649104'}
        result = set(tpar.get_physical_wwpns(self.adpt))
        self.assertSetEqual(expected, result)
        self.mock_vios_get.assert_called_once_with(
            self.adpt, xag=[c.XAG.VIO_STOR])
        # Test caching
        self.mock_vios_get.reset_mock()
        result = set(tpar.get_physical_wwpns(self.adpt, force_refresh=False))
        self.assertSetEqual(expected, result)
        self.mock_vios_get.assert_not_called()
        # Test force_refresh
        result = set(tpar.get_physical_wwpns(self.adpt, force_refresh=True))
        self.assertSetEqual(expected, result)
        self.mock_vios_get.assert_called_once_with(
            self.adpt, xag=[c.XAG.VIO_STOR])

    @mock.patch('pypowervm.tasks.partition.get_active_vioses')
    @mock.patch('pypowervm.utils.transaction.FeedTask')
    def test_build_active_vio_feed_task(self, mock_feed_task,
                                        mock_get_active_vioses):
        mock_get_active_vioses.return_value = ['vios1', 'vios2']
        mock_feed_task.return_value = 'mock_feed'
        self.assertEqual('mock_feed', tpar.build_active_vio_feed_task('adpt'))
        mock_get_active_vioses.assert_called_once_with(
            'adpt', xag=(c.XAG.VIO_STOR, c.XAG.VIO_SMAP, c.XAG.VIO_FMAP),
            find_min=1)

    @mock.patch('pypowervm.tasks.partition.get_active_vioses')
    def test_build_tx_feed_task_w_empty_feed(self, mock_get_active_vioses):
        mock_get_active_vioses.return_value = []
        self.assertRaises(
            ex.FeedTaskEmptyFeed, tpar.build_active_vio_feed_task,
            mock.MagicMock())

    def _mk_mock_vioses(self):
        # No
        mock_vios1 = mock_vios('vios1', bp.LPARState.NOT_ACTIVATED,
                               bp.RMCState.INACTIVE)
        # No
        mock_vios2 = mock_vios('vios2', bp.LPARState.RUNNING,
                               bp.RMCState.BUSY)
        # Yes
        mock_vios3 = mock_vios('vios3', bp.LPARState.RUNNING,
                               bp.RMCState.UNKNOWN)
        # No
        mock_vios4 = mock_vios('vios4', bp.LPARState.UNKNOWN,
                               bp.RMCState.ACTIVE)
        # No
        mock_vios5 = mock_vios('vios5', bp.LPARState.RUNNING,
                               bp.RMCState.ACTIVE)
        # Yes
        mock_vios6 = mock_vios('vios6', bp.LPARState.RUNNING,
                               bp.RMCState.INACTIVE)

        # No
        mock_vios7 = mock_vios('vios7', bp.LPARState.RUNNING,
                               bp.RMCState.INACTIVE, is_mgmt=True)

        return [mock_vios1, mock_vios2, mock_vios3, mock_vios4, mock_vios5,
                mock_vios6, mock_vios7]

    @mock.patch('pypowervm.tasks.partition.LOG.warning')
    def test_timeout_short(self, mock_warn):
        """Short timeout because relevant VIOSes have been up a while."""

        self.mock_vios_get.return_value = self._mk_mock_vioses()

        tpar.validate_vios_ready('adap')
        # We slept 120s, (24 x 5s) because all VIOSes have been up >1h
        self.assertEqual(24, self.mock_sleep.call_count)
        self.mock_sleep.assert_called_with(5)
        # We wound up with rmc_down_vioses
        mock_warn.assert_called_once_with(mock.ANY, {'time': 120,
                                                     'vioses': 'vios3, vios6'})
        # We didn't raise - because some VIOSes were okay.

    @mock.patch('pypowervm.tasks.partition.LOG.warning')
    def test_rmc_down_vioses(self, mock_warn):
        """Time out waiting for up/inactive partitions, but succeed."""

        vioses = self._mk_mock_vioses()
        # This one booted "recently"
        vioses[5].uptime = 3559
        self.mock_vios_get.return_value = vioses

        tpar.validate_vios_ready('adap')
        # We slept 600s, (120 x 5s) because one VIOS booted "recently"
        self.assertEqual(120, self.mock_sleep.call_count)
        self.mock_sleep.assert_called_with(5)
        # We wound up with rmc_down_vioses
        mock_warn.assert_called_once_with(mock.ANY, {'time': 600,
                                                     'vioses': 'vios3, vios6'})
        # We didn't raise - because some VIOSes were okay.

    @mock.patch('pypowervm.tasks.partition.LOG.warning')
    def test_no_vioses(self, mock_warn):
        """In the (highly unusual) case of no VIOSes, no warning, but raise."""
        self.mock_vios_get.return_value = []
        self.assertRaises(ex.ViosNotAvailable, tpar.validate_vios_ready, 'adp')
        mock_warn.assert_not_called()

    @mock.patch('pypowervm.tasks.partition.LOG.warning')
    def test_max_wait_on_exception(self, mock_warn):
        """VIOS.get raises repeatedly until max_wait_time is exceeded."""
        self.mock_vios_get.side_effect = ValueError('foo')
        self.assertRaises(ex.ViosNotAvailable, tpar.validate_vios_ready, 'adp',
                          10)
        self.assertEqual(mock_warn.call_count, 3)

    @mock.patch('pypowervm.tasks.partition.LOG.warning')
    def test_exception_and_good_path(self, mock_warn):
        """VIOS.get raises, then succeeds with some halfsies, then succeeds."""
        vios1_good = mock_vios('vios1', bp.LPARState.RUNNING,
                               bp.RMCState.BUSY)
        vios2_bad = mock_vios('vios2', bp.LPARState.RUNNING,
                              bp.RMCState.UNKNOWN)
        vios2_good = mock_vios('vios2', bp.LPARState.RUNNING,
                               bp.RMCState.ACTIVE)
        self.mock_vios_get.side_effect = (ValueError('foo'),
                                          [vios1_good, vios2_bad],
                                          [vios1_good, vios2_good])
        tpar.validate_vios_ready('adap')
        self.assertEqual(3, self.mock_vios_get.call_count)
        self.assertEqual(2, self.mock_sleep.call_count)
        mock_warn.assert_called_once_with(mock.ANY)

    @mock.patch('pypowervm.tasks.partition.get_mgmt_partition')
    @mock.patch('pypowervm.wrappers.logical_partition.LPAR.get')
    @mock.patch('pypowervm.wrappers.virtual_io_server.VIOS.get')
    def test_get_partitions(self, mock_vio_get, mock_lpar_get, mock_mgmt_get):
        adpt = mock.Mock()

        # Test with the MGMT as a VIOS
        mgmt = mock.Mock(uuid='1')
        vioses = [mock.Mock(uuid='2'), mgmt]
        lpars = [mock.Mock(uuid='3'), mock.Mock(uuid='4')]

        mock_mgmt_get.return_value = mgmt
        mock_vio_get.return_value = vioses
        mock_lpar_get.return_value = lpars

        # Basic case
        self.assertEqual(vioses + lpars, tpar.get_partitions(adpt))

        # Different permutations
        self.assertEqual(lpars + [mgmt], tpar.get_partitions(
            adpt, vioses=False, mgmt=True))
        self.assertEqual(vioses, tpar.get_partitions(
            adpt, lpars=False, mgmt=True))

        # Now test with the MGMT as a LPAR
        vioses = [mock.Mock(uuid='2')]
        lpars = [mock.Mock(uuid='3'), mock.Mock(uuid='4'), mgmt]

        mock_vio_get.return_value = vioses
        mock_lpar_get.return_value = lpars

        # Basic case
        self.assertEqual(vioses + lpars, tpar.get_partitions(adpt))

        # Different permutations
        self.assertEqual(lpars, tpar.get_partitions(
            adpt, vioses=False, mgmt=True))
        self.assertEqual(vioses + [mgmt], tpar.get_partitions(
            adpt, lpars=False, mgmt=True))

    @mock.patch('pypowervm.wrappers.job.Job.run_job')
    def test_clone_uuid(self, mock_run_job):

        mock_resp = mock.MagicMock()
        mock_resp.entry = ent.Entry(
            {}, ent.Element('Dummy', self.adpt), self.adpt)
        self.adpt.read.side_effect = [mock_resp]
        mock_run_job.side_effect = tju.get_parm_checker(
            self, '1234', [('targetLparName', 'abc')])
        tpar.clone_uuid(self.adpt, '1234', 'abc')
        self.adpt.read.assert_called_once_with('LogicalPartition',
                                               root_id='1234',
                                               suffix_type='do',
                                               suffix_parm='CloneUUID')