# Copyright (c) 2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
# Permission is hereby granted, free of charge, to any person obtaining a
# copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish, dis-
# tribute, sublicense, and/or sell copies of the Software, and to permit
# persons to whom the Software is furnished to do so, subject to the fol-
# lowing conditions:
#
# The above copyright notice and this permission notice shall be included
# in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL-
# ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
# SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
# IN THE SOFTWARE.

import os
import contextlib
import copy
import mock

from botocore.exceptions import DataNotFoundError, UnknownServiceError
from botocore.loaders import JSONFileLoader
from botocore.loaders import Loader, create_loader
from botocore.loaders import ExtrasProcessor

from tests import BaseEnvVar


class TestJSONFileLoader(BaseEnvVar):
    def setUp(self):
        super(TestJSONFileLoader, self).setUp()
        self.data_path = os.path.join(os.path.dirname(__file__), 'data')
        self.file_loader = JSONFileLoader()
        self.valid_file_path = os.path.join(self.data_path, 'foo')

    def test_load_file(self):
        data = self.file_loader.load_file(self.valid_file_path)
        self.assertEqual(len(data), 3)
        self.assertTrue('test_key_1' in data)

    def test_load_json_file_does_not_exist_returns_none(self):
        # None is used to indicate that the loader could not find a
        # file to load.
        self.assertIsNone(self.file_loader.load_file('fooasdfasdfasdf'))

    def test_file_exists_check(self):
        self.assertTrue(self.file_loader.exists(self.valid_file_path))

    def test_file_does_not_exist_returns_false(self):
        self.assertFalse(self.file_loader.exists(
            os.path.join(self.data_path, 'does', 'not', 'exist')))

    def test_file_with_non_ascii(self):
        try:
            filename = os.path.join(self.data_path, 'non_ascii')
            self.assertTrue(self.file_loader.load_file(filename) is not None)
        except UnicodeDecodeError:
            self.fail('Fail to handle data file with non-ascii characters')


