mirror of
https://github.com/simonw/datasette.git
synced 2026-09-08 09:34:05 +02:00
Compare commits
3 commits
codex/http
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
bdc9731740 |
||
|
|
3e018bb1b5 |
||
|
|
e78b8a2e6a |
8 changed files with 591 additions and 59 deletions
5
.github/workflows/test.yml
vendored
5
.github/workflows/test.yml
vendored
|
|
@ -11,16 +11,17 @@ jobs:
|
||||||
strategy:
|
strategy:
|
||||||
fail-fast: false
|
fail-fast: false
|
||||||
matrix:
|
matrix:
|
||||||
python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
|
python-version: ["3.10", "3.11", "3.12", "3.13", "3.14", "3.15"]
|
||||||
steps:
|
steps:
|
||||||
- uses: actions/checkout@v7
|
- uses: actions/checkout@v7
|
||||||
- name: Set up Python ${{ matrix.python-version }}
|
- name: Set up Python ${{ matrix.python-version }}
|
||||||
uses: actions/setup-python@v6
|
uses: actions/setup-python@v7
|
||||||
with:
|
with:
|
||||||
python-version: ${{ matrix.python-version }}
|
python-version: ${{ matrix.python-version }}
|
||||||
allow-prereleases: true
|
allow-prereleases: true
|
||||||
cache: pip
|
cache: pip
|
||||||
cache-dependency-path: pyproject.toml
|
cache-dependency-path: pyproject.toml
|
||||||
|
check-latest: true
|
||||||
- name: Build extension for --load-extension test
|
- name: Build extension for --load-extension test
|
||||||
run: |-
|
run: |-
|
||||||
(cd tests && gcc ext.c -fPIC -shared -o ext.so)
|
(cd tests && gcc ext.c -fPIC -shared -o ext.so)
|
||||||
|
|
|
||||||
|
|
@ -453,8 +453,10 @@ class Datasette:
|
||||||
self.databases = collections.OrderedDict()
|
self.databases = collections.OrderedDict()
|
||||||
self.actions = {} # .invoke_startup() will populate this
|
self.actions = {} # .invoke_startup() will populate this
|
||||||
self._column_types = {} # .invoke_startup() will populate this
|
self._column_types = {} # .invoke_startup() will populate this
|
||||||
|
self._setup_db_done = False
|
||||||
try:
|
try:
|
||||||
self._refresh_schemas_lock = asyncio.Lock()
|
self._refresh_schemas_lock = asyncio.Lock()
|
||||||
|
self._startup_lock = asyncio.Lock()
|
||||||
except RuntimeError as rex:
|
except RuntimeError as rex:
|
||||||
# Workaround for intermittent test failure, see:
|
# Workaround for intermittent test failure, see:
|
||||||
# https://github.com/simonw/datasette/issues/1802
|
# https://github.com/simonw/datasette/issues/1802
|
||||||
|
|
@ -462,6 +464,7 @@ class Datasette:
|
||||||
loop = asyncio.new_event_loop()
|
loop = asyncio.new_event_loop()
|
||||||
asyncio.set_event_loop(loop)
|
asyncio.set_event_loop(loop)
|
||||||
self._refresh_schemas_lock = asyncio.Lock()
|
self._refresh_schemas_lock = asyncio.Lock()
|
||||||
|
self._startup_lock = asyncio.Lock()
|
||||||
else:
|
else:
|
||||||
raise
|
raise
|
||||||
self.crossdb = crossdb
|
self.crossdb = crossdb
|
||||||
|
|
@ -2803,24 +2806,52 @@ class Datasette:
|
||||||
raise RowNotFound(db.name, table_name, pk_values)
|
raise RowNotFound(db.name, table_name, pk_values)
|
||||||
return ResolvedRow(db, table_name, sql, params, pks, pk_values, results.first())
|
return ResolvedRow(db, table_name, sql, params, pks, pk_values, results.first())
|
||||||
|
|
||||||
|
async def _startup_sequence(self):
|
||||||
|
"""Idempotently run the full startup sequence: table counts for
|
||||||
|
immutable databases, then invoke_startup(). Safe to call more than
|
||||||
|
once and safe to call concurrently - callers block until whichever
|
||||||
|
call got there first has finished.
|
||||||
|
|
||||||
|
This is the single entry point used by both AsgiLifespan (so
|
||||||
|
real deployments finish startup before accepting requests) and
|
||||||
|
AsgiRunOnFirstRequest (the fallback for hosts that never send
|
||||||
|
lifespan events, e.g. DatasetteClient's httpx.ASGITransport), and
|
||||||
|
`datasette serve` (cli.py) calls it too. The fast path below checks
|
||||||
|
both `_startup_invoked` and `_setup_db_done` - not just the former -
|
||||||
|
so that a bare `await ds.invoke_startup()` made by a caller ahead of
|
||||||
|
`_startup_sequence()` (which only sets `_startup_invoked`) can't
|
||||||
|
make this method skip the immutable-database table-count precompute.
|
||||||
|
"""
|
||||||
|
if self._startup_invoked and self._setup_db_done:
|
||||||
|
return
|
||||||
|
async with self._startup_lock:
|
||||||
|
if self._startup_invoked and self._setup_db_done:
|
||||||
|
return
|
||||||
|
if not self._setup_db_done:
|
||||||
|
# First time server starts up, calculate table counts for
|
||||||
|
# immutable databases
|
||||||
|
for database in self.databases.values():
|
||||||
|
if not database.is_mutable:
|
||||||
|
await database.table_counts(limit=60 * 60 * 1000)
|
||||||
|
self._setup_db_done = True
|
||||||
|
await self.invoke_startup()
|
||||||
|
|
||||||
def app(self):
|
def app(self):
|
||||||
"""Returns an ASGI app function that serves the whole of Datasette"""
|
"""Returns an ASGI app function that serves the whole of Datasette"""
|
||||||
routes = self._routes()
|
routes = self._routes()
|
||||||
|
|
||||||
async def setup_db():
|
|
||||||
# First time server starts up, calculate table counts for immutable databases
|
|
||||||
for database in self.databases.values():
|
|
||||||
if not database.is_mutable:
|
|
||||||
await database.table_counts(limit=60 * 60 * 1000)
|
|
||||||
|
|
||||||
async def _close_on_shutdown():
|
async def _close_on_shutdown():
|
||||||
self.close()
|
self.close()
|
||||||
|
|
||||||
asgi = CrossOriginProtectionMiddleware(DatasetteRouter(self, routes), self)
|
asgi = CrossOriginProtectionMiddleware(DatasetteRouter(self, routes), self)
|
||||||
if self.setting("trace_debug"):
|
if self.setting("trace_debug"):
|
||||||
asgi = AsgiTracer(asgi)
|
asgi = AsgiTracer(asgi)
|
||||||
asgi = AsgiLifespan(asgi, on_shutdown=[_close_on_shutdown])
|
asgi = AsgiLifespan(
|
||||||
asgi = AsgiRunOnFirstRequest(asgi, on_startup=[setup_db, self.invoke_startup])
|
asgi,
|
||||||
|
on_startup=[self._startup_sequence],
|
||||||
|
on_shutdown=[_close_on_shutdown],
|
||||||
|
)
|
||||||
|
asgi = AsgiRunOnFirstRequest(asgi, on_startup=[self._startup_sequence])
|
||||||
for wrapper in pm.hook.asgi_wrapper(datasette=self):
|
for wrapper in pm.hook.asgi_wrapper(datasette=self):
|
||||||
asgi = wrapper(asgi)
|
asgi = wrapper(asgi)
|
||||||
return asgi
|
return asgi
|
||||||
|
|
|
||||||
|
|
@ -663,16 +663,6 @@ def serve(
|
||||||
# Private utility mechanism for writing unit tests
|
# Private utility mechanism for writing unit tests
|
||||||
return ds
|
return ds
|
||||||
|
|
||||||
# Run async soundness checks before startup hooks, since invoke_startup
|
|
||||||
# now populates internal tables which requires querying each database
|
|
||||||
run_sync(lambda: check_databases(ds))
|
|
||||||
|
|
||||||
# Run the "startup" plugin hooks
|
|
||||||
try:
|
|
||||||
run_sync(ds.invoke_startup)
|
|
||||||
except StartupError as e:
|
|
||||||
raise click.ClickException(e.args[0])
|
|
||||||
|
|
||||||
if headers and not get:
|
if headers and not get:
|
||||||
raise click.ClickException("--headers can only be used with --get")
|
raise click.ClickException("--headers can only be used with --get")
|
||||||
|
|
||||||
|
|
@ -680,6 +670,14 @@ def serve(
|
||||||
raise click.ClickException("--token can only be used with --get")
|
raise click.ClickException("--token can only be used with --get")
|
||||||
|
|
||||||
if get:
|
if get:
|
||||||
|
# --get means we don't run Uvicorn at all
|
||||||
|
run_sync(lambda: check_databases(ds))
|
||||||
|
|
||||||
|
try:
|
||||||
|
run_sync(ds.invoke_startup)
|
||||||
|
except StartupError as e:
|
||||||
|
raise click.ClickException(e.args[0])
|
||||||
|
|
||||||
client = TestClient(ds)
|
client = TestClient(ds)
|
||||||
request_headers = {}
|
request_headers = {}
|
||||||
if token:
|
if token:
|
||||||
|
|
@ -704,34 +702,54 @@ def serve(
|
||||||
sys.exit(exit_code)
|
sys.exit(exit_code)
|
||||||
return
|
return
|
||||||
|
|
||||||
# Start the server
|
# check_databases, invoke_startup() and the uvicorn server all run on a
|
||||||
url = None
|
# single event loop, so that anything a plugin's "startup" hook schedules
|
||||||
if root:
|
# on the loop (asyncio.create_task, Lock/Queue/Event objects, ...) is
|
||||||
ds.root_enabled = True
|
# still alive when the server starts handling requests.
|
||||||
url = "http://{}:{}{}?token={}".format(
|
async def _serve_async():
|
||||||
host, port, ds.urls.path("-/auth-token"), ds._root_token
|
# Populate internal catalog tables before invoke_startup
|
||||||
)
|
await check_databases(ds)
|
||||||
click.echo(url)
|
|
||||||
if open_browser:
|
# Run the full startup sequence (immutable-database table-count
|
||||||
if url is None:
|
# precompute + the "startup" plugin hooks) via the same entry point
|
||||||
# Figure out most convenient URL - to table, database or homepage
|
# AsgiLifespan/AsgiRunOnFirstRequest use, so it's not skipped when
|
||||||
path = run_sync(lambda: initial_path_for_datasette(ds))
|
# uvicorn's lifespan.startup fires moments later.
|
||||||
url = f"http://{host}:{port}{path}"
|
try:
|
||||||
webbrowser.open(url)
|
await ds._startup_sequence()
|
||||||
uvicorn_kwargs = {
|
except StartupError as e:
|
||||||
"host": host,
|
raise click.ClickException(e.args[0])
|
||||||
"port": port,
|
|
||||||
"log_level": "info",
|
# Start the server
|
||||||
"lifespan": "on",
|
url = None
|
||||||
"workers": 1,
|
if root:
|
||||||
}
|
ds.root_enabled = True
|
||||||
if uds:
|
url = "http://{}:{}{}?token={}".format(
|
||||||
uvicorn_kwargs["uds"] = uds
|
host, port, ds.urls.path("-/auth-token"), ds._root_token
|
||||||
if ssl_keyfile:
|
)
|
||||||
uvicorn_kwargs["ssl_keyfile"] = ssl_keyfile
|
click.echo(url)
|
||||||
if ssl_certfile:
|
if open_browser:
|
||||||
uvicorn_kwargs["ssl_certfile"] = ssl_certfile
|
if url is None:
|
||||||
uvicorn.run(ds.app(), **uvicorn_kwargs)
|
# Figure out most convenient URL - to table, database or homepage
|
||||||
|
path = await initial_path_for_datasette(ds)
|
||||||
|
url = f"http://{host}:{port}{path}"
|
||||||
|
webbrowser.open(url)
|
||||||
|
uvicorn_kwargs = {
|
||||||
|
"host": host,
|
||||||
|
"port": port,
|
||||||
|
"log_level": "info",
|
||||||
|
"lifespan": "on",
|
||||||
|
"workers": 1,
|
||||||
|
}
|
||||||
|
if uds:
|
||||||
|
uvicorn_kwargs["uds"] = uds
|
||||||
|
if ssl_keyfile:
|
||||||
|
uvicorn_kwargs["ssl_keyfile"] = ssl_keyfile
|
||||||
|
if ssl_certfile:
|
||||||
|
uvicorn_kwargs["ssl_certfile"] = ssl_certfile
|
||||||
|
server = uvicorn.Server(uvicorn.Config(ds.app(), **uvicorn_kwargs))
|
||||||
|
await server.serve()
|
||||||
|
|
||||||
|
asyncio.run(_serve_async())
|
||||||
|
|
||||||
|
|
||||||
@cli.command()
|
@cli.command()
|
||||||
|
|
|
||||||
|
|
@ -1,3 +1,4 @@
|
||||||
|
import asyncio
|
||||||
import json
|
import json
|
||||||
import re
|
import re
|
||||||
from http.cookies import Morsel, SimpleCookie
|
from http.cookies import Morsel, SimpleCookie
|
||||||
|
|
@ -300,12 +301,24 @@ class AsgiLifespan:
|
||||||
while True:
|
while True:
|
||||||
message = await receive()
|
message = await receive()
|
||||||
if message["type"] == "lifespan.startup":
|
if message["type"] == "lifespan.startup":
|
||||||
for fn in self.on_startup:
|
try:
|
||||||
await fn()
|
for fn in self.on_startup:
|
||||||
|
await fn()
|
||||||
|
except Exception as e: # noqa: BLE001
|
||||||
|
await send(
|
||||||
|
{"type": "lifespan.startup.failed", "message": str(e)}
|
||||||
|
)
|
||||||
|
return
|
||||||
await send({"type": "lifespan.startup.complete"})
|
await send({"type": "lifespan.startup.complete"})
|
||||||
elif message["type"] == "lifespan.shutdown":
|
elif message["type"] == "lifespan.shutdown":
|
||||||
for fn in self.on_shutdown:
|
try:
|
||||||
await fn()
|
for fn in self.on_shutdown:
|
||||||
|
await fn()
|
||||||
|
except Exception as e: # noqa: BLE001
|
||||||
|
await send(
|
||||||
|
{"type": "lifespan.shutdown.failed", "message": str(e)}
|
||||||
|
)
|
||||||
|
return
|
||||||
await send({"type": "lifespan.shutdown.complete"})
|
await send({"type": "lifespan.shutdown.complete"})
|
||||||
return
|
return
|
||||||
else:
|
else:
|
||||||
|
|
@ -624,10 +637,23 @@ class AsgiRunOnFirstRequest:
|
||||||
self.asgi = asgi
|
self.asgi = asgi
|
||||||
self.on_startup = on_startup
|
self.on_startup = on_startup
|
||||||
self._started = False
|
self._started = False
|
||||||
|
# Guards against concurrent early requests interleaving with startup:
|
||||||
|
# without this, several requests could all observe `_started is
|
||||||
|
# False` and proceed before any of them finish running the hooks.
|
||||||
|
self._lock = asyncio.Lock()
|
||||||
|
|
||||||
async def __call__(self, scope, receive, send):
|
async def __call__(self, scope, receive, send):
|
||||||
if not self._started:
|
# Leave "lifespan" scope events alone - this shim only exists as a
|
||||||
self._started = True
|
# fallback for hosts that never send them. It wraps AsgiLifespan, so
|
||||||
for hook in self.on_startup:
|
# if it ran on_startup here too, a startup exception would escape
|
||||||
await hook()
|
# before AsgiLifespan's own try/except got a chance to turn it into
|
||||||
|
# a lifespan.startup.failed message.
|
||||||
|
if scope["type"] != "lifespan" and not self._started:
|
||||||
|
async with self._lock:
|
||||||
|
# Re-check: another request may have finished startup while
|
||||||
|
# we were waiting for the lock.
|
||||||
|
if not self._started:
|
||||||
|
for hook in self.on_startup:
|
||||||
|
await hook()
|
||||||
|
self._started = True
|
||||||
return await self.asgi(scope, receive, send)
|
return await self.asgi(scope, receive, send)
|
||||||
|
|
|
||||||
|
|
@ -30,7 +30,7 @@ dependencies = [
|
||||||
"hupper>=1.9",
|
"hupper>=1.9",
|
||||||
"httpx>=0.20,<1.0",
|
"httpx>=0.20,<1.0",
|
||||||
"pluggy>=1.0",
|
"pluggy>=1.0",
|
||||||
"uvicorn>=0.11",
|
"uvicorn>=0.29",
|
||||||
"aiofiles>=0.4",
|
"aiofiles>=0.4",
|
||||||
"PyYAML>=5.3",
|
"PyYAML>=5.3",
|
||||||
"mergedeep>=1.1.1",
|
"mergedeep>=1.1.1",
|
||||||
|
|
|
||||||
|
|
@ -2,6 +2,7 @@ import importlib.metadata
|
||||||
import os
|
import os
|
||||||
import pathlib
|
import pathlib
|
||||||
import re
|
import re
|
||||||
|
import socket
|
||||||
import subprocess
|
import subprocess
|
||||||
import sys
|
import sys
|
||||||
import tempfile
|
import tempfile
|
||||||
|
|
@ -32,17 +33,31 @@ UNDOCUMENTED_PERMISSIONS = {
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
def wait_until_responds(url, timeout=5.0, client=httpx, **kwargs):
|
def wait_until_responds(url, timeout=5.0, client=httpx, process=None, **kwargs):
|
||||||
start = time.time()
|
start = time.time()
|
||||||
while time.time() - start < timeout:
|
while time.time() - start < timeout:
|
||||||
|
# If the server died there is no point waiting out the timeout - fail
|
||||||
|
# now, with its output, instead of after `timeout` seconds of silence
|
||||||
|
if process is not None and process.poll() is not None:
|
||||||
|
raise AssertionError(
|
||||||
|
"Server exited early with returncode {}\n{}".format(
|
||||||
|
process.returncode, process.stdout.read().decode("utf-8")
|
||||||
|
)
|
||||||
|
)
|
||||||
try:
|
try:
|
||||||
client.get(url, **kwargs)
|
client.get(url, **kwargs)
|
||||||
return
|
return
|
||||||
except httpx.ConnectError:
|
except httpx.TransportError:
|
||||||
time.sleep(0.1)
|
time.sleep(0.1)
|
||||||
raise AssertionError(f"Timed out waiting for {url} to respond")
|
raise AssertionError(f"Timed out waiting for {url} to respond")
|
||||||
|
|
||||||
|
|
||||||
|
def find_free_port():
|
||||||
|
with socket.socket() as sock:
|
||||||
|
sock.bind(("127.0.0.1", 0))
|
||||||
|
return sock.getsockname()[1]
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def bare_ds():
|
def bare_ds():
|
||||||
"""
|
"""
|
||||||
|
|
@ -301,6 +316,71 @@ def ds_unix_domain_socket_server(tmp_path_factory):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def serve_with_plugins(tmp_path):
|
||||||
|
"""Factory fixture for starting ``datasette serve`` in a subprocess with
|
||||||
|
plugins written to a temporary ``--plugins-dir``.
|
||||||
|
|
||||||
|
For tests that need the real serve path: event-loop wiring, exit codes,
|
||||||
|
signals. The usual in-process ``pm.register`` plugin pattern can't reach
|
||||||
|
a subprocess, so plugin source is written out as importable files instead.
|
||||||
|
|
||||||
|
Unlike ``ds_localhost_http_server`` this is function-scoped and takes a
|
||||||
|
fresh port each time, because each test needs its own plugins. Call it as::
|
||||||
|
|
||||||
|
proc, port = serve_with_plugins({"my_plugin": PLUGIN_SOURCE})
|
||||||
|
|
||||||
|
``plugins`` maps module name to Python source. Pass
|
||||||
|
``wait_for_startup=False`` when the server is expected to fail during
|
||||||
|
startup rather than begin serving. Extra CLI arguments are passed through.
|
||||||
|
Every process started is terminated when the test ends.
|
||||||
|
"""
|
||||||
|
processes = []
|
||||||
|
|
||||||
|
def start(plugins, *extra_args, wait_for_startup=True):
|
||||||
|
plugins_dir = tmp_path / "plugins"
|
||||||
|
plugins_dir.mkdir(exist_ok=True)
|
||||||
|
for module_name, source in plugins.items():
|
||||||
|
(plugins_dir / f"{module_name}.py").write_text(source, "utf-8")
|
||||||
|
port = find_free_port()
|
||||||
|
proc = subprocess.Popen(
|
||||||
|
[
|
||||||
|
sys.executable,
|
||||||
|
"-m",
|
||||||
|
"datasette",
|
||||||
|
"--memory",
|
||||||
|
"--plugins-dir",
|
||||||
|
str(plugins_dir),
|
||||||
|
"-h",
|
||||||
|
"127.0.0.1",
|
||||||
|
"-p",
|
||||||
|
str(port),
|
||||||
|
*extra_args,
|
||||||
|
],
|
||||||
|
stdout=subprocess.PIPE,
|
||||||
|
stderr=subprocess.STDOUT,
|
||||||
|
# Avoid FileNotFoundError: [Errno 2] No such file or directory:
|
||||||
|
cwd=tempfile.gettempdir(),
|
||||||
|
)
|
||||||
|
processes.append(proc)
|
||||||
|
if wait_for_startup:
|
||||||
|
wait_until_responds(
|
||||||
|
f"http://127.0.0.1:{port}/-/versions.json", process=proc
|
||||||
|
)
|
||||||
|
return proc, port
|
||||||
|
|
||||||
|
yield start
|
||||||
|
|
||||||
|
for proc in processes:
|
||||||
|
if proc.poll() is None:
|
||||||
|
proc.terminate()
|
||||||
|
try:
|
||||||
|
proc.wait(timeout=5)
|
||||||
|
except subprocess.TimeoutExpired:
|
||||||
|
proc.kill()
|
||||||
|
proc.wait()
|
||||||
|
|
||||||
|
|
||||||
# Import fixtures from fixtures.py to make them available
|
# Import fixtures from fixtures.py to make them available
|
||||||
from .fixtures import ( # noqa: F401
|
from .fixtures import ( # noqa: F401
|
||||||
TEMP_PLUGIN_SECRET_FILE,
|
TEMP_PLUGIN_SECRET_FILE,
|
||||||
|
|
|
||||||
|
|
@ -1,4 +1,5 @@
|
||||||
import socket
|
import socket
|
||||||
|
import time
|
||||||
|
|
||||||
import httpx
|
import httpx
|
||||||
import pytest
|
import pytest
|
||||||
|
|
@ -28,3 +29,119 @@ def test_serve_unix_domain_socket(ds_unix_domain_socket_server):
|
||||||
"path": "/_memory",
|
"path": "/_memory",
|
||||||
"tables": [],
|
"tables": [],
|
||||||
}.items() <= response.json().items()
|
}.items() <= response.json().items()
|
||||||
|
|
||||||
|
|
||||||
|
# Shaped after datasette-litestream's startup hook, which schedules a
|
||||||
|
# background task with asyncio.get_running_loop().create_task(...):
|
||||||
|
# https://github.com/datasette/datasette-litestream
|
||||||
|
MARKER_TASK_PLUGIN = """
|
||||||
|
import asyncio
|
||||||
|
from datasette import hookimpl
|
||||||
|
from datasette.utils.asgi import Response
|
||||||
|
|
||||||
|
|
||||||
|
@hookimpl
|
||||||
|
def startup(datasette):
|
||||||
|
datasette._startup_calls = getattr(datasette, "_startup_calls", 0) + 1
|
||||||
|
|
||||||
|
async def _mark():
|
||||||
|
# Must await before setting the flag: a task with no internal
|
||||||
|
# await point could finish on the throwaway loop before it
|
||||||
|
# closed, masking the regression this test guards against.
|
||||||
|
await asyncio.sleep(0.2)
|
||||||
|
datasette._marker_task_ran = True
|
||||||
|
|
||||||
|
asyncio.get_running_loop().create_task(_mark())
|
||||||
|
|
||||||
|
|
||||||
|
@hookimpl
|
||||||
|
def register_routes():
|
||||||
|
async def marker_status(datasette):
|
||||||
|
return Response.json(
|
||||||
|
{
|
||||||
|
"marker_task_ran": getattr(datasette, "_marker_task_ran", False),
|
||||||
|
"startup_calls": getattr(datasette, "_startup_calls", 0),
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
return [(r"^/-/marker-task-ran$", marker_status)]
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
|
STARTUP_ERROR_PLUGIN = """
|
||||||
|
from datasette import hookimpl
|
||||||
|
from datasette.utils import StartupError
|
||||||
|
|
||||||
|
|
||||||
|
@hookimpl
|
||||||
|
def startup(datasette):
|
||||||
|
raise StartupError("boom from plugin")
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.serial
|
||||||
|
def test_startup_hook_background_task_runs_on_serving_loop(serve_with_plugins):
|
||||||
|
"""
|
||||||
|
Litestream-shaped regression test: a startup hook that does
|
||||||
|
asyncio.get_running_loop().create_task(...) must have that task
|
||||||
|
actually execute before/while the server is handling requests. This
|
||||||
|
only holds if invoke_startup() and uvicorn.Server.serve() share one
|
||||||
|
event loop. This test fails against unmodified main, where
|
||||||
|
invoke_startup() runs on a throwaway loop that is closed before
|
||||||
|
uvicorn opens its own loop to serve.
|
||||||
|
"""
|
||||||
|
_, port = serve_with_plugins({"marker_task_plugin": MARKER_TASK_PLUGIN})
|
||||||
|
# The fixture has already waited for the server to answer requests. The
|
||||||
|
# marker task deliberately awaits before setting its flag, so poll for a
|
||||||
|
# moment rather than assuming it landed before the first request arrived.
|
||||||
|
deadline = time.time() + 3.0
|
||||||
|
payload = {}
|
||||||
|
while time.time() < deadline:
|
||||||
|
payload = httpx.get(
|
||||||
|
f"http://127.0.0.1:{port}/-/marker-task-ran", timeout=1.0
|
||||||
|
).json()
|
||||||
|
if payload["marker_task_ran"]:
|
||||||
|
break
|
||||||
|
time.sleep(0.05)
|
||||||
|
assert payload.get("marker_task_ran"), (
|
||||||
|
"The startup hook's asyncio.create_task(...) never ran - "
|
||||||
|
"invoke_startup() and the server are not sharing an event loop"
|
||||||
|
)
|
||||||
|
# Polling above means this test would also pass if the startup hook were
|
||||||
|
# re-run on the serving loop by the first-request fallback - which would
|
||||||
|
# hide exactly the bug being tested. invoke_startup() is idempotent today
|
||||||
|
# so that cannot happen; assert it explicitly so that if the idempotency
|
||||||
|
# guard is ever removed this test fails loudly instead of silently
|
||||||
|
# becoming a no-op.
|
||||||
|
assert payload["startup_calls"] == 1, (
|
||||||
|
"startup hook ran {} times - the marker may have been set by a "
|
||||||
|
"re-run on the serving loop rather than by the original task".format(
|
||||||
|
payload["startup_calls"]
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.serial
|
||||||
|
def test_startup_error_fails_fast_before_port_binds(serve_with_plugins):
|
||||||
|
"""
|
||||||
|
A "startup" plugin hook that raises StartupError must fail fast: print
|
||||||
|
the message, exit non-zero, and never accept a connection on the port -
|
||||||
|
the failure must happen before uvicorn.Server binds the socket.
|
||||||
|
"""
|
||||||
|
proc, port = serve_with_plugins(
|
||||||
|
{"startup_error_plugin": STARTUP_ERROR_PLUGIN}, wait_for_startup=False
|
||||||
|
)
|
||||||
|
stdout, _ = proc.communicate(timeout=15)
|
||||||
|
output = stdout.decode("utf-8")
|
||||||
|
assert proc.returncode not in (0, None), output
|
||||||
|
assert "boom from plugin" in output, output
|
||||||
|
|
||||||
|
# Nothing is listening on the port now the process has exited. This
|
||||||
|
# confirms the socket was not left bound; on its own it cannot prove the
|
||||||
|
# failure preceded the bind, since a port nothing ever touched also
|
||||||
|
# refuses connections.
|
||||||
|
with (
|
||||||
|
pytest.raises(OSError),
|
||||||
|
socket.create_connection(("127.0.0.1", port), timeout=0.2),
|
||||||
|
):
|
||||||
|
pass
|
||||||
|
|
|
||||||
259
tests/test_lifespan.py
Normal file
259
tests/test_lifespan.py
Normal file
|
|
@ -0,0 +1,259 @@
|
||||||
|
"""
|
||||||
|
Tests for wiring Datasette startup (setup_db table counts + invoke_startup)
|
||||||
|
into the ASGI lifespan protocol.
|
||||||
|
|
||||||
|
These exercise Datasette._startup_sequence() via three different callers:
|
||||||
|
- AsgiLifespan, by hand-driving lifespan.startup messages (no HTTP request)
|
||||||
|
- AsgiRunOnFirstRequest, the fallback for hosts that never send lifespan
|
||||||
|
events (this is what DatasetteClient / plain httpx.ASGITransport uses)
|
||||||
|
- Both at once, to prove startup hooks run at most once
|
||||||
|
"""
|
||||||
|
|
||||||
|
import asyncio
|
||||||
|
import contextlib
|
||||||
|
import sqlite3
|
||||||
|
|
||||||
|
import httpx
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from datasette import hookimpl
|
||||||
|
from datasette.app import Datasette
|
||||||
|
from datasette.database import Database
|
||||||
|
from datasette.plugins import pm
|
||||||
|
|
||||||
|
|
||||||
|
async def _drive_lifespan_startup(app):
|
||||||
|
"""Send a single lifespan.startup message into app's ASGI lifespan loop
|
||||||
|
and return the list of messages sent back - without ever sending
|
||||||
|
lifespan.shutdown. Mirrors what a real server does: after startup
|
||||||
|
completes it parks waiting for the next event. We cancel that wait
|
||||||
|
once we've observed the startup response, rather than closing the
|
||||||
|
Datasette instance down with a shutdown message.
|
||||||
|
"""
|
||||||
|
messages_sent = []
|
||||||
|
startup_responded = asyncio.Event()
|
||||||
|
delivered = False
|
||||||
|
|
||||||
|
async def receive():
|
||||||
|
nonlocal delivered
|
||||||
|
if not delivered:
|
||||||
|
delivered = True
|
||||||
|
return {"type": "lifespan.startup"}
|
||||||
|
# No further messages: block until the task is cancelled below,
|
||||||
|
# same as a real server parked waiting for lifespan.shutdown.
|
||||||
|
await asyncio.Event().wait()
|
||||||
|
|
||||||
|
async def send(message):
|
||||||
|
messages_sent.append(message)
|
||||||
|
startup_responded.set()
|
||||||
|
|
||||||
|
task = asyncio.create_task(app({"type": "lifespan"}, receive, send))
|
||||||
|
try:
|
||||||
|
await asyncio.wait_for(startup_responded.wait(), timeout=5)
|
||||||
|
finally:
|
||||||
|
task.cancel()
|
||||||
|
with contextlib.suppress(asyncio.CancelledError):
|
||||||
|
await task
|
||||||
|
return messages_sent
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_lifespan_startup_runs_before_any_request():
|
||||||
|
ds = Datasette(memory=True)
|
||||||
|
assert ds._startup_invoked is False
|
||||||
|
app = ds.app()
|
||||||
|
|
||||||
|
messages = await _drive_lifespan_startup(app)
|
||||||
|
|
||||||
|
assert {"type": "lifespan.startup.complete"} in messages
|
||||||
|
assert ds._startup_invoked is True
|
||||||
|
# Internal catalog tables should be populated too, entirely without an
|
||||||
|
# HTTP request having been made.
|
||||||
|
internal_db = ds.get_internal_database()
|
||||||
|
databases = await internal_db.execute("select * from catalog_databases")
|
||||||
|
assert len(databases.rows) >= 1
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_lifespan_startup_failure_reports_lifespan_startup_failed():
|
||||||
|
class RaisingStartupPlugin:
|
||||||
|
__name__ = "RaisingStartupPlugin"
|
||||||
|
|
||||||
|
@hookimpl
|
||||||
|
def startup(self, datasette):
|
||||||
|
async def inner():
|
||||||
|
raise RuntimeError("boom from startup hook")
|
||||||
|
|
||||||
|
return inner
|
||||||
|
|
||||||
|
ds = Datasette(memory=True)
|
||||||
|
pm.register(RaisingStartupPlugin(), name="raising_startup_plugin")
|
||||||
|
try:
|
||||||
|
app = ds.app()
|
||||||
|
messages = await _drive_lifespan_startup(app)
|
||||||
|
finally:
|
||||||
|
pm.unregister(name="raising_startup_plugin")
|
||||||
|
|
||||||
|
assert messages == [
|
||||||
|
{"type": "lifespan.startup.failed", "message": "boom from startup hook"}
|
||||||
|
]
|
||||||
|
# The exception happened before invoke_startup() got to the end of its
|
||||||
|
# body, so startup is not considered to have completed.
|
||||||
|
assert ds._startup_invoked is False
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_startup_runs_exactly_once_across_lifespan_and_first_request():
|
||||||
|
call_count = {"n": 0}
|
||||||
|
|
||||||
|
class CountingStartupPlugin:
|
||||||
|
__name__ = "CountingStartupPlugin"
|
||||||
|
|
||||||
|
@hookimpl
|
||||||
|
def startup(self, datasette):
|
||||||
|
async def inner():
|
||||||
|
call_count["n"] += 1
|
||||||
|
|
||||||
|
return inner
|
||||||
|
|
||||||
|
ds = Datasette(memory=True)
|
||||||
|
pm.register(CountingStartupPlugin(), name="counting_startup_plugin")
|
||||||
|
try:
|
||||||
|
# Build the ASGI app once, the way a real deployment does - and
|
||||||
|
# reuse the SAME app instance for both the lifespan drive and the
|
||||||
|
# HTTP requests below, since a fresh ds.app() call would reset the
|
||||||
|
# AsgiRunOnFirstRequest fallback's state.
|
||||||
|
app = ds.app()
|
||||||
|
|
||||||
|
messages = await _drive_lifespan_startup(app)
|
||||||
|
assert {"type": "lifespan.startup.complete"} in messages
|
||||||
|
assert call_count["n"] == 1
|
||||||
|
|
||||||
|
# A first HTTP request (as if the host never sent lifespan events,
|
||||||
|
# or lifespan already ran) should not run the hook again.
|
||||||
|
transport = httpx.ASGITransport(app=app)
|
||||||
|
async with httpx.AsyncClient(
|
||||||
|
transport=transport, base_url="http://localhost"
|
||||||
|
) as client:
|
||||||
|
response1 = await client.get("/-/versions.json")
|
||||||
|
assert response1.status_code == 200
|
||||||
|
# ... nor should a second, repeat request.
|
||||||
|
response2 = await client.get("/-/versions.json")
|
||||||
|
assert response2.status_code == 200
|
||||||
|
finally:
|
||||||
|
pm.unregister(name="counting_startup_plugin")
|
||||||
|
|
||||||
|
assert call_count["n"] == 1
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_no_lifespan_first_request_still_triggers_startup():
|
||||||
|
# Pin today's behavior: a client that never drives ASGI lifespan events
|
||||||
|
# at all (like httpx.ASGITransport, which DatasetteClient uses) still
|
||||||
|
# gets startup armed by the AsgiRunOnFirstRequest fallback.
|
||||||
|
ds = Datasette(memory=True)
|
||||||
|
assert ds._startup_invoked is False
|
||||||
|
app = ds.app()
|
||||||
|
transport = httpx.ASGITransport(app=app)
|
||||||
|
async with httpx.AsyncClient(
|
||||||
|
transport=transport, base_url="http://localhost"
|
||||||
|
) as client:
|
||||||
|
response = await client.get("/-/versions.json")
|
||||||
|
assert response.status_code == 200
|
||||||
|
|
||||||
|
assert ds._startup_invoked is True
|
||||||
|
internal_db = ds.get_internal_database()
|
||||||
|
databases = await internal_db.execute("select * from catalog_databases")
|
||||||
|
assert len(databases.rows) >= 1
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_datasette_client_first_request_triggers_startup():
|
||||||
|
# Same as above, but through the real DatasetteClient (ds.client) that
|
||||||
|
# plugins and tests actually use, to confirm nothing regressed there.
|
||||||
|
ds = Datasette(memory=True)
|
||||||
|
assert ds._startup_invoked is False
|
||||||
|
response = await ds.client.get("/-/versions.json")
|
||||||
|
assert response.status_code == 200
|
||||||
|
assert ds._startup_invoked is True
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_concurrent_first_requests_all_wait_for_slow_startup():
|
||||||
|
call_count = {"n": 0}
|
||||||
|
|
||||||
|
class SlowStartupPlugin:
|
||||||
|
__name__ = "SlowStartupPlugin"
|
||||||
|
|
||||||
|
@hookimpl
|
||||||
|
def startup(self, datasette):
|
||||||
|
async def inner():
|
||||||
|
call_count["n"] += 1
|
||||||
|
await asyncio.sleep(0.2)
|
||||||
|
|
||||||
|
return inner
|
||||||
|
|
||||||
|
ds = Datasette(memory=True)
|
||||||
|
pm.register(SlowStartupPlugin(), name="slow_startup_plugin")
|
||||||
|
try:
|
||||||
|
app = ds.app()
|
||||||
|
transport = httpx.ASGITransport(app=app)
|
||||||
|
async with httpx.AsyncClient(
|
||||||
|
transport=transport, base_url="http://localhost"
|
||||||
|
) as client:
|
||||||
|
responses = await asyncio.gather(
|
||||||
|
*[client.get("/-/versions.json") for _ in range(10)]
|
||||||
|
)
|
||||||
|
finally:
|
||||||
|
pm.unregister(name="slow_startup_plugin")
|
||||||
|
|
||||||
|
# Every one of the 10 simultaneous first requests must have blocked
|
||||||
|
# until startup actually finished, not raced ahead of it.
|
||||||
|
assert all(response.status_code == 200 for response in responses)
|
||||||
|
assert call_count["n"] == 1
|
||||||
|
assert ds._startup_invoked is True
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_setup_db_still_runs_when_invoke_startup_ran_first(tmp_path, monkeypatch):
|
||||||
|
# Regression test: `datasette serve` (cli.py _serve_async) calls
|
||||||
|
# ds.invoke_startup() directly, before uvicorn ever sends a
|
||||||
|
# lifespan.startup event that drives _startup_sequence(). If
|
||||||
|
# _startup_sequence()'s fast path only checked `_startup_invoked`, it
|
||||||
|
# would see startup already done and skip the immutable-database
|
||||||
|
# table-count precompute (setup_db) entirely - a silent regression
|
||||||
|
# versus main, where AsgiRunOnFirstRequest ran setup_db unconditionally
|
||||||
|
# on request #1.
|
||||||
|
db_path = tmp_path / "immutable.db"
|
||||||
|
conn = sqlite3.connect(str(db_path))
|
||||||
|
conn.execute("create table t (id integer primary key)")
|
||||||
|
conn.commit()
|
||||||
|
conn.close()
|
||||||
|
|
||||||
|
ds = Datasette([], immutables=[str(db_path)])
|
||||||
|
|
||||||
|
call_count = {"n": 0}
|
||||||
|
original_table_counts = Database.table_counts
|
||||||
|
|
||||||
|
async def counting_table_counts(self, *args, **kwargs):
|
||||||
|
call_count["n"] += 1
|
||||||
|
return await original_table_counts(self, *args, **kwargs)
|
||||||
|
|
||||||
|
monkeypatch.setattr(Database, "table_counts", counting_table_counts)
|
||||||
|
|
||||||
|
# Simulate the CLI path: invoke_startup() runs directly and completes
|
||||||
|
# BEFORE _startup_sequence() ever gets a chance to run setup_db.
|
||||||
|
await ds.invoke_startup()
|
||||||
|
assert ds._startup_invoked is True
|
||||||
|
assert call_count["n"] == 0
|
||||||
|
|
||||||
|
# The lifespan/first-request path (or the CLI itself, per the fix)
|
||||||
|
# calling the shared entry point afterwards must still precompute
|
||||||
|
# table counts for immutable databases.
|
||||||
|
await ds._startup_sequence()
|
||||||
|
assert call_count["n"] == 1
|
||||||
|
assert ds._setup_db_done is True
|
||||||
|
|
||||||
|
# Idempotency: a second call must not recompute.
|
||||||
|
await ds._startup_sequence()
|
||||||
|
assert call_count["n"] == 1
|
||||||
Loading…
Add table
Add a link
Reference in a new issue