# Copyright 2016, Google Inc.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met:
#
#     * Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
#     * Redistributions in binary form must reproduce the above
# copyright notice, this list of conditions and the following disclaimer
# in the documentation and/or other materials provided with the
# distribution.
#     * Neither the name of Google Inc. nor the names of its
# contributors may be used to endorse or promote products derived from
# this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
"""Test of RPCs made against gRPC Python's application-layer API."""

import unittest

import grpc

from tests.unit.framework.common import test_constants

_SERIALIZE_REQUEST = lambda bytestring: bytestring * 2
_DESERIALIZE_REQUEST = lambda bytestring: bytestring[len(bytestring) // 2:]
_SERIALIZE_RESPONSE = lambda bytestring: bytestring * 3
_DESERIALIZE_RESPONSE = lambda bytestring: bytestring[:len(bytestring) // 3]

_UNARY_UNARY = '/test/UnaryUnary'
_UNARY_STREAM = '/test/UnaryStream'
_STREAM_UNARY = '/test/StreamUnary'
_STREAM_STREAM = '/test/StreamStream'


def _unary_unary_multi_callable(channel):
    return channel.unary_unary(_UNARY_UNARY)


def _unary_stream_multi_callable(channel):
    return channel.unary_stream(
        _UNARY_STREAM,
        request_serializer=_SERIALIZE_REQUEST,
        response_deserializer=_DESERIALIZE_RESPONSE)


def _stream_unary_multi_callable(channel):
    return channel.stream_unary(
        _STREAM_UNARY,
        request_serializer=_SERIALIZE_REQUEST,
        response_deserializer=_DESERIALIZE_RESPONSE)


def _stream_stream_multi_callable(channel):
    return channel.stream_stream(_STREAM_STREAM)


class InvalidMetadataTest(unittest.TestCase):

    def setUp(self):
        self._channel = grpc.insecure_channel('localhost:8080')
        self._unary_unary = _unary_unary_multi_callable(self._channel)
        self._unary_stream = _unary_stream_multi_callable(self._channel)
        self._stream_unary = _stream_unary_multi_callable(self._channel)
        self._stream_stream = _stream_stream_multi_callable(self._channel)

    def testUnaryRequestBlockingUnaryResponse(self):
        request = b'\x07\x08'
        metadata = (('InVaLiD', 'UnaryRequestBlockingUnaryResponse'),)
        expected_error_details = "metadata was invalid: %s" % metadata
        with self.assertRaises(ValueError) as exception_context:
            self._unary_unary(request, metadata=metadata)
        self.assertIn(expected_error_details, str(exception_context.exception))

    def testUnaryRequestBlockingUnaryResponseWithCall(self):
        request = b'\x07\x08'
        metadata = (('InVaLiD', 'UnaryRequestBlockingUnaryResponseWithCall'),)
        expected_error_details = "metadata was invalid: %s" % metadata
        with self.assertRaises(ValueError) as exception_context:
            self._unary_unary.with_call(request, metadata=metadata)
        self.assertIn(expected_error_details, str(exception_context.exception))

    def testUnaryRequestFutureUnaryResponse(self):
        request = b'\x07\x08'
        metadata = (('InVaLiD', 'UnaryRequestFutureUnaryResponse'),)
        expected_error_details = "metadata was invalid: %s" % metadata
        response_future = self._unary_unary.future(request, metadata=metadata)
        with self.assertRaises(grpc.RpcError) as exception_context:
            response_future.result()
        self.assertEqual(exception_context.exception.details(),
                         expected_error_details)
        self.assertEqual(exception_context.exception.code(),
                         grpc.StatusCode.INTERNAL)
        self.assertEqual(response_future.details(), expected_error_details)
        self.assertEqual(response_future.code(), grpc.StatusCode.INTERNAL)

    def testUnaryRequestStreamResponse(self):
        request = b'\x37\x58'
        metadata = (('InVaLiD', 'UnaryRequestStreamResponse'),)
        expected_error_details = "metadata was invalid: %s" % metadata
        response_iterator = self._unary_stream(request, metadata=metadata)
        with self.assertRaises(grpc.RpcError) as exception_context:
            next(response_iterator)
        self.assertEqual(exception_context.exception.details(),
                         expected_error_details)
        self.assertEqual(exception_context.exception.code(),
                         grpc.StatusCode.INTERNAL)
        self.assertEqual(response_iterator.details(), expected_error_details)
        self.assertEqual(response_iterator.code(), grpc.StatusCode.INTERNAL)

    def testStreamRequestBlockingUnaryResponse(self):
        request_iterator = (b'\x07\x08'
                            for _ in range(test_constants.STREAM_LENGTH))
        metadata = (('InVaLiD', 'StreamRequestBlockingUnaryResponse'),)
        expected_error_details = "metadata was invalid: %s" % metadata
        with self.assertRaises(ValueError) as exception_context:
            self._stream_unary(request_iterator, metadata=metadata)
        self.assertIn(expected_error_details, str(exception_context.exception))

    def testStreamRequestBlockingUnaryResponseWithCall(self):
        request_iterator = (b'\x07\x08'
                            for _ in range(test_constants.STREAM_LENGTH))
        metadata = (('InVaLiD', 'StreamRequestBlockingUnaryResponseWithCall'),)
        expected_error_details = "metadata was invalid: %s" % metadata
        multi_callable = _stream_unary_multi_callable(self._channel)
        with self.assertRaises(ValueError) as exception_context:
            multi_callable.with_call(request_iterator, metadata=metadata)
        self.assertIn(expected_error_details, str(exception_context.exception))

    def testStreamRequestFutureUnaryResponse(self):
        request_iterator = (b'\x07\x08'
                            for _ in range(test_constants.STREAM_LENGTH))
        metadata = (('InVaLiD', 'StreamRequestFutureUnaryResponse'),)
        expected_error_details = "metadata was invalid: %s" % metadata
        response_future = self._stream_unary.future(
            request_iterator, metadata=metadata)
        with self.assertRaises(grpc.RpcError) as exception_context:
            response_future.result()
        self.assertEqual(exception_context.exception.details(),
                         expected_error_details)
        self.assertEqual(exception_context.exception.code(),
                         grpc.StatusCode.INTERNAL)
        self.assertEqual(response_future.details(), expected_error_details)
        self.assertEqual(response_future.code(), grpc.StatusCode.INTERNAL)

    def testStreamRequestStreamResponse(self):
        request_iterator = (b'\x07\x08'
                            for _ in range(test_constants.STREAM_LENGTH))
        metadata = (('InVaLiD', 'StreamRequestStreamResponse'),)
        expected_error_details = "metadata was invalid: %s" % metadata
        response_iterator = self._stream_stream(
            request_iterator, metadata=metadata)
        with self.assertRaises(grpc.RpcError) as exception_context:
            next(response_iterator)
        self.assertEqual(exception_context.exception.details(),
                         expected_error_details)
        self.assertEqual(exception_context.exception.code(),
                         grpc.StatusCode.INTERNAL)
        self.assertEqual(response_iterator.details(), expected_error_details)
        self.assertEqual(response_iterator.code(), grpc.StatusCode.INTERNAL)


if __name__ == '__main__':
    unittest.main(verbosity=2)