# Copyright 2015, 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 the GRPC-backed ForeLink and RearLink."""

import threading
import unittest

from grpc._adapter import _proto_scenarios
from grpc._adapter import _test_links
from grpc._adapter import fore
from grpc._adapter import rear
from grpc.framework.base import interfaces
from grpc.framework.base.packets import packets as tickets
from grpc.framework.foundation import logging_pool

_IDENTITY = lambda x: x
_TIMEOUT = 2


class RoundTripTest(unittest.TestCase):

  def setUp(self):
    self.fore_link_pool = logging_pool.pool(80)
    self.rear_link_pool = logging_pool.pool(80)

  def tearDown(self):
    self.rear_link_pool.shutdown(wait=True)
    self.fore_link_pool.shutdown(wait=True)

  def testZeroMessageRoundTrip(self):
    test_operation_id = object()
    test_method = 'test method'
    test_fore_link = _test_links.ForeLink(None, None)
    def rear_action(front_to_back_ticket, fore_link):
      if front_to_back_ticket.kind in (
          tickets.Kind.COMPLETION, tickets.Kind.ENTIRE):
        back_to_front_ticket = tickets.BackToFrontPacket(
            front_to_back_ticket.operation_id, 0, tickets.Kind.COMPLETION, None)
        fore_link.accept_back_to_front_ticket(back_to_front_ticket)
    test_rear_link = _test_links.RearLink(rear_action, None)

    fore_link = fore.ForeLink(
        self.fore_link_pool, {test_method: None}, {test_method: None}, None, ())
    fore_link.join_rear_link(test_rear_link)
    test_rear_link.join_fore_link(fore_link)
    fore_link.start()
    port = fore_link.port()

    rear_link = rear.RearLink(
        'localhost', port, self.rear_link_pool, {test_method: None},
        {test_method: None}, False, None, None, None)
    rear_link.join_fore_link(test_fore_link)
    test_fore_link.join_rear_link(rear_link)
    rear_link.start()

    front_to_back_ticket = tickets.FrontToBackPacket(
        test_operation_id, 0, tickets.Kind.ENTIRE, test_method,
        interfaces.ServicedSubscription.Kind.FULL, None, None, _TIMEOUT)
    rear_link.accept_front_to_back_ticket(front_to_back_ticket)

    with test_fore_link.condition:
      while (not test_fore_link.tickets or
             test_fore_link.tickets[-1].kind is tickets.Kind.CONTINUATION):
        test_fore_link.condition.wait()

    rear_link.stop()
    fore_link.stop()

    with test_fore_link.condition:
      self.assertIs(test_fore_link.tickets[-1].kind, tickets.Kind.COMPLETION)

  def testEntireRoundTrip(self):
    test_operation_id = object()
    test_method = 'test method'
    test_front_to_back_datum = b'\x07'
    test_back_to_front_datum = b'\x08'
    test_fore_link = _test_links.ForeLink(None, None)
    rear_sequence_number = [0]
    def rear_action(front_to_back_ticket, fore_link):
      if front_to_back_ticket.payload is None:
        payload = None
      else:
        payload = test_back_to_front_datum
      terminal = front_to_back_ticket.kind in (
          tickets.Kind.COMPLETION, tickets.Kind.ENTIRE)
      if payload is not None or terminal:
        back_to_front_ticket = tickets.BackToFrontPacket(
            front_to_back_ticket.operation_id, rear_sequence_number[0],
            tickets.Kind.COMPLETION if terminal else tickets.Kind.CONTINUATION,
            payload)
        rear_sequence_number[0] += 1
        fore_link.accept_back_to_front_ticket(back_to_front_ticket)
    test_rear_link = _test_links.RearLink(rear_action, None)

    fore_link = fore.ForeLink(
        self.fore_link_pool, {test_method: _IDENTITY},
        {test_method: _IDENTITY}, None, ())
    fore_link.join_rear_link(test_rear_link)
    test_rear_link.join_fore_link(fore_link)
    fore_link.start()
    port = fore_link.port()

    rear_link = rear.RearLink(
        'localhost', port, self.rear_link_pool, {test_method: _IDENTITY},
        {test_method: _IDENTITY}, False, None, None, None)
    rear_link.join_fore_link(test_fore_link)
    test_fore_link.join_rear_link(rear_link)
    rear_link.start()

    front_to_back_ticket = tickets.FrontToBackPacket(
        test_operation_id, 0, tickets.Kind.ENTIRE, test_method,
        interfaces.ServicedSubscription.Kind.FULL, None,
        test_front_to_back_datum, _TIMEOUT)
    rear_link.accept_front_to_back_ticket(front_to_back_ticket)

    with test_fore_link.condition:
      while (not test_fore_link.tickets or
             test_fore_link.tickets[-1].kind is not tickets.Kind.COMPLETION):
        test_fore_link.condition.wait()

    rear_link.stop()
    fore_link.stop()

    with test_rear_link.condition:
      front_to_back_payloads = tuple(
          ticket.payload for ticket in test_rear_link.tickets
          if ticket.payload is not None)
    with test_fore_link.condition:
      back_to_front_payloads = tuple(
          ticket.payload for ticket in test_fore_link.tickets
          if ticket.payload is not None)
    self.assertTupleEqual((test_front_to_back_datum,), front_to_back_payloads)
    self.assertTupleEqual((test_back_to_front_datum,), back_to_front_payloads)

  def _perform_scenario_test(self, scenario):
    test_operation_id = object()
    test_method = scenario.method()
    test_fore_link = _test_links.ForeLink(None, None)
    rear_lock = threading.Lock()
    rear_sequence_number = [0]
    def rear_action(front_to_back_ticket, fore_link):
      with rear_lock:
        if front_to_back_ticket.payload is not None:
          response = scenario.response_for_request(front_to_back_ticket.payload)
        else:
          response = None
      terminal = front_to_back_ticket.kind in (
          tickets.Kind.COMPLETION, tickets.Kind.ENTIRE)
      if response is not None or terminal:
        back_to_front_ticket = tickets.BackToFrontPacket(
            front_to_back_ticket.operation_id, rear_sequence_number[0],
            tickets.Kind.COMPLETION if terminal else tickets.Kind.CONTINUATION,
            response)
        rear_sequence_number[0] += 1
        fore_link.accept_back_to_front_ticket(back_to_front_ticket)
    test_rear_link = _test_links.RearLink(rear_action, None)

    fore_link = fore.ForeLink(
        self.fore_link_pool, {test_method: scenario.deserialize_request},
        {test_method: scenario.serialize_response}, None, ())
    fore_link.join_rear_link(test_rear_link)
    test_rear_link.join_fore_link(fore_link)
    fore_link.start()
    port = fore_link.port()

    rear_link = rear.RearLink(
        'localhost', port, self.rear_link_pool,
        {test_method: scenario.serialize_request},
        {test_method: scenario.deserialize_response}, False, None, None, None)
    rear_link.join_fore_link(test_fore_link)
    test_fore_link.join_rear_link(rear_link)
    rear_link.start()

    commencement_ticket = tickets.FrontToBackPacket(
        test_operation_id, 0, tickets.Kind.COMMENCEMENT, test_method,
        interfaces.ServicedSubscription.Kind.FULL, None, None, _TIMEOUT)
    fore_sequence_number = 1
    rear_link.accept_front_to_back_ticket(commencement_ticket)
    for request in scenario.requests():
      continuation_ticket = tickets.FrontToBackPacket(
          test_operation_id, fore_sequence_number, tickets.Kind.CONTINUATION,
          None, None, None, request, None)
      fore_sequence_number += 1
      rear_link.accept_front_to_back_ticket(continuation_ticket)
    completion_ticket = tickets.FrontToBackPacket(
        test_operation_id, fore_sequence_number, tickets.Kind.COMPLETION, None,
        None, None, None, None)
    fore_sequence_number += 1
    rear_link.accept_front_to_back_ticket(completion_ticket)

    with test_fore_link.condition:
      while (not test_fore_link.tickets or
             test_fore_link.tickets[-1].kind is not tickets.Kind.COMPLETION):
        test_fore_link.condition.wait()

    rear_link.stop()
    fore_link.stop()

    with test_rear_link.condition:
      requests = tuple(
          ticket.payload for ticket in test_rear_link.tickets
          if ticket.payload is not None)
    with test_fore_link.condition:
      responses = tuple(
          ticket.payload for ticket in test_fore_link.tickets
          if ticket.payload is not None)
    self.assertTrue(scenario.verify_requests(requests))
    self.assertTrue(scenario.verify_responses(responses))

  def testEmptyScenario(self):
    self._perform_scenario_test(_proto_scenarios.EmptyScenario())

  def testBidirectionallyUnaryScenario(self):
    self._perform_scenario_test(_proto_scenarios.BidirectionallyUnaryScenario())

  def testBidirectionallyStreamingScenario(self):
    self._perform_scenario_test(
        _proto_scenarios.BidirectionallyStreamingScenario())


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

