#
#   Licensed under the Apache License, Version 2.0 (the "License"); you may
#   not use this file except in compliance with the License. You may obtain
#   a copy of the License at
#
#        http://www.apache.org/licenses/LICENSE-2.0
#
#   Unless required by applicable law or agreed to in writing, software
#   distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#   WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#   License for the specific language governing permissions and limitations
#   under the License.

from unittest import mock

from openstack.block_storage.v2 import snapshot as _snapshot
from openstack.block_storage.v3 import volume as _volume
from openstack import exceptions as sdk_exceptions
from openstack.test import fakes as sdk_fakes
from osc_lib.cli import format_columns
from osc_lib import exceptions

from openstackclient.tests.unit.identity.v3 import fakes as project_fakes
from openstackclient.tests.unit import utils as test_utils
from openstackclient.tests.unit.volume.v2 import fakes as volume_fakes
from openstackclient.volume.v2 import volume_snapshot


class TestVolumeSnapshot(volume_fakes.TestVolume):
    def setUp(self):
        super().setUp()

        self.snapshots_mock = self.volume_client.volume_snapshots
        self.snapshots_mock.reset_mock()
        self.volumes_mock = self.volume_client.volumes
        self.volumes_mock.reset_mock()
        self.project_mock = self.identity_client.projects
        self.project_mock.reset_mock()


class TestVolumeSnapshotCreate(volume_fakes.TestVolume):
    columns = (
        'created_at',
        'description',
        'id',
        'name',
        'properties',
        'size',
        'status',
        'volume_id',
    )

    def setUp(self):
        super().setUp()

        self.volume = sdk_fakes.generate_fake_resource(_volume.Volume)
        self.volume_sdk_client.find_volume.return_value = self.volume
        self.snapshot = sdk_fakes.generate_fake_resource(
            _snapshot.Snapshot, volume_id=self.volume.id
        )
        self.volume_sdk_client.create_snapshot.return_value = self.snapshot
        self.volume_sdk_client.manage_snapshot.return_value = self.snapshot

        self.data = (
            self.snapshot.created_at,
            self.snapshot.description,
            self.snapshot.id,
            self.snapshot.name,
            format_columns.DictColumn(self.snapshot.metadata),
            self.snapshot.size,
            self.snapshot.status,
            self.snapshot.volume_id,
        )

        self.cmd = volume_snapshot.CreateVolumeSnapshot(self.app, None)

    def test_snapshot_create(self):
        arglist = [
            "--volume",
            self.snapshot.volume_id,
            "--description",
            self.snapshot.description,
            "--force",
            '--property',
            'Alpha=a',
            '--property',
            'Beta=b',
            self.snapshot.name,
        ]
        verifylist = [
            ("volume", self.snapshot.volume_id),
            ("description", self.snapshot.description),
            ("force", True),
            ('properties', {'Alpha': 'a', 'Beta': 'b'}),
            ("snapshot_name", self.snapshot.name),
        ]
        parsed_args = self.check_parser(self.cmd, arglist, verifylist)

        columns, data = self.cmd.take_action(parsed_args)

        self.assertEqual(self.columns, columns)
        self.assertEqual(self.data, data)
        self.volume_sdk_client.find_volume.assert_called_once_with(
            self.snapshot.volume_id, ignore_missing=False
        )
        self.volume_sdk_client.create_snapshot.assert_called_with(
            volume_id=self.snapshot.volume_id,
            force=True,
            name=self.snapshot.name,
            description=self.snapshot.description,
            metadata={'Alpha': 'a', 'Beta': 'b'},
        )

    def test_snapshot_create_without_name(self):
        arglist = [
            "--volume",
            self.snapshot.volume_id,
        ]
        verifylist = [
            ("volume", self.snapshot.volume_id),
        ]
        self.assertRaises(
            test_utils.ParserException,
            self.check_parser,
            self.cmd,
            arglist,
            verifylist,
        )

    def test_snapshot_create_without_volume(self):
        arglist = [
            "--description",
            self.snapshot.description,
            "--force",
            self.snapshot.name,
        ]
        verifylist = [
            ("description", self.snapshot.description),
            ("force", True),
            ("snapshot_name", self.snapshot.name),
        ]
        parsed_args = self.check_parser(self.cmd, arglist, verifylist)

        columns, data = self.cmd.take_action(parsed_args)

        self.assertEqual(self.columns, columns)
        self.assertEqual(self.data, data)
        self.volume_sdk_client.find_volume.assert_called_once_with(
            self.snapshot.name, ignore_missing=False
        )
        self.volume_sdk_client.create_snapshot.assert_called_once_with(
            volume_id=self.snapshot.volume_id,
            force=True,
            name=self.snapshot.name,
            description=self.snapshot.description,
            metadata=None,
        )

    def test_snapshot_create_with_remote_source(self):
        arglist = [
            '--remote-source',
            'source-name=test_source_name',
            '--remote-source',
            'source-id=test_source_id',
            '--volume',
            self.snapshot.volume_id,
            self.snapshot.name,
        ]
        ref_dict = {
            'source-name': 'test_source_name',
            'source-id': 'test_source_id',
        }
        verifylist = [
            ('remote_source', ref_dict),
            ('volume', self.snapshot.volume_id),
            ("snapshot_name", self.snapshot.name),
        ]
        parsed_args = self.check_parser(self.cmd, arglist, verifylist)

        columns, data = self.cmd.take_action(parsed_args)

        self.assertEqual(self.columns, columns)
        self.assertEqual(self.data, data)
        self.volume_sdk_client.find_volume.assert_called_once_with(
            self.snapshot.volume_id, ignore_missing=False
        )
        self.volume_sdk_client.manage_snapshot.assert_called_with(
            volume_id=self.snapshot.volume_id,
            ref=ref_dict,
            name=self.snapshot.name,
            description=None,
            metadata=None,
        )
        self.volume_sdk_client.create_snapshot.assert_not_called()


