Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge pull request #739 from soltanmm/py-update

Browse Source 
Updated Python protoc plugin and Python plugin testing.
pull/400/merge
Nathaniel Manista 10 years ago
parent 886144544c 59d9ff4d9f
commit afae27cf5e
6 changed files with 443 additions and 344 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 367
      src/compiler/python_generator.cc
  2. 3
      src/compiler/python_generator.h
  3. 16
      src/compiler/python_plugin.cc
  4. 395
      test/compiler/python_plugin_test.py
  5. 3
      test/compiler/test.proto
  6. 3
      tools/run_tests/run_python.sh

367
src/compiler/python_generator.cc
Unescape View File

@ -33,9 +33,11 @@
#include <cassert>
#include <cctype>
#include <cstring>
#include <map>
#include <ostream>
#include <sstream>
#include <vector>
#include "src/compiler/python_generator.h"
#include <google/protobuf/io/printer.h>
@ -43,14 +45,19 @@
#include <google/protobuf/descriptor.pb.h>
#include <google/protobuf/descriptor.h>
using google::protobuf::Descriptor;
using google::protobuf::FileDescriptor;
using google::protobuf::ServiceDescriptor;
using google::protobuf::MethodDescriptor;
using google::protobuf::io::Printer;
using google::protobuf::io::StringOutputStream;
using std::initializer_list;
using std::make_pair;
using std::map;
using std::pair;
using std::string;
using std::strlen;
using std::vector;
namespace grpc_python_generator {
namespace {
@ -99,62 +106,81 @@ class IndentScope {
// END FORMATTING BOILERPLATE //
////////////////////////////////
void PrintService(const ServiceDescriptor* service,
Printer* out) {
bool PrintServicer(const ServiceDescriptor* service,
Printer* out) {
string doc = "<fill me in later!>";
map<string, string> dict = ListToDict({
"Service", service->name(),
"Documentation", doc,
});
out->Print(dict, "class $Service$Service(object):\n");
out->Print(dict, "class EarlyAdopter$Service$Servicer(object):\n");
{
IndentScope raii_class_indent(out);
out->Print(dict, "\"\"\"$Documentation$\"\"\"\n");
out->Print("def __init__(self):\n");
{
IndentScope raii_method_indent(out);
out->Print("pass\n");
out->Print("__metaclass__ = abc.ABCMeta\n");
for (int i = 0; i < service->method_count(); ++i) {
auto meth = service->method(i);
string arg_name = meth->client_streaming() ?
"request_iterator" : "request";
out->Print("@abc.abstractmethod\n");
out->Print("def $Method$(self, $ArgName$):\n",
"Method", meth->name(), "ArgName", arg_name);
{
IndentScope raii_method_indent(out);
out->Print("raise NotImplementedError()\n");
}
}
}
return true;
}
void PrintServicer(const ServiceDescriptor* service,
Printer* out) {
bool PrintServer(const ServiceDescriptor* service, Printer* out) {
string doc = "<fill me in later!>";
map<string, string> dict = ListToDict({
"Service", service->name(),
"Documentation", doc,
});
out->Print(dict, "class $Service$Servicer(object):\n");
out->Print(dict, "class EarlyAdopter$Service$Server(object):\n");
{
IndentScope raii_class_indent(out);
out->Print(dict, "\"\"\"$Documentation$\"\"\"\n");
for (int i = 0; i < service->method_count(); ++i) {
auto meth = service->method(i);
out->Print("def $Method$(self, arg):\n", "Method", meth->name());
{
IndentScope raii_method_indent(out);
out->Print("raise NotImplementedError()\n");
}
out->Print("__metaclass__ = abc.ABCMeta\n");
out->Print("@abc.abstractmethod\n");
out->Print("def start(self):\n");
{
IndentScope raii_method_indent(out);
out->Print("raise NotImplementedError()\n");
}
out->Print("@abc.abstractmethod\n");
out->Print("def stop(self):\n");
{
IndentScope raii_method_indent(out);
out->Print("raise NotImplementedError()\n");
}
}
return true;
}
void PrintStub(const ServiceDescriptor* service,
bool PrintStub(const ServiceDescriptor* service,
Printer* out) {
string doc = "<fill me in later!>";
map<string, string> dict = ListToDict({
"Service", service->name(),
"Documentation", doc,
});
out->Print(dict, "class $Service$Stub(object):\n");
out->Print(dict, "class EarlyAdopter$Service$Stub(object):\n");
{
IndentScope raii_class_indent(out);
out->Print(dict, "\"\"\"$Documentation$\"\"\"\n");
out->Print("__metaclass__ = abc.ABCMeta\n");
for (int i = 0; i < service->method_count(); ++i) {
const MethodDescriptor* meth = service->method(i);
auto methdict = ListToDict({"Method", meth->name()});
out->Print(methdict, "def $Method$(self, arg):\n");
string arg_name = meth->client_streaming() ?
"request_iterator" : "request";
auto methdict = ListToDict({"Method", meth->name(), "ArgName", arg_name});
out->Print("@abc.abstractmethod\n");
out->Print(methdict, "def $Method$(self, $ArgName$):\n");
{
IndentScope raii_method_indent(out);
out->Print("raise NotImplementedError()\n");
@ -162,169 +188,190 @@ void PrintStub(const ServiceDescriptor* service,
out->Print(methdict, "$Method$.async = None\n");
}
}
return true;
}
void PrintStubImpl(const ServiceDescriptor* service,
Printer* out) {
map<string, string> dict = ListToDict({
"Service", service->name(),
});
out->Print(dict, "class _$Service$Stub($Service$Stub):\n");
{
IndentScope raii_class_indent(out);
out->Print("def __init__(self, face_stub, default_timeout):\n");
{
IndentScope raii_method_indent(out);
out->Print("self._face_stub = face_stub\n"
"self._default_timeout = default_timeout\n"
"stub_self = self\n");
bool GetModuleAndMessagePath(const Descriptor* type,
pair<string, string>* out) {
const Descriptor* path_elem_type = type;
vector<const Descriptor*> message_path;
do {
message_path.push_back(path_elem_type);
path_elem_type = path_elem_type->containing_type();
} while (path_elem_type != nullptr);
string file_name = type->file()->name();
string module_name;
static const int proto_suffix_length = strlen(".proto");
if (!(file_name.size() > static_cast<size_t>(proto_suffix_length) &&
file_name.find_last_of(".proto") == file_name.size() - 1)) {
return false;
}
module_name = file_name.substr(
0, file_name.size() - proto_suffix_length) + "_pb2";
string package = type->file()->package();
string module = (package.empty() ? "" : package + ".") +
module_name;
string message_type;
for (auto path_iter = message_path.rbegin();
path_iter != message_path.rend(); ++path_iter) {
message_type += (*path_iter)->name() + ".";
}
message_type.pop_back();
*out = make_pair(module, message_type);
return true;
}
for (int i = 0; i < service->method_count(); ++i) {
const MethodDescriptor* meth = service->method(i);
bool server_streaming = meth->server_streaming();
bool client_streaming = meth->client_streaming();
std::string blocking_call, future_call;
if (server_streaming) {
if (client_streaming) {
blocking_call = "stub_self._face_stub.inline_stream_in_stream_out";
future_call = blocking_call;
} else {
blocking_call = "stub_self._face_stub.inline_value_in_stream_out";
future_call = blocking_call;
}
} else {
if (client_streaming) {
blocking_call = "stub_self._face_stub.blocking_stream_in_value_out";
future_call = "stub_self._face_stub.future_stream_in_value_out";
} else {
blocking_call = "stub_self._face_stub.blocking_value_in_value_out";
future_call = "stub_self._face_stub.future_value_in_value_out";
}
}
// TODO(atash): use the solution described at
// http://stackoverflow.com/a/2982 to bind 'async' attribute
// functions to def'd functions instead of using callable attributes.
auto methdict = ListToDict({
"Method", meth->name(),
"BlockingCall", blocking_call,
"FutureCall", future_call
});
out->Print(methdict, "class $Method$(object):\n");
{
IndentScope raii_callable_indent(out);
out->Print("def __call__(self, arg):\n");
{
IndentScope raii_callable_call_indent(out);
out->Print(methdict,
"return $BlockingCall$(\"$Method$\", arg, "
"stub_self._default_timeout)\n");
}
out->Print("def async(self, arg):\n");
{
IndentScope raii_callable_async_indent(out);
out->Print(methdict,
"return $FutureCall$(\"$Method$\", arg, "
"stub_self._default_timeout)\n");
}
}
out->Print(methdict, "self.$Method$ = $Method$()\n");
bool PrintServerFactory(const ServiceDescriptor* service, Printer* out) {
out->Print("def early_adopter_create_$Service$_server(servicer, port, "
"root_certificates, key_chain_pairs):\n",
"Service", service->name());
{
IndentScope raii_create_server_indent(out);
map<string, pair<string, string>> method_to_module_and_message;
out->Print("method_implementations = {\n");
for (int i = 0; i < service->method_count(); ++i) {
IndentScope raii_implementations_indent(out);
const MethodDescriptor* meth = service->method(i);
string meth_type =
string(meth->client_streaming() ? "stream" : "unary") +
string(meth->server_streaming() ? "_stream" : "_unary") + "_inline";
out->Print("\"$Method$\": utilities.$Type$(servicer.$Method$),\n",
"Method", meth->name(),
"Type", meth_type);
// Maintain information on the input type of the service method for later
// use in constructing the service assembly's activated fore link.
const Descriptor* input_type = meth->input_type();
pair<string, string> module_and_message;
if (!GetModuleAndMessagePath(input_type, &module_and_message)) {
return false;
}
method_to_module_and_message.emplace(
meth->name(), module_and_message);
}
out->Print("}\n");
// Ensure that we've imported all of the relevant messages.
for (auto& meth_vals : method_to_module_and_message) {
out->Print("import $Module$\n",
"Module", meth_vals.second.first);
}
out->Print("request_deserializers = {\n");
for (auto& meth_vals : method_to_module_and_message) {
IndentScope raii_serializers_indent(out);
string full_input_type_path = meth_vals.second.first + "." +
meth_vals.second.second;
out->Print("\"$Method$\": $Type$.FromString,\n",
"Method", meth_vals.first,
"Type", full_input_type_path);
}
out->Print("}\n");
out->Print("response_serializers = {\n");
for (auto& meth_vals : method_to_module_and_message) {
IndentScope raii_serializers_indent(out);
out->Print("\"$Method$\": lambda x: x.SerializeToString(),\n",
"Method", meth_vals.first);
}
out->Print("}\n");
out->Print("link = fore.activated_fore_link(port, request_deserializers, "
"response_serializers, root_certificates, key_chain_pairs)\n");
out->Print("return implementations.assemble_service("
"method_implementations, link)\n");
}
return true;
}
void PrintStubGenerators(const ServiceDescriptor* service, Printer* out) {
bool PrintStubFactory(const ServiceDescriptor* service, Printer* out) {
map<string, string> dict = ListToDict({
"Service", service->name(),
});
// Write out a generator of linked pairs of Server/Stub
out->Print(dict, "def mock_$Service$(servicer, default_timeout):\n");
out->Print(dict, "def early_adopter_create_$Service$_stub(host, port):\n");
{
IndentScope raii_mock_indent(out);
out->Print("value_in_value_out = {}\n"
"value_in_stream_out = {}\n"
"stream_in_value_out = {}\n"
"stream_in_stream_out = {}\n");
IndentScope raii_create_server_indent(out);
map<string, pair<string, string>> method_to_module_and_message;
out->Print("method_implementations = {\n");
for (int i = 0; i < service->method_count(); ++i) {
IndentScope raii_implementations_indent(out);
const MethodDescriptor* meth = service->method(i);
std::string super_interface, meth_dict;
bool server_streaming = meth->server_streaming();
bool client_streaming = meth->client_streaming();
if (server_streaming) {
if (client_streaming) {
super_interface = "InlineStreamInStreamOutMethod";
meth_dict = "stream_in_stream_out";
} else {
super_interface = "InlineValueInStreamOutMethod";
meth_dict = "value_in_stream_out";
}
} else {
if (client_streaming) {
super_interface = "InlineStreamInValueOutMethod";
meth_dict = "stream_in_value_out";
} else {
super_interface = "InlineValueInValueOutMethod";
meth_dict = "value_in_value_out";
}
}
map<string, string> methdict = ListToDict({
"Method", meth->name(),
"SuperInterface", super_interface,
"MethodDict", meth_dict
});
out->Print(
methdict, "class $Method$(_face_interfaces.$SuperInterface$):\n");
{
IndentScope raii_inline_class_indent(out);
out->Print("def service(self, request, context):\n");
{
IndentScope raii_inline_class_fn_indent(out);
out->Print(methdict, "return servicer.$Method$(request)\n");
}
string meth_type =
string(meth->client_streaming() ? "stream" : "unary") +
string(meth->server_streaming() ? "_stream" : "_unary") + "_inline";
// TODO(atash): once the expected input to assemble_dynamic_inline_stub is
// cleaned up, change this to the expected argument's dictionary values.
out->Print("\"$Method$\": utilities.$Type$(None),\n",
"Method", meth->name(),
"Type", meth_type);
// Maintain information on the input type of the service method for later
// use in constructing the service assembly's activated fore link.
const Descriptor* output_type = meth->output_type();
pair<string, string> module_and_message;
if (!GetModuleAndMessagePath(output_type, &module_and_message)) {
return false;
}
out->Print(methdict, "$MethodDict$['$Method$'] = $Method$()\n");
method_to_module_and_message.emplace(
meth->name(), module_and_message);
}
out->Print(
"face_linked_pair = _face_testing.server_and_stub(default_timeout,"
"inline_value_in_value_out_methods=value_in_value_out,"
"inline_value_in_stream_out_methods=value_in_stream_out,"
"inline_stream_in_value_out_methods=stream_in_value_out,"
"inline_stream_in_stream_out_methods=stream_in_stream_out)\n");
out->Print("class LinkedPair(object):\n");
{
IndentScope raii_linked_pair(out);
out->Print("def __init__(self, server, stub):\n");
{
IndentScope raii_linked_pair_init(out);
out->Print("self.server = server\n"
"self.stub = stub\n");
}
out->Print("}\n");
// Ensure that we've imported all of the relevant messages.
for (auto& meth_vals : method_to_module_and_message) {
out->Print("import $Module$\n",
"Module", meth_vals.second.first);
}
out->Print(
dict,
"stub = _$Service$Stub(face_linked_pair.stub, default_timeout)\n");
out->Print("return LinkedPair(None, stub)\n");
out->Print("response_deserializers = {\n");
for (auto& meth_vals : method_to_module_and_message) {
IndentScope raii_serializers_indent(out);
string full_output_type_path = meth_vals.second.first + "." +
meth_vals.second.second;
out->Print("\"$Method$\": $Type$.FromString,\n",
"Method", meth_vals.first,
"Type", full_output_type_path);
}
out->Print("}\n");
out->Print("request_serializers = {\n");
for (auto& meth_vals : method_to_module_and_message) {
IndentScope raii_serializers_indent(out);
out->Print("\"$Method$\": lambda x: x.SerializeToString(),\n",
"Method", meth_vals.first);
}
out->Print("}\n");
out->Print("link = rear.activated_rear_link("
"host, port, request_serializers, response_deserializers)\n");
out->Print("return implementations.assemble_dynamic_inline_stub("
"method_implementations, link)\n");
}
return true;
}
bool PrintPreamble(const FileDescriptor* file, Printer* out) {
out->Print("import abc\n");
out->Print("from grpc._adapter import fore\n");
out->Print("from grpc._adapter import rear\n");
out->Print("from grpc.framework.assembly import implementations\n");
out->Print("from grpc.framework.assembly import utilities\n");
return true;
}
} // namespace
string GetServices(const FileDescriptor* file) {
pair<bool, string> GetServices(const FileDescriptor* file) {
string output;
StringOutputStream output_stream(&output);
Printer out(&output_stream, '$');
out.Print("from grpc.framework.face import demonstration as _face_testing\n");
out.Print("from grpc.framework.face import interfaces as _face_interfaces\n");
for (int i = 0; i < file->service_count(); ++i) {
auto service = file->service(i);
PrintService(service, &out);
PrintServicer(service, &out);
PrintStub(service, &out);
PrintStubImpl(service, &out);
PrintStubGenerators(service, &out);
{
// Scope the output stream so it closes and finalizes output to the string.
StringOutputStream output_stream(&output);
Printer out(&output_stream, '$');
if (!PrintPreamble(file, &out)) {
return make_pair(false, "");
}
for (int i = 0; i < file->service_count(); ++i) {
auto service = file->service(i);
if (!(PrintServicer(service, &out) &&
PrintServer(service, &out) &&
PrintStub(service, &out) &&
PrintServerFactory(service, &out) &&
PrintStubFactory(service, &out))) {
return make_pair(false, "");
}
}
}
return output;
return make_pair(true, std::move(output));
}
} // namespace grpc_python_generator

3
src/compiler/python_generator.h
Unescape View File

@ -35,6 +35,7 @@
#define __GRPC_COMPILER_PYTHON_GENERATOR_H__
#include <string>
#include <utility>
namespace google {
namespace protobuf {
@ -44,7 +45,7 @@ class FileDescriptor;
namespace grpc_python_generator {
std::string GetServices(const google::protobuf::FileDescriptor* file);
std::pair<bool, std::string> GetServices(const google::protobuf::FileDescriptor* file);
} // namespace grpc_python_generator

16
src/compiler/python_plugin.cc
Unescape View File

@ -33,6 +33,7 @@
// Generates a Python gRPC service interface out of Protobuf IDL.
#include <cstring>
#include <memory>
#include <string>
@ -50,6 +51,7 @@ using google::protobuf::compiler::PluginMain;
using google::protobuf::io::CodedOutputStream;
using google::protobuf::io::ZeroCopyOutputStream;
using std::string;
using std::strlen;
class PythonGrpcGenerator : public CodeGenerator {
public:
@ -62,7 +64,7 @@ class PythonGrpcGenerator : public CodeGenerator {
string* error) const override {
// Get output file name.
string file_name;
static const int proto_suffix_length = 6; // length of ".proto"
static const int proto_suffix_length = strlen(".proto");
if (file->name().size() > static_cast<size_t>(proto_suffix_length) &&
file->name().find_last_of(".proto") == file->name().size() - 1) {
file_name = file->name().substr(
@ -75,9 +77,15 @@ class PythonGrpcGenerator : public CodeGenerator {
std::unique_ptr<ZeroCopyOutputStream> output(
context->OpenForInsert(file_name, "module_scope"));
CodedOutputStream coded_out(output.get());
string code = grpc_python_generator::GetServices(file);
coded_out.WriteRaw(code.data(), code.size());
return true;
bool success = false;
string code = "";
tie(success, code) = grpc_python_generator::GetServices(file);
if (success) {
coded_out.WriteRaw(code.data(), code.size());
return true;
} else {
return false;
}
}
};

395
test/compiler/python_plugin_test.py
Unescape View File

@ -40,8 +40,24 @@ import unittest
from grpc.framework.face import exceptions
from grpc.framework.foundation import future
# Identifiers of entities we expect to find in the generated module.
SERVICER_IDENTIFIER = 'EarlyAdopterTestServiceServicer'
SERVER_IDENTIFIER = 'EarlyAdopterTestServiceServer'
STUB_IDENTIFIER = 'EarlyAdopterTestServiceStub'
SERVER_FACTORY_IDENTIFIER = 'early_adopter_create_TestService_server'
STUB_FACTORY_IDENTIFIER = 'early_adopter_create_TestService_stub'
# Timeouts and delays.
SHORT_TIMEOUT = 0.1
NORMAL_TIMEOUT = 1
LONG_TIMEOUT = 2
DOES_NOT_MATTER_DELAY = 0
NO_DELAY = 0
LONG_DELAY = 1
# Assigned in __main__.
_build_mode = None
_port = None
class _ServicerMethods(object):
@ -71,14 +87,14 @@ class _ServicerMethods(object):
while self._paused:
time.sleep(0)
def UnaryCall(self, request):
def UnaryCall(self, request, context):
response = self.test_pb2.SimpleResponse()
response.payload.payload_type = self.test_pb2.COMPRESSABLE
response.payload.payload_compressable = 'a' * request.response_size
self._control()
return response
def StreamingOutputCall(self, request):
def StreamingOutputCall(self, request, context):
for parameter in request.response_parameters:
response = self.test_pb2.StreamingOutputCallResponse()
response.payload.payload_type = self.test_pb2.COMPRESSABLE
@ -86,7 +102,7 @@ class _ServicerMethods(object):
self._control()
yield response
def StreamingInputCall(self, request_iter):
def StreamingInputCall(self, request_iter, context):
response = self.test_pb2.StreamingInputCallResponse()
aggregated_payload_size = 0
for request in request_iter:
@ -95,7 +111,7 @@ class _ServicerMethods(object):
self._control()
return response
def FullDuplexCall(self, request_iter):
def FullDuplexCall(self, request_iter, context):
for request in request_iter:
for parameter in request.response_parameters:
response = self.test_pb2.StreamingOutputCallResponse()
@ -104,7 +120,7 @@ class _ServicerMethods(object):
self._control()
yield response
def HalfDuplexCall(self, request_iter):
def HalfDuplexCall(self, request_iter, context):
responses = []
for request in request_iter:
for parameter in request.response_parameters:
@ -117,7 +133,7 @@ class _ServicerMethods(object):
yield response
def CreateService(test_pb2, delay=0, timeout=1):
def _CreateService(test_pb2, delay):
"""Provides a servicer backend and a stub.
The servicer is just the implementation
@ -136,28 +152,30 @@ def CreateService(test_pb2, delay=0, timeout=1):
A two-tuple (servicer, stub), where the servicer is the back-end of the
service bound to the stub.
"""
class Servicer(test_pb2.TestServiceServicer):
servicer_methods = _ServicerMethods(test_pb2, delay)
def UnaryCall(self, request):
return servicer_methods.UnaryCall(request)
class Servicer(getattr(test_pb2, SERVICER_IDENTIFIER)):
def StreamingOutputCall(self, request):
return servicer_methods.StreamingOutputCall(request)
def UnaryCall(self, request, context):
return servicer_methods.UnaryCall(request, context)
def StreamingInputCall(self, request_iter):
return servicer_methods.StreamingInputCall(request_iter)
def StreamingOutputCall(self, request, context):
return servicer_methods.StreamingOutputCall(request, context)
def FullDuplexCall(self, request_iter):
return servicer_methods.FullDuplexCall(request_iter)
def StreamingInputCall(self, request_iter, context):
return servicer_methods.StreamingInputCall(request_iter, context)
def HalfDuplexCall(self, request_iter):
return servicer_methods.HalfDuplexCall(request_iter)
def FullDuplexCall(self, request_iter, context):
return servicer_methods.FullDuplexCall(request_iter, context)
def HalfDuplexCall(self, request_iter, context):
return servicer_methods.HalfDuplexCall(request_iter, context)
servicer_methods = _ServicerMethods(test_pb2, delay)
servicer = Servicer()
linked_pair = test_pb2.mock_TestService(servicer, timeout)
stub = linked_pair.stub
return servicer_methods, stub
server = getattr(test_pb2, SERVER_FACTORY_IDENTIFIER)(servicer, _port,
None, None)
stub = getattr(test_pb2, STUB_FACTORY_IDENTIFIER)('localhost', _port)
return servicer_methods, stub, server
def StreamingInputRequest(test_pb2):
@ -198,19 +216,20 @@ class PythonPluginTest(unittest.TestCase):
def setUp(self):
protoc_command = '../../bins/%s/protobuf/protoc' % _build_mode
protoc_plugin_filename = '../../bins/%s/grpc_python_plugin' % _build_mode
test_proto_filename = '../cpp/interop/test.proto'
test_proto_filename = './test.proto'
if not os.path.isfile(protoc_command):
# Assume that if we haven't built protoc that it's on the system.
protoc_command = 'protoc'
# ensure that the output directory exists
outdir = '../../gens/test/compiler/python/'
# Ensure that the output directory exists.
outdir = '../../gens/test/compiler/python'
try:
os.makedirs(outdir)
except OSError as exception:
if exception.errno != errno.EEXIST:
raise
# Invoke protoc with the plugin.
cmd = [
protoc_command,
'--plugin=protoc-gen-python-grpc=%s' % protoc_plugin_filename,
@ -222,215 +241,231 @@ class PythonPluginTest(unittest.TestCase):
subprocess.call(' '.join(cmd), shell=True)
sys.path.append(outdir)
self.delay = 1 # seconds
self.timeout = 2 # seconds
# TODO(atash): Figure out which of theses tests is hanging flakily with small
# probability.
def testImportAttributes(self):
# check that we can access the members
# check that we can access the generated module and its members.
import test_pb2 # pylint: disable=g-import-not-at-top
self.assertIsNotNone(getattr(test_pb2, 'TestServiceServicer', None))
self.assertIsNotNone(getattr(test_pb2, 'TestServiceService', None))
self.assertIsNotNone(getattr(test_pb2, 'TestServiceStub', None))
self.assertIsNotNone(getattr(test_pb2, SERVICER_IDENTIFIER, None))
self.assertIsNotNone(getattr(test_pb2, SERVER_IDENTIFIER, None))
self.assertIsNotNone(getattr(test_pb2, STUB_IDENTIFIER, None))
self.assertIsNotNone(getattr(test_pb2, SERVER_FACTORY_IDENTIFIER, None))
self.assertIsNotNone(getattr(test_pb2, STUB_FACTORY_IDENTIFIER, None))
def testUpDown(self):
import test_pb2
servicer, stub, server = _CreateService(test_pb2, DOES_NOT_MATTER_DELAY)
request = test_pb2.SimpleRequest(response_size=13)
with server, stub:
pass
def testUnaryCall(self):
import test_pb2 # pylint: disable=g-import-not-at-top
servicer, stub = CreateService(test_pb2)
servicer, stub, server = _CreateService(test_pb2, NO_DELAY)
request = test_pb2.SimpleRequest(response_size=13)
response = stub.UnaryCall(request)
expected_response = servicer.UnaryCall(request)
with server, stub:
response = stub.UnaryCall(request, NORMAL_TIMEOUT)
expected_response = servicer.UnaryCall(request, None)
self.assertEqual(expected_response, response)
def testUnaryCallAsync(self):
import test_pb2 # pylint: disable=g-import-not-at-top
servicer, stub = CreateService(
test_pb2, delay=self.delay, timeout=self.timeout)
servicer, stub, server = _CreateService(test_pb2, LONG_DELAY)
request = test_pb2.SimpleRequest(response_size=13)
# TODO(atash): consider using the 'profile' module? Does it even work here?
start_time = time.clock()
response_future = stub.UnaryCall.async(request)
self.assertGreater(self.delay, time.clock() - start_time)
response = response_future.result()
expected_response = servicer.UnaryCall(request)
with server, stub:
start_time = time.clock()
response_future = stub.UnaryCall.async(request, LONG_TIMEOUT)
# Check that we didn't block on the asynchronous call.
self.assertGreater(LONG_DELAY, time.clock() - start_time)
response = response_future.result()
expected_response = servicer.UnaryCall(request, None)
self.assertEqual(expected_response, response)
def testUnaryCallAsyncExpired(self):
import test_pb2 # pylint: disable=g-import-not-at-top
# set the timeout super low...
servicer, stub = CreateService(test_pb2, delay=1, timeout=0.1)
servicer, stub, server = _CreateService(test_pb2,
delay=DOES_NOT_MATTER_DELAY)
request = test_pb2.SimpleRequest(response_size=13)
with servicer.pause():
response_future = stub.UnaryCall.async(request)
with self.assertRaises(exceptions.ExpirationError):
response_future.result()
with server, stub:
with servicer.pause():
response_future = stub.UnaryCall.async(request, SHORT_TIMEOUT)
with self.assertRaises(exceptions.ExpirationError):
response_future.result()
def testUnaryCallAsyncCancelled(self):
import test_pb2 # pylint: disable=g-import-not-at-top
servicer, stub = CreateService(test_pb2)
servicer, stub, server = _CreateService(test_pb2, DOES_NOT_MATTER_DELAY)
request = test_pb2.SimpleRequest(response_size=13)
with servicer.pause():
response_future = stub.UnaryCall.async(request)
response_future.cancel()
self.assertTrue(response_future.cancelled())
with server, stub:
with servicer.pause():
response_future = stub.UnaryCall.async(request, 1)
response_future.cancel()
self.assertTrue(response_future.cancelled())
def testUnaryCallAsyncFailed(self):
import test_pb2 # pylint: disable=g-import-not-at-top
servicer, stub = CreateService(test_pb2)
servicer, stub, server = _CreateService(test_pb2, DOES_NOT_MATTER_DELAY)
request = test_pb2.SimpleRequest(response_size=13)
with servicer.fail():
response_future = stub.UnaryCall.async(request)
self.assertIsNotNone(response_future.exception())
with server, stub:
with servicer.fail():
response_future = stub.UnaryCall.async(request, NORMAL_TIMEOUT)
self.assertIsNotNone(response_future.exception())
def testStreamingOutputCall(self):
import test_pb2 # pylint: disable=g-import-not-at-top
servicer, stub = CreateService(test_pb2)
servicer, stub, server = _CreateService(test_pb2, NO_DELAY)
request = StreamingOutputRequest(test_pb2)
responses = stub.StreamingOutputCall(request)
expected_responses = servicer.StreamingOutputCall(request)
for check in itertools.izip_longest(expected_responses, responses):
expected_response, response = check
self.assertEqual(expected_response, response)
def testStreamingOutputCallAsync(self):
import test_pb2 # pylint: disable=g-import-not-at-top
servicer, stub = CreateService(test_pb2, timeout=self.timeout)
request = StreamingOutputRequest(test_pb2)
responses = stub.StreamingOutputCall.async(request)
expected_responses = servicer.StreamingOutputCall(request)
for check in itertools.izip_longest(expected_responses, responses):
expected_response, response = check
self.assertEqual(expected_response, response)
def testStreamingOutputCallAsyncExpired(self):
with server, stub:
responses = stub.StreamingOutputCall(request, NORMAL_TIMEOUT)
expected_responses = servicer.StreamingOutputCall(request, None)
for check in itertools.izip_longest(expected_responses, responses):
expected_response, response = check
self.assertEqual(expected_response, response)
def testStreamingOutputCallExpired(self):
import test_pb2 # pylint: disable=g-import-not-at-top
servicer, stub = CreateService(test_pb2, timeout=0.1)
servicer, stub, server = _CreateService(test_pb2, DOES_NOT_MATTER_DELAY)
request = StreamingOutputRequest(test_pb2)
with servicer.pause():
responses = stub.StreamingOutputCall.async(request)
with self.assertRaises(exceptions.ExpirationError):
list(responses)
with server, stub:
with servicer.pause():
responses = stub.StreamingOutputCall(request, SHORT_TIMEOUT)
with self.assertRaises(exceptions.ExpirationError):
list(responses)
def testStreamingOutputCallAsyncCancelled(self):
def testStreamingOutputCallCancelled(self):
import test_pb2 # pylint: disable=g-import-not-at-top
_, stub = CreateService(test_pb2, timeout=0.1)
unused_servicer, stub, server = _CreateService(test_pb2,
DOES_NOT_MATTER_DELAY)
request = StreamingOutputRequest(test_pb2)
responses = stub.StreamingOutputCall.async(request)
next(responses)
responses.cancel()
with self.assertRaises(future.CancelledError):
with server, stub:
responses = stub.StreamingOutputCall(request, SHORT_TIMEOUT)
next(responses)
responses.cancel()
with self.assertRaises(future.CancelledError):
next(responses)
def testStreamingOutputCallAsyncFailed(self):
@unittest.skip('TODO(atash,nathaniel): figure out why this times out '
'instead of raising the proper error.')
def testStreamingOutputCallFailed(self):
import test_pb2 # pylint: disable=g-import-not-at-top
servicer, stub = CreateService(test_pb2, timeout=0.1)
servicer, stub, server = _CreateService(test_pb2, DOES_NOT_MATTER_DELAY)
request = StreamingOutputRequest(test_pb2)
with servicer.fail():
responses = stub.StreamingOutputCall.async(request)
self.assertIsNotNone(responses)
with self.assertRaises(exceptions.ServicerError):
next(responses)
with server, stub:
with servicer.fail():
responses = stub.StreamingOutputCall(request, 1)
self.assertIsNotNone(responses)
with self.assertRaises(exceptions.ServicerError):
next(responses)
def testStreamingInputCall(self):
import test_pb2 # pylint: disable=g-import-not-at-top
servicer, stub = CreateService(test_pb2)
response = stub.StreamingInputCall(StreamingInputRequest(test_pb2))
servicer, stub, server = _CreateService(test_pb2, NO_DELAY)
with server, stub:
response = stub.StreamingInputCall(StreamingInputRequest(test_pb2),
NORMAL_TIMEOUT)
expected_response = servicer.StreamingInputCall(
StreamingInputRequest(test_pb2))
StreamingInputRequest(test_pb2), None)
self.assertEqual(expected_response, response)
def testStreamingInputCallAsync(self):
import test_pb2 # pylint: disable=g-import-not-at-top
servicer, stub = CreateService(
test_pb2, delay=self.delay, timeout=self.timeout)
start_time = time.clock()
response_future = stub.StreamingInputCall.async(
StreamingInputRequest(test_pb2))
self.assertGreater(self.delay, time.clock() - start_time)
response = response_future.result()
servicer, stub, server = _CreateService(
test_pb2, LONG_DELAY)
with server, stub:
start_time = time.clock()
response_future = stub.StreamingInputCall.async(
StreamingInputRequest(test_pb2), LONG_TIMEOUT)
self.assertGreater(LONG_DELAY, time.clock() - start_time)
response = response_future.result()
expected_response = servicer.StreamingInputCall(
StreamingInputRequest(test_pb2))
StreamingInputRequest(test_pb2), None)
self.assertEqual(expected_response, response)
def testStreamingInputCallAsyncExpired(self):
import test_pb2 # pylint: disable=g-import-not-at-top
# set the timeout super low...
servicer, stub = CreateService(test_pb2, delay=1, timeout=0.1)
with servicer.pause():
response_future = stub.StreamingInputCall.async(
StreamingInputRequest(test_pb2))
with self.assertRaises(exceptions.ExpirationError):
response_future.result()
self.assertIsInstance(
response_future.exception(), exceptions.ExpirationError)
servicer, stub, server = _CreateService(test_pb2, DOES_NOT_MATTER_DELAY)
with server, stub:
with servicer.pause():
response_future = stub.StreamingInputCall.async(
StreamingInputRequest(test_pb2), SHORT_TIMEOUT)
with self.assertRaises(exceptions.ExpirationError):
response_future.result()
self.assertIsInstance(
response_future.exception(), exceptions.ExpirationError)
def testStreamingInputCallAsyncCancelled(self):
import test_pb2 # pylint: disable=g-import-not-at-top
servicer, stub = CreateService(test_pb2)
with servicer.pause():
response_future = stub.StreamingInputCall.async(
StreamingInputRequest(test_pb2))
response_future.cancel()
self.assertTrue(response_future.cancelled())
with self.assertRaises(future.CancelledError):
response_future.result()
servicer, stub, server = _CreateService(test_pb2, DOES_NOT_MATTER_DELAY)
with server, stub:
with servicer.pause():
response_future = stub.StreamingInputCall.async(
StreamingInputRequest(test_pb2), NORMAL_TIMEOUT)
response_future.cancel()
self.assertTrue(response_future.cancelled())
with self.assertRaises(future.CancelledError):
response_future.result()
def testStreamingInputCallAsyncFailed(self):
import test_pb2 # pylint: disable=g-import-not-at-top
servicer, stub = CreateService(test_pb2)
with servicer.fail():
response_future = stub.StreamingInputCall.async(
StreamingInputRequest(test_pb2))
self.assertIsNotNone(response_future.exception())
servicer, stub, server = _CreateService(test_pb2, DOES_NOT_MATTER_DELAY)
with server, stub:
with servicer.fail():
response_future = stub.StreamingInputCall.async(
StreamingInputRequest(test_pb2), SHORT_TIMEOUT)
self.assertIsNotNone(response_future.exception())
def testFullDuplexCall(self):
import test_pb2 # pylint: disable=g-import-not-at-top
servicer, stub = CreateService(test_pb2)
responses = stub.FullDuplexCall(FullDuplexRequest(test_pb2))
expected_responses = servicer.FullDuplexCall(FullDuplexRequest(test_pb2))
for check in itertools.izip_longest(expected_responses, responses):
expected_response, response = check
self.assertEqual(expected_response, response)
def testFullDuplexCallAsync(self):
servicer, stub, server = _CreateService(test_pb2, NO_DELAY)
with server, stub:
responses = stub.FullDuplexCall(FullDuplexRequest(test_pb2),
NORMAL_TIMEOUT)
expected_responses = servicer.FullDuplexCall(FullDuplexRequest(test_pb2),
None)
for check in itertools.izip_longest(expected_responses, responses):
expected_response, response = check
self.assertEqual(expected_response, response)
def testFullDuplexCallExpired(self):
import test_pb2 # pylint: disable=g-import-not-at-top
servicer, stub = CreateService(test_pb2, timeout=self.timeout)
responses = stub.FullDuplexCall.async(FullDuplexRequest(test_pb2))
expected_responses = servicer.FullDuplexCall(FullDuplexRequest(test_pb2))
for check in itertools.izip_longest(expected_responses, responses):
expected_response, response = check
self.assertEqual(expected_response, response)
def testFullDuplexCallAsyncExpired(self):
import test_pb2 # pylint: disable=g-import-not-at-top
servicer, stub = CreateService(test_pb2, timeout=0.1)
servicer, stub, server = _CreateService(test_pb2, DOES_NOT_MATTER_DELAY)
request = FullDuplexRequest(test_pb2)
with servicer.pause():
responses = stub.FullDuplexCall.async(request)
with self.assertRaises(exceptions.ExpirationError):
list(responses)
with server, stub:
with servicer.pause():
responses = stub.FullDuplexCall(request, SHORT_TIMEOUT)
with self.assertRaises(exceptions.ExpirationError):
list(responses)
def testFullDuplexCallAsyncCancelled(self):
def testFullDuplexCallCancelled(self):
import test_pb2 # pylint: disable=g-import-not-at-top
_, stub = CreateService(test_pb2, timeout=0.1)
request = FullDuplexRequest(test_pb2)
responses = stub.FullDuplexCall.async(request)
next(responses)
responses.cancel()
with self.assertRaises(future.CancelledError):
unused_servicer, stub, server = _CreateService(test_pb2, NO_DELAY)
with server, stub:
request = FullDuplexRequest(test_pb2)
responses = stub.FullDuplexCall(request, NORMAL_TIMEOUT)
next(responses)
responses.cancel()
with self.assertRaises(future.CancelledError):
next(responses)
def testFullDuplexCallAsyncFailed(self):
@unittest.skip('TODO(atash,nathaniel): figure out why this hangs forever '
'and fix.')
def testFullDuplexCallFailed(self):
import test_pb2 # pylint: disable=g-import-not-at-top
servicer, stub = CreateService(test_pb2, timeout=0.1)
servicer, stub, server = _CreateService(test_pb2, DOES_NOT_MATTER_DELAY)
request = FullDuplexRequest(test_pb2)
with servicer.fail():
responses = stub.FullDuplexCall.async(request)
self.assertIsNotNone(responses)
with self.assertRaises(exceptions.ServicerError):
next(responses)
with server, stub:
with servicer.fail():
responses = stub.FullDuplexCall(request, NORMAL_TIMEOUT)
self.assertIsNotNone(responses)
with self.assertRaises(exceptions.ServicerError):
next(responses)
def testHalfDuplexCall(self):
import test_pb2 # pylint: disable=g-import-not-at-top
servicer, stub = CreateService(test_pb2)
servicer, stub, server = _CreateService(test_pb2, NO_DELAY)
def HalfDuplexRequest():
request = test_pb2.StreamingOutputCallRequest()
request.response_parameters.add(size=1, interval_us=0)
@ -439,15 +474,16 @@ class PythonPluginTest(unittest.TestCase):
request.response_parameters.add(size=2, interval_us=0)
request.response_parameters.add(size=3, interval_us=0)
yield request
responses = stub.HalfDuplexCall(HalfDuplexRequest())
expected_responses = servicer.HalfDuplexCall(HalfDuplexRequest())
for check in itertools.izip_longest(expected_responses, responses):
expected_response, response = check
self.assertEqual(expected_response, response)
def testHalfDuplexCallAsyncWedged(self):
with server, stub:
responses = stub.HalfDuplexCall(HalfDuplexRequest(), NORMAL_TIMEOUT)
expected_responses = servicer.HalfDuplexCall(HalfDuplexRequest(), None)
for check in itertools.izip_longest(expected_responses, responses):
expected_response, response = check
self.assertEqual(expected_response, response)
def testHalfDuplexCallWedged(self):
import test_pb2 # pylint: disable=g-import-not-at-top
_, stub = CreateService(test_pb2, timeout=1)
_, stub, server = _CreateService(test_pb2, NO_DELAY)
wait_flag = [False]
@contextlib.contextmanager
def wait(): # pylint: disable=invalid-name
@ -461,20 +497,25 @@ class PythonPluginTest(unittest.TestCase):
yield request
while wait_flag[0]:
time.sleep(0.1)
with wait():
responses = stub.HalfDuplexCall.async(HalfDuplexRequest())
# half-duplex waits for the client to send all info
with self.assertRaises(exceptions.ExpirationError):
next(responses)
with server, stub:
with wait():
responses = stub.HalfDuplexCall(HalfDuplexRequest(), NORMAL_TIMEOUT)
# half-duplex waits for the client to send all info
with self.assertRaises(exceptions.ExpirationError):
next(responses)
if __name__ == '__main__':
os.chdir(os.path.dirname(sys.argv[0]))
parser = argparse.ArgumentParser(description='Run Python compiler plugin test.')
parser.add_argument('--build_mode', dest='build_mode', type=str, default='dbg',
help='The build mode of the targets to test, e.g. '
'"dbg", "opt", "asan", etc.')
parser = argparse.ArgumentParser(
description='Run Python compiler plugin test.')
parser.add_argument(
'--build_mode', dest='build_mode', type=str, default='dbg',
help='The build mode of the targets to test, e.g. "dbg", "opt", "asan", '
'etc.')
parser.add_argument('--port', dest='port', type=int, default=0)
args, remainder = parser.parse_known_args()
_build_mode = args.build_mode
_port = args.port
sys.argv[1:] = remainder
unittest.main()

3
test/compiler/test.proto
Unescape View File

@ -32,7 +32,8 @@
// This file is duplicated around the code base. See GitHub issue #526.
syntax = "proto2";
package grpc.testing;
// TODO(atash): Investigate this statement's utility.
// package grpc.testing;
enum PayloadType {
// Compressable text format.

3
tools/run_tests/run_python.sh
Unescape View File

@ -37,7 +37,8 @@ root=`pwd`
export LD_LIBRARY_PATH=$root/libs/opt
source python2.7_virtual_environment/bin/activate
# TODO(issue 215): Properly itemize these in run_tests.py so that they can be parallelized.
python2.7 -B test/compiler/python_plugin_test.py
# TODO(atash): Enable dynamic unused port discovery for this test.
python2.7 -B test/compiler/python_plugin_test.py --build_mode=opt --port=40987
python2.7 -B -m grpc._adapter._blocking_invocation_inline_service_test
python2.7 -B -m grpc._adapter._c_test
python2.7 -B -m grpc._adapter._event_invocation_synchronous_event_service_test

Write Preview
Loading…
Cancel
Save