class TestLoader(BaseEnvVar):

    def test_default_search_paths(self):
        loader = Loader()
        self.assertEqual(len(loader.search_paths), 2)
        # We should also have ~/.aws/models added to
        # the search path.  To deal with cross platform
        # issues we'll just check for a path that ends
        # with .aws/models.
        home_dir_path = os.path.join('.aws', 'models')
        self.assertTrue(
            any(p.endswith(home_dir_path) for p in
                loader.search_paths))

    def test_can_add_to_search_path(self):
        loader = Loader()
        loader.search_paths.append('mypath')
        self.assertIn('mypath', loader.search_paths)

    def test_can_initialize_with_search_paths(self):
        loader = Loader(extra_search_paths=['foo', 'bar'])
        # Note that the extra search paths are before
        # the customer/builtin data paths.
        self.assertEqual(
            loader.search_paths,
            ['foo', 'bar', loader.CUSTOMER_DATA_PATH,
             loader.BUILTIN_DATA_PATH])

    # The file loader isn't consulted unless the current
    # search path exists, so we're patching isdir to always
    # say that a directory exists.
    @mock.patch('os.path.isdir', mock.Mock(return_value=True))
    def test_load_data_uses_loader(self):
        search_paths = ['foo', 'bar', 'baz']

        class FakeLoader(object):
            def load_file(self, name):
                expected_ending = os.path.join('bar', 'baz')
                if name.endswith(expected_ending):
                    return ['loaded data']

        loader = Loader(extra_search_paths=search_paths,
                        file_loader=FakeLoader())
        loaded = loader.load_data('baz')
        self.assertEqual(loaded, ['loaded data'])

    def test_data_not_found_raises_exception(self):
        class FakeLoader(object):
            def load_file(self, name):
                # Returning None indicates that the
                # loader couldn't find anything.
                return None
        loader = Loader(file_loader=FakeLoader())
        with self.assertRaises(DataNotFoundError):
            loader.load_data('baz')

    @mock.patch('os.path.isdir', mock.Mock(return_value=True))
    def test_error_raised_if_service_does_not_exist(self):
        loader = Loader(extra_search_paths=[],
                        include_default_search_paths=False)
        with self.assertRaises(DataNotFoundError):
            loader.determine_latest_version('unknownservice', 'service-2')

    @mock.patch('os.path.isdir', mock.Mock(return_value=True))
    def test_load_service_model(self):
        class FakeLoader(object):
            def load_file(self, name):
                return ['loaded data']

        loader = Loader(extra_search_paths=['foo'],
                        file_loader=FakeLoader(),
                        include_default_search_paths=False,
                        include_default_extras=False)
        loader.determine_latest_version = mock.Mock(return_value='2015-03-01')
        loader.list_available_services = mock.Mock(return_value=['baz'])
        loaded = loader.load_service_model('baz', type_name='service-2')
        self.assertEqual(loaded, ['loaded data'])

    @mock.patch('os.path.isdir', mock.Mock(return_value=True))
    def test_load_service_model_enforces_case(self):
        class FakeLoader(object):
            def load_file(self, name):
                return ['loaded data']

        loader = Loader(extra_search_paths=['foo'],
                        file_loader=FakeLoader(),
                        include_default_search_paths=False)
        loader.determine_latest_version = mock.Mock(return_value='2015-03-01')
        loader.list_available_services = mock.Mock(return_value=['baz'])

        # Should have a) the unknown service name and b) list of valid
        # service names.
        with self.assertRaisesRegexp(UnknownServiceError,
                                     'Unknown service.*BAZ.*baz'):
            loader.load_service_model('BAZ', type_name='service-2')

    def test_load_service_model_uses_provided_type_name(self):
        loader = Loader(extra_search_paths=['foo'],
                        file_loader=mock.Mock(),
                        include_default_search_paths=False)
        loader.list_available_services = mock.Mock(return_value=['baz'])

        # Should have a) the unknown service name and b) list of valid
        # service names.
        provided_type_name = 'not-service-2'
        with self.assertRaisesRegexp(UnknownServiceError,
                                     'Unknown service.*BAZ.*baz'):
            loader.load_service_model(
                'BAZ', type_name=provided_type_name)

        loader.list_available_services.assert_called_with(provided_type_name)

    def test_create_loader_parses_data_path(self):
        search_path = os.pathsep.join(['foo', 'bar', 'baz'])
        loader = create_loader(search_path)
        self.assertIn('foo', loader.search_paths)
        self.assertIn('bar', loader.search_paths)
        self.assertIn('baz', loader.search_paths)


