Delete grpcio/Test_Grpcio.py

grpcio/Test_Grpcio.py

@@ -1,163 +0,0 @@
-"""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)