Clean up Serve proxy files (ray-project#45486)

This PR includes 2 changes:
1. Cleaning up the proxy file to improve consistency for the HTTP and
GRPC proxy.
2. Add missing properties to testing classes to better simulate e2e
tests.

---------

Signed-off-by: Kamen Shah <kamenshah@gmail.com>
Signed-off-by: Ryan O'Leary <ryanaoleary@google.com>

python/ray/serve/_private/proxy.py

@@ -710,25 +710,25 @@ async def send_request_to_replica(
                 context._set_on_grpc_context(proxy_request.context)
                 yield result
 
-            yield ResponseStatus(code=grpc.StatusCode.OK)
+            status = ResponseStatus(code=grpc.StatusCode.OK)
         except TimeoutError:
             message = f"Request timed out after {self.request_timeout_s}s."
             logger.warning(message)
-            yield ResponseStatus(
+            status = ResponseStatus(
                 code=grpc.StatusCode.DEADLINE_EXCEEDED,
                 is_error=True,
                 message=message,
             )
         except asyncio.CancelledError:
             message = f"Client for request {request_id} disconnected."
             logger.info(message)
-            yield ResponseStatus(
+            status = ResponseStatus(
                 code=grpc.StatusCode.CANCELLED,
                 is_error=True,
                 message=message,
             )
         except BackPressureError as e:
-            yield ResponseStatus(
+            status = ResponseStatus(
                 code=grpc.StatusCode.UNAVAILABLE,
                 is_error=True,
                 message=e.message,
@@ -738,12 +738,16 @@ async def send_request_to_replica(
                 logger.warning(f"Request failed: {e}", extra={"log_to_stderr": False})
             else:
                 logger.exception("Request failed due to unexpected error.")
-            yield ResponseStatus(
+            status = ResponseStatus(
                 code=grpc.StatusCode.INTERNAL,
                 is_error=True,
                 message=str(e),
             )
 
+        # The status code should always be set.
+        assert status is not None
+        yield status
+
 
 class HTTPProxy(GenericProxy):
     """This class is meant to be instantiated and run by an ASGI HTTP server."""

python/ray/serve/tests/test_proxy.py

@@ -92,6 +92,14 @@ def streaming_call():
     def options(self, *args, **kwargs):
         return self
 
+    @property
+    def deployment_name(self) -> str:
+        return self.deployment_id.name
+
+    @property
+    def app_name(self) -> str:
+        return self.deployment_id.app_name
+
 
 class FakeProxyRouter(ProxyRouter):
     def __init__(self, *args, **kwargs):
@@ -184,6 +192,14 @@ async def remote(self, *args, **kwargs):
     def options(self, *args, **kwargs):
         return self
 
+    @property
+    def deployment_name(self) -> str:
+        return self.deployment_id.name
+
+    @property
+    def app_name(self) -> str:
+        return self.deployment_id.app_name
+
 
 class FakeHttpReceive:
     def __init__(self, messages=None):