class TestVolumeSnapshotDelete(volume_fakes.TestVolume):
    def setUp(self):
        super().setUp()

        self.snapshots = list(
            sdk_fakes.generate_fake_resources(_snapshot.Snapshot)
        )
        self.volume_sdk_client.find_snapshot.side_effect = self.snapshots
        self.volume_sdk_client.delete_snapshot.return_value = None

        self.cmd = volume_snapshot.DeleteVolumeSnapshot(self.app, None)

    def test_snapshot_delete(self):
        arglist = [self.snapshots[0].id]
        verifylist = [("snapshots", [self.snapshots[0].id])]
        parsed_args = self.check_parser(self.cmd, arglist, verifylist)

        result = self.cmd.take_action(parsed_args)
        self.assertIsNone(result)

        self.volume_sdk_client.find_snapshot.assert_called_once_with(
            self.snapshots[0].id, ignore_missing=False
        )
        self.volume_sdk_client.delete_snapshot.assert_called_once_with(
            self.snapshots[0].id, force=False
        )

    def test_snapshot_delete_with_force(self):
        arglist = ['--force', self.snapshots[0].id]
        verifylist = [('force', True), ("snapshots", [self.snapshots[0].id])]
        parsed_args = self.check_parser(self.cmd, arglist, verifylist)

        result = self.cmd.take_action(parsed_args)
        self.assertIsNone(result)

        self.volume_sdk_client.find_snapshot.assert_called_once_with(
            self.snapshots[0].id, ignore_missing=False
        )
        self.volume_sdk_client.delete_snapshot.assert_called_once_with(
            self.snapshots[0].id, force=True
        )

    def test_delete_multiple_snapshots(self):
        arglist = []
        for s in self.snapshots:
            arglist.append(s.id)
        verifylist = [
            ('snapshots', arglist),
        ]
        parsed_args = self.check_parser(self.cmd, arglist, verifylist)

        result = self.cmd.take_action(parsed_args)
        self.assertIsNone(result)

        self.volume_sdk_client.find_snapshot.assert_has_calls(
            [mock.call(x.id, ignore_missing=False) for x in self.snapshots]
        )
        self.volume_sdk_client.delete_snapshot.assert_has_calls(
            [mock.call(x.id, force=False) for x in self.snapshots]
        )

    def test_delete_multiple_snapshots_with_exception(self):
        self.volume_sdk_client.find_snapshot.side_effect = [
            self.snapshots[0],
            sdk_exceptions.NotFoundException(),
        ]

        arglist = [
            self.snapshots[0].id,
            'unexist_snapshot',
        ]
        verifylist = [
            ('snapshots', arglist),
        ]

        parsed_args = self.check_parser(self.cmd, arglist, verifylist)

        exc = self.assertRaises(
            exceptions.CommandError,
            self.cmd.take_action,
            parsed_args,
        )
        self.assertEqual('1 of 2 snapshots failed to delete.', str(exc))

        self.volume_sdk_client.find_snapshot.assert_has_calls(
            [
                mock.call(self.snapshots[0].id, ignore_missing=False),
                mock.call('unexist_snapshot', ignore_missing=False),
            ]
        )
        self.volume_sdk_client.delete_snapshot.assert_has_calls(
            [
                mock.call(self.snapshots[0].id, force=False),
            ]
        )


