Upload grpcio/Test_Grpcio.py with huggingface_hub

grpcio/Test_Grpcio.py

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