# Copyright 2012 OpenStack Foundation
# Copyright 2013 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.

import pytest
import allure
import operator

from tests.api.volume import base
from lib.common.utils import data_utils


class TestVolumesListJSON(base.BaseVolumeTest):
    """Test listing volumes

    NOTE: This test creates a number of 1G volumes. To run it successfully,
    ensure that the backing file for the volume group that Cinder uses
    has space for at least 3 1G volumes!
    If you are running a Devstack environment, ensure that the
    VOLUME_BACKING_FILE_SIZE is at least 4G in your localrc
    """

    VOLUME_FIELDS = ('id', 'name')

    @classmethod
    def _remove_volatile_fields(cls, fetched_list):
        """Remove fields that should not be compared.

        This method makes sure that Tempest does not compare e.g.
        the volume's "updated_at" field that may change for any reason
        internal to the operation of Cinder.
        """
        for volume in fetched_list:
            for field in ('updated_at',):
                if field in volume:
                    del volume[field]

    def _assert_volumes_in(self, fetched_list, expected_list, fields=None):
        """Check out the list.

        This function is aim at check out whether all of the volumes in
        expected_list are in fetched_list.
        """
        if fields:
            fieldsgetter = operator.itemgetter(*fields)
            expected_list = map(fieldsgetter, expected_list)
            fetched_list = [fieldsgetter(item) for item in fetched_list]

        # Hopefully the expected_list has already been cleaned.
        self._remove_volatile_fields(fetched_list)
        missing_vols = [v for v in expected_list if v not in fetched_list]
        if not missing_vols:
            return True, ""

        def str_vol(vol):
            return "%s:%s" % (vol['id'], vol['name'])

        raw_msg = "Could not find volumes %s in expected list %s; fetched %s"
        fail_msg = (raw_msg % [str_vol(v) for v in missing_vols],
                             [str_vol(v) for v in expected_list],
                             [str_vol(v) for v in fetched_list])
        return False, fail_msg

    @classmethod
    def resource_setup(cls):
        super(TestVolumesListJSON, cls).resource_setup()

        existing_volumes = cls.volumes_client.list_volumes()['volumes']
        cls.volume_id_list = [vol['id'] for vol in existing_volumes]

        # Create 3 test volumes
        cls.volume_list = []
        cls.metadata = {'Type': 'work'}
        for _ in range(3):
            volume = cls.create_volume(metadata=cls.metadata)
            volume = cls.volumes_client.show_volume(volume['id'])['volume']
            cls.volume_list.append(volume)
            cls.volume_id_list.append(volume['id'])
        cls._remove_volatile_fields(cls.volume_list)

    def _list_by_param_value_and_assert(self, params, with_detail=False):
        """list or list_details with given params and validates result"""
        if with_detail:
            fetched_vol_list = \
                self.volumes_client.list_volumes(detail=True,
                                                 params=params)['volumes']
        else:
            fetched_vol_list = self.volumes_client.list_volumes(
                params=params)['volumes']

        # Validating params of fetched volumes
        if with_detail:
            for volume in fetched_vol_list:
                for key in params:
                    msg = "Failed to list volumes %s by %s" % \
                          ('details' if with_detail else '', key)
                    if key == 'metadata':
                        if not set(params[key].items()).issubset(volume[key].items()):
                            return False, msg
                    else:
                        if params[key] != volume[key]:
                            return False, msg
        return True, ""

    @pytest.mark.smoke
    @pytest.mark.positive
    def test_volume_list(self):
        """Test getting a list of volumes

        :ref: `test_volume_list <https://docs.openstack.org/tempest/latest/tests/volume/volume.html#volume.test_volumes_list.VolumesListTestJSON.test_volume_list>`__
        """
        # Fetch all volumes
        fetched_list = self.volumes_client.list_volumes()['volumes']
        with allure.step("Verify volume list with expected list"):
            is_success, msg = self._assert_volumes_in(fetched_list, self.volume_list,
                                    fields=self.VOLUME_FIELDS)
            assert is_success, msg

    @pytest.mark.positive
    def test_volume_list_by_name(self):
        """Test getting a list of volumes filtered by volume name

        :ref: `test_volume_list_by_name <https://docs.openstack.org/tempest/latest/tests/volume/volume.html#volume.test_volumes_list.VolumesListTestJSON.test_volume_list_by_name>`__
        """
        volume = self.volume_list[data_utils.rand_int_id(0, 2)]
        params = {'name': volume['name']}
        fetched_vol = self.volumes_client.list_volumes(
            params=params)['volumes']
        with allure.step("Verify list of volume filtered by name has not duplicate name"):
            assert 1 == len(fetched_vol), str(fetched_vol)
        with allure.step("Verify list of volume filtered by name is matched expected volume name"):
            assert fetched_vol[0]['name'] == volume['name']

    @pytest.mark.positive
    def test_volume_list_with_details(self):
        """Test getting a list of detailed volumes

        :ref: `test_volume_list_with_details <https://docs.openstack.org/tempest/latest/tests/volume/volume.html#volume.test_volumes_list.VolumesListTestJSON.test_volume_list_with_details>`__
        """
        # Fetch all Volumes
        fetched_list = self.volumes_client.list_volumes(detail=True)['volumes']
        with allure.step("Verify list of detailed volumes with expected volume list"):
            is_success, msg = self._assert_volumes_in(fetched_list, self.volume_list)
            assert is_success, msg

    @pytest.mark.positive
    def test_volumes_list_details_by_status(self):
        """Test getting a list of detailed volumes filtered by status

        :ref: `test_volumes_list_details_by_status <https://docs.openstack.org/tempest/latest/tests/volume/volume.html#volume.test_volumes_list.VolumesListTestJSON.test_volumes_list_details_by_status>`__
        """
        params = {'status': 'available'}
        fetched_list = self.volumes_client.list_volumes(
            detail=True, params=params)['volumes']
        with allure.step("Verify list of detailed volume's status is 'available'"):
            for volume in fetched_list:
                assert 'available' == volume['status']
        with allure.step("Verify list of detailed volumes filtered by status with expected volume list"):
            is_success, msg = self._assert_volumes_in(fetched_list, self.volume_list)
            assert is_success, msg

    @pytest.mark.positive
    def test_volumes_list_by_bootable(self):
        """Check out volumes.

        This test function is aim at check out whether all of the volumes
        in volume_list are not a bootable volume.

        :ref: `test_volumes_list_by_bootable <https://docs.openstack.org/tempest/latest/tests/volume/volume.html#volume.test_volumes_list.VolumesListTestJSON.test_volumes_list_by_bootable>`__
        """
        params = {'bootable': 'false'}
        fetched_list = self.volumes_client.list_volumes(
            params=params)['volumes']
        with allure.step("List volume with bootable params"):
            is_success, msg = self._list_by_param_value_and_assert(params)
            assert is_success, msg
        with allure.step("Verify list of volumes filtered by bootable with expected volume list"):
            is_success, msg = self._assert_volumes_in(fetched_list, self.volume_list,
                                    fields=self.VOLUME_FIELDS)
            assert is_success, msg

    @pytest.mark.positive
    def test_volumes_list_details_by_bootable(self):
        """Test getting a list of detailed volumes filtered by bootable

        :ref: `test_volumes_list_details_by_bootable <https://docs.openstack.org/tempest/latest/tests/volume/volume.html#volume.test_volumes_list.VolumesListTestJSON.test_volumes_list_details_by_bootable>`__
        """
        params = {'bootable': 'false'}
        fetched_list = self.volumes_client.list_volumes(
            detail=True, params=params)['volumes']
        with allure.step(f"Verify list details of volume's bootable is {params['bootable']}"):
            for volume in fetched_list:
                assert 'false' == volume['bootable']
        with allure.step("Verify list details of volumes filtered by bootable with expected volume list"):
            is_success, msg = self._assert_volumes_in(fetched_list, self.volume_list)
            assert is_success, msg

    @pytest.mark.positive
    def test_volumes_list_by_availability_zone(self):
        """Test getting a list of volumes filtered by availability zone

        :ref: `test_volumes_list_by_availability_zone <https://docs.openstack.org/tempest/latest/tests/volume/volume.html#volume.test_volumes_list.VolumesListTestJSON.test_volumes_list_by_availability_zone>`__
        """
        volume = self.volume_list[data_utils.rand_int_id(0, 2)]
        zone = volume['availability_zone']
        params = {'availability_zone': zone}
        fetched_list = self.volumes_client.list_volumes(
            params=params)['volumes']
        with allure.step("List volume with availability_zone params"):
            is_success, msg = self._list_by_param_value_and_assert(params)
            assert is_success, msg
        with allure.step("Verify list of volumes filtered by availability_zone with expected volume list"):
            is_success, msg = self._assert_volumes_in(fetched_list, self.volume_list,
                                    fields=self.VOLUME_FIELDS)
            assert is_success, msg

    @pytest.mark.positive
    def test_volumes_list_details_by_availability_zone(self):
        """Test getting a list of detailed volumes by availability zone

        :ref: `test_volumes_list_details_by_availability_zone <https://docs.openstack.org/tempest/latest/tests/volume/volume.html#volume.test_volumes_list.VolumesListTestJSON.test_volumes_list_details_by_availability_zone>`__
        """
        volume = self.volume_list[data_utils.rand_int_id(0, 2)]
        zone = volume['availability_zone']
        params = {'availability_zone': zone}
        fetched_list = self.volumes_client.list_volumes(
            detail=True, params=params)['volumes']
        with allure.step(f"Validate volume list details with availability_zone is {zone}"):
            for volume in fetched_list:
                assert zone == volume['availability_zone']
        with allure.step("Verify list details of volumes filtered by availability_zone with expected volume list"):
            is_sucess, msg = self._assert_volumes_in(fetched_list, self.volume_list)
            assert is_sucess, msg

    @pytest.mark.positive
    def test_volume_list_with_param_metadata(self):
        """Test listing volumes when metadata param is given

        :ref: `test_volume_list_with_param_metadata <https://docs.openstack.org/tempest/latest/tests/volume/volume.html#volume.test_volumes_list.VolumesListTestJSON.test_volume_list_with_param_metadata>`__
        """
        params = {'metadata': self.metadata}
        with allure.step("Verify List volume with param metadata"):
            is_success, msg = self._list_by_param_value_and_assert(params)
            assert is_success, msg

    @pytest.mark.positive
    def test_volume_list_with_detail_param_metadata(self):
        """Test listing volumes details when metadata param is given

        :ref: `test_volume_list_with_detail_param_metadata <https://docs.openstack.org/tempest/latest/tests/volume/volume.html#volume.test_volumes_list.VolumesListTestJSON.test_volume_list_with_detail_param_metadata>`__
        """
        params = {'metadata': self.metadata}
        with allure.step("Verify List volume with detail param metadata"):
            is_success, msg = self._list_by_param_value_and_assert(params, with_detail=True)
            assert is_success, msg

    @pytest.mark.positive
    def test_volume_list_param_display_name_and_status(self):
        """Test listing volume when display name and status param is given

        :ref: `test_volume_list_param_display_name_and_status <https://docs.openstack.org/tempest/latest/tests/volume/volume.html#volume.test_volumes_list.VolumesListTestJSON.test_volume_list_param_display_name_and_status>`__
        """
        volume = self.volume_list[data_utils.rand_int_id(0, 2)]
        params = {'name': volume['name'],
                  'status': 'available'}
        with allure.step("Verify List volume with param display name, status"):
            is_success, msg = self._list_by_param_value_and_assert(params)
            assert is_success, msg

    @pytest.mark.positive
    def test_volume_list_with_detail_param_display_name_and_status(self):
        """Test listing volume when name and status param is given

        :ref: `test_volume_list_with_detail_param_display_name_and_status <https://docs.openstack.org/tempest/latest/tests/volume/volume.html#volume.test_volumes_list.VolumesListTestJSON.test_volume_list_with_detail_param_display_name_and_status>`__
        """
        volume = self.volume_list[data_utils.rand_int_id(0, 2)]
        params = {'name': volume['name'],
                  'status': 'available'}
        with allure.step("Verify List volume with detail param display name, status"):
            is_success, msg = self._list_by_param_value_and_assert(params, with_detail=True)
            assert is_success, msg