class TestMergeExtras(BaseEnvVar):
    def setUp(self):
        super(TestMergeExtras, self).setUp()
        self.file_loader = mock.Mock()
        self.data_loader = Loader(
            extra_search_paths=['datapath'], file_loader=self.file_loader,
            include_default_search_paths=False)
        self.data_loader.determine_latest_version = mock.Mock(
            return_value='2015-03-01')
        self.data_loader.list_available_services = mock.Mock(
            return_value=['myservice'])

        isdir_mock = mock.Mock(return_value=True)
        self.isdir_patch = mock.patch('os.path.isdir', isdir_mock)
        self.isdir_patch.start()

    def tearDown(self):
        super(TestMergeExtras, self).tearDown()
        self.isdir_patch.stop()

    def test_merge_extras(self):
        service_data = {'foo': 'service', 'bar': 'service'}
        sdk_extras = {'merge': {'foo': 'sdk'}}
        self.file_loader.load_file.side_effect = [service_data, sdk_extras]

        loaded = self.data_loader.load_service_model('myservice', 'service-2')
        expected = {'foo': 'sdk', 'bar': 'service'}
        self.assertEqual(loaded, expected)

        call_args = self.file_loader.load_file.call_args_list
        call_args = [c[0][0] for c in call_args]
        base_path = os.path.join('datapath', 'myservice', '2015-03-01')
        expected_call_args = [
            os.path.join(base_path, 'service-2'),
            os.path.join(base_path, 'service-2.sdk-extras')
        ]
        self.assertEqual(call_args, expected_call_args)

    def test_extras_not_found(self):
        service_data = {'foo': 'service', 'bar': 'service'}
        service_data_copy = copy.copy(service_data)
        self.file_loader.load_file.side_effect = [service_data, None]

        loaded = self.data_loader.load_service_model('myservice', 'service-2')
        self.assertEqual(loaded, service_data_copy)

    def test_no_merge_in_extras(self):
        service_data = {'foo': 'service', 'bar': 'service'}
        service_data_copy = copy.copy(service_data)
        self.file_loader.load_file.side_effect = [service_data, {}]

        loaded = self.data_loader.load_service_model('myservice', 'service-2')
        self.assertEqual(loaded, service_data_copy)

    def test_include_default_extras(self):
        self.data_loader = Loader(
            extra_search_paths=['datapath'], file_loader=self.file_loader,
            include_default_search_paths=False,
            include_default_extras=False)
        self.data_loader.determine_latest_version = mock.Mock(
            return_value='2015-03-01')
        self.data_loader.list_available_services = mock.Mock(
            return_value=['myservice'])

        service_data = {'foo': 'service', 'bar': 'service'}
        service_data_copy = copy.copy(service_data)
        sdk_extras = {'merge': {'foo': 'sdk'}}
        self.file_loader.load_file.side_effect = [service_data, sdk_extras]

        loaded = self.data_loader.load_service_model('myservice', 'service-2')
        self.assertEqual(loaded, service_data_copy)

    def test_append_extra_type(self):
        service_data = {'foo': 'service', 'bar': 'service'}
        sdk_extras = {'merge': {'foo': 'sdk'}}
        cli_extras = {'merge': {'cli': True}}
        self.file_loader.load_file.side_effect = [
            service_data, sdk_extras, cli_extras]

        self.data_loader.extras_types.append('cli')

        loaded = self.data_loader.load_service_model('myservice', 'service-2')
        expected = {'foo': 'sdk', 'bar': 'service', 'cli': True}
        self.assertEqual(loaded, expected)

        call_args = self.file_loader.load_file.call_args_list
        call_args = [c[0][0] for c in call_args]
        base_path = os.path.join('datapath', 'myservice', '2015-03-01')
        expected_call_args = [
            os.path.join(base_path, 'service-2'),
            os.path.join(base_path, 'service-2.sdk-extras'),
            os.path.join(base_path, 'service-2.cli-extras')
        ]
        self.assertEqual(call_args, expected_call_args)

    def test_sdk_empty_extras_skipped(self):
        service_data = {'foo': 'service', 'bar': 'service'}
        cli_extras = {'merge': {'foo': 'cli'}}
        self.file_loader.load_file.side_effect = [
            service_data, None, cli_extras]

        self.data_loader.extras_types.append('cli')

        loaded = self.data_loader.load_service_model('myservice', 'service-2')
        expected = {'foo': 'cli', 'bar': 'service'}
        self.assertEqual(loaded, expected)