class TestVolumeSnapshotList(TestVolumeSnapshot):
    volume = volume_fakes.create_one_volume()
    project = project_fakes.FakeProject.create_one_project()
    snapshots = volume_fakes.create_snapshots(
        attrs={'volume_id': volume.name}, count=3
    )

    columns = ["ID", "Name", "Description", "Status", "Size"]
    columns_long = columns + ["Created At", "Volume", "Properties"]

    data = []
    for s in snapshots:
        data.append(
            (
                s.id,
                s.name,
                s.description,
                s.status,
                s.size,
            )
        )
    data_long = []
    for s in snapshots:
        data_long.append(
            (
                s.id,
                s.name,
                s.description,
                s.status,
                s.size,
                s.created_at,
                volume_snapshot.VolumeIdColumn(
                    s.volume_id, volume_cache={volume.id: volume}
                ),
                format_columns.DictColumn(s.metadata),
            )
        )

    def setUp(self):
        super().setUp()

        self.volumes_mock.list.return_value = [self.volume]
        self.volumes_mock.get.return_value = self.volume
        self.project_mock.get.return_value = self.project
        self.snapshots_mock.list.return_value = self.snapshots
        # Get the command to test
        self.cmd = volume_snapshot.ListVolumeSnapshot(self.app, None)

    def test_snapshot_list_without_options(self):
        arglist = []
        verifylist = [('all_projects', False), ('long', False)]
        parsed_args = self.check_parser(self.cmd, arglist, verifylist)

        columns, data = self.cmd.take_action(parsed_args)

        self.snapshots_mock.list.assert_called_once_with(
            limit=None,
            marker=None,
            search_opts={
                'all_tenants': False,
                'name': None,
                'status': None,
                'project_id': None,
                'volume_id': None,
            },
        )
        self.assertEqual(self.columns, columns)
        self.assertEqual(self.data, list(data))

    def test_snapshot_list_with_options(self):
        arglist = [
            "--long",
            "--limit",
            "2",
            "--project",
            self.project.id,
            "--marker",
            self.snapshots[0].id,
        ]
        verifylist = [
            ("long", True),
            ("limit", 2),
            ("project", self.project.id),
            ("marker", self.snapshots[0].id),
            ('all_projects', False),
        ]
        parsed_args = self.check_parser(self.cmd, arglist, verifylist)

        columns, data = self.cmd.take_action(parsed_args)

        self.snapshots_mock.list.assert_called_once_with(
            limit=2,
            marker=self.snapshots[0].id,
            search_opts={
                'all_tenants': True,
                'project_id': self.project.id,
                'name': None,
                'status': None,
                'volume_id': None,
            },
        )
        self.assertEqual(self.columns_long, columns)
        self.assertEqual(self.data_long, list(data))

    def test_snapshot_list_all_projects(self):
        arglist = [
            '--all-projects',
        ]
        verifylist = [('long', False), ('all_projects', True)]
        parsed_args = self.check_parser(self.cmd, arglist, verifylist)

        columns, data = self.cmd.take_action(parsed_args)

        self.snapshots_mock.list.assert_called_once_with(
            limit=None,
            marker=None,
            search_opts={
                'all_tenants': True,
                'name': None,
                'status': None,
                'project_id': None,
                'volume_id': None,
            },
        )
        self.assertEqual(self.columns, columns)
        self.assertEqual(self.data, list(data))

    def test_snapshot_list_name_option(self):
        arglist = [
            '--name',
            self.snapshots[0].name,
        ]
        verifylist = [
            ('all_projects', False),
            ('long', False),
            ('name', self.snapshots[0].name),
        ]
        parsed_args = self.check_parser(self.cmd, arglist, verifylist)

        columns, data = self.cmd.take_action(parsed_args)

        self.snapshots_mock.list.assert_called_once_with(
            limit=None,
            marker=None,
            search_opts={
                'all_tenants': False,
                'name': self.snapshots[0].name,
                'status': None,
                'project_id': None,
                'volume_id': None,
            },
        )
        self.assertEqual(self.columns, columns)
        self.assertEqual(self.data, list(data))

    def test_snapshot_list_status_option(self):
        arglist = [
            '--status',
            self.snapshots[0].status,
        ]
        verifylist = [
            ('all_projects', False),
            ('long', False),
            ('status', self.snapshots[0].status),
        ]
        parsed_args = self.check_parser(self.cmd, arglist, verifylist)

        columns, data = self.cmd.take_action(parsed_args)

        self.snapshots_mock.list.assert_called_once_with(
            limit=None,
            marker=None,
            search_opts={
                'all_tenants': False,
                'name': None,
                'status': self.snapshots[0].status,
                'project_id': None,
                'volume_id': None,
            },
        )
        self.assertEqual(self.columns, columns)
        self.assertEqual(self.data, list(data))

    def test_snapshot_list_volumeid_option(self):
        arglist = [
            '--volume',
            self.volume.id,
        ]
        verifylist = [
            ('all_projects', False),
            ('long', False),
            ('volume', self.volume.id),
        ]
        parsed_args = self.check_parser(self.cmd, arglist, verifylist)

        columns, data = self.cmd.take_action(parsed_args)

        self.snapshots_mock.list.assert_called_once_with(
            limit=None,
            marker=None,
            search_opts={
                'all_tenants': False,
                'name': None,
                'status': None,
                'project_id': None,
                'volume_id': self.volume.id,
            },
        )
        self.assertEqual(self.columns, columns)
        self.assertEqual(self.data, list(data))

    def test_snapshot_list_negative_limit(self):
        arglist = [
            "--limit",
            "-2",
        ]
        verifylist = [
            ("limit", -2),
        ]
        self.assertRaises(
            test_utils.ParserException,
            self.check_parser,
            self.cmd,
            arglist,
            verifylist,
        )


