# 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. require 'grpc' def wakey_thread(&blk) n = GRPC::Notifier.new t = Thread.new do blk.call(n) end n.wait t end def load_test_certs test_root = File.join(File.dirname(File.dirname(__FILE__)), 'testdata') files = ['ca.pem', 'server1.key', 'server1.pem'] files.map { |f| File.open(File.join(test_root, f)).read } end include GRPC::Core::StatusCodes include GRPC::Core::TimeConsts include GRPC::Core::CallOps describe 'ClientStub' do let(:noop) { proc { |x| x } } before(:each) do Thread.abort_on_exception = true @server = nil @server_queue = nil @method = 'an_rpc_method' @pass = OK @fail = INTERNAL @cq = GRPC::Core::CompletionQueue.new end after(:each) do @server.close(@server_queue) unless @server_queue.nil? end describe '#new' do let(:fake_host) { 'localhost:0' } it 'can be created from a host and args' do opts = { channel_args: { a_channel_arg: 'an_arg' } } blk = proc do GRPC::ClientStub.new(fake_host, @cq, :this_channel_is_insecure, **opts) end expect(&blk).not_to raise_error end it 'can be created with an channel override' do opts = { channel_args: { a_channel_arg: 'an_arg' }, channel_override: @ch } blk = proc do GRPC::ClientStub.new(fake_host, @cq, :this_channel_is_insecure, **opts) end expect(&blk).not_to raise_error end it 'cannot be created with a bad channel override' do blk = proc do opts = { channel_args: { a_channel_arg: 'an_arg' }, channel_override: Object.new } GRPC::ClientStub.new(fake_host, @cq, :this_channel_is_insecure, **opts) end expect(&blk).to raise_error end it 'cannot be created with bad credentials' do blk = proc do opts = { channel_args: { a_channel_arg: 'an_arg' } } GRPC::ClientStub.new(fake_host, @cq, Object.new, **opts) end expect(&blk).to raise_error end it 'can be created with test test credentials' do certs = load_test_certs blk = proc do opts = { channel_args: { GRPC::Core::Channel::SSL_TARGET => 'foo.test.google.fr', a_channel_arg: 'an_arg' } } creds = GRPC::Core::ChannelCredentials.new(certs[0], nil, nil) GRPC::ClientStub.new(fake_host, @cq, creds, **opts) end expect(&blk).to_not raise_error end end describe '#request_response' do before(:each) do @sent_msg, @resp = 'a_msg', 'a_reply' end shared_examples 'request response' do it 'should send a request to/receive a reply from a server' do server_port = create_test_server th = run_request_response(@sent_msg, @resp, @pass) stub = GRPC::ClientStub.new("localhost:#{server_port}", @cq, :this_channel_is_insecure) expect(get_response(stub)).to eq(@resp) th.join end it 'should send metadata to the server ok' do server_port = create_test_server host = "localhost:#{server_port}" th = run_request_response(@sent_msg, @resp, @pass, k1: 'v1', k2: 'v2') stub = GRPC::ClientStub.new(host, @cq, :this_channel_is_insecure) expect(get_response(stub)).to eq(@resp) th.join end it 'should send a request when configured using an override channel' do server_port = create_test_server alt_host = "localhost:#{server_port}" th = run_request_response(@sent_msg, @resp, @pass) ch = GRPC::Core::Channel.new(alt_host, nil, :this_channel_is_insecure) stub = GRPC::ClientStub.new('ignored-host', @cq, :this_channel_is_insecure, channel_override: ch) expect(get_response(stub)).to eq(@resp) th.join end it 'should raise an error if the status is not OK' do server_port = create_test_server host = "localhost:#{server_port}" th = run_request_response(@sent_msg, @resp, @fail) stub = GRPC::ClientStub.new(host, @cq, :this_channel_is_insecure) blk = proc { get_response(stub) } expect(&blk).to raise_error(GRPC::BadStatus) th.join end end describe 'without a call operation' do def get_response(stub) stub.request_response(@method, @sent_msg, noop, noop, metadata: { k1: 'v1', k2: 'v2' }) end it_behaves_like 'request response' end describe 'via a call operation' do def get_response(stub) op = stub.request_response(@method, @sent_msg, noop, noop, return_op: true, metadata: { k1: 'v1', k2: 'v2' }) expect(op).to be_a(GRPC::ActiveCall::Operation) op.execute end it_behaves_like 'request response' end end describe '#client_streamer' do shared_examples 'client streaming' do before(:each) do server_port = create_test_server host = "localhost:#{server_port}" @stub = GRPC::ClientStub.new(host, @cq, :this_channel_is_insecure) @metadata = { k1: 'v1', k2: 'v2' } @sent_msgs = Array.new(3) { |i| 'msg_' + (i + 1).to_s } @resp = 'a_reply' end it 'should send requests to/receive a reply from a server' do th = run_client_streamer(@sent_msgs, @resp, @pass) expect(get_response(@stub)).to eq(@resp) th.join end it 'should send metadata to the server ok' do th = run_client_streamer(@sent_msgs, @resp, @pass, **@metadata) expect(get_response(@stub)).to eq(@resp) th.join end it 'should raise an error if the status is not ok' do th = run_client_streamer(@sent_msgs, @resp, @fail) blk = proc { get_response(@stub) } expect(&blk).to raise_error(GRPC::BadStatus) th.join end it 'should raise ArgumentError if metadata contains invalid values' do @metadata.merge!(k3: 3) expect do get_response(@stub) end.to raise_error(ArgumentError, /Header values must be of type string or array/) end end describe 'without a call operation' do def get_response(stub) stub.client_streamer(@method, @sent_msgs, noop, noop, metadata: @metadata) end it_behaves_like 'client streaming' end describe 'via a call operation' do def get_response(stub) op = stub.client_streamer(@method, @sent_msgs, noop, noop, return_op: true, metadata: @metadata) expect(op).to be_a(GRPC::ActiveCall::Operation) op.execute end it_behaves_like 'client streaming' end end describe '#server_streamer' do shared_examples 'server streaming' do before(:each) do @sent_msg = 'a_msg' @replys = Array.new(3) { |i| 'reply_' + (i + 1).to_s } end it 'should send a request to/receive replies from a server' do server_port = create_test_server host = "localhost:#{server_port}" th = run_server_streamer(@sent_msg, @replys, @pass) stub = GRPC::ClientStub.new(host, @cq, :this_channel_is_insecure) expect(get_responses(stub).collect { |r| r }).to eq(@replys) th.join end it 'should raise an error if the status is not ok' do server_port = create_test_server host = "localhost:#{server_port}" th = run_server_streamer(@sent_msg, @replys, @fail) stub = GRPC::ClientStub.new(host, @cq, :this_channel_is_insecure) e = get_responses(stub) expect { e.collect { |r| r } }.to raise_error(GRPC::BadStatus) th.join end it 'should send metadata to the server ok' do server_port = create_test_server host = "localhost:#{server_port}" th = run_server_streamer(@sent_msg, @replys, @fail, k1: 'v1', k2: 'v2') stub = GRPC::ClientStub.new(host, @cq, :this_channel_is_insecure) e = get_responses(stub) expect { e.collect { |r| r } }.to raise_error(GRPC::BadStatus) th.join end end describe 'without a call operation' do def get_responses(stub) e = stub.server_streamer(@method, @sent_msg, noop, noop, metadata: { k1: 'v1', k2: 'v2' }) expect(e).to be_a(Enumerator) e end it_behaves_like 'server streaming' end describe 'via a call operation' do def get_responses(stub) op = stub.server_streamer(@method, @sent_msg, noop, noop, return_op: true, metadata: { k1: 'v1', k2: 'v2' }) expect(op).to be_a(GRPC::ActiveCall::Operation) e = op.execute expect(e).to be_a(Enumerator) e end it_behaves_like 'server streaming' end end describe '#bidi_streamer' do shared_examples 'bidi streaming' do before(:each) do @sent_msgs = Array.new(3) { |i| 'msg_' + (i + 1).to_s } @replys = Array.new(3) { |i| 'reply_' + (i + 1).to_s } server_port = create_test_server @host = "localhost:#{server_port}" end it 'supports sending all the requests first', bidi: true do th = run_bidi_streamer_handle_inputs_first(@sent_msgs, @replys, @pass) stub = GRPC::ClientStub.new(@host, @cq, :this_channel_is_insecure) e = get_responses(stub) expect(e.collect { |r| r }).to eq(@replys) th.join end it 'supports client-initiated ping pong', bidi: true do th = run_bidi_streamer_echo_ping_pong(@sent_msgs, @pass, true) stub = GRPC::ClientStub.new(@host, @cq, :this_channel_is_insecure) e = get_responses(stub) expect(e.collect { |r| r }).to eq(@sent_msgs) th.join end it 'supports a server-initiated ping pong', bidi: true do th = run_bidi_streamer_echo_ping_pong(@sent_msgs, @pass, false) stub = GRPC::ClientStub.new(@host, @cq, :this_channel_is_insecure) e = get_responses(stub) expect(e.collect { |r| r }).to eq(@sent_msgs) th.join end end describe 'without a call operation' do def get_responses(stub) e = stub.bidi_streamer(@method, @sent_msgs, noop, noop) expect(e).to be_a(Enumerator) e end it_behaves_like 'bidi streaming' end describe 'via a call operation' do def get_responses(stub) op = stub.bidi_streamer(@method, @sent_msgs, noop, noop, return_op: true) expect(op).to be_a(GRPC::ActiveCall::Operation) e = op.execute expect(e).to be_a(Enumerator) e end it_behaves_like 'bidi streaming' end describe 'without enough time to run' do before(:each) do @sent_msgs = Array.new(3) { |i| 'msg_' + (i + 1).to_s } @replys = Array.new(3) { |i| 'reply_' + (i + 1).to_s } server_port = create_test_server @host = "localhost:#{server_port}" end it 'should fail with DeadlineExceeded', bidi: true do @server.start stub = GRPC::ClientStub.new(@host, @cq, :this_channel_is_insecure) blk = proc do e = stub.bidi_streamer(@method, @sent_msgs, noop, noop, deadline: from_relative_time(0.001)) e.collect { |r| r } end expect(&blk).to raise_error GRPC::BadStatus, /Deadline Exceeded/ end end end def run_server_streamer(expected_input, replys, status, **kw) wanted_metadata = kw.clone wakey_thread do |notifier| c = expect_server_to_be_invoked(notifier) wanted_metadata.each do |k, v| expect(c.metadata[k.to_s]).to eq(v) end expect(c.remote_read).to eq(expected_input) replys.each { |r| c.remote_send(r) } c.send_status(status, status == @pass ? 'OK' : 'NOK', true) end end def run_bidi_streamer_handle_inputs_first(expected_inputs, replys, status) wakey_thread do |notifier| c = expect_server_to_be_invoked(notifier) expected_inputs.each { |i| expect(c.remote_read).to eq(i) } replys.each { |r| c.remote_send(r) } c.send_status(status, status == @pass ? 'OK' : 'NOK', true) end end def run_bidi_streamer_echo_ping_pong(expected_inputs, status, client_starts) wakey_thread do |notifier| c = expect_server_to_be_invoked(notifier) expected_inputs.each do |i| if client_starts expect(c.remote_read).to eq(i) c.remote_send(i) else c.remote_send(i) expect(c.remote_read).to eq(i) end end c.send_status(status, status == @pass ? 'OK' : 'NOK', true) end end def run_client_streamer(expected_inputs, resp, status, **kw) wanted_metadata = kw.clone wakey_thread do |notifier| c = expect_server_to_be_invoked(notifier) expected_inputs.each { |i| expect(c.remote_read).to eq(i) } wanted_metadata.each do |k, v| expect(c.metadata[k.to_s]).to eq(v) end c.remote_send(resp) c.send_status(status, status == @pass ? 'OK' : 'NOK', true) end end def run_request_response(expected_input, resp, status, **kw) wanted_metadata = kw.clone wakey_thread do |notifier| c = expect_server_to_be_invoked(notifier) expect(c.remote_read).to eq(expected_input) wanted_metadata.each do |k, v| expect(c.metadata[k.to_s]).to eq(v) end c.remote_send(resp) c.send_status(status, status == @pass ? 'OK' : 'NOK', true) end end def create_test_server @server_queue = GRPC::Core::CompletionQueue.new @server = GRPC::Core::Server.new(@server_queue, nil) @server.add_http2_port('0.0.0.0:0', :this_port_is_insecure) end def expect_server_to_be_invoked(notifier) @server.start notifier.notify(nil) server_tag = Object.new recvd_rpc = @server.request_call(@server_queue, server_tag, INFINITE_FUTURE) recvd_call = recvd_rpc.call recvd_call.metadata = recvd_rpc.metadata recvd_call.run_batch(@server_queue, server_tag, Time.now + 2, SEND_INITIAL_METADATA => nil) GRPC::ActiveCall.new(recvd_call, @server_queue, noop, noop, INFINITE_FUTURE) end end