测试 API 端点

测试对于确保代码在各种条件下按预期运行非常重要。创建 BentoML 项目后,您可以设计不同的测试来验证机器学习 (ML) 模型的功能以及服务的操作方面。

测试提供了多项好处,包括

  • 可靠性: 确保您的 BentoML 服务按预期运行。

  • 定期性: 促进定期自动检查代码库中的错误。

  • 可重构性: 使代码库更易于维护,并能适应变化。

本文档解释了如何为 BentoML 服务设计和运行测试。它以摘要服务为例进行测试。

先决条件

可以使用 pytest 等测试运行器运行测试。如果您尚未安装,请通过 pip 安装 pytest

pip install pytest

有关更多信息,请参阅 pytest 文档

服务验证

创建 service.py 文件后,您可以使用一个简单明了的验证脚本来快速确保您的服务按预期运行。这非常适合在 Jupyter Notebook 或类似环境中使用,允许您在开发过程中交互式地调试和验证代码更改。您可以这样实现它

service_verification.py
from service import Summarization, EXAMPLE_INPUT # Imported from the Summarization service.py file

# Initialize the Service
svc = Summarization()

# Invoke the Service with example input and print the output
output = svc.summarize(EXAMPLE_INPUT)
print(output)

此验证测试通过初始化服务并将模型加载到内存中来提供即时反馈,这有助于您验证服务的所有组件(包括外部依赖项和模型本身)是否已正确设置。

如果模型过大或需要 GPU 资源,则在笔记本电脑或资源有限的环境(例如 GitHub Actions)中运行此测试可能不可行。在这种情况下,您可以连接到远程 GPU 机器或利用托管在具有充足资源的服务器上的 Jupyter Notebook。对于加载模型不切实际的情况,请阅读下面的单元测试部分,以使用模拟技术来模拟模型推理。

单元测试

单元测试独立于代码的其余部分,验证项目中最小的可测试部分,例如函数或方法。目的是确保每个组件按设计正确运行。

在处理 ML 模型或像摘要这样的服务时(输出可能并非完全固定),您可以模拟依赖项和输出,并专注于测试服务代码的行为和逻辑,而不是模型的输出。您可能不会直接测试模型的输出,但会确保 BentoML 服务与模型流水线正确交互,并按预期处理输入和输出。

示例

test_unit.py
from unittest.mock import patch, MagicMock
from service import Summarization, EXAMPLE_INPUT # Imported from the Summarization service.py file


@patch('service.pipeline')
def test_summarization(mock_pipeline):
    # Setup a mock return value that resembles the model's output structure
    mock_pipeline.return_value = MagicMock(return_value=[{"summary_text": "Mock summary"}])

    service = Summarization()
    summary = service.summarize(EXAMPLE_INPUT)

    # Check that the mocked pipeline method was called exactly once
    mock_pipeline.assert_called_once()
    # Check the type of the response
    assert isinstance(summary, str), "The output should be a string."
    # Verify the length of the summarized text is less than the original input
    assert len(summary) < len(EXAMPLE_INPUT), "The summarized text should be shorter than the input."

此单元测试执行以下操作

  1. 使用 unittest.mock.patch 模拟 Transformers 库中的 pipeline 函数。

  2. 创建一个模拟对象,模拟真实 pipeline 函数返回的可调用对象的行为。无论何时调用此模拟可调用对象,它都会返回一个包含单个字典的列表,该字典的键为 "summary_text",值为 "Mock summary"。有关更多信息,请参阅 模拟对象库

  3. 进行断言以确保服务正常运行。

注意

当输出固定且已知时(例如,返回常量值或基于输入返回可预测结果的函数),您可以编写直接断言预期输出的测试。在这种情况下,仍然可以使用模拟来将函数与其任何依赖项隔离开,但测试的重点可以是断言函数返回准确的预期值。

运行单元测试

pytest test_unit.py -v

预期输出

====================================================================== test session starts ======================================================================
platform linux -- Python 3.11.7, pytest-8.0.2, pluggy-1.4.0 -- /home/demo/Documents/summarization/summarization/bin/python
cachedir: .pytest_cache
rootdir: /home/demo/Documents/summarization
plugins: anyio-4.3.0
collected 1 item

test_unit.py::test_summarization PASSED                                                                                                                   [100%]

======================================================================= 1 passed in 2.08s =======================================================================

集成测试

集成测试评估两个或多个组件的组合操作。目标是确保项目的不同部分按预期协同工作,包括与数据库、外部 API 和其他服务的交互。