class TestVolumeSnapshotSet(volume_fakes.TestVolume):
    def setUp(self):
        super().setUp()

        self.snapshot = sdk_fakes.generate_fake_resource(
            _snapshot.Snapshot, metadata={'foo': 'bar'}
        )
        self.volume_sdk_client.find_snapshot.return_value = self.snapshot
        self.volume_sdk_client.delete_snapshot_metadata.return_value = None
        self.volume_sdk_client.set_snapshot_metadata.return_value = None
        self.volume_sdk_client.update_snapshot.return_value = None

        self.cmd = volume_snapshot.SetVolumeSnapshot(self.app, None)

    def test_snapshot_set_no_option(self):
        arglist = [
            self.snapshot.id,
        ]
        verifylist = [
            ("snapshot", self.snapshot.id),
        ]
        parsed_args = self.check_parser(self.cmd, arglist, verifylist)

        result = self.cmd.take_action(parsed_args)

        self.assertIsNone(result)
        self.volume_sdk_client.find_snapshot.assert_called_once_with(
            parsed_args.snapshot, ignore_missing=False
        )
        self.volume_sdk_client.reset_snapshot_status.assert_not_called()
        self.volume_sdk_client.update_snapshot.assert_not_called()
        self.volume_sdk_client.set_snapshot_metadata.assert_not_called()

    def test_snapshot_set_name_and_property(self):
        arglist = [
            "--name",
            "new_snapshot",
            "--property",
            "x=y",
            "--property",
            "foo=foo",
            self.snapshot.id,
        ]
        verifylist = [
            ("name", "new_snapshot"),
            ("properties", {"x": "y", "foo": "foo"}),
            ("snapshot", self.snapshot.id),
        ]
        parsed_args = self.check_parser(self.cmd, arglist, verifylist)

        result = self.cmd.take_action(parsed_args)

        self.assertIsNone(result)
        self.volume_sdk_client.update_snapshot.assert_called_with(
            self.snapshot.id, name="new_snapshot"
        )
        self.volume_sdk_client.set_snapshot_metadata.assert_called_with(
            self.snapshot.id, x="y", foo="foo"
        )

    def test_snapshot_set_with_no_property(self):
        arglist = [
            "--no-property",
            self.snapshot.id,
        ]
        verifylist = [
            ("no_property", True),
            ("snapshot", self.snapshot.id),
        ]
        parsed_args = self.check_parser(self.cmd, arglist, verifylist)

        result = self.cmd.take_action(parsed_args)

        self.assertIsNone(result)
        self.volume_sdk_client.find_snapshot.assert_called_once_with(
            parsed_args.snapshot, ignore_missing=False
        )
        self.volume_sdk_client.reset_snapshot_status.assert_not_called()
        self.volume_sdk_client.update_snapshot.assert_not_called()
        self.volume_sdk_client.set_snapshot_metadata.assert_not_called()
        self.volume_sdk_client.delete_snapshot_metadata.assert_called_with(
            self.snapshot.id, keys=["foo"]
        )

    def test_snapshot_set_with_no_property_and_property(self):
        arglist = [
            "--no-property",
            "--property",
            "foo_1=bar_1",
            self.snapshot.id,
        ]
        verifylist = [
            ("no_property", True),
            ("properties", {"foo_1": "bar_1"}),
            ("snapshot", self.snapshot.id),
        ]
        parsed_args = self.check_parser(self.cmd, arglist, verifylist)

        result = self.cmd.take_action(parsed_args)

        self.assertIsNone(result)
        self.volume_sdk_client.find_snapshot.assert_called_once_with(
            parsed_args.snapshot, ignore_missing=False
        )
        self.volume_sdk_client.reset_snapshot_status.assert_not_called()
        self.volume_sdk_client.update_snapshot.assert_not_called()
        self.volume_sdk_client.delete_snapshot_metadata.assert_called_with(
            self.snapshot.id, keys=["foo"]
        )
        self.volume_sdk_client.set_snapshot_metadata.assert_called_once_with(
            self.snapshot.id,
            foo_1="bar_1",
        )

    def test_snapshot_set_state_to_error(self):
        arglist = ["--state", "error", self.snapshot.id]
        verifylist = [("state", "error"), ("snapshot", self.snapshot.id)]
        parsed_args = self.check_parser(self.cmd, arglist, verifylist)

        result = self.cmd.take_action(parsed_args)

        self.assertIsNone(result)
        self.volume_sdk_client.reset_snapshot_status.assert_called_with(
            self.snapshot.id, "error"
        )

    def test_volume_set_state_failed(self):
        self.volume_sdk_client.reset_snapshot_status.side_effect = (
            exceptions.CommandError()
        )
        arglist = ['--state', 'error', self.snapshot.id]
        verifylist = [('state', 'error'), ('snapshot', self.snapshot.id)]

        parsed_args = self.check_parser(self.cmd, arglist, verifylist)

        exc = self.assertRaises(
            exceptions.CommandError, self.cmd.take_action, parsed_args
        )
        self.assertEqual('One or more of the set operations failed', str(exc))
        self.volume_sdk_client.reset_snapshot_status.assert_called_once_with(
            self.snapshot.id, 'error'
        )

    def test_volume_set_name_and_state_failed(self):
        self.volume_sdk_client.reset_snapshot_status.side_effect = (
            exceptions.CommandError()
        )
        arglist = [
            '--state',
            'error',
            "--name",
            "new_snapshot",
            self.snapshot.id,
        ]
        verifylist = [
            ('state', 'error'),
            ("name", "new_snapshot"),
            ('snapshot', self.snapshot.id),
        ]
        parsed_args = self.check_parser(self.cmd, arglist, verifylist)

        exc = self.assertRaises(
            exceptions.CommandError,
            self.cmd.take_action,
            parsed_args,
        )

        self.assertEqual('One or more of the set operations failed', str(exc))
        self.volume_sdk_client.update_snapshot.assert_called_once_with(
            self.snapshot.id, name="new_snapshot"
        )
        self.volume_sdk_client.reset_snapshot_status.assert_called_once_with(
            self.snapshot.id, 'error'
        )