[((249, 2, 249, 17), 'unittest.main', 'unittest.main', ({}, {}), '()', False, 'import unittest\n'), ((50, 26, 50, 47), 'grpc.framework.foundation.logging_pool.pool', 'logging_pool.pool', ({(50, 44, 50, 46): '80'}, {}), '(80)', False, 'from grpc.framework.foundation import logging_pool\n'), ((51, 26, 51, 47), 'grpc.framework.foundation.logging_pool.pool', 'logging_pool.pool', ({(51, 44, 51, 46): '80'}, {}), '(80)', False, 'from grpc.framework.foundation import logging_pool\n'), ((60, 21, 60, 53), 'grpc._adapter._test_links.ForeLink', '_test_links.ForeLink', ({(60, 42, 60, 46): 'None', (60, 48, 60, 52): 'None'}, {}), '(None, None)', False, 'from grpc._adapter import _test_links\n'), ((67, 21, 67, 60), 'grpc._adapter._test_links.RearLink', '_test_links.RearLink', ({(67, 42, 67, 53): 'rear_action', (67, 55, 67, 59): 'None'}, {}), '(rear_action, None)', False, 'from grpc._adapter import _test_links\n'), ((69, 16, 70, 80), 'grpc._adapter.fore.ForeLink', 'fore.ForeLink', ({(70, 8, 70, 27): 'self.fore_link_pool', (70, 29, 70, 48): '{test_method: None}', (70, 50, 70, 69): '{test_method: None}', (70, 71, 70, 75): 'None', (70, 77, 70, 79): '()'}, {}), '(self.fore_link_pool, {test_method: None}, {test_method: None},\n    None, ())', False, 'from grpc._adapter import fore\n'), ((76, 16, 78, 53), 'grpc._adapter.rear.RearLink', 'rear.RearLink', ({(77, 8, 77, 19): '"""localhost"""', (77, 21, 77, 25): 'port', (77, 27, 77, 46): 'self.rear_link_pool', (77, 48, 77, 67): '{test_method: None}', (78, 8, 78, 27): '{test_method: None}', (78, 29, 78, 34): 'False', (78, 36, 78, 40): 'None', (78, 42, 78, 46): 'None', (78, 48, 78, 52): 'None'}, {}), "('localhost', port, self.rear_link_pool, {test_method: None},\n    {test_method: None}, False, None, None, None)", False, 'from grpc._adapter import rear\n'), ((83, 27, 85, 72), 'grpc.framework.base.packets.packets.FrontToBackPacket', 'tickets.FrontToBackPacket', ({(84, 8, 84, 25): 'test_operation_id', (84, 27, 84, 28): '0', (84, 30, 84, 49): 'tickets.Kind.ENTIRE', (84, 51, 84, 62): 'test_method', (85, 8, 85, 49): 'interfaces.ServicedSubscription.Kind.FULL', (85, 51, 85, 55): 'None', (85, 57, 85, 61): 'None', (85, 63, 85, 71): '_TIMEOUT'}, {}), '(test_operation_id, 0, tickets.Kind.ENTIRE,\n    test_method, interfaces.ServicedSubscription.Kind.FULL, None, None,\n    _TIMEOUT)', True, 'from grpc.framework.base.packets import packets as tickets\n'), ((104, 21, 104, 53), 'grpc._adapter._test_links.ForeLink', '_test_links.ForeLink', ({(104, 42, 104, 46): 'None', (104, 48, 104, 52): 'None'}, {}), '(None, None)', False, 'from grpc._adapter import _test_links\n'), ((120, 21, 120, 60), 'grpc._adapter._test_links.RearLink', '_test_links.RearLink', ({(120, 42, 120, 53): 'rear_action', (120, 55, 120, 59): 'None'}, {}), '(rear_action, None)', False, 'from grpc._adapter import _test_links\n'), ((122, 16, 124, 43), 'grpc._adapter.fore.ForeLink', 'fore.ForeLink', ({(123, 8, 123, 27): 'self.fore_link_pool', (123, 29, 123, 53): '{test_method: _IDENTITY}', (124, 8, 124, 32): '{test_method: _IDENTITY}', (124, 34, 124, 38): 'None', (124, 40, 124, 42): '()'}, {}), '(self.fore_link_pool, {test_method: _IDENTITY}, {test_method:\n    _IDENTITY}, None, ())', False, 'from grpc._adapter import fore\n'), ((130, 16, 132, 58), 'grpc._adapter.rear.RearLink', 'rear.RearLink', ({(131, 8, 131, 19): '"""localhost"""', (131, 21, 131, 25): 'port', (131, 27, 131, 46): 'self.rear_link_pool', (131, 48, 131, 72): '{test_method: _IDENTITY}', (132, 8, 132, 32): '{test_method: _IDENTITY}', (132, 34, 132, 39): 'False', (132, 41, 132, 45): 'None', (132, 47, 132, 51): 'None', (132, 53, 132, 57): 'None'}, {}), "('localhost', port, self.rear_link_pool, {test_method:\n    _IDENTITY}, {test_method: _IDENTITY}, False, None, None, None)", False, 'from grpc._adapter import rear\n'), ((137, 27, 140, 43), 'grpc.framework.base.packets.packets.FrontToBackPacket', 'tickets.FrontToBackPacket', ({(138, 8, 138, 25): 'test_operation_id', (138, 27, 138, 28): '0', (138, 30, 138, 49): 'tickets.Kind.ENTIRE', (138, 51, 138, 62): 'test_method', (139, 8, 139, 49): 'interfaces.ServicedSubscription.Kind.FULL', (139, 51, 139, 55): 'None', (140, 8, 140, 32): 'test_front_to_back_datum', (140, 34, 140, 42): '_TIMEOUT'}, {}), '(test_operation_id, 0, tickets.Kind.ENTIRE,\n    test_method, interfaces.ServicedSubscription.Kind.FULL, None,\n    test_front_to_back_datum, _TIMEOUT)', True, 'from grpc.framework.base.packets import packets as tickets\n'), ((165, 21, 165, 53), 'grpc._adapter._test_links.ForeLink', '_test_links.ForeLink', ({(165, 42, 165, 46): 'None', (165, 48, 165, 52): 'None'}, {}), '(None, None)', False, 'from grpc._adapter import _test_links\n'), ((166, 16, 166, 32), 'threading.Lock', 'threading.Lock', ({}, {}), '()', False, 'import threading\n'), ((183, 21, 183, 60), 'grpc._adapter._test_links.RearLink', '_test_links.RearLink', ({(183, 42, 183, 53): 'rear_action', (183, 55, 183, 59): 'None'}, {}), '(rear_action, None)', False, 'from grpc._adapter import _test_links\n'), ((185, 16, 187, 61), 'grpc._adapter.fore.ForeLink', 'fore.ForeLink', ({(186, 8, 186, 27): 'self.fore_link_pool', (186, 29, 186, 72): '{test_method: scenario.deserialize_request}', (187, 8, 187, 50): '{test_method: scenario.serialize_response}', (187, 52, 187, 56): 'None', (187, 58, 187, 60): '()'}, {}), '(self.fore_link_pool, {test_method: scenario.\n    deserialize_request}, {test_method: scenario.serialize_response}, None, ())', False, 'from grpc._adapter import fore\n'), ((193, 16, 196, 78), 'grpc._adapter.rear.RearLink', 'rear.RearLink', ({(194, 8, 194, 19): '"""localhost"""', (194, 21, 194, 25): 'port', (194, 27, 194, 46): 'self.rear_link_pool', (195, 8, 195, 49): '{test_method: scenario.serialize_request}', (196, 8, 196, 52): '{test_method: scenario.deserialize_response}', (196, 54, 196, 59): 'False', (196, 61, 196, 65): 'None', (196, 67, 196, 71): 'None', (196, 73, 196, 77): 'None'}, {}), "('localhost', port, self.rear_link_pool, {test_method:\n    scenario.serialize_request}, {test_method: scenario.\n    deserialize_response}, False, None, None, None)", False, 'from grpc._adapter import rear\n'), ((201, 26, 203, 72), 'grpc.framework.base.packets.packets.FrontToBackPacket', 'tickets.FrontToBackPacket', ({(202, 8, 202, 25): 'test_operation_id', (202, 27, 202, 28): '0', (202, 30, 202, 55): 'tickets.Kind.COMMENCEMENT', (202, 57, 202, 68): 'test_method', (203, 8, 203, 49): 'interfaces.ServicedSubscription.Kind.FULL', (203, 51, 203, 55): 'None', (203, 57, 203, 61): 'None', (203, 63, 203, 71): '_TIMEOUT'}, {}), '(test_operation_id, 0, tickets.Kind.COMMENCEMENT,\n    test_method, interfaces.ServicedSubscription.Kind.FULL, None, None,\n    _TIMEOUT)', True, 'from grpc.framework.base.packets import packets as tickets\n'), ((212, 24, 214, 31), 'grpc.framework.base.packets.packets.FrontToBackPacket', 'tickets.FrontToBackPacket', ({(213, 8, 213, 25): 'test_operation_id', (213, 27, 213, 47): 'fore_sequence_number', (213, 49, 213, 72): 'tickets.Kind.COMPLETION', (213, 74, 213, 78): 'None', (214, 8, 214, 12): 'None', (214, 14, 214, 18): 'None', (214, 20, 214, 24): 'None', (214, 26, 214, 30): 'None'}, {}), '(test_operation_id, fore_sequence_number, tickets.\n    Kind.COMPLETION, None, None, None, None, None)', True, 'from grpc.framework.base.packets import packets as tickets\n'), ((207, 28, 209, 42), 'grpc.framework.base.packets.packets.FrontToBackPacket', 'tickets.FrontToBackPacket', ({(208, 10, 208, 27): 'test_operation_id', (208, 29, 208, 49): 'fore_sequence_number', (208, 51, 208, 76): 'tickets.Kind.CONTINUATION', (209, 10, 209, 14): 'None', (209, 16, 209, 20): 'None', (209, 22, 209, 26): 'None', (209, 28, 209, 35): 'request', (209, 37, 209, 41): 'None'}, {}), '(test_operation_id, fore_sequence_number, tickets.\n    Kind.CONTINUATION, None, None, None, request, None)', True, 'from grpc.framework.base.packets import packets as tickets\n'), ((238, 32, 238, 64), 'grpc._adapter._proto_scenarios.EmptyScenario', '_proto_scenarios.EmptyScenario', ({}, {}), '()', False, 'from grpc._adapter import _proto_scenarios\n'), ((241, 32, 241, 79), 'grpc._adapter._proto_scenarios.BidirectionallyUnaryScenario', '_proto_scenarios.BidirectionallyUnaryScenario', ({}, {}), '()', False, 'from grpc._adapter import _proto_scenarios\n'), ((245, 8, 245, 59), 'grpc._adapter._proto_scenarios.BidirectionallyStreamingScenario', '_proto_scenarios.BidirectionallyStreamingScenario', ({}, {}), '()', False, 'from grpc._adapter import _proto_scenarios\n'), ((64, 31, 65, 80), 'grpc.framework.base.packets.packets.BackToFrontPacket', 'tickets.BackToFrontPacket', ({(65, 12, 65, 45): 'front_to_back_ticket.operation_id', (65, 47, 65, 48): '0', (65, 50, 65, 73): 'tickets.Kind.COMPLETION', (65, 75, 65, 79): 'None'}, {}), '(front_to_back_ticket.operation_id, 0, tickets.\n    Kind.COMPLETION, None)', True, 'from grpc.framework.base.packets import packets as tickets\n'), ((114, 31, 117, 20), 'grpc.framework.base.packets.packets.BackToFrontPacket', 'tickets.BackToFrontPacket', ({(115, 12, 115, 45): 'front_to_back_ticket.operation_id', (115, 47, 115, 70): 'rear_sequence_number[0]', (116, 12, 116, 78): 'tickets.Kind.COMPLETION if terminal else tickets.Kind.CONTINUATION', (117, 12, 117, 19): 'payload'}, {}), '(front_to_back_ticket.operation_id,\n    rear_sequence_number[0], tickets.Kind.COMPLETION if terminal else\n    tickets.Kind.CONTINUATION, payload)', True, 'from grpc.framework.base.packets import packets as tickets\n'), ((177, 31, 180, 21), 'grpc.framework.base.packets.packets.BackToFrontPacket', 'tickets.BackToFrontPacket', ({(178, 12, 178, 45): 'front_to_back_ticket.operation_id', (178, 47, 178, 70): 'rear_sequence_number[0]', (179, 12, 179, 78): 'tickets.Kind.COMPLETION if terminal else tickets.Kind.CONTINUATION', (180, 12, 180, 20): 'response'}, {}), '(front_to_back_ticket.operation_id,\n    rear_sequence_number[0], tickets.Kind.COMPLETION if terminal else\n    tickets.Kind.CONTINUATION, response)', True, 'from grpc.framework.base.packets import packets as tickets\n')]