BentoML 服务的集成测试可以涉及启动服务并发送 HTTP 请求以验证其响应。

示例

test_integration.py
import bentoml
import subprocess

from service import EXAMPLE_INPUT # Imported from the Summarization service.py file

def test_summarization_service_integration():
    with subprocess.Popen(["bentoml", "serve", "service:Summarization", "-p", "50001"]) as server_proc:
        try:
            client = bentoml.SyncHTTPClient("https://:50001", server_ready_timeout=10)
            summarized_text = client.summarize(text=EXAMPLE_INPUT)

            # Ensure the summarized text is not empty
            assert summarized_text, "The summarized text should not be empty."
            # Check the type of the response
            assert isinstance(summarized_text, str), "The response should be a string."
            # Verify the length of the summarized text is less than the original input
            assert len(summarized_text) < len(EXAMPLE_INPUT), "The summarized text should be shorter than the input."
        finally:
            server_proc.terminate()

此集成测试执行以下操作

  1. 使用 subprocess 模块在单独的进程中启动端口 50001 上的摘要服务。

  2. 创建一个客户端并发送请求。server_ready_timeout=10 意味着客户端将等待 10 秒,直到服务器准备就绪,然后才继续调用。

  3. 进行断言以确保服务正常运行。

运行集成测试

pytest test_integration.py -v

预期输出

====================================================================== test session starts ======================================================================
platform linux -- Python 3.11.7, pytest-8.0.2, pluggy-1.4.0 -- /home/demo/Documents/summarization/summarization/bin/python
cachedir: .pytest_cache
rootdir: /home/demo/Documents/summarization
plugins: anyio-4.3.0
collected 1 item

test_integration.py::test_summarization_service_integration PASSED                                                                                        [100%]

====================================================================== 1 passed in 19.29s =======================================================================

HTTP 行为测试

要测试 BentoML 服务的 HTTP 行为,您可以模拟 HTTP 请求并断言响应与预期结果匹配。

您可以使用 starlette.testclient 模块创建测试客户端。这允许您直接向 BentoML 服务发送 HTTP 请求,该服务可以通过 to_asgi() 方法转换为 ASGI 应用程序。测试客户端公开的接口与任何其他 httpx 会话相同。

示例

test_http.py
from starlette.testclient import TestClient
from service import Summarization, EXAMPLE_INPUT # Imported from the Summarization service.py file
import pytest

def test_request():
    # Initialize the ASGI app with the Summarization Service
    app = Summarization.to_asgi()
    # Create a test client to interact with the ASGI app
    # The TestClient must be used as a context manager in order to initialize the ASGI app
    with TestClient(app=app) as test_client:
        response = test_client.post("/summarize", json={"text": EXAMPLE_INPUT})
        # Retrieve the text from the response for validation
        summarized_text = response.text
        # Assert that the HTTP response status code is 200, indicating success
        assert response.status_code == 200
        # Assert that the summarized text is not empty
        assert summarized_text, "The summary should not be empty"

此测试执行以下操作

  • 创建一个Starlette 测试客户端,该客户端通过 to_asgi() 与从摘要服务转换而来的 ASGI 应用程序交互。

  • /summarize 端点发送 POST 请求。它模拟客户端向摘要服务发送输入数据进行处理。

  • 进行断言以确保服务正常运行。

运行 HTTP 行为测试

pytest test_http.py -v

预期输出

================================================================================== test session starts ===================================================================================
platform linux -- Python 3.11.7, pytest-8.0.2, pluggy-1.4.0 -- /home/demo/Documents/summarization/summarization/bin/python
cachedir: .pytest_cache
rootdir: /home/demo/Documents/summarization
plugins: anyio-4.3.0
asyncio: mode=Mode.STRICT
collected 1 item

test_http.py::test_request PASSED                                                                                                                                                  [100%]

=================================================================================== 1 passed in 6.13s ====================================================================================

端到端测试

端到端测试对于确保您的 AI 应用不仅在受控测试条件下表现良好,而且能在真实的生产环境中有效运行非常重要。

将 BentoML 服务部署到 BentoCloud 时,您可以在端到端测试中实现以下内容。

  1. 创建测试部署: 将您的 BentoML 服务部署到 BentoCloud。

  2. 等待部署就绪: 确保部署已完全准备好处理请求。

  3. 发送测试请求并验证输出: 通过发送测试请求并验证响应来与部署交互,以确保服务按预期运行。

  4. 关闭并删除部署: 通过关闭和删除测试部署来清理资源,以避免不必要的成本。

