Add /-/tasks introspection endpoint for supervised background tasks (#2892)

Co-authored-by: Simon Willison <swillison@gmail.com>

tests/test_permissions.py

@@ -520,6 +520,7 @@ def view_instance_client():
         "/-/plugins",
         "/-/settings",
         "/-/threads",
+        "/-/tasks",
         "/-/databases",
         "/-/permissions",
         "/-/messages",

tests/test_success_envelope.py

@@ -59,6 +59,7 @@ async def test_success_object_has_ok_true(ds_client, path):
         "/-/rules.json?action=view-instance",
         "/-/check.json?action=view-instance",
         "/-/threads.json",
+        "/-/tasks.json",
     ),
 )
 async def test_permission_debug_success_has_ok_true(ds_envelope, path):

tests/test_tasks_endpoint.py

@@ -0,0 +1,122 @@
+"""
+Tests for the /-/tasks introspection endpoint.
+
+/-/tasks exposes datasette._background_tasks (see tests/test_background_tasks.py
+for the supervisor machinery itself) the same way /-/threads exposes threading
+internals: gated behind the permissions-debug permission, JSON-only.
+"""
+
+import asyncio
+import contextlib
+import functools
+
+import pytest
+
+from datasette.app import Datasette
+
+
+async def example_task(datasette):
+    pass
+
+
+class ExampleWorker:
+    async def run(self, datasette):
+        pass
+
+    async def __call__(self, datasette):
+        pass
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "func, qualified_name",
+    [
+        (example_task, "example_task"),
+        (functools.partial(example_task), "example_task"),
+        (ExampleWorker().run, "ExampleWorker.run"),
+        (ExampleWorker(), "ExampleWorker.__call__"),
+    ],
+)
+async def test_task_function_path(func, qualified_name):
+    ds = Datasette(memory=True)
+    ds.root_enabled = True
+    handle = ds.add_background_task(func, name="custom-name")
+    try:
+        response = await ds.client.get("/-/tasks.json", actor={"id": "root"})
+        assert response.status_code == 200
+        task = response.json()["tasks"][0]
+        assert task["name"] == "custom-name"
+        assert task["function"] == f"{__name__}.{qualified_name}"
+        assert handle.function == task["function"]
+        assert "plugin" not in task
+        await handle.task
+        html = await ds.client.get("/-/tasks", actor={"id": "root"})
+        assert html.status_code == 200
+        assert task["function"] in html.text
+    finally:
+        await ds.invoke_shutdown()
+
+
+@pytest.mark.asyncio
+async def test_tasks_requires_permissions_debug():
+    ds = Datasette(memory=True)
+    ds.root_enabled = True
+
+    denied = await ds.client.get("/-/tasks.json")
+    assert denied.status_code == 403
+
+    allowed = await ds.client.get("/-/tasks.json", actor={"id": "root"})
+    assert allowed.status_code == 200
+    data = allowed.json()
+    assert data["ok"] is True
+    assert "tasks" in data
+    assert "launched" in data
+
+
+@pytest.mark.asyncio
+async def test_running_and_crashed_task_states():
+    ds = Datasette(memory=True)
+    ds.root_enabled = True
+
+    async def long_running(datasette):
+        await asyncio.Event().wait()
+
+    async def crashing_task(datasette):
+        raise RuntimeError("kaboom")
+
+    long_handle = ds.add_background_task(long_running, name="long-runner")
+    crash_handle = ds.add_background_task(crashing_task, name="crashing_task")
+
+    await ds.start_background_tasks()
+
+    # Let the crashing_task run to completion and its done-callback (which sets
+    # handle.state = "crashed") actually fire before we read state back out.
+    await asyncio.wait_for(
+        asyncio.gather(crash_handle.task, return_exceptions=True), timeout=5
+    )
+    await asyncio.sleep(0)
+
+    try:
+        response = await ds.client.get("/-/tasks.json", actor={"id": "root"})
+        assert response.status_code == 200
+        data = response.json()
+        assert data["launched"] is True
+
+        by_name = {t["name"]: t for t in data["tasks"]}
+        assert by_name["long-runner"]["state"] == "running"
+        assert by_name["long-runner"]["exception"] is None
+        assert by_name["long-runner"]["started_at"] is not None
+
+        crashed = by_name["crashing_task"]
+        assert crashed["function"] == (
+            f"{__name__}.test_running_and_crashed_task_states.<locals>.crashing_task"
+        )
+        assert crashed["state"] == "crashed"
+        assert crashed["exception"] is not None
+        assert isinstance(crashed["exception"], str)
+        assert "kaboom" in crashed["exception"]
+        assert "RuntimeError" in crashed["exception"]
+    finally:
+        long_handle.cancel()
+        with contextlib.suppress(asyncio.CancelledError):
+            await long_handle.task