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

from google.api_core import grpc_helpers
import google.auth.credentials
from google.protobuf import empty_pb2
import mock

import google.cloud.logging
from google.cloud.logging import _gapic
from google.cloud.logging_v2.gapic import config_service_v2_client
from google.cloud.logging_v2.gapic import logging_service_v2_client
from google.cloud.logging_v2.gapic import metrics_service_v2_client
from google.cloud.logging_v2.proto import log_entry_pb2
from google.cloud.logging_v2.proto import logging_pb2
from google.cloud.logging_v2.proto import logging_config_pb2
from google.cloud.logging_v2.proto import logging_metrics_pb2


PROJECT = 'PROJECT'
PROJECT_PATH = 'projects/%s' % (PROJECT,)
FILTER = 'logName:syslog AND severity>=ERROR'


class Test_LoggingAPI(object):
    LOG_NAME = 'log_name'
    LOG_PATH = 'projects/%s/logs/%s' % (PROJECT, LOG_NAME)

    @staticmethod
    def make_logging_api():
        channel = grpc_helpers.ChannelStub()
        gapic_client = logging_service_v2_client.LoggingServiceV2Client(
            channel=channel)
        handwritten_client = mock.Mock()
        api = _gapic._LoggingAPI(gapic_client, handwritten_client)
        return channel, api

    def test_ctor(self):
        channel = grpc_helpers.ChannelStub()
        gapic_client = logging_service_v2_client.LoggingServiceV2Client(
            channel=channel)
        api = _gapic._LoggingAPI(gapic_client, mock.sentinel.client)
        assert api._gapic_api is gapic_client
        assert api._client is mock.sentinel.client

    def test_list_entries(self):
        channel, api = self.make_logging_api()

        log_entry_msg = log_entry_pb2.LogEntry(
            log_name=self.LOG_PATH,
            text_payload='text')
        channel.ListLogEntries.response = logging_pb2.ListLogEntriesResponse(
            entries=[log_entry_msg])
        result = api.list_entries(
            [PROJECT], FILTER, google.cloud.logging.DESCENDING)

        entries = list(result)

        # Check the response
        assert len(entries) == 1
        entry = entries[0]
        assert isinstance(entry, google.cloud.logging.entries.TextEntry)
        assert entry.payload == 'text'

        # Check the request
        assert len(channel.ListLogEntries.requests) == 1
        request = channel.ListLogEntries.requests[0]
        assert request.project_ids == [PROJECT]
        assert request.filter == FILTER
        assert request.order_by == google.cloud.logging.DESCENDING

    def test_list_entries_with_options(self):
        channel, api = self.make_logging_api()

        channel.ListLogEntries.response = logging_pb2.ListLogEntriesResponse(
            entries=[])

        result = api.list_entries(
            [PROJECT], FILTER, google.cloud.logging.ASCENDING, page_size=42,
            page_token='token')

        list(result)

        # Check the request
        assert len(channel.ListLogEntries.requests) == 1
        request = channel.ListLogEntries.requests[0]
        assert request.project_ids == [PROJECT]
        assert request.filter == FILTER
        assert request.order_by == google.cloud.logging.ASCENDING
        assert request.page_size == 42
        assert request.page_token == 'token'

    def test_write_entries_single(self):
        channel, api = self.make_logging_api()

        channel.WriteLogEntries.response = empty_pb2.Empty()

        entry = {
            'logName': self.LOG_PATH,
            'resource': {'type': 'global'},
            'textPayload': 'text',
        }

        api.write_entries([entry])

        # Check the request
        assert len(channel.WriteLogEntries.requests) == 1
        request = channel.WriteLogEntries.requests[0]
        assert request.partial_success is False
        assert len(request.entries) == 1
        assert request.entries[0].log_name == entry['logName']
        assert request.entries[0].resource.type == entry['resource']['type']
        assert request.entries[0].text_payload == 'text'

    def test_logger_delete(self):
        channel, api = self.make_logging_api()

        channel.DeleteLog.response = empty_pb2.Empty()

        api.logger_delete(PROJECT, self.LOG_NAME)

        assert len(channel.DeleteLog.requests) == 1
        request = channel.DeleteLog.requests[0]
        assert request.log_name == self.LOG_PATH