示例

test_e2e.py
import pytest
import bentoml
from service import Summarization, EXAMPLE_INPUT  # Imported from the Summarization service.py file

@pytest.fixture(scope="session")
def bentoml_client():
    # Deploy the Summarization Service to BentoCloud
    deployment = bentoml.deployment.create(
        bento="./path_to_your_project", # Alternatively, use an existing Bento tag
        name="test-summarization",
        scaling_min=1,
        scaling_max=1
    )
    try:
        # Wait until the Deployment is ready
        deployment.wait_until_ready(timeout=3600)

        # Provide the Deployment's client for testing
        yield deployment.get_client()
    finally:
        # Clean up
        bentoml.deployment.terminate(name="test-summarization")
        bentoml.deployment.delete(name="test-summarization")

def test_summarization_service(bentoml_client):
    # Send a request to the deployed Summarization service
    summarized_text: str = bentoml_client.summarize(text=EXAMPLE_INPUT)
    # Ensure the summarized text is not empty
    assert summarized_text, "The summarized text should not be empty."
    # Check the type of the response
    assert isinstance(summarized_text, str), "The response should be a string."
    # Verify the length of the summarized text is less than the original input
    assert len(summarized_text) < len(EXAMPLE_INPUT), "The summarized text should be shorter than the input."

此测试执行以下操作

  • 使用 bentoml_client fixture 在 BentoCloud 上设置摘要服务的部署。它确保在提供用于测试的客户端之前,部署已被创建并准备就绪。

  • 使用客户端与摘要服务交互并进行断言,以确保服务正常运行。

  • 测试后通过终止和删除部署来清理资源,以防止因未使用资源而产生的持续费用。

运行端到端测试

pytest test_e2e.py -v

预期结果

=================================================================================================== test session starts ===================================================================================================
platform linux -- Python 3.11.7, pytest-8.1.1, pluggy-1.4.0 -- /home/demo/Documents/summarization/summarization/bin/python
cachedir: .pytest_cache
rootdir: /home/demo/Documents/summarization/test
plugins: anyio-4.3.0
collected 1 item

test_e2e.py::test_summarization_service PASSED                                                                                                                                                                      [100%]

============================================================================================== 1 passed in 120.65s (0:02:00) ==============================================================================================

有关更多信息,请参阅配置部署管理部署

最佳实践

设计测试时请考虑以下几点

  • 保持单元测试独立;模拟外部依赖项,确保测试不受外部因素影响。

  • 使用 CI/CD 流水线自动化测试,以确保定期运行。

  • 保持测试简单且专注。理想情况下,一个测试应该验证一个行为。

  • 确保您的测试环境与生产环境紧密匹配,以避免出现“在我的机器上可以运行”的问题。

  • 自定义或配置 pytest 并使您的测试过程更高效、更符合您的需求,您可以创建一个 pytest.ini 配置文件。通过在 pytest.ini 中指定设置,您可以确保 pytest 在不同环境和设置中始终识别您的项目结构和偏好。示例如下

    [pytest]
    # Add current directory to PYTHONPATH for easy module imports
    pythonpath = .
    
    # Specify where pytest should look for tests, in this case, a directory named `test`
    testpaths = test
    
    # Optionally, configure pytest to use specific markers
    markers =
       integration: mark tests as integration tests.
       unit: mark tests as unit tests.
    

    导航到项目的根目录(即 pytest.ini 所在的目录),然后运行以下命令开始测试

    pytest -v
    

    预期输出

    ================================================================================== test session starts ===================================================================================
    platform linux -- Python 3.11.7, pytest-8.0.2, pluggy-1.4.0 -- /home/demo/Documents/summarization/summarization/bin/python
    cachedir: .pytest_cache
    rootdir: /home/demo/Documents/summarization
    configfile: pytest.ini
    testpaths: test
    plugins: anyio-4.3.0, asyncio-0.23.5.post1
    asyncio: mode=Mode.STRICT
    collected 3 items
    
    test/test_http.py::test_request PASSED                                                                                                                                             [ 33%]
    test/test_integration.py::test_summarization_service_integration PASSED                                                                                                            [ 66%]
    test/test_unit.py::test_summarization PASSED                                                                                                                                       [100%]
    
    =================================================================================== 3 passed in 17.57s ===================================================================================