测试插件

我们建议使用 pytest 来为你的插件编写自动化测试。

如果你使用 使用 cookiecutter 启动可安装插件 中描述的模板，你的插件将在 tests/ 目录中包含一个测试文件，其内容如下

from datasette.app import Datasette
import pytest


@pytest.mark.asyncio
async def test_plugin_is_installed():
    datasette = Datasette(memory=True)
    response = await datasette.client.get("/-/plugins.json")
    assert response.status_code == 200
    installed_plugins = {p["name"] for p in response.json()}
    assert (
        "datasette-plugin-template-demo"
        in installed_plugins
    )

这个测试使用 datasette.client 对象来测试 Datasette 的测试实例。datasette.client 是 HTTPX Python 库的一个封装，它可以使用 ASGI 来模拟 HTTP 请求。这是针对 Datasette 实例编写测试的推荐方法。

这个测试还使用了 pytest-asyncio 包来支持在 pytest 下运行的 async def 测试函数。

你可以这样安装这些包

pip install pytest pytest-asyncio

如果你正在构建一个可安装的包，你可以将它们作为测试依赖添加到你的 setup.py 模块中，像这样

setup(
    name="datasette-my-plugin",
    # ...
    extras_require={"test": ["pytest", "pytest-asyncio"]},
    tests_require=["datasette-my-plugin[test]"],
)

然后你可以这样安装测试依赖

pip install -e '.[test]'

然后像这样使用 pytest 运行测试

pytest

设置 Datasette 测试实例

上面的例子展示了针对 Datasette 实例开始编写测试的最简单方法

from datasette.app import Datasette
import pytest


@pytest.mark.asyncio
async def test_plugin_is_installed():
    datasette = Datasette(memory=True)
    response = await datasette.client.get("/-/plugins.json")
    assert response.status_code == 200

像这样创建一个 Datasette() 实例在测试中是一个有用的捷径，但你需要注意一个细节。确保在该实例上调用了异步方法 .invoke_startup() 非常重要。你可以这样做

datasette = Datasette(memory=True)
await datasette.invoke_startup()

这个方法会注册任何可能需要进行异步调用的 startup(datasette) 或 prepare_jinja2_environment(env, datasette) 插件。

如果你正在使用 await datasette.client.get() 和类似的方法，那么你无需担心这个问题 - Datasette 在第一次处理请求时会自动调用 invoke_startup()。

使用 pdb 调试 Datasette 内部抛出的错误

如果在测试期间 Datasette 内部发生异常，返回给你的插件的响应将具有一个 response.status_code 值 500。

你可以在 Datasette 构造函数中添加 pdb=True，以便在你的测试运行中进入 Python 调试器会话，而不是收到 500 响应码。这相当于使用 --pdb 选项运行 datasette 命令行工具。

在测试函数中看起来像这样

def test_that_opens_the_debugger_or_errors():
    ds = Datasette([db_path], pdb=True)
    response = await ds.client.get("/")

如果你使用这种模式，你需要在运行 pytest 时带上 -s 选项，以避免捕获 stdin/stdout，从而能够与调试器提示符进行交互。

使用 pytest fixtures

Pytest fixtures 可用于创建初始可测试对象，这些对象随后可由多个测试使用。

Datasette 插件的一个常见模式是创建一个 fixture，该 fixture 设置一个临时测试数据库并将其封装在 Datasette 实例中。

这是一个使用 sqlite-utils 库来填充临时测试数据库的示例。它还使用模拟的 metadata.json 配置来设置该表的标题

from datasette.app import Datasette
import pytest
import sqlite_utils


@pytest.fixture(scope="session")
def datasette(tmp_path_factory):
    db_directory = tmp_path_factory.mktemp("dbs")
    db_path = db_directory / "test.db"
    db = sqlite_utils.Database(db_path)
    db["dogs"].insert_all(
        [
            {"id": 1, "name": "Cleo", "age": 5},
            {"id": 2, "name": "Pancakes", "age": 4},
        ],
        pk="id",
    )
    datasette = Datasette(
        [db_path],
        metadata={
            "databases": {
                "test": {
                    "tables": {
                        "dogs": {"title": "Some dogs"}
                    }
                }
            }
        },
    )
    return datasette