class TestVolumeSnapshotShow(TestVolumeSnapshot):
    columns = (
        'created_at',
        'description',
        'id',
        'name',
        'properties',
        'size',
        'status',
        'volume_id',
    )

    def setUp(self):
        super().setUp()

        self.snapshot = volume_fakes.create_one_snapshot()

        self.data = (
            self.snapshot.created_at,
            self.snapshot.description,
            self.snapshot.id,
            self.snapshot.name,
            format_columns.DictColumn(self.snapshot.metadata),
            self.snapshot.size,
            self.snapshot.status,
            self.snapshot.volume_id,
        )

        self.snapshots_mock.get.return_value = self.snapshot
        # Get the command object to test
        self.cmd = volume_snapshot.ShowVolumeSnapshot(self.app, None)

    def test_snapshot_show(self):
        arglist = [self.snapshot.id]
        verifylist = [("snapshot", self.snapshot.id)]
        parsed_args = self.check_parser(self.cmd, arglist, verifylist)

        columns, data = self.cmd.take_action(parsed_args)
        self.snapshots_mock.get.assert_called_with(self.snapshot.id)

        self.assertEqual(self.columns, columns)
        self.assertCountEqual(self.data, data)


class TestVolumeSnapshotUnset(volume_fakes.TestVolume):
    def setUp(self):
        super().setUp()

        self.snapshot = sdk_fakes.generate_fake_resource(_snapshot.Snapshot)
        self.volume_sdk_client.find_snapshot.return_value = self.snapshot
        self.volume_sdk_client.delete_snapshot_metadata.return_value = None

        self.cmd = volume_snapshot.UnsetVolumeSnapshot(self.app, None)

    def test_snapshot_unset(self):
        arglist = [
            "--property",
            "foo",
            self.snapshot.id,
        ]
        verifylist = [
            ("properties", ["foo"]),
            ("snapshot", self.snapshot.id),
        ]

        parsed_args = self.check_parser(self.cmd, arglist, verifylist)

        result = self.cmd.take_action(parsed_args)

        self.assertIsNone(result)
        self.volume_sdk_client.delete_snapshot_metadata.assert_called_with(
            self.snapshot.id, keys=["foo"]
        )