# Copyright 2016 Google LLC 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 unittest

import mock


class TestInstance(unittest.TestCase):

    PROJECT = "project"
    PARENT = "projects/" + PROJECT
    INSTANCE_ID = "instance-id"
    INSTANCE_NAME = PARENT + "/instances/" + INSTANCE_ID
    CONFIG_NAME = "configuration-name"
    LOCATION = "projects/" + PROJECT + "/locations/" + CONFIG_NAME
    DISPLAY_NAME = "display_name"
    NODE_COUNT = 5
    OP_ID = 8915
    OP_NAME = "operations/projects/%s/instances/%soperations/%d" % (
        PROJECT,
        INSTANCE_ID,
        OP_ID,
    )
    TABLE_ID = "table_id"
    TABLE_NAME = INSTANCE_NAME + "/tables/" + TABLE_ID
    TIMEOUT_SECONDS = 1
    DATABASE_ID = "database_id"
    DATABASE_NAME = "%s/databases/%s" % (INSTANCE_NAME, DATABASE_ID)

    def _getTargetClass(self):
        from google.cloud.spanner_v1.instance import Instance

        return Instance

    def _make_one(self, *args, **kwargs):
        return self._getTargetClass()(*args, **kwargs)

    def test_constructor_defaults(self):
        from google.cloud.spanner_v1.instance import DEFAULT_NODE_COUNT

        client = object()
        instance = self._make_one(self.INSTANCE_ID, client)
        self.assertEqual(instance.instance_id, self.INSTANCE_ID)
        self.assertIs(instance._client, client)
        self.assertIs(instance.configuration_name, None)
        self.assertEqual(instance.node_count, DEFAULT_NODE_COUNT)
        self.assertEqual(instance.display_name, self.INSTANCE_ID)

    def test_constructor_non_default(self):
        DISPLAY_NAME = "display_name"
        client = object()

        instance = self._make_one(
            self.INSTANCE_ID,
            client,
            configuration_name=self.CONFIG_NAME,
            node_count=self.NODE_COUNT,
            display_name=DISPLAY_NAME,
        )
        self.assertEqual(instance.instance_id, self.INSTANCE_ID)
        self.assertIs(instance._client, client)
        self.assertEqual(instance.configuration_name, self.CONFIG_NAME)
        self.assertEqual(instance.node_count, self.NODE_COUNT)
        self.assertEqual(instance.display_name, DISPLAY_NAME)

    def test_copy(self):
        DISPLAY_NAME = "display_name"

        client = _Client(self.PROJECT)
        instance = self._make_one(
            self.INSTANCE_ID, client, self.CONFIG_NAME, display_name=DISPLAY_NAME
        )
        new_instance = instance.copy()

        # Make sure the client copy succeeded.
        self.assertIsNot(new_instance._client, client)
        self.assertEqual(new_instance._client, client)
        # Make sure the client got copied to a new instance.
        self.assertIsNot(instance, new_instance)
        self.assertEqual(instance, new_instance)

    def test__update_from_pb_success(self):
        from google.cloud.spanner_admin_instance_v1.proto import (
            spanner_instance_admin_pb2 as admin_v1_pb2,
        )

        display_name = "display_name"
        instance_pb = admin_v1_pb2.Instance(display_name=display_name)

        instance = self._make_one(None, None, None, None)
        self.assertEqual(instance.display_name, None)
        instance._update_from_pb(instance_pb)
        self.assertEqual(instance.display_name, display_name)

    def test__update_from_pb_no_display_name(self):
        from google.cloud.spanner_admin_instance_v1.proto import (
            spanner_instance_admin_pb2 as admin_v1_pb2,
        )

        instance_pb = admin_v1_pb2.Instance()
        instance = self._make_one(None, None, None, None)
        self.assertEqual(instance.display_name, None)
        with self.assertRaises(ValueError):
            instance._update_from_pb(instance_pb)
        self.assertEqual(instance.display_name, None)

    def test_from_pb_bad_instance_name(self):
        from google.cloud.spanner_admin_instance_v1.proto import (
            spanner_instance_admin_pb2 as admin_v1_pb2,
        )

        instance_name = "INCORRECT_FORMAT"
        instance_pb = admin_v1_pb2.Instance(name=instance_name)

        klass = self._getTargetClass()
        with self.assertRaises(ValueError):
            klass.from_pb(instance_pb, None)

    def test_from_pb_project_mistmatch(self):
        from google.cloud.spanner_admin_instance_v1.proto import (
            spanner_instance_admin_pb2 as admin_v1_pb2,
        )

        ALT_PROJECT = "ALT_PROJECT"
        client = _Client(project=ALT_PROJECT)

        self.assertNotEqual(self.PROJECT, ALT_PROJECT)

        instance_pb = admin_v1_pb2.Instance(name=self.INSTANCE_NAME)

        klass = self._getTargetClass()
        with self.assertRaises(ValueError):
            klass.from_pb(instance_pb, client)

    def test_from_pb_success(self):
        from google.cloud.spanner_admin_instance_v1.proto import (
            spanner_instance_admin_pb2 as admin_v1_pb2,
        )

        client = _Client(project=self.PROJECT)

        instance_pb = admin_v1_pb2.Instance(
            name=self.INSTANCE_NAME,
            config=self.CONFIG_NAME,
            display_name=self.INSTANCE_ID,
        )

        klass = self._getTargetClass()
        instance = klass.from_pb(instance_pb, client)
        self.assertTrue(isinstance(instance, klass))
        self.assertEqual(instance._client, client)
        self.assertEqual(instance.instance_id, self.INSTANCE_ID)
        self.assertEqual(instance.configuration_name, self.CONFIG_NAME)

    def test_name_property(self):
        client = _Client(project=self.PROJECT)

        instance = self._make_one(self.INSTANCE_ID, client, self.CONFIG_NAME)
        self.assertEqual(instance.name, self.INSTANCE_NAME)

    def test___eq__(self):
        client = object()
        instance1 = self._make_one(self.INSTANCE_ID, client, self.CONFIG_NAME)
        instance2 = self._make_one(self.INSTANCE_ID, client, self.CONFIG_NAME)
        self.assertEqual(instance1, instance2)

    def test___eq__type_differ(self):
        client = object()
        instance1 = self._make_one(self.INSTANCE_ID, client, self.CONFIG_NAME)
        instance2 = object()
        self.assertNotEqual(instance1, instance2)

    def test___ne__same_value(self):
        client = object()
        instance1 = self._make_one(self.INSTANCE_ID, client, self.CONFIG_NAME)
        instance2 = self._make_one(self.INSTANCE_ID, client, self.CONFIG_NAME)
        comparison_val = instance1 != instance2
        self.assertFalse(comparison_val)

    def test___ne__(self):
        instance1 = self._make_one("instance_id1", "client1", self.CONFIG_NAME)
        instance2 = self._make_one("instance_id2", "client2", self.CONFIG_NAME)
        self.assertNotEqual(instance1, instance2)

    def test_create_grpc_error(self):
        from google.api_core.exceptions import Unknown

        client = _Client(self.PROJECT)
        client.instance_admin_api = _FauxInstanceAdminAPI(_rpc_error=True)
        instance = self._make_one(
            self.INSTANCE_ID, client, configuration_name=self.CONFIG_NAME
        )

        with self.assertRaises(Unknown):
            instance.create()

    def test_create_already_exists(self):
        from google.cloud.exceptions import Conflict

        client = _Client(self.PROJECT)
        api = client.instance_admin_api = _FauxInstanceAdminAPI(
            _create_instance_conflict=True
        )
        instance = self._make_one(
            self.INSTANCE_ID, client, configuration_name=self.CONFIG_NAME
        )

        with self.assertRaises(Conflict):
            instance.create()

        (parent, instance_id, instance, metadata) = api._created_instance
        self.assertEqual(parent, self.PARENT)
        self.assertEqual(instance_id, self.INSTANCE_ID)
        self.assertEqual(instance.name, self.INSTANCE_NAME)
        self.assertEqual(instance.config, self.CONFIG_NAME)
        self.assertEqual(instance.display_name, self.INSTANCE_ID)
        self.assertEqual(instance.node_count, 1)
        self.assertEqual(metadata, [("google-cloud-resource-prefix", instance.name)])

    def test_create_success(self):
        op_future = _FauxOperationFuture()
        client = _Client(self.PROJECT)
        api = client.instance_admin_api = _FauxInstanceAdminAPI(
            _create_instance_response=op_future
        )
        instance = self._make_one(
            self.INSTANCE_ID,
            client,
            configuration_name=self.CONFIG_NAME,
            display_name=self.DISPLAY_NAME,
            node_count=self.NODE_COUNT,
        )

        future = instance.create()

        self.assertIs(future, op_future)

        (parent, instance_id, instance, metadata) = api._created_instance
        self.assertEqual(parent, self.PARENT)
        self.assertEqual(instance_id, self.INSTANCE_ID)
        self.assertEqual(instance.name, self.INSTANCE_NAME)
        self.assertEqual(instance.config, self.CONFIG_NAME)
        self.assertEqual(instance.display_name, self.DISPLAY_NAME)
        self.assertEqual(instance.node_count, self.NODE_COUNT)
        self.assertEqual(metadata, [("google-cloud-resource-prefix", instance.name)])

    def test_exists_instance_grpc_error(self):
        from google.api_core.exceptions import Unknown

        client = _Client(self.PROJECT)
        client.instance_admin_api = _FauxInstanceAdminAPI(_rpc_error=True)
        instance = self._make_one(self.INSTANCE_ID, client, self.CONFIG_NAME)

        with self.assertRaises(Unknown):
            instance.exists()

    def test_exists_instance_not_found(self):
        client = _Client(self.PROJECT)
        api = client.instance_admin_api = _FauxInstanceAdminAPI(
            _instance_not_found=True
        )
        api._instance_not_found = True
        instance = self._make_one(self.INSTANCE_ID, client, self.CONFIG_NAME)

        self.assertFalse(instance.exists())

        name, metadata = api._got_instance
        self.assertEqual(name, self.INSTANCE_NAME)
        self.assertEqual(metadata, [("google-cloud-resource-prefix", instance.name)])

    def test_exists_success(self):
        from google.cloud.spanner_admin_instance_v1.proto import (
            spanner_instance_admin_pb2 as admin_v1_pb2,
        )

        client = _Client(self.PROJECT)
        instance_pb = admin_v1_pb2.Instance(
            name=self.INSTANCE_NAME,
            config=self.CONFIG_NAME,
            display_name=self.DISPLAY_NAME,
            node_count=self.NODE_COUNT,
        )
        api = client.instance_admin_api = _FauxInstanceAdminAPI(
            _get_instance_response=instance_pb
        )
        instance = self._make_one(self.INSTANCE_ID, client)

        self.assertTrue(instance.exists())

        name, metadata = api._got_instance
        self.assertEqual(name, self.INSTANCE_NAME)
        self.assertEqual(metadata, [("google-cloud-resource-prefix", instance.name)])

    def test_reload_instance_grpc_error(self):
        from google.api_core.exceptions import Unknown

        client = _Client(self.PROJECT)
        client.instance_admin_api = _FauxInstanceAdminAPI(_rpc_error=True)
        instance = self._make_one(self.INSTANCE_ID, client, self.CONFIG_NAME)

        with self.assertRaises(Unknown):
            instance.reload()

    def test_reload_instance_not_found(self):
        from google.cloud.exceptions import NotFound

        client = _Client(self.PROJECT)
        api = client.instance_admin_api = _FauxInstanceAdminAPI(
            _instance_not_found=True
        )
        api._instance_not_found = True
        instance = self._make_one(self.INSTANCE_ID, client, self.CONFIG_NAME)

        with self.assertRaises(NotFound):
            instance.reload()

        name, metadata = api._got_instance
        self.assertEqual(name, self.INSTANCE_NAME)
        self.assertEqual(metadata, [("google-cloud-resource-prefix", instance.name)])

    def test_reload_success(self):
        from google.cloud.spanner_admin_instance_v1.proto import (
            spanner_instance_admin_pb2 as admin_v1_pb2,
        )

        client = _Client(self.PROJECT)
        instance_pb = admin_v1_pb2.Instance(
            name=self.INSTANCE_NAME,
            config=self.CONFIG_NAME,
            display_name=self.DISPLAY_NAME,
            node_count=self.NODE_COUNT,
        )
        api = client.instance_admin_api = _FauxInstanceAdminAPI(
            _get_instance_response=instance_pb
        )
        instance = self._make_one(self.INSTANCE_ID, client)

        instance.reload()

        self.assertEqual(instance.configuration_name, self.CONFIG_NAME)
        self.assertEqual(instance.node_count, self.NODE_COUNT)
        self.assertEqual(instance.display_name, self.DISPLAY_NAME)

        name, metadata = api._got_instance
        self.assertEqual(name, self.INSTANCE_NAME)
        self.assertEqual(metadata, [("google-cloud-resource-prefix", instance.name)])

    def test_update_grpc_error(self):
        from google.api_core.exceptions import Unknown

        client = _Client(self.PROJECT)
        client.instance_admin_api = _FauxInstanceAdminAPI(_rpc_error=True)
        instance = self._make_one(
            self.INSTANCE_ID, client, configuration_name=self.CONFIG_NAME
        )

        with self.assertRaises(Unknown):
            instance.update()

    def test_update_not_found(self):
        from google.cloud.exceptions import NotFound
        from google.cloud.spanner_v1.instance import DEFAULT_NODE_COUNT

        client = _Client(self.PROJECT)
        api = client.instance_admin_api = _FauxInstanceAdminAPI(
            _instance_not_found=True
        )
        instance = self._make_one(
            self.INSTANCE_ID, client, configuration_name=self.CONFIG_NAME
        )

        with self.assertRaises(NotFound):
            instance.update()

        instance, field_mask, metadata = api._updated_instance
        self.assertEqual(field_mask.paths, ["config", "display_name", "node_count"])
        self.assertEqual(instance.name, self.INSTANCE_NAME)
        self.assertEqual(instance.config, self.CONFIG_NAME)
        self.assertEqual(instance.display_name, self.INSTANCE_ID)
        self.assertEqual(instance.node_count, DEFAULT_NODE_COUNT)
        self.assertEqual(metadata, [("google-cloud-resource-prefix", instance.name)])

    def test_update_success(self):
        op_future = _FauxOperationFuture()
        client = _Client(self.PROJECT)
        api = client.instance_admin_api = _FauxInstanceAdminAPI(
            _update_instance_response=op_future
        )
        instance = self._make_one(
            self.INSTANCE_ID,
            client,
            configuration_name=self.CONFIG_NAME,
            node_count=self.NODE_COUNT,
            display_name=self.DISPLAY_NAME,
        )

        future = instance.update()

        self.assertIs(future, op_future)

        instance, field_mask, metadata = api._updated_instance
        self.assertEqual(field_mask.paths, ["config", "display_name", "node_count"])
        self.assertEqual(instance.name, self.INSTANCE_NAME)
        self.assertEqual(instance.config, self.CONFIG_NAME)
        self.assertEqual(instance.display_name, self.DISPLAY_NAME)
        self.assertEqual(instance.node_count, self.NODE_COUNT)
        self.assertEqual(metadata, [("google-cloud-resource-prefix", instance.name)])

    def test_delete_grpc_error(self):
        from google.api_core.exceptions import Unknown

        client = _Client(self.PROJECT)
        client.instance_admin_api = _FauxInstanceAdminAPI(_rpc_error=True)
        instance = self._make_one(self.INSTANCE_ID, client)

        with self.assertRaises(Unknown):
            instance.delete()

    def test_delete_not_found(self):
        from google.cloud.exceptions import NotFound

        client = _Client(self.PROJECT)
        api = client.instance_admin_api = _FauxInstanceAdminAPI(
            _instance_not_found=True
        )
        instance = self._make_one(self.INSTANCE_ID, client)

        with self.assertRaises(NotFound):
            instance.delete()

        name, metadata = api._deleted_instance
        self.assertEqual(name, self.INSTANCE_NAME)
        self.assertEqual(metadata, [("google-cloud-resource-prefix", instance.name)])

    def test_delete_success(self):
        from google.protobuf.empty_pb2 import Empty

        client = _Client(self.PROJECT)
        api = client.instance_admin_api = _FauxInstanceAdminAPI(
            _delete_instance_response=Empty()
        )
        instance = self._make_one(self.INSTANCE_ID, client, self.CONFIG_NAME)

        instance.delete()

        name, metadata = api._deleted_instance
        self.assertEqual(name, self.INSTANCE_NAME)
        self.assertEqual(metadata, [("google-cloud-resource-prefix", instance.name)])

    def test_database_factory_defaults(self):
        from google.cloud.spanner_v1.database import Database
        from google.cloud.spanner_v1.pool import BurstyPool

        client = _Client(self.PROJECT)
        instance = self._make_one(self.INSTANCE_ID, client, self.CONFIG_NAME)
        DATABASE_ID = "database-id"

        database = instance.database(DATABASE_ID)

        self.assertTrue(isinstance(database, Database))
        self.assertEqual(database.database_id, DATABASE_ID)
        self.assertIs(database._instance, instance)
        self.assertEqual(list(database.ddl_statements), [])
        self.assertIsInstance(database._pool, BurstyPool)
        pool = database._pool
        self.assertIs(pool._database, database)

    def test_database_factory_explicit(self):
        from google.cloud.spanner_v1.database import Database
        from tests._fixtures import DDL_STATEMENTS

        client = _Client(self.PROJECT)
        instance = self._make_one(self.INSTANCE_ID, client, self.CONFIG_NAME)
        DATABASE_ID = "database-id"
        pool = _Pool()

        database = instance.database(
            DATABASE_ID, ddl_statements=DDL_STATEMENTS, pool=pool
        )

        self.assertTrue(isinstance(database, Database))
        self.assertEqual(database.database_id, DATABASE_ID)
        self.assertIs(database._instance, instance)
        self.assertEqual(list(database.ddl_statements), DDL_STATEMENTS)
        self.assertIs(database._pool, pool)
        self.assertIs(pool._bound, database)

    def test_list_databases(self):
        from google.cloud.spanner_admin_database_v1.gapic import database_admin_client
        from google.cloud.spanner_admin_database_v1.proto import (
            spanner_database_admin_pb2,
        )
        from google.cloud.spanner_v1.database import Database

        api = database_admin_client.DatabaseAdminClient(mock.Mock())
        client = _Client(self.PROJECT)
        client.database_admin_api = api
        instance = self._make_one(self.INSTANCE_ID, client)

        databases_pb = spanner_database_admin_pb2.ListDatabasesResponse(
            databases=[
                spanner_database_admin_pb2.Database(
                    name="{}/databases/aa".format(self.INSTANCE_NAME)
                ),
                spanner_database_admin_pb2.Database(
                    name="{}/databases/bb".format(self.INSTANCE_NAME)
                ),
            ]
        )

        ld_api = api._inner_api_calls["list_databases"] = mock.Mock(
            return_value=databases_pb
        )

        response = instance.list_databases()
        databases = list(response)

        self.assertIsInstance(databases[0], Database)
        self.assertTrue(databases[0].name.endswith("/aa"))
        self.assertTrue(databases[1].name.endswith("/bb"))

        expected_metadata = [
            ("google-cloud-resource-prefix", instance.name),
            ("x-goog-request-params", "parent={}".format(instance.name)),
        ]
        ld_api.assert_called_once_with(
            spanner_database_admin_pb2.ListDatabasesRequest(parent=self.INSTANCE_NAME),
            metadata=expected_metadata,
            retry=mock.ANY,
            timeout=mock.ANY,
        )

    def test_list_databases_w_options(self):
        from google.cloud.spanner_admin_database_v1.gapic import database_admin_client
        from google.cloud.spanner_admin_database_v1.proto import (
            spanner_database_admin_pb2,
        )

        api = database_admin_client.DatabaseAdminClient(mock.Mock())
        client = _Client(self.PROJECT)
        client.database_admin_api = api
        instance = self._make_one(self.INSTANCE_ID, client)

        databases_pb = spanner_database_admin_pb2.ListDatabasesResponse(databases=[])

        ld_api = api._inner_api_calls["list_databases"] = mock.Mock(
            return_value=databases_pb
        )

        page_size = 42
        page_token = "token"
        response = instance.list_databases(page_size=page_size, page_token=page_token)
        databases = list(response)

        self.assertEqual(databases, [])

        expected_metadata = [
            ("google-cloud-resource-prefix", instance.name),
            ("x-goog-request-params", "parent={}".format(instance.name)),
        ]
        ld_api.assert_called_once_with(
            spanner_database_admin_pb2.ListDatabasesRequest(
                parent=self.INSTANCE_NAME, page_size=page_size, page_token=page_token
            ),
            metadata=expected_metadata,
            retry=mock.ANY,
            timeout=mock.ANY,
        )