class Test_SinksAPI(object):
    SINK_NAME = 'sink_name'
    SINK_PATH = 'projects/%s/sinks/%s' % (PROJECT, SINK_NAME)
    DESTINATION_URI = 'faux.googleapis.com/destination'
    SINK_WRITER_IDENTITY = 'serviceAccount:project-123@example.com'

    @staticmethod
    def make_sinks_api():
        channel = grpc_helpers.ChannelStub()
        gapic_client = config_service_v2_client.ConfigServiceV2Client(
            channel=channel)
        handwritten_client = mock.Mock()
        api = _gapic._SinksAPI(gapic_client, handwritten_client)
        return channel, api

    def test_ctor(self):
        channel = grpc_helpers.ChannelStub()
        gapic_client = config_service_v2_client.ConfigServiceV2Client(
            channel=channel)
        api = _gapic._SinksAPI(gapic_client, mock.sentinel.client)
        assert api._gapic_api is gapic_client
        assert api._client is mock.sentinel.client

    def test_list_sinks(self):
        channel, api = self.make_sinks_api()

        sink_msg = logging_config_pb2.LogSink(
            name=self.SINK_PATH,
            destination=self.DESTINATION_URI,
            filter=FILTER)
        channel.ListSinks.response = logging_config_pb2.ListSinksResponse(
            sinks=[sink_msg])

        result = api.list_sinks(PROJECT)
        sinks = list(result)

        # Check the response
        assert len(sinks) == 1
        sink = sinks[0]
        assert isinstance(sink, google.cloud.logging.sink.Sink)
        assert sink.name == self.SINK_PATH
        assert sink.destination == self.DESTINATION_URI
        assert sink.filter_ == FILTER

        # Check the request
        assert len(channel.ListSinks.requests) == 1
        request = channel.ListSinks.requests[0]
        assert request.parent == PROJECT_PATH

    def test_list_sinks_with_options(self):
        channel, api = self.make_sinks_api()

        channel.ListSinks.response = logging_config_pb2.ListSinksResponse(
            sinks=[])

        result = api.list_sinks(PROJECT, page_size=42, page_token='token')
        list(result)

        # Check the request
        assert len(channel.ListSinks.requests) == 1
        request = channel.ListSinks.requests[0]
        assert request.parent == 'projects/%s' % PROJECT
        assert request.page_size == 42
        assert request.page_token == 'token'

    def test_sink_create(self):
        channel, api = self.make_sinks_api()

        channel.CreateSink.response = logging_config_pb2.LogSink(
            name=self.SINK_NAME,
            destination=self.DESTINATION_URI,
            filter=FILTER,
            writer_identity=self.SINK_WRITER_IDENTITY,
        )

        result = api.sink_create(
            PROJECT,
            self.SINK_NAME,
            FILTER,
            self.DESTINATION_URI,
            unique_writer_identity=True,
        )

        # Check response
        assert result == {
            'name': self.SINK_NAME,
            'filter': FILTER,
            'destination': self.DESTINATION_URI,
            'writerIdentity': self.SINK_WRITER_IDENTITY,
        }

        # Check request
        assert len(channel.CreateSink.requests) == 1
        request = channel.CreateSink.requests[0]
        assert request.parent == PROJECT_PATH
        assert request.unique_writer_identity is True
        assert request.sink.name == self.SINK_NAME
        assert request.sink.filter == FILTER
        assert request.sink.destination == self.DESTINATION_URI

    def test_sink_get(self):
        channel, api = self.make_sinks_api()

        channel.GetSink.response = logging_config_pb2.LogSink(
            name=self.SINK_PATH,
            destination=self.DESTINATION_URI,
            filter=FILTER)

        response = api.sink_get(PROJECT, self.SINK_NAME)

        # Check response
        assert response == {
            'name': self.SINK_PATH,
            'filter': FILTER,
            'destination': self.DESTINATION_URI
        }

        # Check request
        assert len(channel.GetSink.requests) == 1
        request = channel.GetSink.requests[0]
        assert request.sink_name == self.SINK_PATH

    def test_sink_update(self):
        channel, api = self.make_sinks_api()

        channel.UpdateSink.response = logging_config_pb2.LogSink(
            name=self.SINK_NAME,
            destination=self.DESTINATION_URI,
            filter=FILTER,
            writer_identity=self.SINK_WRITER_IDENTITY,
        )

        result = api.sink_update(
            PROJECT,
            self.SINK_NAME,
            FILTER,
            self.DESTINATION_URI,
            unique_writer_identity=True)

        # Check response
        assert result == {
            'name': self.SINK_NAME,
            'filter': FILTER,
            'destination': self.DESTINATION_URI,
            'writerIdentity': self.SINK_WRITER_IDENTITY,
        }

        # Check request
        assert len(channel.UpdateSink.requests) == 1
        request = channel.UpdateSink.requests[0]
        assert request.sink_name == self.SINK_PATH
        assert request.unique_writer_identity is True
        assert request.sink.name == self.SINK_PATH
        assert request.sink.filter == FILTER
        assert request.sink.destination == self.DESTINATION_URI

    def test_sink_delete(self):
        channel, api = self.make_sinks_api()

        channel.DeleteSink.response = empty_pb2.Empty()

        api.sink_delete(PROJECT, self.SINK_NAME)

        assert len(channel.DeleteSink.requests) == 1
        request = channel.DeleteSink.requests[0]
        assert request.sink_name == self.SINK_PATH


