Abstract the validation into a common function

src/python/grpcio/grpc/_common.py

@@ -61,6 +61,9 @@ STATUS_CODE_TO_CYGRPC_STATUS_CODE = {
 
 MAXIMUM_WAIT_TIMEOUT = 0.1
 
+_ERROR_MESSAGE_PORT_BINDING_FAILED = 'Failed to bind to address %s; set ' \
+    'GRPC_VERBOSITY=debug environment variable to see detailed error message.'
+
 
 def encode(s):
     if isinstance(s, bytes):
@@ -144,3 +147,22 @@ def wait(wait_fn, wait_complete_fn, timeout=None, spin_cb=None):
                 return True
             _wait_once(wait_fn, remaining, spin_cb)
     return False
+
+
+def validate_port_binding_result(address, port):
+    """Validates if the port binding succeed.
+
+    If the port returned by Core is 0, the binding is failed. However, in that
+    case, the Core API doesn't return a detailed failing reason. The best we
+    can do is raising an exception to prevent further confusion.
+
+    Args:
+        address: The address string to be bound.
+        port: An int returned by core
+    """
+    if port == 0:
+        # The Core API doesn't return a failure message. The best we can do
+        # is raising an exception to prevent further confusion.
+        raise RuntimeError(_ERROR_MESSAGE_PORT_BINDING_FAILED % address)
+    else:
+        return port

src/python/grpcio/grpc/_server.py

@@ -958,27 +958,14 @@ class _Server(grpc.Server):
         _add_generic_handlers(self._state, generic_rpc_handlers)
 
     def add_insecure_port(self, address):
-        port = _add_insecure_port(self._state, _common.encode(address))
-        if port == 0:
-            # The Core API doesn't return a failure message. The best we can do
-            # is raising an exception to prevent further confusion.
-            raise RuntimeError('Failed to bind to address %s; set '
-                               'GRPC_VERBOSITY=debug env to see detailed error '
-                               'message.' % address)
-        else:
-            return port
+        return _common.validate_port_binding_result(
+            address, _add_insecure_port(self._state, _common.encode(address)))
 
     def add_secure_port(self, address, server_credentials):
-        port = _add_secure_port(self._state, _common.encode(address),
-                                server_credentials)
-        if port == 0:
-            # The Core API doesn't return a failure message. The best we can do
-            # is raising an exception to prevent further confusion.
-            raise RuntimeError('Failed to bind to address %s; set '
-                               'GRPC_VERBOSITY=debug env to see detailed error '
-                               'message.' % address)
-        else:
-            return port
+        return _common.validate_port_binding_result(
+            address,
+            _add_secure_port(self._state, _common.encode(address),
+                             server_credentials))
 
     def start(self):
         _start(self._state)

src/python/grpcio/grpc/experimental/aio/_server.py

@@ -80,15 +80,8 @@ class Server(_base_server.Server):
         Returns:
           An integer port on which the server will accept RPC requests.
         """
-        port = self._server.add_insecure_port(_common.encode(address))
-        if port == 0:
-            # The Core API doesn't return a failure message. The best we can do
-            # is raising an exception to prevent further confusion.
-            raise RuntimeError('Failed to bind to address %s; set '
-                               'GRPC_VERBOSITY=debug env to see detailed error '
-                               'message.' % address)
-        else:
-            return port
+        return _common.validate_port_binding_result(
+            address, self._server.add_insecure_port(_common.encode(address)))
 
     def add_secure_port(self, address: str,
                         server_credentials: grpc.ServerCredentials) -> int:
@@ -105,16 +98,10 @@ class Server(_base_server.Server):
         Returns:
           An integer port on which the server will accept RPC requests.
         """
-        port = self._server.add_secure_port(_common.encode(address),
-                                            server_credentials)
-        if port == 0:
-            # The Core API doesn't return a failure message. The best we can do
-            # is raising an exception to prevent further confusion.
-            raise RuntimeError('Failed to bind to address %s; set '
-                               'GRPC_VERBOSITY=debug env to see detailed error '
-                               'message.' % address)
-        else:
-            return port
+        return _common.validate_port_binding_result(
+            address,
+            self._server.add_secure_port(_common.encode(address),
+                                         server_credentials))
 
     async def start(self) -> None:
         """Starts this Server.