class _Client(object):
    def __init__(self, project, timeout_seconds=None):
        self.project = project
        self.project_name = "projects/" + self.project
        self.timeout_seconds = timeout_seconds

    def copy(self):
        from copy import deepcopy

        return deepcopy(self)

    def __eq__(self, other):
        return (
            other.project == self.project
            and other.project_name == self.project_name
            and other.timeout_seconds == self.timeout_seconds
        )


class _FauxInstanceAdminAPI(object):

    _create_instance_conflict = False
    _instance_not_found = False
    _rpc_error = False

    def __init__(self, **kwargs):
        self.__dict__.update(**kwargs)

    def create_instance(self, parent, instance_id, instance, metadata=None):
        from google.api_core.exceptions import AlreadyExists, Unknown

        self._created_instance = (parent, instance_id, instance, metadata)
        if self._rpc_error:
            raise Unknown("error")
        if self._create_instance_conflict:
            raise AlreadyExists("conflict")
        return self._create_instance_response

    def get_instance(self, name, metadata=None):
        from google.api_core.exceptions import NotFound, Unknown

        self._got_instance = (name, metadata)
        if self._rpc_error:
            raise Unknown("error")
        if self._instance_not_found:
            raise NotFound("error")
        return self._get_instance_response

    def update_instance(self, instance, field_mask, metadata=None):
        from google.api_core.exceptions import NotFound, Unknown

        self._updated_instance = (instance, field_mask, metadata)
        if self._rpc_error:
            raise Unknown("error")
        if self._instance_not_found:
            raise NotFound("error")
        return self._update_instance_response

    def delete_instance(self, name, metadata=None):
        from google.api_core.exceptions import NotFound, Unknown

        self._deleted_instance = name, metadata
        if self._rpc_error:
            raise Unknown("error")
        if self._instance_not_found:
            raise NotFound("error")
        return self._delete_instance_response


class _FauxOperationFuture(object):
    pass


class _Pool(object):
    _bound = None

    def bind(self, database):
        self._bound = database
