mirror of
https://github.com/simonw/datasette.git
synced 2026-10-25 17:56:57 +01:00
Merge spike/budget into spike/final
Conflicts: - database.py: keep both class attributes (is_scratch from scratch, immutable_count_time_limit_ms from budget). - cli.py check_databases(): take budget's catalog-first check and keep scratch's rule that scratch databases are never checked (they are skipped in _databases_needing_check()); the --crossdb warning no longer counts scratch databases, which are never ATTACHed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EwkLo2rJWidpkDVT3HkvNv
This commit is contained in:
Claude
commit
c786d91cf9
20 changed files with 1934 additions and 259 deletions
204
tests/connection_budget.py
Normal file
204
tests/connection_budget.py
Normal file
|
|
@ -0,0 +1,204 @@
|
|||
"""Record which databases a block of code touches, and assert a budget.
|
||||
|
||||
With a capped, idle-reaped connection pool, anything that loops over every
|
||||
attached database opens (and evicts) a connection per database at a 0% hit
|
||||
rate. These helpers make "this request only touches the databases it is
|
||||
about" a testable property::
|
||||
|
||||
async with connection_budget() as budget:
|
||||
await ds.client.get("/db1/t1.json")
|
||||
budget.assert_only("db1") # plus _internal, always allowed
|
||||
|
||||
What counts as touching a database:
|
||||
|
||||
* ``opened``: ``Database.connect()`` calls - pooled read connections, write
|
||||
connections, isolated connections and the SchemaWatcher's short-lived
|
||||
scan/pragma connections (``track=False``)
|
||||
* ``reads``: ``Database._execute_fn()`` calls (every ``execute()``,
|
||||
``execute_fn()`` and introspection helper), whether or not the pool had an
|
||||
idle connection - so a warm pool does not hide a loop over every database
|
||||
* ``writes``: ``execute_write*()``, ``execute_isolated_fn()`` and work run on
|
||||
the write connection
|
||||
* ``attached``: databases ATTACHed to a connection while it was prepared
|
||||
(``--crossdb``)
|
||||
* ``probes``: ``sqlite3.connect()`` calls that bypass ``Database.connect()``
|
||||
(``:memory:`` feature probes and the like), keyed by filename
|
||||
|
||||
The patches are installed process-wide for the duration of the block, so
|
||||
background work (the SchemaWatcher's polling task, the read pool's reaper)
|
||||
that runs during the block is counted too. Tests that want to measure one
|
||||
request should turn polling off (``schema_watch_interval_ms=0``) or measure
|
||||
it separately.
|
||||
"""
|
||||
|
||||
import collections
|
||||
import contextlib
|
||||
import threading
|
||||
|
||||
from datasette import app as app_module
|
||||
from datasette.database import Database
|
||||
from datasette.utils import sqlite3
|
||||
|
||||
INTERNAL = "__INTERNAL__"
|
||||
|
||||
_local = threading.local()
|
||||
|
||||
|
||||
class ConnectionBudget:
|
||||
def __init__(self):
|
||||
self._lock = threading.Lock()
|
||||
self.opened = collections.Counter()
|
||||
self.reads = collections.Counter()
|
||||
self.writes = collections.Counter()
|
||||
self.attached = collections.Counter()
|
||||
self.probes = collections.Counter()
|
||||
|
||||
def _add(self, counter, key):
|
||||
with self._lock:
|
||||
counter[key] += 1
|
||||
|
||||
def touched(self):
|
||||
"""Names of every database opened, read, written or ATTACHed."""
|
||||
names = set()
|
||||
for counter in (self.opened, self.reads, self.writes, self.attached):
|
||||
names.update(counter)
|
||||
return names
|
||||
|
||||
def summary(self):
|
||||
return {
|
||||
"opened": dict(self.opened),
|
||||
"reads": dict(self.reads),
|
||||
"writes": dict(self.writes),
|
||||
"attached": dict(self.attached),
|
||||
"probes": dict(self.probes),
|
||||
}
|
||||
|
||||
def assert_only(self, *names, internal=True, probes=0):
|
||||
"""Assert that nothing outside ``names`` was touched.
|
||||
|
||||
``_internal`` is allowed unless ``internal=False``. ``probes`` is the
|
||||
number of direct ``sqlite3.connect()`` calls allowed.
|
||||
"""
|
||||
allowed = set(names)
|
||||
if internal:
|
||||
allowed.add(INTERNAL)
|
||||
extra = self.touched() - allowed
|
||||
assert not extra, (
|
||||
f"touched {len(extra)} database(s) outside the budget "
|
||||
f"{sorted(allowed)}: {sorted(extra)[:10]}"
|
||||
f"{' ...' if len(extra) > 10 else ''}\n{self.summary()}"
|
||||
)
|
||||
total_probes = sum(self.probes.values())
|
||||
assert (
|
||||
total_probes <= probes
|
||||
), f"{total_probes} direct sqlite3.connect() call(s): {dict(self.probes)}"
|
||||
|
||||
def assert_at_most(self, n, *, internal=True):
|
||||
"""Assert that at most ``n`` user databases were touched."""
|
||||
touched = self.touched()
|
||||
if internal:
|
||||
touched.discard(INTERNAL)
|
||||
assert len(touched) <= n, (
|
||||
f"touched {len(touched)} databases, budget {n}: "
|
||||
f"{sorted(touched)[:10]}\n{self.summary()}"
|
||||
)
|
||||
|
||||
def assert_no_connections(self, internal=True):
|
||||
"""Assert that no connection was opened at all (``_internal`` aside)."""
|
||||
opened = {
|
||||
k: v for k, v in self.opened.items() if not (internal and k == INTERNAL)
|
||||
}
|
||||
assert not opened, f"opened connections: {opened}"
|
||||
assert not self.attached, f"attached: {dict(self.attached)}"
|
||||
assert not self.probes, f"probes: {dict(self.probes)}"
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def _patched(budget):
|
||||
original_connect = Database.connect
|
||||
original_execute_fn = Database._execute_fn
|
||||
original_execute_write_fn = Database._execute_write_fn
|
||||
original_isolated = Database.execute_isolated_fn
|
||||
original_on_write_conn = Database._execute_on_write_connection
|
||||
original_prepare = app_module.Datasette._prepare_connection
|
||||
original_sqlite_connect = sqlite3.connect
|
||||
|
||||
def connect(self, *args, **kwargs):
|
||||
budget._add(budget.opened, self.name)
|
||||
_local.depth = getattr(_local, "depth", 0) + 1
|
||||
try:
|
||||
return original_connect(self, *args, **kwargs)
|
||||
finally:
|
||||
_local.depth -= 1
|
||||
|
||||
async def _execute_fn(self, fn):
|
||||
budget._add(budget.reads, self.name)
|
||||
return await original_execute_fn(self, fn)
|
||||
|
||||
async def _execute_write_fn(self, fn, *args, **kwargs):
|
||||
budget._add(budget.writes, self.name)
|
||||
return await original_execute_write_fn(self, fn, *args, **kwargs)
|
||||
|
||||
async def execute_isolated_fn(self, fn):
|
||||
budget._add(budget.writes, self.name)
|
||||
return await original_isolated(self, fn)
|
||||
|
||||
async def _execute_on_write_connection(self, fn):
|
||||
budget._add(budget.writes, self.name)
|
||||
return await original_on_write_conn(self, fn)
|
||||
|
||||
def _prepare_connection(self, conn, database, **kwargs):
|
||||
result = original_prepare(self, conn, database, **kwargs)
|
||||
try:
|
||||
rows = conn.execute("PRAGMA database_list").fetchall()
|
||||
except Exception: # noqa: BLE001
|
||||
rows = []
|
||||
for row in rows:
|
||||
name = row[1]
|
||||
if name not in ("main", "temp"):
|
||||
budget._add(budget.attached, name)
|
||||
return result
|
||||
|
||||
def sqlite_connect(database, *args, **kwargs):
|
||||
if not getattr(_local, "depth", 0):
|
||||
budget._add(budget.probes, str(database))
|
||||
return original_sqlite_connect(database, *args, **kwargs)
|
||||
|
||||
Database.connect = connect
|
||||
Database._execute_fn = _execute_fn
|
||||
Database._execute_write_fn = _execute_write_fn
|
||||
Database.execute_isolated_fn = execute_isolated_fn
|
||||
Database._execute_on_write_connection = _execute_on_write_connection
|
||||
app_module.Datasette._prepare_connection = _prepare_connection
|
||||
sqlite3.connect = sqlite_connect
|
||||
try:
|
||||
yield budget
|
||||
finally:
|
||||
Database.connect = original_connect
|
||||
Database._execute_fn = original_execute_fn
|
||||
Database._execute_write_fn = original_execute_write_fn
|
||||
Database.execute_isolated_fn = original_isolated
|
||||
Database._execute_on_write_connection = original_on_write_conn
|
||||
app_module.Datasette._prepare_connection = original_prepare
|
||||
sqlite3.connect = original_sqlite_connect
|
||||
|
||||
|
||||
class connection_budget:
|
||||
"""Context manager (sync or async) yielding a ConnectionBudget."""
|
||||
|
||||
def __init__(self):
|
||||
self.budget = ConnectionBudget()
|
||||
self._cm = None
|
||||
|
||||
def __enter__(self):
|
||||
self._cm = _patched(self.budget)
|
||||
return self._cm.__enter__()
|
||||
|
||||
def __exit__(self, *exc):
|
||||
return self._cm.__exit__(*exc)
|
||||
|
||||
async def __aenter__(self):
|
||||
return self.__enter__()
|
||||
|
||||
async def __aexit__(self, *exc):
|
||||
return self.__exit__(*exc)
|
||||
652
tests/test_connection_budget.py
Normal file
652
tests/test_connection_budget.py
Normal file
|
|
@ -0,0 +1,652 @@
|
|||
"""Requests, startup and background work only touch the databases they need.
|
||||
|
||||
With a capped, idle-reaped connection pool, a loop over every attached
|
||||
database opens (and evicts) one connection per database. These tests attach
|
||||
50 databases and use tests/connection_budget.py to assert which databases
|
||||
each endpoint opens, reads or writes. Every test uses a fresh instance, so
|
||||
caches that are only warm after a first request do not hide a sweep.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import sqlite3
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
from datasette.app import Datasette
|
||||
from datasette.utils.catalog import (
|
||||
catalog_relationship_counts,
|
||||
catalog_summaries,
|
||||
catalog_table_details,
|
||||
)
|
||||
from datasette.utils.sqlite import (
|
||||
sqlite_derived_table_dependencies,
|
||||
supports_table_list,
|
||||
)
|
||||
|
||||
from .connection_budget import INTERNAL, connection_budget
|
||||
|
||||
N = 50
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def only_default_plugins():
|
||||
"""Plugins other tests leave registered (tests/plugins/my_plugin.py and
|
||||
friends query datasette.get_database() from their hooks) are not part
|
||||
of what these tests measure: unregister them for the duration."""
|
||||
from datasette.plugins import DEFAULT_PLUGINS, pm
|
||||
|
||||
removed = [
|
||||
(name, plugin)
|
||||
for name, plugin in pm.list_name_plugin()
|
||||
if name not in DEFAULT_PLUGINS
|
||||
]
|
||||
for name, _ in removed:
|
||||
pm.unregister(name=name)
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
for name, plugin in removed:
|
||||
if pm.get_plugin(name) is None:
|
||||
pm.register(plugin, name=name)
|
||||
|
||||
|
||||
def _make_db(path, i):
|
||||
conn = sqlite3.connect(path)
|
||||
conn.executescript("""
|
||||
create table t2 (id integer primary key, title text);
|
||||
create table t1 (
|
||||
id integer primary key, name text, other_id integer references t2(id)
|
||||
);
|
||||
""")
|
||||
conn.executemany(
|
||||
"insert into t1 (name, other_id) values (?, ?)",
|
||||
[(f"n{j}", j % 3) for j in range(20)],
|
||||
)
|
||||
conn.executemany(
|
||||
"insert into t2 (title) values (?)", [(f"t{j}",) for j in range(3)]
|
||||
)
|
||||
if i % 5 == 0:
|
||||
conn.executescript("""
|
||||
create virtual table t1_fts using fts5(
|
||||
name, content="t1", content_rowid='id'
|
||||
);
|
||||
insert into t1_fts(t1_fts) values('rebuild');
|
||||
""")
|
||||
if i % 3 == 0:
|
||||
conn.execute("create view v1 as select * from t1")
|
||||
if i % 7 == 0:
|
||||
conn.execute("create table _hidden (x)")
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def db_files(tmp_path_factory):
|
||||
directory = tmp_path_factory.mktemp("budget")
|
||||
paths = []
|
||||
for i in range(N):
|
||||
path = str(directory / f"db{i:03d}.db")
|
||||
_make_db(path, i)
|
||||
paths.append(path)
|
||||
# Older than the SchemaWatcher's racy window, so a sweep is stat-only
|
||||
old = time.time() - 60
|
||||
for path in paths:
|
||||
os.utime(path, (old, old))
|
||||
return paths
|
||||
|
||||
|
||||
def _ds(paths, **kwargs):
|
||||
settings = {"schema_watch_interval_ms": 0}
|
||||
settings.update(kwargs.pop("settings", {}))
|
||||
ds = Datasette(paths, settings=settings, **kwargs)
|
||||
ds.root_enabled = True
|
||||
return ds
|
||||
|
||||
|
||||
def _root(ds):
|
||||
return {"ds_actor": ds.client.actor_cookie({"id": "root"})}
|
||||
|
||||
|
||||
async def _get(ds, path, cookies=None):
|
||||
async with connection_budget() as budget:
|
||||
response = await ds.client.get(path, cookies=cookies or {})
|
||||
assert response.status_code == 200, (path, response.status_code, response.text)
|
||||
return response, budget
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("path", ["/", "/.json", "/-/", "/-/.json"])
|
||||
@pytest.mark.parametrize("as_root", [False, True])
|
||||
async def test_index_page_opens_no_database(db_files, path, as_root):
|
||||
ds = _ds(db_files)
|
||||
await ds.invoke_startup()
|
||||
response, budget = await _get(ds, path, _root(ds) if as_root else None)
|
||||
budget.assert_only()
|
||||
budget.assert_no_connections()
|
||||
if path.endswith(".json"):
|
||||
data = response.json()
|
||||
assert len(data["databases"]) == N
|
||||
# More databases than COUNT_MAX_DATABASES: no row counts
|
||||
assert not any(d["show_table_row_counts"] for d in data["databases"])
|
||||
db000 = next(d for d in data["databases"] if d["name"] == "db000")
|
||||
assert db000["tables_count"] == 2 # t1, t2
|
||||
# _hidden, and t1_fts (an external content FTS table is hidden).
|
||||
# Its fts5 shadow tables are not listed at all: they derive from
|
||||
# t1_fts, which is itself derived (the one-hop permission policy)
|
||||
assert db000["hidden_tables_count"] == 2
|
||||
assert db000["views_count"] == 1
|
||||
shown = {t["name"]: t for t in db000["tables_and_views_truncated"]}
|
||||
assert shown["t1"]["columns"] == ["id", "name", "other_id"]
|
||||
assert shown["t1"]["primary_keys"] == ["id"]
|
||||
assert shown["t1"]["fts_table"] == "t1_fts"
|
||||
# Sorted by foreign key relationships when there are no counts
|
||||
assert shown["t1"]["num_relationships_for_sorting"] == 1
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_index_page_counts_rows_for_few_databases(db_files):
|
||||
# At most COUNT_MAX_DATABASES databases: rows are counted, and only the
|
||||
# databases listed are touched
|
||||
ds = _ds(db_files[:3])
|
||||
await ds.invoke_startup()
|
||||
response, budget = await _get(ds, "/.json")
|
||||
budget.assert_only("db000", "db001", "db002")
|
||||
data = response.json()
|
||||
assert all(d["show_table_row_counts"] for d in data["databases"])
|
||||
assert data["databases"][0]["table_rows_sum"] == 23
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_index_page_uses_cached_immutable_counts(db_files):
|
||||
# Immutable databases' counts are computed once, when first needed, and
|
||||
# reused even when there are too many databases to count
|
||||
ds = Datasette(
|
||||
db_files[3:],
|
||||
immutables=db_files[:3],
|
||||
settings={"schema_watch_interval_ms": 0},
|
||||
)
|
||||
await ds.invoke_startup()
|
||||
for name in ("db000", "db001", "db002"):
|
||||
await ds.get_database(name).table_counts()
|
||||
response, budget = await _get(ds, "/.json")
|
||||
budget.assert_no_connections()
|
||||
budget.assert_only()
|
||||
by_name = {d["name"]: d for d in response.json()["databases"]}
|
||||
assert by_name["db000"]["show_table_row_counts"]
|
||||
assert not by_name["db010"]["show_table_row_counts"]
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("path", ["/-/databases", "/-/databases.json"])
|
||||
async def test_databases_endpoint(db_files, path):
|
||||
ds = _ds(db_files)
|
||||
await ds.invoke_startup()
|
||||
_, budget = await _get(ds, path)
|
||||
budget.assert_only()
|
||||
budget.assert_no_connections()
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"path",
|
||||
[
|
||||
"/db001",
|
||||
"/db001.json",
|
||||
"/db000.json?_search=n",
|
||||
"/db001/t1",
|
||||
"/db001/t1.json",
|
||||
"/db001/t1?_facet=name",
|
||||
"/db001/t1.json?_facet=name",
|
||||
"/db000/t1_fts.json?_search=n1",
|
||||
"/db001/-/query?sql=select+1",
|
||||
"/db001/-/query.json?sql=select+1",
|
||||
"/db001/-/query",
|
||||
"/db001/t1/1",
|
||||
"/db001/t1/1.json",
|
||||
"/db001/-/schema.json",
|
||||
"/db001/t1/-/autocomplete?q=n",
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize("as_root", [False, True])
|
||||
async def test_database_scoped_pages_touch_only_their_database(db_files, path, as_root):
|
||||
ds = _ds(db_files)
|
||||
await ds.invoke_startup()
|
||||
database = path.split("/")[1].split(".")[0]
|
||||
_, budget = await _get(ds, path, _root(ds) if as_root else None)
|
||||
budget.assert_only(database)
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"path,as_root",
|
||||
[
|
||||
("/-/versions", False),
|
||||
("/-/versions.json", False),
|
||||
("/-/plugins.json", False),
|
||||
("/-/settings.json", False),
|
||||
("/-/config.json", False),
|
||||
("/-/actor.json", False),
|
||||
("/-/threads", True),
|
||||
("/-/threads.json", True),
|
||||
("/-/tasks.json", True),
|
||||
("/-/actions.json", True),
|
||||
("/-/jump?q=t1", False),
|
||||
("/-/jump.json?q=t1", False),
|
||||
("/-/jump.json", False),
|
||||
("/-/allowed.json?action=view-table", False),
|
||||
("/-/allowed.json?action=view-table", True),
|
||||
("/-/allowed.json?action=view-database", False),
|
||||
("/-/allowed?action=view-table", True),
|
||||
("/-/rules.json?action=view-table", True),
|
||||
("/-/check.json?action=view-table&parent=db001&child=t1", True),
|
||||
("/-/permissions", True),
|
||||
("/-/allow-debug", False),
|
||||
("/-/api", False),
|
||||
("/-/api", True),
|
||||
("/-/create-token", True),
|
||||
("/-/queries.json", False),
|
||||
("/-/patterns", False),
|
||||
],
|
||||
)
|
||||
async def test_instance_pages_open_no_user_database(db_files, path, as_root):
|
||||
ds = _ds(db_files)
|
||||
await ds.invoke_startup()
|
||||
# /-/versions probes the SQLite library once per instance
|
||||
await ds.client.get("/-/versions.json")
|
||||
_, budget = await _get(ds, path, _root(ds) if as_root else None)
|
||||
budget.assert_only()
|
||||
budget.assert_no_connections()
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_versions_probes_sqlite_once(db_files):
|
||||
ds = _ds(db_files[:12], crossdb=True)
|
||||
await ds.invoke_startup()
|
||||
async with connection_budget() as budget:
|
||||
for _ in range(3):
|
||||
assert (await ds.client.get("/-/versions.json")).status_code == 200
|
||||
# One :memory: connection for the first request, none after; it is not
|
||||
# given the --crossdb ATTACHes
|
||||
assert sum(budget.probes.values()) == 1
|
||||
assert not budget.attached
|
||||
budget.assert_only(probes=1)
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_detect_json1_is_cached(db_files):
|
||||
ds = _ds(db_files[:2])
|
||||
await ds.invoke_startup()
|
||||
await ds.client.get("/db001/t1")
|
||||
_, budget = await _get(ds, "/db001/t1?_facet=name")
|
||||
assert not budget.probes
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_instance_schema_reads_each_visible_database_once(db_files):
|
||||
# /-/schema dumps every visible database's sqlite_master: one read each,
|
||||
# nothing for databases the actor cannot see
|
||||
ds = _ds(
|
||||
db_files[:10],
|
||||
config={"databases": {"db003": {"allow": False}}},
|
||||
)
|
||||
await ds.invoke_startup()
|
||||
_, budget = await _get(ds, "/-/schema.json")
|
||||
expected = {f"db{i:03d}" for i in range(10)} - {"db003"}
|
||||
assert budget.touched() - {INTERNAL} == expected
|
||||
assert all(budget.reads[name] == 1 for name in expected)
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_autocomplete_debug_is_bounded(db_files):
|
||||
ds = _ds(db_files)
|
||||
await ds.invoke_startup()
|
||||
_, budget = await _get(ds, "/-/debug/autocomplete")
|
||||
# Only databases holding one of the (at most 5) suggestions
|
||||
budget.assert_at_most(5)
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_execute_write_touches_only_its_database(db_files, tmp_path):
|
||||
paths = []
|
||||
for path in db_files[:20]:
|
||||
target = tmp_path / os.path.basename(path)
|
||||
shutil.copy(path, target)
|
||||
paths.append(str(target))
|
||||
ds = _ds(paths)
|
||||
await ds.invoke_startup()
|
||||
async with connection_budget() as budget:
|
||||
response = await ds.client.post(
|
||||
"/db004/-/execute-write",
|
||||
actor={"id": "root"},
|
||||
data={"sql": "create table new_table (id integer primary key)"},
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
assert "Query executed" in response.text
|
||||
budget.assert_only("db004")
|
||||
# The catalog already has the new table
|
||||
rows = await ds.get_internal_database().execute(
|
||||
"select 1 from catalog_tables where database_name = 'db004' and table_name = 'new_table'"
|
||||
)
|
||||
assert rows.rows
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_telemetry_metric_callbacks_open_nothing(db_files):
|
||||
from datasette import telemetry
|
||||
|
||||
ds = _ds(db_files)
|
||||
await ds.invoke_startup()
|
||||
telemetry.register_datasette(ds)
|
||||
try:
|
||||
with connection_budget() as budget:
|
||||
observations = 0
|
||||
for callback in (
|
||||
telemetry.observe_sql_thread_limit,
|
||||
telemetry.observe_sql_thread_queue_depth,
|
||||
telemetry.observe_pending_queries,
|
||||
telemetry.observe_write_queue_depth,
|
||||
telemetry.observe_open_connections,
|
||||
):
|
||||
observations += len(list(callback()))
|
||||
assert observations
|
||||
budget.assert_only(internal=False)
|
||||
budget.assert_no_connections(internal=False)
|
||||
finally:
|
||||
telemetry.unregister_datasette(ds)
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("num_sql_threads", [0, 3])
|
||||
async def test_startup_opens_each_database_at_most_once(db_files, num_sql_threads):
|
||||
imm = db_files[:5]
|
||||
with connection_budget() as budget:
|
||||
ds = Datasette(
|
||||
db_files[5:],
|
||||
immutables=imm,
|
||||
settings={
|
||||
"schema_watch_interval_ms": 0,
|
||||
"num_sql_threads": num_sql_threads,
|
||||
},
|
||||
)
|
||||
await ds._startup_sequence()
|
||||
# One catalog scan connection per database, no pooled reads or writes
|
||||
# (immutable table counts are no longer precomputed at startup)
|
||||
assert set(budget.opened) - {INTERNAL, "_memory"} == {
|
||||
f"db{i:03d}" for i in range(N)
|
||||
}
|
||||
assert all(
|
||||
count == 1 for name, count in budget.opened.items() if name != INTERNAL
|
||||
), budget.opened
|
||||
assert set(budget.reads) <= {INTERNAL}, budget.reads
|
||||
assert set(budget.writes) <= {INTERNAL}, budget.writes
|
||||
assert not budget.attached
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_startup_with_crossdb_attaches_nothing(db_files):
|
||||
with connection_budget() as budget:
|
||||
ds = Datasette(
|
||||
db_files[:12],
|
||||
settings={"schema_watch_interval_ms": 0},
|
||||
crossdb=True,
|
||||
)
|
||||
await ds._startup_sequence()
|
||||
assert not budget.attached
|
||||
# A crossdb query still sees the attached databases
|
||||
response = await ds.client.get(
|
||||
"/_memory/-/query.json?sql=select+count(*)+from+db001.t1&_shape=array"
|
||||
)
|
||||
assert response.status_code == 200
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_restart_with_persisted_catalog_opens_nothing(
|
||||
db_files, tmp_path, monkeypatch
|
||||
):
|
||||
from datasette import schema_watcher
|
||||
|
||||
# The files' ctimes are recent (os.utime() in the fixture): without this
|
||||
# their stored fingerprints would be racy, and so rescanned
|
||||
monkeypatch.setattr(schema_watcher, "RACY_WINDOW_NS", 0)
|
||||
internal = str(tmp_path / "internal.db")
|
||||
ds = Datasette(db_files, internal=internal)
|
||||
await ds.invoke_startup()
|
||||
ds.close()
|
||||
with connection_budget() as budget:
|
||||
ds = Datasette(
|
||||
db_files, internal=internal, settings={"schema_watch_interval_ms": 0}
|
||||
)
|
||||
await ds._startup_sequence()
|
||||
response = await ds.client.get("/.json")
|
||||
assert response.status_code == 200
|
||||
budget.assert_only()
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_idle_sweeps_open_nothing(db_files):
|
||||
ds = _ds(db_files)
|
||||
await ds.invoke_startup()
|
||||
with connection_budget() as budget:
|
||||
for _ in range(3):
|
||||
await ds._refresh_schemas(background=True)
|
||||
await ds.refresh_schemas(force=True)
|
||||
budget.assert_only(internal=False)
|
||||
budget.assert_no_connections(internal=False)
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_failed_scan_is_not_retried_by_idle_sweeps(
|
||||
db_files, tmp_path, monkeypatch
|
||||
):
|
||||
from datasette import schema_watcher
|
||||
|
||||
# The file was just written: do not wait out the racy window
|
||||
monkeypatch.setattr(schema_watcher, "RACY_WINDOW_NS", 0)
|
||||
bad = tmp_path / "bad.db"
|
||||
bad.write_bytes(b"this is not a SQLite database" * 10)
|
||||
old = time.time() - 60
|
||||
os.utime(bad, (old, old))
|
||||
ds = _ds(db_files[:3] + [str(bad)])
|
||||
await ds.invoke_startup()
|
||||
assert ds.get_database("bad")._watch_state.error is not None
|
||||
with connection_budget() as budget:
|
||||
for _ in range(3):
|
||||
await ds._refresh_schemas(background=True)
|
||||
assert "bad" not in budget.touched()
|
||||
# ...until the file changes
|
||||
good = sqlite3.connect(tmp_path / "good.db")
|
||||
good.execute("create table fixed (id integer primary key)")
|
||||
good.commit()
|
||||
good.close()
|
||||
os.replace(tmp_path / "good.db", bad)
|
||||
await ds._refresh_schemas(background=True)
|
||||
rows = await ds.get_internal_database().execute(
|
||||
"select table_name from catalog_tables where database_name = 'bad'"
|
||||
)
|
||||
assert [r[0] for r in rows.rows] == ["fixed"]
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_databases_reuses_catalog_scan(db_files, tmp_path):
|
||||
from datasette.cli import check_databases
|
||||
from datasette.utils import ConnectionProblem
|
||||
|
||||
ds = _ds(db_files)
|
||||
with connection_budget() as budget:
|
||||
await check_databases(ds)
|
||||
# The catalog scan opened each database once; no second pass
|
||||
assert all(count == 1 for name, count in budget.opened.items() if name != INTERNAL)
|
||||
assert set(budget.reads) <= {INTERNAL, "_memory"}, budget.reads
|
||||
ds.close()
|
||||
|
||||
bad = tmp_path / "bad.db"
|
||||
bad.write_bytes(b"this is not a SQLite database" * 10)
|
||||
ds = _ds(db_files[:3] + [str(bad)])
|
||||
with pytest.raises(Exception) as excinfo:
|
||||
await check_databases(ds)
|
||||
assert "bad.db" in str(excinfo.value)
|
||||
assert "file is not a database" in str(excinfo.value)
|
||||
assert not isinstance(excinfo.value, ConnectionProblem)
|
||||
ds.close()
|
||||
|
||||
|
||||
# Parity: the catalog helpers answer exactly what the live helpers do
|
||||
|
||||
|
||||
MISC_SQL = """
|
||||
create table docs (id integer primary key, title text, body text);
|
||||
create virtual table docs_fts using fts4(title, body, content="docs");
|
||||
create virtual table docs_vocab using fts4aux(docs_fts);
|
||||
create virtual table plain using fts5(x);
|
||||
create virtual table geo using rtree(id, minx, maxx);
|
||||
create table _private (x);
|
||||
create table geometry_columns (f_table_name text);
|
||||
create table idx_foo_bar (x);
|
||||
create table IDXA (x);
|
||||
create table parent (id integer primary key);
|
||||
create table child (
|
||||
id integer primary key,
|
||||
p integer references parent(id),
|
||||
q integer references missing(id)
|
||||
);
|
||||
create table comp (a, b, foreign key (a, b) references parent(id, id));
|
||||
create table self_ref (id integer primary key, up integer references self_ref(id));
|
||||
create table wide (b text, a integer, c, primary key (a, b));
|
||||
create view v_docs as select * from docs;
|
||||
"""
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("name", ["misc", "fixtures"])
|
||||
async def test_catalog_helpers_match_live_introspection(tmp_path, name):
|
||||
from datasette.fixtures import write_fixture_database
|
||||
from datasette.utils.catalog import (
|
||||
all_derived_table_dependencies,
|
||||
catalog_all_foreign_keys,
|
||||
)
|
||||
|
||||
path = str(tmp_path / f"{name}.db")
|
||||
if name == "fixtures":
|
||||
write_fixture_database(path)
|
||||
config_hidden = set()
|
||||
config = {}
|
||||
else:
|
||||
conn = sqlite3.connect(path)
|
||||
conn.executescript(MISC_SQL)
|
||||
conn.close()
|
||||
config_hidden = {"docs"}
|
||||
config = {"databases": {"misc": {"tables": {"docs": {"hidden": True}}}}}
|
||||
ds = Datasette([path], config=config, settings={"schema_watch_interval_ms": 0})
|
||||
await ds.invoke_startup()
|
||||
db = ds.get_database(name)
|
||||
tables = await db.table_names()
|
||||
summary = (await catalog_summaries(ds, [name]))[name]
|
||||
# hidden_table_names() also lists SpatiaLite names that do not exist
|
||||
hidden = set(await db.hidden_table_names()) & set(tables)
|
||||
assert {t for t in tables if summary.is_hidden(t, config_hidden)} == hidden
|
||||
assert summary.views == set(await db.view_names())
|
||||
|
||||
details = await catalog_table_details(ds, [(name, t) for t in tables])
|
||||
for table in tables:
|
||||
assert details[(name, table)]["columns"] == await db.table_columns(table)
|
||||
assert details[(name, table)]["primary_keys"] == await db.primary_keys(table)
|
||||
assert details[(name, table)]["fts_table"] == await db.fts_table(table)
|
||||
|
||||
all_foreign_keys = await db.get_all_foreign_keys()
|
||||
assert await catalog_all_foreign_keys(ds, name) == all_foreign_keys
|
||||
relationships = (await catalog_relationship_counts(ds, [name])).get(name, {})
|
||||
for table, fks in all_foreign_keys.items():
|
||||
assert relationships.get(table, 0) == len(fks["incoming"] + fks["outgoing"])
|
||||
|
||||
derived = await all_derived_table_dependencies(ds)
|
||||
live = await db.execute_fn(sqlite_derived_table_dependencies)
|
||||
assert derived.get(name, {}) == live
|
||||
# Restored from the catalog rather than remembered from the scan
|
||||
db._cached_derived_table_dependencies = None
|
||||
assert await db.derived_table_dependencies() == live
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.skipif(not supports_table_list(), reason="needs PRAGMA table_list")
|
||||
async def test_catalog_records_table_types(tmp_path):
|
||||
path = str(tmp_path / "types.db")
|
||||
conn = sqlite3.connect(path)
|
||||
conn.executescript("""
|
||||
create table plain (id integer primary key);
|
||||
create virtual table f using fts5(x);
|
||||
""")
|
||||
conn.close()
|
||||
ds = Datasette([path], settings={"schema_watch_interval_ms": 0})
|
||||
await ds.invoke_startup()
|
||||
rows = await ds.get_internal_database().execute(
|
||||
"select table_name, type from catalog_tables where database_name = 'types'"
|
||||
)
|
||||
types = dict(rows.rows)
|
||||
assert types["plain"] == "table"
|
||||
assert types["f"] == "virtual"
|
||||
assert types["f_data"] == "shadow"
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_catalog_derived_dependencies_follow_catalog_changes(tmp_path):
|
||||
path = str(tmp_path / "live.db")
|
||||
sqlite3.connect(path).execute("create table docs (body text)").connection.close()
|
||||
ds = Datasette([path], settings={"schema_watch_interval_ms": 0})
|
||||
await ds.invoke_startup()
|
||||
from datasette.utils.catalog import all_derived_table_dependencies
|
||||
|
||||
assert "live" not in await all_derived_table_dependencies(ds)
|
||||
db = ds.get_database("live")
|
||||
await db.execute_write(
|
||||
"create virtual table docs_fts using fts5(body, content='docs')"
|
||||
)
|
||||
derived = await all_derived_table_dependencies(ds)
|
||||
assert derived["live"]["docs_fts"] == "docs"
|
||||
assert derived["live"]["docs_fts_data"] == "docs_fts"
|
||||
ds.close()
|
||||
|
||||
|
||||
def test_connection_budget_helper_counts(db_files):
|
||||
# The helper itself: a direct execute_fn is a read, connect() an open
|
||||
import asyncio
|
||||
|
||||
async def run():
|
||||
ds = _ds(db_files[:2])
|
||||
await ds.invoke_startup()
|
||||
db = ds.get_database("db001")
|
||||
async with connection_budget() as budget:
|
||||
await db.execute("select 1")
|
||||
conn = sqlite3.connect(":memory:")
|
||||
conn.close()
|
||||
assert budget.reads["db001"] == 1
|
||||
assert budget.touched() == {"db001"}
|
||||
with pytest.raises(AssertionError, match="db001"):
|
||||
budget.assert_only("db000")
|
||||
ds.close()
|
||||
return budget
|
||||
|
||||
budget = asyncio.run(run())
|
||||
assert sum(budget.probes.values()) == 1
|
||||
assert json.dumps(budget.summary())
|
||||
|
|
@ -95,6 +95,11 @@ async def test_derived_permission_discovery_error_is_retried(monkeypatch, listin
|
|||
action="view-table", resource=TableResource("data", "documents")
|
||||
)
|
||||
|
||||
# While the catalog is current, dependencies come from its rows and
|
||||
# nothing is discovered live. Live discovery - the path that can fail -
|
||||
# happens while a rescan of the database is pending.
|
||||
db._watch_state.needs_scan = True
|
||||
db._cached_derived_table_dependencies = None
|
||||
token = _permission_check_cache.set({})
|
||||
try:
|
||||
with monkeypatch.context() as patch:
|
||||
|
|
|
|||
|
|
@ -31,6 +31,7 @@ async def test_internal_migrations_recorded(ds_client):
|
|||
assert [tuple(row) for row in migrations.rows] == [
|
||||
("datasette_internal", "0001_initial"),
|
||||
("datasette_internal", "0002_catalog_fingerprint"),
|
||||
("datasette_internal", "0003_catalog_table_type"),
|
||||
]
|
||||
|
||||
|
||||
|
|
@ -64,6 +65,7 @@ async def test_internal_migrations_adopt_existing_internal_db(tmp_path):
|
|||
assert [tuple(row) for row in migrations.rows] == [
|
||||
("datasette_internal", "0001_initial"),
|
||||
("datasette_internal", "0002_catalog_fingerprint"),
|
||||
("datasette_internal", "0003_catalog_table_type"),
|
||||
]
|
||||
|
||||
ds.close()
|
||||
|
|
@ -75,7 +77,13 @@ async def test_internal_tables(ds_client):
|
|||
tables = await internal_db.execute("select * from catalog_tables")
|
||||
assert len(tables) > 5
|
||||
table = tables.rows[0]
|
||||
assert set(table.keys()) == {"rootpage", "table_name", "database_name", "sql"}
|
||||
assert set(table.keys()) == {
|
||||
"rootpage",
|
||||
"table_name",
|
||||
"database_name",
|
||||
"sql",
|
||||
"type",
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
|
|
|||
|
|
@ -215,18 +215,15 @@ async def test_concurrent_first_requests_all_wait_for_slow_startup():
|
|||
|
||||
|
||||
@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.
|
||||
async def test_immutable_table_counts_computed_when_first_needed(tmp_path, monkeypatch):
|
||||
# Immutable databases used to have their table counts precomputed by
|
||||
# _startup_sequence(), opening every immutable database at startup.
|
||||
# They are now computed the first time a page needs them, with the same
|
||||
# generous time limit, and cached.
|
||||
db_path = tmp_path / "immutable.db"
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
conn.execute("create table t (id integer primary key)")
|
||||
conn.execute("insert into t default values")
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
|
|
@ -241,22 +238,22 @@ async def test_setup_db_still_runs_when_invoke_startup_ran_first(tmp_path, monke
|
|||
|
||||
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()
|
||||
await ds._startup_sequence()
|
||||
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
|
||||
assert call_count["n"] == 0
|
||||
assert ds.get_database("immutable").cached_table_counts is None
|
||||
|
||||
# Idempotency: a second call must not recompute.
|
||||
await ds._startup_sequence()
|
||||
response = await ds.client.get("/.json")
|
||||
assert response.json()["databases"][0]["table_rows_sum"] == 1
|
||||
assert call_count["n"] == 1
|
||||
assert ds.get_database("immutable").cached_table_counts == {"t": 1}
|
||||
|
||||
# Cached from then on
|
||||
await ds.client.get("/immutable.json")
|
||||
assert ds.get_database("immutable").cached_table_counts == {"t": 1}
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue