grpcio/Test_Grpcio.py

"""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)