class TestExtrasProcessor(BaseEnvVar):
    def setUp(self):
        super(TestExtrasProcessor, self).setUp()
        self.processor = ExtrasProcessor()
        self.service_data = {
            'shapes': {
                'StringShape': {'type': 'string'},
            }
        }
        self.service_data_copy = copy.deepcopy(self.service_data)

    def test_process_empty_list(self):
        self.processor.process(self.service_data, [])
        self.assertEqual(self.service_data, self.service_data_copy)

    def test_process_empty_extras(self):
        self.processor.process(self.service_data, [{}])
        self.assertEqual(self.service_data, self.service_data_copy)

    def test_process_merge_key(self):
        extras = {'merge': {'shapes': {'BooleanShape': {'type': 'boolean'}}}}
        self.processor.process(self.service_data, [extras])
        self.assertNotEqual(self.service_data, self.service_data_copy)

        boolean_shape = self.service_data['shapes'].get('BooleanShape')
        self.assertEqual(boolean_shape, {'type': 'boolean'})

    def test_process_in_order(self):
        extras = [
            {'merge': {'shapes': {'BooleanShape': {'type': 'boolean'}}}},
            {'merge': {'shapes': {'BooleanShape': {'type': 'string'}}}}
        ]
        self.processor.process(self.service_data, extras)
        self.assertNotEqual(self.service_data, self.service_data_copy)

        boolean_shape = self.service_data['shapes'].get('BooleanShape')
        self.assertEqual(boolean_shape, {'type': 'string'})


class TestLoadersWithDirectorySearching(BaseEnvVar):
    def setUp(self):
        super(TestLoadersWithDirectorySearching, self).setUp()
        self.fake_directories = {}

    def tearDown(self):
        super(TestLoadersWithDirectorySearching, self).tearDown()

    @contextlib.contextmanager
    def loader_with_fake_dirs(self):
        mock_file_loader = mock.Mock()
        mock_file_loader.exists = self.fake_exists
        search_paths = list(self.fake_directories)
        loader = Loader(extra_search_paths=search_paths,
                        include_default_search_paths=False,
                        file_loader=mock_file_loader)
        with mock.patch('os.listdir', self.fake_listdir):
            with mock.patch('os.path.isdir', mock.Mock(return_value=True)):
                yield loader


    def fake_listdir(self, dirname):
        parts = dirname.split(os.path.sep)
        result = self.fake_directories
        while parts:
            current = parts.pop(0)
            result = result[current]
        return list(result)

    def fake_exists(self, path):
        parts = path.split(os.sep)
        result = self.fake_directories
        while len(parts) > 1:
            current = parts.pop(0)
            result = result[current]
        return parts[0] in result

    def test_list_available_services(self):
        self.fake_directories = {
            'foo': {
                'ec2': {
                    '2010-01-01': ['service-2'],
                    '2014-10-01': ['service-1'],
                },
                'dynamodb': {
                    '2010-01-01': ['service-2'],
                },
            },
            'bar': {
                'ec2': {
                    '2015-03-01': ['service-1'],
                },
                'rds': {
                    # This will not show up in
                    # list_available_services() for type
                    # service-2 because it does not contains
                    # a service-2.
                    '2012-01-01': ['resource-1'],
                },
            },
        }
        with self.loader_with_fake_dirs() as loader:
            self.assertEqual(
                loader.list_available_services(type_name='service-2'),
                ['dynamodb', 'ec2'])
            self.assertEqual(
                loader.list_available_services(type_name='resource-1'),
                ['rds'])

    def test_determine_latest(self):
        # Fake mapping of directories to subdirectories.
        # In this example, we can see that the 'bar' directory
        # contains the latest EC2 API version, 2015-03-01,
        # so loader.determine_latest('ec2') should return
        # this value 2015-03-01.
        self.fake_directories = {
            'foo': {
                'ec2': {
                    '2010-01-01': ['service-2'],
                    # This directory contains the latest API version
                    # for EC2 because its the highest API directory
                    # that contains a service-2.
                    '2014-10-01': ['service-2'],
                },
            },
            'bar': {
                'ec2': {
                    '2012-01-01': ['service-2'],
                    # 2015-03-1 is *not* the latest for service-2,
                    # because its directory only has service-1.json.
                    '2015-03-01': ['service-1'],
                },
            },
        }
        with self.loader_with_fake_dirs() as loader:
            latest = loader.determine_latest_version('ec2', 'service-2')
            self.assertEqual(loader.determine_latest_version('ec2', 'service-2'),
                             '2014-10-01')
            self.assertEqual(loader.determine_latest_version('ec2', 'service-1'),
                             '2015-03-01')