@pytest.mark.asyncio
async def test_example_table_json(datasette):
    response = await datasette.client.get(
        "/test/dogs.json?_shape=array"
    )
    assert response.status_code == 200
    assert response.json() == [
        {"id": 1, "name": "Cleo", "age": 5},
        {"id": 2, "name": "Pancakes", "age": 4},
    ]


@pytest.mark.asyncio
async def test_example_table_html(datasette):
    response = await datasette.client.get("/test/dogs")
    assert ">Some dogs</h1>" in response.text

这里的 datasette() 函数定义了 fixture，然后根据 pytest 自动匹配其 datasette 函数参数，将该 fixture 自动传递给两个测试函数。

这里的 @pytest.fixture(scope="session") 行确保了该 fixture 在整个 pytest 执行会话中被复用。这意味着临时数据库文件将创建一次并用于每个测试。

如果你想为每个单独的测试函数重复创建该测试数据库，可以像这样编写 fixture 函数。如果你的插件以某种方式修改了数据库内容，你可能希望这样做

@pytest.fixture
def datasette(tmp_path_factory):
    # This fixture will be executed repeatedly for every test
    ...

使用 pytest-httpx 测试出站 HTTP 调用

如果你的插件进行了出站 HTTP 调用 - 例如 datasette-auth-github 或 datasette-import-table - 你可能需要在测试中模拟这些 HTTP 请求。

的 pytest-httpx 包是一个用于模拟调用的有用库。但它与 Datasette 一起使用时可能会很棘手，因为它模拟了所有 HTTPX 请求，而 Datasette 自身的测试机制内部使用了 HTTPX。

为了避免破坏你的测试，你可以从 non_mocked_hosts() fixture 中返回 ["localhost"]。

举个例子，这是一个非常简单的插件，它执行一个 HTTP 响应并返回结果内容

from datasette import hookimpl
from datasette.utils.asgi import Response
import httpx


@hookimpl
def register_routes():
    return [
        (r"^/-/fetch-url$", fetch_url),
    ]


async def fetch_url(datasette, request):
    if request.method == "GET":
        return Response.html(
            """
            <form action="/-/fetch-url" method="post">
            <input type="hidden" name="csrftoken" value="{}">
            <input name="url"><input type="submit">
        </form>""".format(
                request.scope["csrftoken"]()
            )
        )
    vars = await request.post_vars()
    url = vars["url"]
    return Response.text(httpx.get(url).text)

这是一个针对该插件的测试，它模拟了 HTTPX 出站请求

from datasette.app import Datasette
import pytest


@pytest.fixture
def non_mocked_hosts():
    # This ensures httpx-mock will not affect Datasette's own
    # httpx calls made in the tests by datasette.client:
    return ["localhost"]


async def test_outbound_http_call(httpx_mock):
    httpx_mock.add_response(
        url="https://www.example.com/",
        text="Hello world",
    )
    datasette = Datasette([], memory=True)
    response = await datasette.client.post(
        "/-/fetch-url",
        data={"url": "https://www.example.com/"},
    )
    assert response.text == "Hello world"

    outbound_request = httpx_mock.get_request()
    assert (
        outbound_request.url == "https://www.example.com/"
    )

在测试期间注册插件

在编写插件测试时，你可能会发现只在单个测试期间注册一个测试插件很有用。你可以使用 pm.register() 和 pm.unregister() 来实现，像这样

from datasette import hookimpl
from datasette.app import Datasette
from datasette.plugins import pm
import pytest


@pytest.mark.asyncio
async def test_using_test_plugin():
    class TestPlugin:
        __name__ = "TestPlugin"

        # Use hookimpl and method names to register hooks
        @hookimpl
        def register_routes(self):
            return [
                (r"^/error$", lambda: 1 / 0),
            ]

    pm.register(TestPlugin(), name="undo")
    try:
        # The test implementation goes here
        datasette = Datasette()
        response = await datasette.client.get("/error")
        assert response.status_code == 500
    finally:
        pm.unregister(name="undo")