class Test_MetricsAPI(object):
    METRIC_NAME = 'metric_name'
    METRIC_PATH = 'projects/%s/metrics/%s' % (PROJECT, METRIC_NAME)
    DESCRIPTION = 'Description'

    @staticmethod
    def make_metrics_api():
        channel = grpc_helpers.ChannelStub()
        gapic_client = metrics_service_v2_client.MetricsServiceV2Client(
            channel=channel)
        handwritten_client = mock.Mock()
        api = _gapic._MetricsAPI(gapic_client, handwritten_client)
        return channel, api

    def test_ctor(self):
        channel = grpc_helpers.ChannelStub()
        gapic_client = metrics_service_v2_client.MetricsServiceV2Client(
            channel=channel)
        api = _gapic._MetricsAPI(gapic_client, mock.sentinel.client)
        assert api._gapic_api is gapic_client
        assert api._client is mock.sentinel.client

    def test_list_metrics(self):
        channel, api = self.make_metrics_api()

        sink_msg = logging_metrics_pb2.LogMetric(
            name=self.METRIC_PATH,
            description=self.DESCRIPTION,
            filter=FILTER)
        channel.ListLogMetrics.response = (
            logging_metrics_pb2.ListLogMetricsResponse(
                metrics=[sink_msg]))

        result = api.list_metrics(PROJECT)
        metrics = list(result)

        # Check the response
        assert len(metrics) == 1
        metric = metrics[0]
        assert isinstance(metric, google.cloud.logging.metric.Metric)
        assert metric.name == self.METRIC_PATH
        assert metric.description == self.DESCRIPTION
        assert metric.filter_ == FILTER

        # Check the request
        assert len(channel.ListLogMetrics.requests) == 1
        request = channel.ListLogMetrics.requests[0]
        assert request.parent == PROJECT_PATH

    def test_list_metrics_options(self):
        channel, api = self.make_metrics_api()

        channel.ListLogMetrics.response = (
            logging_metrics_pb2.ListLogMetricsResponse(
                metrics=[]))

        result = api.list_metrics(PROJECT, page_size=42, page_token='token')
        list(result)

        # Check the request
        assert len(channel.ListLogMetrics.requests) == 1
        request = channel.ListLogMetrics.requests[0]
        assert request.parent == PROJECT_PATH
        assert request.page_size == 42
        assert request.page_token == 'token'

    def test_metric_create(self):
        channel, api = self.make_metrics_api()

        channel.CreateLogMetric.response = empty_pb2.Empty()

        api.metric_create(
            PROJECT, self.METRIC_NAME, FILTER, self.DESCRIPTION)

        # Check the request
        assert len(channel.CreateLogMetric.requests) == 1
        request = channel.CreateLogMetric.requests[0]
        assert request.parent == PROJECT_PATH
        assert request.metric.name == self.METRIC_NAME
        assert request.metric.filter == FILTER
        assert request.metric.description == self.DESCRIPTION

    def test_metric_get(self):
        channel, api = self.make_metrics_api()

        channel.GetLogMetric.response = logging_metrics_pb2.LogMetric(
            name=self.METRIC_PATH,
            description=self.DESCRIPTION,
            filter=FILTER)

        response = api.metric_get(PROJECT, self.METRIC_NAME)

        # Check the response
        assert response == {
            'name': self.METRIC_PATH,
            'filter': FILTER,
            'description': self.DESCRIPTION,
        }

        # Check the request
        assert len(channel.GetLogMetric.requests) == 1
        request = channel.GetLogMetric.requests[0]
        assert request.metric_name == self.METRIC_PATH

    def test_metric_update(self):
        channel, api = self.make_metrics_api()

        channel.UpdateLogMetric.response = logging_metrics_pb2.LogMetric(
            name=self.METRIC_PATH,
            description=self.DESCRIPTION,
            filter=FILTER)

        response = api.metric_update(
            PROJECT, self.METRIC_NAME, FILTER, self.DESCRIPTION)

        # Check the response
        assert response == {
            'name': self.METRIC_PATH,
            'filter': FILTER,
            'description': self.DESCRIPTION,
        }

        # Check the request
        assert len(channel.UpdateLogMetric.requests) == 1
        request = channel.UpdateLogMetric.requests[0]
        assert request.metric_name == self.METRIC_PATH
        assert request.metric.name == self.METRIC_PATH
        assert request.metric.filter == FILTER
        assert request.metric.description == self.DESCRIPTION

    def test_metric_delete(self):
        channel, api = self.make_metrics_api()

        channel.DeleteLogMetric.response = empty_pb2.Empty()

        api.metric_delete(PROJECT, self.METRIC_NAME)

        assert len(channel.DeleteLogMetric.requests) == 1
        request = channel.DeleteLogMetric.requests[0]
        assert request.metric_name == self.METRIC_PATH


