File size: 5,498 Bytes
4cc0907 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 | """Test_Grpcio.py - device test for grpcio 1.83.1 Android wheels.
Runs on-device via PipManager Scripts folder (needs the grpcio wheel with
the bundled C++ core + Cython cygrpc extension; needs `typing_extensions`
installed from PyPI/official wheel).
Exit-code contract: exit 0 iff every check PASSes (SKIPs allowed), else 1.
Each check prints [PASS] / [FAIL] / [SKIP] with the measured values.
Generated by RIMI
"""
import sys
import traceback
PASS = 0
FAIL = 0
SKIP = 0
def check(name, cond, detail=""):
global PASS, FAIL
if cond:
PASS += 1
print("[PASS] %s %s" % (name, detail))
else:
FAIL += 1
print("[FAIL] %s %s" % (name, detail))
def skip(name, detail=""):
global SKIP
SKIP += 1
print("[SKIP] %s %s" % (name, detail))
def is_net_restricted(exc):
msg = repr(exc)
for token in ("EACCES", "EPERM", "Permission denied", "Network is unreachable",
"EADDRNOTAVAIL", "ENETUNREACH", "FutureTimeoutError",
"TimeoutError", "timed out", "Deadline Exceeded",
"UNAVAILABLE"):
if token in msg:
return True
return False
print("grpcio device test - python %s" % sys.version.split()[0])
# 1. import + native extension present
try:
import grpc
import grpc._cython.cygrpc as cygrpc
check("import grpc", True, "version=%s" % grpc.__version__)
check("native cygrpc", True, cygrpc.__name__)
except Exception as e:
check("import grpc", False, repr(e))
print("RESULT: %d PASS, %d FAIL, %d SKIP" % (PASS, FAIL, SKIP))
traceback.print_exc()
sys.exit(1)
check("grpc version 1.83.1", grpc.__version__ == "1.83.1",
"got=%s" % grpc.__version__)
# 2. runtime dep present
try:
import typing_extensions
try:
from importlib.metadata import version as _dv
_tev = _dv("typing_extensions")
except Exception:
_tev = getattr(typing_extensions, "__version__", "unknown")
check("import typing_extensions", True,
"version=%s" % _tev)
except Exception as e:
check("import typing_extensions", False, repr(e))
# 3. channel object creation (no network traffic)
try:
ch0 = grpc.insecure_channel("127.0.0.1:1")
check("insecure_channel create", True, "target=127.0.0.1:1")
ch0.close()
except Exception as e:
check("insecure_channel create", False, repr(e))
# 4. loopback unary-unary echo roundtrip (raw bytes, no protobuf, no
# external network). Generic handler echoes b'echo:' + request.
server = None
try:
from concurrent import futures
import socket
def _echo_handler(request, context):
return b"echo:" + request
handler = grpc.unary_unary_rpc_method_handler(
_echo_handler,
request_deserializer=lambda b: b,
response_serializer=lambda b: b,
)
generic = grpc.method_handlers_generic_handler(
"test.TestService", {"Echo": handler})
server = grpc.server(futures.ThreadPoolExecutor(max_workers=2))
server.add_generic_rpc_handlers((generic,))
port = server.add_insecure_port("127.0.0.1:0")
check("loopback server bind", port > 0, "port=%d" % port)
server.start()
try:
ch = grpc.insecure_channel("127.0.0.1:%d" % port)
try:
fut = grpc.channel_ready_future(ch)
fut.result(timeout=15)
check("channel ready (loopback)", True, "port=%d" % port)
except Exception as e:
if is_net_restricted(e):
skip("channel ready (loopback)", repr(e))
else:
check("channel ready (loopback)", False, repr(e))
try:
resp = ch.unary_unary("/test.TestService/Echo")(b"ping", timeout=15)
check("unary echo roundtrip", resp == b"echo:ping",
"got=%r" % (resp,))
except Exception as e:
if is_net_restricted(e):
skip("unary echo roundtrip", repr(e))
else:
check("unary echo roundtrip", False, repr(e))
try:
resp2 = ch.unary_unary("/test.TestService/Echo")(
b"hello-grpc", timeout=15)
check("unary echo roundtrip #2", resp2 == b"echo:hello-grpc",
"got=%r" % (resp2,))
except Exception as e:
if is_net_restricted(e):
skip("unary echo roundtrip #2", repr(e))
else:
check("unary echo roundtrip #2", False, repr(e))
ch.close()
finally:
server.stop(None)
server = None
except Exception as e:
if is_net_restricted(e):
skip("loopback server bind", repr(e))
skip("channel ready (loopback)", "server unavailable")
skip("unary echo roundtrip", "server unavailable")
skip("unary echo roundtrip #2", "server unavailable")
else:
check("loopback server bind", False, repr(e))
finally:
if server is not None:
try:
server.stop(None)
except Exception:
pass
# 5. status-code API surface (pure python, no network)
try:
check("StatusCode.OK", grpc.StatusCode.OK.value[0] == 0,
"name=%s" % grpc.StatusCode.OK.name)
except Exception as e:
check("StatusCode.OK", False, repr(e))
print("RESULT: %d PASS, %d FAIL, %d SKIP" % (PASS, FAIL, SKIP))
sys.exit(0 if FAIL == 0 else 1)
|