Ray Serve Unit Testing Examples

_description.md

Ray Serve Unit Test Examples

1. test_deployment_class.py

Demonstrates how to test a Ray serve deployment class directly without spinning up a Ray cluster.

The key method: func_or_class, allows accessing the underlying class directly where normally, the class would be revised to a Ray serve deployment type due to the @serve.deployment decorator.

2. test_deployment_requests.py

Example pattern to test issuing http requests to a mock Ray serve server which includes the deployment under test. Implementing an end to end test asserting the request and response of the server are as expected.

It's suggested to include the conftest.py in your setup to make injection of the Ray serve resource straightforward. Ray serve internal unit tests are a great reference for how to test these objects.

conftest.py

from mock import patch
import pytest

import ray
from ray import serve

# This file instantiates Ray test fixtures to reuse across multiple tests

@pytest.fixture(scope="session")
def ray_serve():
    ray.init(
        # Suggested to hard code num_cpus from docs:
        # https://docs.ray.io/en/latest/ray-core/examples/testing-tips.html#tip-1-fixing-the-resource-quantity-with-ray-init-num-cpus
        num_cpus=1,
    )
    yield serve.start(detached=True, http_options={"host": "0.0.0.0"})
    ray.shutdown()

test_deployment_class.py

from typing import Dict

import pytest
from ray import serve


@serve.deployment
class MyDeployment:
    async def __call__(self, request: Dict) -> Dict:
        response_json = {
            "text": "undefined"
        }
        if request["text"]:
            response_json["text"] = request["text"]
        return response_json


@pytest.mark.asyncio
class TestMyDeployment:
    async def test_call_my_deployment_returns_input_text(self):
        request_json = {"text": "test_text"}
        response = await MyDeployment.func_or_class()(request=request_json)
        assert response["text"] == request_json["text"]

test_deployment_requests.py

from typing import Dict

import pytest
import requests
from ray import serve
from ray.serve.drivers import DAGDriver
from ray.serve.http_adapters import json_request

@serve.deployment
class MyDeployment:
    async def __call__(self, request: Dict) -> Dict:
        response_json = {
            "text": "undefined"
        }
        if request["text"]:
            response_json["text"] = request["text"]
        return response_json


class TestMyDeploymentServer:
    def _setup_endpoint(self):
        serve.run(
            DAGDriver.options(ray_actor_options={"num_cpus": 0}).bind(
                MyDeployment.options(ray_actor_options={"num_cpus": 0}).bind(),
                http_adapter=json_request,
            )
        )

    def _get_url(self, ray_serve) -> str:
        return f"{ray_serve.root_url}/{MyDeployment.name}"

    def test_call_server_returns_200_response(
        self, ray_serve
    ):
        self._setup_endpoint()

        request_json = {"text": "test_text"}

        url = self._get_url(ray_serve)
        response = requests.post(url, json=request_json)
        assert response.status_code == 200

        response_json = response.json()
        assert response_json["text"] == request_json["text"]