# Copyright 2017 gRPC authors.
#
# 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.
"""Test a corner-case at the level of the Cython API."""

import threading
import unittest

from grpc._cython import cygrpc

from tests.unit._cython import _common
from tests.unit._cython import test_utilities


class Test(_common.RpcTest, unittest.TestCase):

    def _do_rpcs(self):
        server_request_call_tag = 'server_request_call_tag'
        server_send_initial_metadata_tag = 'server_send_initial_metadata_tag'
        server_complete_rpc_tag = 'server_complete_rpc_tag'

        with self.server_condition:
            server_request_call_start_batch_result = self.server.request_call(
                self.server_completion_queue, self.server_completion_queue,
                server_request_call_tag)
            self.server_driver.add_due({
                server_request_call_tag,
            })

        client_receive_initial_metadata_tag = 'client_receive_initial_metadata_tag'
        client_complete_rpc_tag = 'client_complete_rpc_tag'
        client_call = self.channel.integrated_call(
            _common.EMPTY_FLAGS, b'/twinkies', None, None,
            _common.INVOCATION_METADATA, None, [
                (
                    [
                        cygrpc.SendInitialMetadataOperation(
                            _common.INVOCATION_METADATA, _common.EMPTY_FLAGS),
                        cygrpc.SendCloseFromClientOperation(
                            _common.EMPTY_FLAGS),
                        cygrpc.ReceiveStatusOnClientOperation(
                            _common.EMPTY_FLAGS),
                    ],
                    client_complete_rpc_tag,
                ),
            ])
        client_call.operate([
            cygrpc.ReceiveInitialMetadataOperation(_common.EMPTY_FLAGS),
        ], client_receive_initial_metadata_tag)

        client_events_future = test_utilities.SimpleFuture(
            lambda: [
                self.channel.next_call_event(),
                self.channel.next_call_event(),])
        server_request_call_event = self.server_driver.event_with_tag(
            server_request_call_tag)

        with self.server_condition:
            server_send_initial_metadata_start_batch_result = (
                server_request_call_event.call.start_server_batch([
                    cygrpc.SendInitialMetadataOperation(
                        _common.INITIAL_METADATA, _common.EMPTY_FLAGS),
                ], server_send_initial_metadata_tag))
            self.server_driver.add_due({
                server_send_initial_metadata_tag,
            })
        server_send_initial_metadata_event = self.server_driver.event_with_tag(
            server_send_initial_metadata_tag)

        with self.server_condition:
            server_complete_rpc_start_batch_result = (
                server_request_call_event.call.start_server_batch([
                    cygrpc.ReceiveCloseOnServerOperation(_common.EMPTY_FLAGS),
                    cygrpc.SendStatusFromServerOperation(
                        _common.TRAILING_METADATA, cygrpc.StatusCode.ok,
                        'test details', _common.EMPTY_FLAGS),
                ], server_complete_rpc_tag))
            self.server_driver.add_due({
                server_complete_rpc_tag,
            })
        server_complete_rpc_event = self.server_driver.event_with_tag(
            server_complete_rpc_tag)

        client_events = client_events_future.result()
        client_receive_initial_metadata_event = client_events[0]
        client_complete_rpc_event = client_events[1]

        return (
            _common.OperationResult(server_request_call_start_batch_result,
                                    server_request_call_event.completion_type,
                                    server_request_call_event.success),
            _common.OperationResult(
                cygrpc.CallError.ok,
                client_receive_initial_metadata_event.completion_type,
                client_receive_initial_metadata_event.success),
            _common.OperationResult(cygrpc.CallError.ok,
                                    client_complete_rpc_event.completion_type,
                                    client_complete_rpc_event.success),
            _common.OperationResult(
                server_send_initial_metadata_start_batch_result,
                server_send_initial_metadata_event.completion_type,
                server_send_initial_metadata_event.success),
            _common.OperationResult(server_complete_rpc_start_batch_result,
                                    server_complete_rpc_event.completion_type,
                                    server_complete_rpc_event.success),
        )

    def test_rpcs(self):
        expecteds = [(
            _common.SUCCESSFUL_OPERATION_RESULT,) * 5] * _common.RPC_COUNT
        actuallys = _common.execute_many_times(self._do_rpcs)
        self.assertSequenceEqual(expecteds, actuallys)


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