class Test__parse_log_entry(unittest.TestCase):

    @staticmethod
    def _call_fut(*args, **kwargs):
        from google.cloud.logging._gapic import _parse_log_entry

        return _parse_log_entry(*args, **kwargs)

    def test_simple(self):
        from google.cloud.logging_v2.proto.log_entry_pb2 import LogEntry

        entry_pb = LogEntry(log_name=u'lol-jk', text_payload=u'bah humbug')
        result = self._call_fut(entry_pb)
        expected = {
            'logName': entry_pb.log_name,
            'textPayload': entry_pb.text_payload,
        }
        self.assertEqual(result, expected)

    @mock.patch('google.cloud.logging._gapic.MessageToDict',
                side_effect=TypeError)
    def test_non_registry_failure(self, msg_to_dict_mock):
        entry_pb = mock.Mock(spec=['HasField'])
        entry_pb.HasField.return_value = False
        with self.assertRaises(TypeError):
            self._call_fut(entry_pb)

        entry_pb.HasField.assert_called_once_with('proto_payload')
        msg_to_dict_mock.assert_called_once_with(entry_pb)

    def test_unregistered_type(self):
        from google.cloud.logging_v2.proto.log_entry_pb2 import LogEntry
        from google.protobuf import any_pb2
        from google.protobuf import descriptor_pool
        from google.protobuf.timestamp_pb2 import Timestamp

        pool = descriptor_pool.Default()
        type_name = 'google.bigtable.admin.v2.UpdateClusterMetadata'
        # Make sure the descriptor is not known in the registry.
        with self.assertRaises(KeyError):
            pool.FindMessageTypeByName(type_name)

        type_url = 'type.googleapis.com/' + type_name
        metadata_bytes = (
            b'\n\n\n\x03foo\x12\x03bar\x12\x06\x08\xbd\xb6\xfb\xc6\x05')
        any_pb = any_pb2.Any(type_url=type_url, value=metadata_bytes)
        timestamp = Timestamp(seconds=61, nanos=1234000)

        entry_pb = LogEntry(proto_payload=any_pb, timestamp=timestamp)
        result = self._call_fut(entry_pb)
        self.assertEqual(len(result), 2)
        self.assertEqual(result['timestamp'], '1970-01-01T00:01:01.001234Z')
        # NOTE: This "hack" is needed on Windows, where the equality check
        #       for an ``Any`` instance fails on unregistered types.
        self.assertEqual(result['protoPayload'].type_url, type_url)
        self.assertEqual(result['protoPayload'].value, metadata_bytes)

    def test_registered_type(self):
        from google.cloud.logging_v2.proto.log_entry_pb2 import LogEntry
        from google.protobuf import any_pb2
        from google.protobuf import descriptor_pool
        from google.protobuf.struct_pb2 import Struct
        from google.protobuf.struct_pb2 import Value

        pool = descriptor_pool.Default()
        type_name = 'google.protobuf.Struct'
        # Make sure the descriptor is known in the registry.
        descriptor = pool.FindMessageTypeByName(type_name)
        self.assertEqual(descriptor.name, 'Struct')

        type_url = 'type.googleapis.com/' + type_name
        field_name = 'foo'
        field_value = u'Bar'
        struct_pb = Struct(
            fields={field_name: Value(string_value=field_value)})
        any_pb = any_pb2.Any(
            type_url=type_url,
            value=struct_pb.SerializeToString(),
        )

        entry_pb = LogEntry(proto_payload=any_pb, log_name=u'all-good')
        result = self._call_fut(entry_pb)
        expected_proto = {
            'logName': entry_pb.log_name,
            'protoPayload': {
                '@type': type_url,
                'value': {field_name: field_value},
            },
        }
        self.assertEqual(result, expected_proto)


