remote/datasette
1
0
Fork 0
mirror of https://github.com/simonw/datasette.git synced 2026-09-09 01:54:15 +02:00
Code Issues Projects Releases Packages Wiki Activity
datasette/tests/conftest.py
Alex Garcia 721cc6d9f9 Review polish: fix stale exemplar context, dedupe rationales, close test gaps 
- The exemplars docs described "the pinned opentelemetry-exporter-prometheus"
  and "Datasette's own Prometheus exporter" - context from demo/plugin work
  that is no longer part of this stack. Reworded to stand alone.
- Saturate a num_sql_threads=1 pool and assert the queue-depth gauge reads
  above zero - the headline alerting metric previously only had an absence
  test, and this also pins the private ThreadPoolExecutor._work_queue
  attribute it depends on.
- Pin error.type on the write path of db.client.operation.duration - the
  write wrappers time a different code path than the read one already tested.
- Isolate the non-threaded-mode gauge test from other live instances instead
  of comparing global observation counts, which a GC pass could shift.
- Halve the metrics banner, point conftest's meter note at it, compact the
  interrupted-counter call-site comment to a registry pointer, note why
  instrument and registry descriptions are separate strings, and stop
  calling the metric dimension a "later phase" now that metrics shipped.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012U7coQfVu8nK2R4q2mCULA
2026-09-02 14:21:05 -07:00

564 lines
19 KiB
Python
Raw Blame History

