# Copyright 2015 Amazon.com, Inc. or its affiliates. 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. A copy of
# the License is located at
#
# http://aws.amazon.com/apache2.0/
#
# or in the "license" file accompanying this file. This file 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 json
import os
import shutil
import tempfile

from botocore.client import ClientCreator
from botocore.compat import OrderedDict
from botocore.configprovider import ConfigValueStore
from botocore.docs.bcdoc.restdoc import DocumentStructure
from botocore.hooks import HierarchicalEmitter
from botocore.loaders import Loader
from botocore.model import OperationModel, ServiceModel
from tests import mock, unittest


class BaseDocsTest(unittest.TestCase):
    def setUp(self):
        self.root_dir = tempfile.mkdtemp()
        self.version_dirs = os.path.join(
            self.root_dir, 'myservice', '2014-01-01'
        )
        os.makedirs(self.version_dirs)
        self.model_file = os.path.join(self.version_dirs, 'service-2.json')
        self.waiter_model_file = os.path.join(
            self.version_dirs, 'waiters-2.json'
        )
        self.paginator_model_file = os.path.join(
            self.version_dirs, 'paginators-1.json'
        )
        self.example_model_file = os.path.join(
            self.version_dirs, 'examples-1.json'
        )

        self.json_model = {}
        self.nested_json_model = {}
        self._setup_models()
        self.build_models()
        self.events = HierarchicalEmitter()
        self.setup_client()
        self.doc_name = 'MyDoc'
        self.doc_structure = DocumentStructure(self.doc_name, target='html')

    def tearDown(self):
        shutil.rmtree(self.root_dir)

    def setup_client(self):
        with open(self.example_model_file, 'w') as f:
            json.dump(self.example_json_model, f)

        with open(self.waiter_model_file, 'w') as f:
            json.dump(self.waiter_json_model, f)

        with open(self.paginator_model_file, 'w') as f:
            json.dump(self.paginator_json_model, f)

        with open(self.model_file, 'w') as f:
            json.dump(self.json_model, f)

        myservice_endpoint_rule_set = {
            "version": "1.3",
            "parameters": {},
            "rules": [
                {
                    "conditions": [],
                    "endpoint": {
                        "url": "https://example.com",
                        "properties": {},
                        "headers": {},
                    },
                    "type": "endpoint",
                }
            ],
        }

        def load_service_mock(*args, **kwargs):
            if args[1] == "endpoint-rule-set-1":
                return myservice_endpoint_rule_set
            else:
                return Loader(
                    extra_search_paths=[self.root_dir]
                ).load_service_model(*args, **kwargs)

        self.loader = Loader(extra_search_paths=[self.root_dir])
        self.loader.load_service_model = mock.Mock()
        self.loader.load_service_model.side_effect = load_service_mock

        endpoint_resolver = mock.Mock()
        endpoint_resolver.construct_endpoint.return_value = {
            'hostname': 'foo.us-east-1',
            'partition': 'aws',
            'endpointName': 'us-east-1',
            'signatureVersions': ['v4'],
        }

        self.creator = ClientCreator(
            loader=self.loader,
            endpoint_resolver=endpoint_resolver,
            user_agent='user-agent',
            event_emitter=self.events,
            retry_handler_factory=mock.Mock(),
            retry_config_translator=mock.Mock(),
            exceptions_factory=mock.Mock(),
            config_store=ConfigValueStore(),
        )

        self.client = self.creator.create_client('myservice', 'us-east-1')

    def _setup_models(self):
        self.json_model = {
            'metadata': {
                'apiVersion': '2014-01-01',
                'endpointPrefix': 'myservice',
                'signatureVersion': 'v4',
                'serviceFullName': 'AWS MyService',
                'uid': 'myservice-2014-01-01',
                'protocol': 'query',
                'serviceId': 'MyService',
            },
            'operations': {
                'SampleOperation': {
                    'name': 'SampleOperation',
                    'input': {'shape': 'SampleOperationInputOutput'},
                    'output': {'shape': 'SampleOperationInputOutput'},
                }
            },
            'shapes': {
                'SampleOperationInputOutput': {
                    'type': 'structure',
                    'members': OrderedDict(),
                },
                'String': {'type': 'string'},
            },
            'documentation': 'AWS MyService Description',
        }

        self.waiter_json_model = {
            "version": 2,
            "waiters": {
                "SampleOperationComplete": {
                    "delay": 15,
                    "operation": "SampleOperation",
                    "maxAttempts": 40,
                    "acceptors": [
                        {
                            "expected": "complete",
                            "matcher": "pathAll",
                            "state": "success",
                            "argument": "Biz",
                        },
                        {
                            "expected": "failed",
                            "matcher": "pathAny",
                            "state": "failure",
                            "argument": "Biz",
                        },
                    ],
                }
            },
        }

        self.paginator_json_model = {
            "pagination": {
                "SampleOperation": {
                    "input_token": "NextResult",
                    "output_token": "NextResult",
                    "limit_key": "MaxResults",
                    "result_key": "Biz",
                }
            }
        }

        self.example_json_model = {
            "version": 1,
            "examples": {
                "SampleOperation": [
                    {
                        "id": "sample-id",
                        "title": "sample-title",
                        "description": "Sample Description.",
                        "input": OrderedDict(
                            [
                                ("Biz", "foo"),
                            ]
                        ),
                        "comments": {
                            "input": {"Biz": "bar"},
                        },
                    }
                ]
            },
        }

    def build_models(self):
        self.service_model = ServiceModel(self.json_model)
        self.operation_model = OperationModel(
            self.json_model['operations']['SampleOperation'],
            self.service_model,
        )

    def add_shape(self, shape):
        shape_name = list(shape.keys())[0]
        self.json_model['shapes'][shape_name] = shape[shape_name]

    def add_shape_to_params(
        self, param_name, shape_name, documentation=None, is_required=False
    ):
        params_shape = self.json_model['shapes']['SampleOperationInputOutput']
        member = {'shape': shape_name}
        if documentation is not None:
            member['documentation'] = documentation
        params_shape['members'][param_name] = member

        if is_required:
            required_list = params_shape.get('required', [])
            required_list.append(param_name)
            params_shape['required'] = required_list

    def add_shape_to_errors(self, shape_name):
        operation = self.json_model['operations']['SampleOperation']
        errors = operation.get('errors', [])
        errors.append({'shape': shape_name})
        operation['errors'] = errors

    def assert_contains_line(self, line):
        contents = self.doc_structure.flush_structure().decode('utf-8')
        self.assertIn(line, contents)

    def assert_contains_lines_in_order(self, lines):
        contents = self.doc_structure.flush_structure().decode('utf-8')
        for line in lines:
            self.assertIn(line, contents)
            beginning = contents.find(line)
            contents = contents[(beginning + len(line)) :]

    def assert_not_contains_line(self, line):
        contents = self.doc_structure.flush_structure().decode('utf-8')
        self.assertNotIn(line, contents)

    def assert_not_contains_lines(self, lines):
        contents = self.doc_structure.flush_structure().decode('utf-8')
        for line in lines:
            self.assertNotIn(line, contents)