class Test__log_entry_mapping_to_pb(unittest.TestCase):

    @staticmethod
    def _call_fut(*args, **kwargs):
        from google.cloud.logging._gapic import _log_entry_mapping_to_pb

        return _log_entry_mapping_to_pb(*args, **kwargs)

    def test_simple(self):
        from google.cloud.logging_v2.proto.log_entry_pb2 import LogEntry

        result = self._call_fut({})
        self.assertEqual(result, LogEntry())

    def test_unregistered_type(self):
        from google.protobuf import descriptor_pool
        from google.protobuf.json_format import ParseError

        pool = descriptor_pool.Default()
        type_name = 'google.bigtable.admin.v2.UpdateClusterMetadata'
        # Make sure the descriptor is not known in the registry.
        with self.assertRaises(KeyError):
            pool.FindMessageTypeByName(type_name)

        type_url = 'type.googleapis.com/' + type_name
        json_mapping = {
            'protoPayload': {
                '@type': type_url,
                'originalRequest': {
                    'name': 'foo',
                    'location': 'bar',
                },
                'requestTime': {
                    'seconds': 1491000125,
                },
            },
        }
        with self.assertRaises(ParseError):
            self._call_fut(json_mapping)

    def test_registered_type(self):
        from google.cloud.logging_v2.proto.log_entry_pb2 import LogEntry
        from google.protobuf import any_pb2
        from google.protobuf import descriptor_pool

        pool = descriptor_pool.Default()
        type_name = 'google.protobuf.Struct'
        # Make sure the descriptor is known in the registry.
        descriptor = pool.FindMessageTypeByName(type_name)
        self.assertEqual(descriptor.name, 'Struct')

        type_url = 'type.googleapis.com/' + type_name
        field_name = 'foo'
        field_value = u'Bar'
        json_mapping = {
            'logName': u'hi-everybody',
            'protoPayload': {
                '@type': type_url,
                'value': {field_name: field_value},
            },
        }
        # Convert to a valid LogEntry.
        result = self._call_fut(json_mapping)
        entry_pb = LogEntry(
            log_name=json_mapping['logName'],
            proto_payload=any_pb2.Any(
                type_url=type_url,
                value=b'\n\014\n\003foo\022\005\032\003Bar',
            ),
        )
        self.assertEqual(result, entry_pb)


@mock.patch(
    'google.cloud.logging._gapic.LoggingServiceV2Client', autospec=True)
def test_make_logging_api(gapic_client):
    client = mock.Mock(spec=['_credentials'])
    api = _gapic.make_logging_api(client)
    assert api._client == client
    assert api._gapic_api == gapic_client.return_value
    gapic_client.assert_called_once_with(
        credentials=client._credentials, client_info=_gapic._CLIENT_INFO)


@mock.patch(
    'google.cloud.logging._gapic.MetricsServiceV2Client', autospec=True)
def test_make_metrics_api(gapic_client):
    client = mock.Mock(spec=['_credentials'])
    api = _gapic.make_metrics_api(client)
    assert api._client == client
    assert api._gapic_api == gapic_client.return_value
    gapic_client.assert_called_once_with(
        credentials=client._credentials, client_info=_gapic._CLIENT_INFO)


@mock.patch(
    'google.cloud.logging._gapic.ConfigServiceV2Client', autospec=True)
def test_make_sinks_api(gapic_client):
    client = mock.Mock(spec=['_credentials'])
    api = _gapic.make_sinks_api(client)
    assert api._client == client
    assert api._gapic_api == gapic_client.return_value
    gapic_client.assert_called_once_with(
        credentials=client._credentials, client_info=_gapic._CLIENT_INFO)