import importlib.metadata
import os
import pathlib
import re
import socket
import subprocess
import sys
import tempfile
import time
from dataclasses import dataclass
import httpx
import pytest
import pytest_asyncio
from datasette import Event, hookimpl
try:
import pysqlite3 as sqlite3
except ImportError:
import sqlite3
UNDOCUMENTED_PERMISSIONS = {
"this_is_allowed",
"this_is_denied",
"this_is_allowed_async",
"this_is_denied_async",
"no_match",
# Test actions from test_hook_register_actions_with_custom_resources
"manage_documents",
"view_document_collection",
"view_document",
}
def wait_until_responds(url, timeout=5.0, client=httpx, process=None, **kwargs):
start = time.time()
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:
client.get(url, **kwargs)
return
except httpx.TransportError:
time.sleep(0.1)
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]
_otel_span_exporter = None
@pytest.fixture(scope="session", autouse=True)
def _otel_provider():
"""
Install a real OTel SDK TracerProvider + InMemorySpanExporter exactly
once, before any span is ever created in this process.
This has to be session-scoped and autouse because
`opentelemetry.trace.set_tracer_provider()` is effectively
once-per-process: a second call logs a warning and is ignored. So the
install must happen exactly once, before anything asserts on spans.
`datasette.telemetry.tracer` is a module-level `ProxyTracer`. Once a
provider exists, the first span it starts resolves a concrete tracer
and caches it permanently. It does *not* cache the no-op tracer, so
any span started before this fixture runs is merely lost rather than
poisoning the tracer for the rest of the process. If the SDK isn't
installed, do nothing: core spans stay no-op `NonRecordingSpan`s and
the rest of the suite is unaffected.
"""
global _otel_span_exporter
try:
from opentelemetry import trace as otel_trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
InMemorySpanExporter,
)
except ImportError:
return
exporter = InMemorySpanExporter()
provider = TracerProvider()
# SimpleSpanProcessor exports synchronously on span end - no background
# batching thread, so assertions immediately after a request never race.
provider.add_span_processor(SimpleSpanProcessor(exporter))
otel_trace.set_tracer_provider(provider)
_otel_span_exporter = exporter
@pytest.fixture
def otel_spans():
"""
Function-scoped access to the finished-spans exporter: clears any spans
left over from previous tests, then yields the exporter so a test can
call `.get_finished_spans()` after making requests. Skips (rather than
fails) if the OTel SDK is not installed.
"""
pytest.importorskip("opentelemetry.sdk")
if _otel_span_exporter is None:
pytest.skip("OpenTelemetry SDK provider was not installed")
_otel_span_exporter.clear()
yield _otel_span_exporter
_otel_metric_reader = None
@pytest.fixture(scope="session", autouse=True)
def _otel_meter_provider():
"""
Install a real OTel SDK MeterProvider + InMemoryMetricReader once per
process.
Unlike the tracer, ordering is not load-bearing here - see the metrics
banner in `datasette/telemetry.py` for the `_ProxyMeter`-vs-`ProxyTracer`
difference. This fixture is still session-scoped and autouse for
symmetry, and so that a single reader collects for the whole run.
DELTA temporality is chosen for counters and histograms so that each
collection reports only what happened since the previous one. With the
SDK default of CUMULATIVE, every metrics test would see every query run
by every earlier test in the session.
"""
global _otel_metric_reader
try:
from opentelemetry import metrics as otel_metrics
from opentelemetry.sdk.metrics import Counter, Histogram, MeterProvider
from opentelemetry.sdk.metrics.export import (
AggregationTemporality,
InMemoryMetricReader,
)
except ImportError:
return
reader = InMemoryMetricReader(
preferred_temporality={
Counter: AggregationTemporality.DELTA,
Histogram: AggregationTemporality.DELTA,
}
)
otel_metrics.set_meter_provider(MeterProvider(metric_readers=[reader]))
_otel_metric_reader = reader
class MetricsCollector:
"""
Thin reader over an `InMemoryMetricReader`.
`collect()` runs a collection cycle - which is what invokes the observable
gauge callbacks - and snapshots the result. Queries then run against that
snapshot rather than re-collecting, so a test that inspects several
metrics sees one consistent moment and does not drain delta state twice.
"""
def __init__(self, reader):
self.reader = reader
self.snapshot = {}
def collect(self):
self.snapshot = {}
data = self.reader.get_metrics_data()
if data is None:
return self.snapshot
for resource_metrics in data.resource_metrics:
for scope_metrics in resource_metrics.scope_metrics:
for metric in scope_metrics.metrics:
self.snapshot.setdefault(metric.name, []).extend(
metric.data.data_points
)
return self.snapshot
def points(self, name, attributes=None):
"Data points for `name` whose attributes are a superset of `attributes`."
found = []
for point in self.snapshot.get(name, []):
point_attributes = dict(point.attributes or {})
if all(point_attributes.get(k) == v for k, v in (attributes or {}).items()):
found.append(point)
return found
def point(self, name, attributes=None):
"The single matching data point, asserting there is exactly one."
found = self.points(name, attributes)
assert len(found) == 1, (
f"expected exactly one {name} point matching {attributes}, "
f"got {len(found)}: {found}"
)
return found[0]
@pytest.fixture
def otel_metrics():
"""
Function-scoped metrics collector. Drains any delta state accumulated by
earlier tests before yielding, so counts start from zero.
"""
pytest.importorskip("opentelemetry.sdk")
if _otel_metric_reader is None:
pytest.skip("OpenTelemetry SDK meter provider was not installed")
_otel_metric_reader.get_metrics_data()
yield MetricsCollector(_otel_metric_reader)
@pytest.fixture
def bare_ds():
"""
Minimal Datasette with no plugins, data, metadata, or config - for tests
that want to exercise core behavior (e.g. middleware) in isolation.
"""
from datasette.app import Datasette
return Datasette(memory=True)
@pytest_asyncio.fixture(scope="session")
async def ds_client():
import secrets
from datasette.app import Datasette
from datasette.database import Database
from .fixtures import CONFIG, METADATA, PLUGINS_DIR
ds = Datasette(
metadata=METADATA,
config=CONFIG,
plugins_dir=PLUGINS_DIR,
settings={
"default_page_size": 50,
"max_returned_rows": 100,
"sql_time_limit_ms": 200,
"facet_suggest_time_limit_ms": 200, # Up from 50 default
# Default is 3 but this results in "too many open files"
# errors when running the full test suite:
"num_sql_threads": 1,
},
)
from datasette.fixtures import populate_fixture_database
# Use a unique memory_name to avoid collisions between different
# Datasette instances in the same process, but use "fixtures" for routing
unique_memory_name = f"fixtures_{secrets.token_hex(8)}"
db = ds.add_database(Database(ds, memory_name=unique_memory_name), name="fixtures")
ds.remove_database("_memory")
def prepare(conn):
if not conn.execute("select count(*) from sqlite_master").fetchone()[0]:
populate_fixture_database(conn)
await db.execute_write_fn(prepare)
await ds.invoke_startup()
return ds.client
def pytest_report_header(config):
conn = sqlite3.connect(":memory:")
version = conn.execute("select sqlite_version()").fetchone()[0]
conn.close()
sqlite_utils_version = importlib.metadata.version("sqlite-utils")
headers = [
f"SQLite: {version}",
f"sqlite-utils: {sqlite_utils_version}",
]
if config.getoption("--playwright"):
try:
browsers = config.getoption("--browser")
except ValueError:
browsers = None
if isinstance(browsers, str):
browsers = [browsers]
if browsers:
headers.append("Playwright browsers: {}".format(", ".join(browsers)))
return headers
def pytest_addoption(parser):
parser.addoption(
"--playwright",
action="store_true",
default=False,
help="run Playwright browser automation tests",
)
def pytest_configure(config):
import sys
sys._called_from_test = True
def pytest_unconfigure(config):
import sys
del sys._called_from_test
def pytest_collection_modifyitems(config, items):
if not config.getoption("--playwright"):
skip_playwright = pytest.mark.skip(reason="need --playwright option to run")
for item in items:
if "playwright" in item.keywords:
item.add_marker(skip_playwright)
# Ensure test_cli.py and test_black.py and test_inspect.py run first before any asyncio code kicks in
move_to_front(items, "test_cli")
move_to_front(items, "test_black")
move_to_front(items, "test_inspect_cli")
move_to_front(items, "test_serve_with_get")
move_to_front(items, "test_serve_with_get_exit_code_for_error")
move_to_front(items, "test_inspect_cli_writes_to_file")
move_to_front(items, "test_spatialite_error_if_attempt_to_open_spatialite")
move_to_front(items, "test_package")
move_to_front(items, "test_package_with_port")
# Same reason: this one shells out to a fresh interpreter. Late in a serial
# run the pytest process holds enough threads that the fork half of
# subprocess' fork+exec crashes the interpreter on macOS/CPython 3.13
# (SIGSEGV/SIGBUS inside _execute_child). Reproduces with any subprocess
# call placed there, on an unmodified tree - running it first avoids it.
move_to_front(items, "test_datasette_package_never_imports_the_sdk")
move_to_front(items, "test_no_provider_takes_the_fast_path")
def move_to_front(items, test_name):
test = [fn for fn in items if fn.name == test_name]
if test:
items.insert(0, items.pop(items.index(test[0])))
@pytest.fixture
def restore_working_directory(tmpdir, request):
try:
previous_cwd = os.getcwd()
except OSError:
# https://github.com/simonw/datasette/issues/1361
previous_cwd = None
tmpdir.chdir()
def return_to_previous():
os.chdir(previous_cwd)
if previous_cwd is not None:
request.addfinalizer(return_to_previous)
@pytest.fixture(scope="session", autouse=True)
def check_actions_are_documented():
from datasette.default_actions import register_actions as default_register_actions
from datasette.plugins import pm
content = (
pathlib.Path(__file__).parent.parent / "docs" / "authentication.rst"
).read_text()
permissions_re = re.compile(r"\.\. _actions_([^\s:]+):")
documented_actions = set(permissions_re.findall(content)).union(
UNDOCUMENTED_PERMISSIONS
)
# Only Datasette core actions need to be documented - actions registered
# by (test) plugins are checked for registration but not documentation
core_actions = {action.name for action in default_register_actions()}
def before(hook_name, hook_impls, kwargs):
if hook_name == "permission_resources_sql":
datasette = kwargs["datasette"]
assert kwargs["action"] in datasette.actions, (
"'{}' has not been registered with register_actions()".format(
kwargs["action"]
)
+ " (or maybe a test forgot to do await ds.invoke_startup())"
)
action = kwargs.get("action").replace("-", "_")
if kwargs["action"] in core_actions:
assert (
action in documented_actions
), f"Undocumented permission action: {action}"
pm.add_hookcall_monitoring(
before=before, after=lambda outcome, hook_name, hook_impls, kwargs: None
)
class TrackEventPlugin:
__name__ = "TrackEventPlugin"
@dataclass
class OneEvent(Event):
name = "one"
extra: str
@hookimpl
def register_events(self, datasette):
async def inner():
return [self.OneEvent]
return inner
@hookimpl
def track_event(self, datasette, event):
datasette._tracked_events = getattr(datasette, "_tracked_events", [])
datasette._tracked_events.append(event)
@pytest.fixture(scope="session", autouse=True)
def install_event_tracking_plugin():
from datasette.plugins import pm
pm.register(TrackEventPlugin(), name="TrackEventPlugin")
@pytest.fixture(scope="session")
def ds_localhost_http_server():
ds_proc = subprocess.Popen(
[sys.executable, "-m", "datasette", "--memory", "-p", "8041"],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
# Avoid FileNotFoundError: [Errno 2] No such file or directory:
cwd=tempfile.gettempdir(),
)
wait_until_responds("http://localhost:8041/")
# Check it started successfully
assert not ds_proc.poll(), ds_proc.stdout.read().decode("utf-8")
yield ds_proc
# Shut it down at the end of the pytest session
ds_proc.terminate()
@pytest.fixture(scope="session")
def ds_unix_domain_socket_server(tmp_path_factory):
# This used to use tmp_path_factory.mktemp("uds") but that turned out to
# produce paths that were too long to use as UDS on macOS, see
# https://github.com/simonw/datasette/issues/1407 - so I switched to
# using tempfile.gettempdir() with a per-process filename.
uds = str(pathlib.Path(tempfile.gettempdir()) / f"datasette-{os.getpid()}.sock")
try:
os.unlink(uds)
except FileNotFoundError:
pass
ds_proc = subprocess.Popen(
[sys.executable, "-m", "datasette", "--memory", "--uds", uds],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
cwd=tempfile.gettempdir(),
)
# Poll until available
transport = httpx.HTTPTransport(uds=uds)
client = httpx.Client(transport=transport)
try:
wait_until_responds(
"http://localhost/_memory.json", timeout=30.0, client=client
)
# Check it started successfully
assert not ds_proc.poll(), ds_proc.stdout.read().decode("utf-8")
yield ds_proc, uds
finally:
client.close()
# Shut it down at the end of the pytest session
ds_proc.terminate()
try:
ds_proc.wait(timeout=5)
except subprocess.TimeoutExpired:
ds_proc.kill()
ds_proc.wait()
try:
os.unlink(uds)
except FileNotFoundError:
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
from .fixtures import ( # noqa: F401
TEMP_PLUGIN_SECRET_FILE,
app_client,
app_client_base_url_prefix,
app_client_conflicting_database_names,
app_client_csv_max_mb_one,
app_client_immutable_and_inspect_file,
app_client_larger_cache_size,
app_client_no_files,
app_client_returned_rows_matches_page_size,
app_client_shorter_time_limit,
app_client_two_attached_databases,
app_client_two_attached_databases_crossdb_enabled,
app_client_two_attached_databases_one_immutable,
app_client_with_cors,
app_client_with_dot,
app_client_with_trace,
make_app_client,
)
Reference in a new issue View git blame Copy permalink