remote/datasette
1
0
Fork 0
mirror of https://github.com/simonw/datasette.git synced 2026-06-01 22:56:58 +02:00
Code Issues Projects Releases Packages Wiki Activity

Fixed the rest of the warnings

Browse source 
Refs https://github.com/simonw/datasette/pull/2615#issuecomment-3649771920
This commit is contained in:
Simon Willison 2025-12-13 12:16:57 -08:00
commit 8895c4a202
8 changed files with 528 additions and 264 deletions
Download patch file Download diff file Expand all files Collapse all files

106
datasette/database.py
View file

@ -28,6 +28,9 @@ connections = threading.local()
AttachedDatabase = namedtuple("AttachedDatabase", ("seq", "name", "file"))
# Sentinel object to signal write thread shutdown
_SHUTDOWN_SENTINEL = object()
class Database:
# For table counts stop at this many rows:
@ -62,10 +65,25 @@ class Database:
# These are used when in non-threaded mode:
self._read_connection = None
self._write_connection = None
# This is used to track all file connections so they can be closed
self._all_file_connections = []
# This is used to track all connections so they can be closed
self._all_connections = []
self._closed = False
self.mode = mode
def __del__(self):
# Ensure connections are closed when Database is garbage collected
# This prevents ResourceWarning about unclosed database connections
if not self._closed:
# Close all tracked connections without executor cleanup
# (executor might already be gone during garbage collection)
for connection in self._all_connections:
try:
connection.close()
except Exception:
pass
self._all_connections.clear()
self._closed = True
@property
def cached_table_counts(self):
if self._cached_table_counts is not None:
@ -103,9 +121,12 @@ class Database:
)
if not write:
conn.execute("PRAGMA query_only=1")
self._all_connections.append(conn)
return conn
if self.is_memory:
return sqlite3.connect(":memory:", uri=True)
conn = sqlite3.connect(":memory:", uri=True, check_same_thread=False)
self._all_connections.append(conn)
return conn
# mode=ro or immutable=1?
if self.is_mutable:
@ -122,13 +143,69 @@ class Database:
conn = sqlite3.connect(
f"file:{self.path}{qs}", uri=True, check_same_thread=False, **extra_kwargs
)
self._all_file_connections.append(conn)
self._all_connections.append(conn)
return conn
def close(self):
# Close all connections - useful to avoid running out of file handles in tests
for connection in self._all_file_connections:
connection.close()
self._closed = True
# First, signal the write thread to shut down if it exists
if self._write_thread is not None and self._write_queue is not None:
self._write_queue.put(_SHUTDOWN_SENTINEL)
self._write_thread.join(timeout=1.0)
# Clear the instance variable references (connections will be closed below)
self._read_connection = None
self._write_connection = None
# Close and clear thread-local connection if it exists in the current thread
main_thread_conn = getattr(connections, self._thread_local_id, None)
if main_thread_conn is not None:
try:
main_thread_conn.close()
except Exception:
pass
delattr(connections, self._thread_local_id)
# If executor is available, use a barrier to ensure cleanup runs on ALL threads
thread_local_id = self._thread_local_id
if self.ds.executor is not None:
import concurrent.futures
max_workers = getattr(self.ds.executor, "_max_workers", None) or 1
barrier = threading.Barrier(max_workers, timeout=2.0)
def clear_thread_local():
# Close and clear this database's thread-local connection in this thread
conn = getattr(connections, thread_local_id, None)
if conn is not None:
try:
conn.close()
except Exception:
pass # Connection might already be closed
delattr(connections, thread_local_id)
# Wait for all threads to reach this point - this ensures
# all threads are processing cleanup simultaneously
try:
barrier.wait()
except threading.BrokenBarrierError:
pass
try:
# Submit exactly max_workers tasks - the barrier ensures all
# threads must be occupied with our cleanup tasks
futures = [
self.ds.executor.submit(clear_thread_local)
for _ in range(max_workers)
]
# Wait for all cleanup tasks to complete
concurrent.futures.wait(futures, timeout=3.0)
except Exception:
pass # Executor might be shutting down
# Close all tracked connections
for connection in self._all_connections:
try:
connection.close()
except Exception:
pass # Connection might already be closed
self._all_connections.clear()
async def execute_write(self, sql, params=None, block=True):
def _inner(conn):
@ -178,7 +255,7 @@ class Database:
finally:
isolated_connection.close()
try:
self._all_file_connections.remove(isolated_connection)
self._all_connections.remove(isolated_connection)
except ValueError:
# Was probably a memory connection
pass
@ -242,6 +319,15 @@ class Database:
conn_exception = e
while True:
task = self._write_queue.get()
# Check for shutdown sentinel
if task is _SHUTDOWN_SENTINEL:
if conn is not None:
conn.close()
try:
self._all_connections.remove(conn)
except ValueError:
pass
return
if conn_exception is not None:
result = conn_exception
else:
@ -256,7 +342,7 @@ class Database:
finally:
isolated_connection.close()
try:
self._all_file_connections.remove(isolated_connection)
self._all_connections.remove(isolated_connection)
except ValueError:
# Was probably a memory connection
pass
@ -284,6 +370,10 @@ class Database:
# threaded mode
def in_thread():
conn = getattr(connections, self._thread_local_id, None)
# Check if database was closed - if so, clear the stale cached connection
if conn and self._closed:
delattr(connections, self._thread_local_id)
conn = None
if not conn:
conn = self.connect()
self.ds._prepare_connection(conn, self.name)

47
tests/conftest.py
View file

@ -30,6 +30,7 @@ UNDOCUMENTED_PERMISSIONS = {
}
_ds_client = None
_ds_instance = None
def wait_until_responds(url, timeout=5.0, client=httpx, **kwargs):
@ -50,7 +51,7 @@ async def ds_client():
from .fixtures import CONFIG, METADATA, PLUGINS_DIR
import secrets
global _ds_client
global _ds_client, _ds_instance
if _ds_client is not None:
return _ds_client
@ -86,6 +87,7 @@ async def ds_client():
await db.execute_write_fn(prepare)
await ds.invoke_startup()
_ds_client = ds.client
_ds_instance = ds
return _ds_client
@ -107,6 +109,19 @@ def pytest_unconfigure(config):
del sys._called_from_test
# Clean up the global ds_client fixture
global _ds_instance
if _ds_instance is not None:
# Close databases first (while executor is still running)
for db in _ds_instance.databases.values():
db.close()
if hasattr(_ds_instance, "_internal_database"):
_ds_instance._internal_database.close()
# Then shut down executor
if _ds_instance.executor is not None:
_ds_instance.executor.shutdown(wait=True)
_ds_instance = None
def pytest_collection_modifyitems(items):
# Ensure test_cli.py and test_black.py and test_inspect.py run first before any asyncio code kicks in
@ -218,6 +233,27 @@ def ds_localhost_http_server():
yield ds_proc
# Shut it down at the end of the pytest session
ds_proc.terminate()
ds_proc.wait()
if ds_proc.stdout:
ds_proc.stdout.close()
def wait_until_uds_responds(uds_path, timeout=5.0):
"""Wait for a Unix domain socket to accept connections."""
import socket as socket_module
start = time.time()
while time.time() - start < timeout:
sock = socket_module.socket(socket_module.AF_UNIX, socket_module.SOCK_STREAM)
try:
sock.connect(uds_path)
# Connection successful, now close and return
sock.close()
return
except (ConnectionRefusedError, FileNotFoundError):
sock.close()
time.sleep(0.1)
raise AssertionError("Timed out waiting for {} to respond".format(uds_path))
@pytest.fixture(scope="session")
@ -233,15 +269,16 @@ def ds_unix_domain_socket_server(tmp_path_factory):
stderr=subprocess.STDOUT,
cwd=tempfile.gettempdir(),
)
# Poll until available
transport = httpx.HTTPTransport(uds=uds)
client = httpx.Client(transport=transport)
wait_until_responds("http://localhost/_memory.json", client=client)
# Poll until available using raw socket to avoid httpx connection pool leaks
wait_until_uds_responds(uds)
# Check it started successfully
assert not ds_proc.poll(), ds_proc.stdout.read().decode("utf-8")
yield ds_proc, uds
# Shut it down at the end of the pytest session
ds_proc.terminate()
ds_proc.wait()
if ds_proc.stdout:
ds_proc.stdout.close()
# Import fixtures from fixtures.py to make them available

18
tests/fixtures.py
View file

@ -171,11 +171,15 @@ def make_app_client(
crossdb=crossdb,
)
yield TestClient(ds)
# Close as many database connections as possible
# to try and avoid too many open files error
# Close all database connections first (while executor is still running)
# This allows db.close() to submit cleanup tasks to executor threads
for db in ds.databases.values():
if not db.is_memory:
db.close()
db.close()
if hasattr(ds, "_internal_database"):
ds._internal_database.close()
# Then shut down executor
if ds.executor is not None:
ds.executor.shutdown(wait=True)
@pytest.fixture(scope="session")
@ -188,8 +192,14 @@ def app_client():
def app_client_no_files():
ds = Datasette([])
yield TestClient(ds)
# Close databases first (while executor is still running)
for db in ds.databases.values():
db.close()
if hasattr(ds, "_internal_database"):
ds._internal_database.close()
# Then shut down executor
if ds.executor is not None:
ds.executor.shutdown(wait=True)
@pytest.fixture(scope="session")

30
tests/test_cli_serve_server.py
View file

@ -5,12 +5,13 @@ import socket
@pytest.mark.serial
def test_serve_localhost_http(ds_localhost_http_server):
response = httpx.get("http://localhost:8041/_memory.json")
assert {
"database": "_memory",
"path": "/_memory",
"tables": [],
}.items() <= response.json().items()
with httpx.Client() as client:
response = client.get("http://localhost:8041/_memory.json")
assert {
"database": "_memory",
"path": "/_memory",
"tables": [],
}.items() <= response.json().items()
@pytest.mark.serial
@ -20,10 +21,13 @@ def test_serve_localhost_http(ds_localhost_http_server):
def test_serve_unix_domain_socket(ds_unix_domain_socket_server):
_, uds = ds_unix_domain_socket_server
transport = httpx.HTTPTransport(uds=uds)
client = httpx.Client(transport=transport)
response = client.get("http://localhost/_memory.json")
assert {
"database": "_memory",
"path": "/_memory",
"tables": [],
}.items() <= response.json().items()
try:
with httpx.Client(transport=transport) as client:
response = client.get("http://localhost/_memory.json")
assert {
"database": "_memory",
"path": "/_memory",
"tables": [],
}.items() <= response.json().items()
finally:
transport.close()

2
tests/test_config_dir.py
View file

@ -100,6 +100,8 @@ def config_dir_client(config_dir):
yield _TestClient(ds)
for db in ds.databases.values():
db.close()
if hasattr(ds, "_internal_database"):
ds._internal_database.close()
def test_settings(config_dir_client):

8
tests/test_docs_plugins.py
View file

@ -23,6 +23,14 @@ async def datasette_with_plugin():
yield datasette
finally:
datasette.pm.unregister(name="undo")
# Close databases first (while executor is still running)
for db in datasette.databases.values():
db.close()
if hasattr(datasette, "_internal_database"):
datasette._internal_database.close()
# Then shut down executor
if datasette.executor is not None:
datasette.executor.shutdown(wait=True)
# -- end datasette_with_plugin_fixture --

348
tests/test_facets.py
View file

@ -407,55 +407,68 @@ async def test_array_facet_results(ds_client):
@pytest.mark.skipif(not detect_json1(), reason="Requires the SQLite json1 module")
async def test_array_facet_handle_duplicate_tags():
ds = Datasette([], memory=True)
db = ds.add_database(Database(ds, memory_name="test_array_facet"))
await db.execute_write("create table otters(name text, tags text)")
for name, tags in (
("Charles", ["friendly", "cunning", "friendly"]),
("Shaun", ["cunning", "empathetic", "friendly"]),
("Tracy", ["empathetic", "eager"]),
):
await db.execute_write(
"insert into otters (name, tags) values (?, ?)", [name, json.dumps(tags)]
)
try:
db = ds.add_database(Database(ds, memory_name="test_array_facet"))
await db.execute_write("create table otters(name text, tags text)")
for name, tags in (
("Charles", ["friendly", "cunning", "friendly"]),
("Shaun", ["cunning", "empathetic", "friendly"]),
("Tracy", ["empathetic", "eager"]),
):
await db.execute_write(
"insert into otters (name, tags) values (?, ?)",
[name, json.dumps(tags)],
)
response = await ds.client.get("/test_array_facet/otters.json?_facet_array=tags")
assert response.json()["facet_results"]["results"]["tags"] == {
"name": "tags",
"type": "array",
"results": [
{
"value": "cunning",
"label": "cunning",
"count": 2,
"toggle_url": "http://localhost/test_array_facet/otters.json?_facet_array=tags&tags__arraycontains=cunning",
"selected": False,
},
{
"value": "empathetic",
"label": "empathetic",
"count": 2,
"toggle_url": "http://localhost/test_array_facet/otters.json?_facet_array=tags&tags__arraycontains=empathetic",
"selected": False,
},
{
"value": "friendly",
"label": "friendly",
"count": 2,
"toggle_url": "http://localhost/test_array_facet/otters.json?_facet_array=tags&tags__arraycontains=friendly",
"selected": False,
},
{
"value": "eager",
"label": "eager",
"count": 1,
"toggle_url": "http://localhost/test_array_facet/otters.json?_facet_array=tags&tags__arraycontains=eager",
"selected": False,
},
],
"hideable": True,
"toggle_url": "/test_array_facet/otters.json",
"truncated": False,
}
response = await ds.client.get(
"/test_array_facet/otters.json?_facet_array=tags"
)
assert response.json()["facet_results"]["results"]["tags"] == {
"name": "tags",
"type": "array",
"results": [
{
"value": "cunning",
"label": "cunning",
"count": 2,
"toggle_url": "http://localhost/test_array_facet/otters.json?_facet_array=tags&tags__arraycontains=cunning",
"selected": False,
},
{
"value": "empathetic",
"label": "empathetic",
"count": 2,
"toggle_url": "http://localhost/test_array_facet/otters.json?_facet_array=tags&tags__arraycontains=empathetic",
"selected": False,
},
{
"value": "friendly",
"label": "friendly",
"count": 2,
"toggle_url": "http://localhost/test_array_facet/otters.json?_facet_array=tags&tags__arraycontains=friendly",
"selected": False,
},
{
"value": "eager",
"label": "eager",
"count": 1,
"toggle_url": "http://localhost/test_array_facet/otters.json?_facet_array=tags&tags__arraycontains=eager",
"selected": False,
},
],
"hideable": True,
"toggle_url": "/test_array_facet/otters.json",
"truncated": False,
}
finally:
# Close databases first (while executor is still running)
for db_obj in ds.databases.values():
db_obj.close()
if hasattr(ds, "_internal_database"):
ds._internal_database.close()
# Then shut down executor
if ds.executor is not None:
ds.executor.shutdown(wait=True)
@pytest.mark.asyncio
@ -513,99 +526,124 @@ async def test_date_facet_results(ds_client):
@pytest.mark.asyncio
async def test_json_array_with_blanks_and_nulls():
ds = Datasette([], memory=True)
db = ds.add_database(Database(ds, memory_name="test_json_array"))
await db.execute_write("create table foo(json_column text)")
for value in ('["a", "b", "c"]', '["a", "b"]', "", None):
await db.execute_write("insert into foo (json_column) values (?)", [value])
response = await ds.client.get("/test_json_array/foo.json?_extra=suggested_facets")
data = response.json()
assert data["suggested_facets"] == [
{
"name": "json_column",
"type": "array",
"toggle_url": "http://localhost/test_json_array/foo.json?_extra=suggested_facets&_facet_array=json_column",
}
]
try:
db = ds.add_database(Database(ds, memory_name="test_json_array"))
await db.execute_write("create table foo(json_column text)")
for value in ('["a", "b", "c"]', '["a", "b"]', "", None):
await db.execute_write("insert into foo (json_column) values (?)", [value])
response = await ds.client.get(
"/test_json_array/foo.json?_extra=suggested_facets"
)
data = response.json()
assert data["suggested_facets"] == [
{
"name": "json_column",
"type": "array",
"toggle_url": "http://localhost/test_json_array/foo.json?_extra=suggested_facets&_facet_array=json_column",
}
]
finally:
# Close databases first (while executor is still running)
for db_obj in ds.databases.values():
db_obj.close()
if hasattr(ds, "_internal_database"):
ds._internal_database.close()
# Then shut down executor
if ds.executor is not None:
ds.executor.shutdown(wait=True)
@pytest.mark.asyncio
async def test_facet_size():
ds = Datasette([], memory=True, settings={"max_returned_rows": 50})
db = ds.add_database(Database(ds, memory_name="test_facet_size"))
await db.execute_write("create table neighbourhoods(city text, neighbourhood text)")
for i in range(1, 51):
for j in range(1, 4):
await db.execute_write(
"insert into neighbourhoods (city, neighbourhood) values (?, ?)",
["City {}".format(i), "Neighbourhood {}".format(j)],
)
response = await ds.client.get(
"/test_facet_size/neighbourhoods.json?_extra=suggested_facets"
)
data = response.json()
assert data["suggested_facets"] == [
{
"name": "neighbourhood",
"toggle_url": "http://localhost/test_facet_size/neighbourhoods.json?_extra=suggested_facets&_facet=neighbourhood",
}
]
# Bump up _facet_size= to suggest city too
response2 = await ds.client.get(
"/test_facet_size/neighbourhoods.json?_facet_size=50&_extra=suggested_facets"
)
data2 = response2.json()
assert sorted(data2["suggested_facets"], key=lambda f: f["name"]) == [
{
"name": "city",
"toggle_url": "http://localhost/test_facet_size/neighbourhoods.json?_facet_size=50&_extra=suggested_facets&_facet=city",
},
{
"name": "neighbourhood",
"toggle_url": "http://localhost/test_facet_size/neighbourhoods.json?_facet_size=50&_extra=suggested_facets&_facet=neighbourhood",
},
]
# Facet by city should return expected number of results
response3 = await ds.client.get(
"/test_facet_size/neighbourhoods.json?_facet_size=50&_facet=city"
)
data3 = response3.json()
assert len(data3["facet_results"]["results"]["city"]["results"]) == 50
# Reduce max_returned_rows and check that it's respected
ds._settings["max_returned_rows"] = 20
response4 = await ds.client.get(
"/test_facet_size/neighbourhoods.json?_facet_size=50&_facet=city"
)
data4 = response4.json()
assert len(data4["facet_results"]["results"]["city"]["results"]) == 20
# Test _facet_size=max
response5 = await ds.client.get(
"/test_facet_size/neighbourhoods.json?_facet_size=max&_facet=city"
)
data5 = response5.json()
assert len(data5["facet_results"]["results"]["city"]["results"]) == 20
# Now try messing with facet_size in the table metadata
orig_config = ds.config
try:
ds.config = {
"databases": {
"test_facet_size": {"tables": {"neighbourhoods": {"facet_size": 6}}}
}
}
response6 = await ds.client.get(
"/test_facet_size/neighbourhoods.json?_facet=city"
db = ds.add_database(Database(ds, memory_name="test_facet_size"))
await db.execute_write(
"create table neighbourhoods(city text, neighbourhood text)"
)
data6 = response6.json()
assert len(data6["facet_results"]["results"]["city"]["results"]) == 6
# Setting it to max bumps it up to 50 again
ds.config["databases"]["test_facet_size"]["tables"]["neighbourhoods"][
"facet_size"
] = "max"
data7 = (
await ds.client.get("/test_facet_size/neighbourhoods.json?_facet=city")
).json()
assert len(data7["facet_results"]["results"]["city"]["results"]) == 20
for i in range(1, 51):
for j in range(1, 4):
await db.execute_write(
"insert into neighbourhoods (city, neighbourhood) values (?, ?)",
["City {}".format(i), "Neighbourhood {}".format(j)],
)
response = await ds.client.get(
"/test_facet_size/neighbourhoods.json?_extra=suggested_facets"
)
data = response.json()
assert data["suggested_facets"] == [
{
"name": "neighbourhood",
"toggle_url": "http://localhost/test_facet_size/neighbourhoods.json?_extra=suggested_facets&_facet=neighbourhood",
}
]
# Bump up _facet_size= to suggest city too
response2 = await ds.client.get(
"/test_facet_size/neighbourhoods.json?_facet_size=50&_extra=suggested_facets"
)
data2 = response2.json()
assert sorted(data2["suggested_facets"], key=lambda f: f["name"]) == [
{
"name": "city",
"toggle_url": "http://localhost/test_facet_size/neighbourhoods.json?_facet_size=50&_extra=suggested_facets&_facet=city",
},
{
"name": "neighbourhood",
"toggle_url": "http://localhost/test_facet_size/neighbourhoods.json?_facet_size=50&_extra=suggested_facets&_facet=neighbourhood",
},
]
# Facet by city should return expected number of results
response3 = await ds.client.get(
"/test_facet_size/neighbourhoods.json?_facet_size=50&_facet=city"
)
data3 = response3.json()
assert len(data3["facet_results"]["results"]["city"]["results"]) == 50
# Reduce max_returned_rows and check that it's respected
ds._settings["max_returned_rows"] = 20
response4 = await ds.client.get(
"/test_facet_size/neighbourhoods.json?_facet_size=50&_facet=city"
)
data4 = response4.json()
assert len(data4["facet_results"]["results"]["city"]["results"]) == 20
# Test _facet_size=max
response5 = await ds.client.get(
"/test_facet_size/neighbourhoods.json?_facet_size=max&_facet=city"
)
data5 = response5.json()
assert len(data5["facet_results"]["results"]["city"]["results"]) == 20
# Now try messing with facet_size in the table metadata
orig_config = ds.config
try:
ds.config = {
"databases": {
"test_facet_size": {"tables": {"neighbourhoods": {"facet_size": 6}}}
}
}
response6 = await ds.client.get(
"/test_facet_size/neighbourhoods.json?_facet=city"
)
data6 = response6.json()
assert len(data6["facet_results"]["results"]["city"]["results"]) == 6
# Setting it to max bumps it up to 50 again
ds.config["databases"]["test_facet_size"]["tables"]["neighbourhoods"][
"facet_size"
] = "max"
data7 = (
await ds.client.get("/test_facet_size/neighbourhoods.json?_facet=city")
).json()
assert len(data7["facet_results"]["results"]["city"]["results"]) == 20
finally:
ds.config = orig_config
finally:
ds.config = orig_config
# Close databases first (while executor is still running)
# This allows db.close() to clear thread-local storage in executor threads
for db_obj in list(ds.databases.values()):
db_obj.close()
if hasattr(ds, "_internal_database"):
ds._internal_database.close()
# Then shut down executor
if ds.executor is not None:
ds.executor.shutdown(wait=True)
def test_other_types_of_facet_in_metadata():
@ -648,20 +686,30 @@ async def test_conflicting_facet_names_json(ds_client):
@pytest.mark.asyncio
async def test_facet_against_in_memory_database():
ds = Datasette()
db = ds.add_memory_database("mem")
await db.execute_write(
"create table t (id integer primary key, name text, name2 text)"
)
to_insert = [{"name": "one", "name2": "1"} for _ in range(800)] + [
{"name": "two", "name2": "2"} for _ in range(300)
]
await db.execute_write_many(
"insert into t (name, name2) values (:name, :name2)", to_insert
)
response1 = await ds.client.get("/mem/t")
assert response1.status_code == 200
response2 = await ds.client.get("/mem/t?_facet=name&_facet=name2")
assert response2.status_code == 200
try:
db = ds.add_memory_database("mem")
await db.execute_write(
"create table t (id integer primary key, name text, name2 text)"
)
to_insert = [{"name": "one", "name2": "1"} for _ in range(800)] + [
{"name": "two", "name2": "2"} for _ in range(300)
]
await db.execute_write_many(
"insert into t (name, name2) values (:name, :name2)", to_insert
)
response1 = await ds.client.get("/mem/t")
assert response1.status_code == 200
response2 = await ds.client.get("/mem/t?_facet=name&_facet=name2")
assert response2.status_code == 200
finally:
# Close databases first (while executor is still running)
for db_obj in ds.databases.values():
db_obj.close()
if hasattr(ds, "_internal_database"):
ds._internal_database.close()
# Then shut down executor
if ds.executor is not None:
ds.executor.shutdown(wait=True)
@pytest.mark.asyncio
@ -698,3 +746,9 @@ async def test_facet_only_considers_first_x_rows():
assert data2["suggested_facets"] == []
finally:
Facet.suggest_consider = original_suggest_consider
if ds.executor is not None:
ds.executor.shutdown(wait=True)
for db_obj in ds.databases.values():
db_obj.close()
if hasattr(ds, "_internal_database"):
ds._internal_database.close()

225
tests/test_permissions.py
View file

@ -42,7 +42,17 @@ async def perms_ds():
await two.execute_write("create table if not exists t1 (id integer primary key)")
# Trigger catalog refresh so allowed_resources() can be called
await ds.client.get("/")
return ds
try:
yield ds
finally:
# Close databases first (while executor is still running)
for db in ds.databases.values():
db.close()
if hasattr(ds, "_internal_database"):
ds._internal_database.close()
# Then shut down executor
if ds.executor is not None:
ds.executor.shutdown(wait=True)
@pytest.mark.parametrize(
@ -946,24 +956,34 @@ async def test_permissions_in_config(
@pytest.mark.asyncio
async def test_actor_endpoint_allows_any_token():
ds = Datasette()
token = ds.sign(
{
"a": "root",
try:
token = ds.sign(
{
"a": "root",
"token": "dstok",
"t": int(time.time()),
"_r": {"a": ["debug-menu"]},
},
namespace="token",
)
response = await ds.client.get(
"/-/actor.json", headers={"Authorization": f"Bearer dstok_{token}"}
)
assert response.status_code == 200
assert response.json()["actor"] == {
"id": "root",
"token": "dstok",
"t": int(time.time()),
"_r": {"a": ["debug-menu"]},
},
namespace="token",
)
response = await ds.client.get(
"/-/actor.json", headers={"Authorization": f"Bearer dstok_{token}"}
)
assert response.status_code == 200
assert response.json()["actor"] == {
"id": "root",
"token": "dstok",
"_r": {"a": ["debug-menu"]},
}
}
finally:
# Close databases first (while executor is still running)
for db in ds.databases.values():
db.close()
if hasattr(ds, "_internal_database"):
ds._internal_database.close()
# Then shut down executor
if ds.executor is not None:
ds.executor.shutdown(wait=True)
@pytest.mark.serial
@ -1341,9 +1361,19 @@ async def test_actor_restrictions(
)
async def test_restrictions_allow_action(restrictions, action, resource, expected):
ds = Datasette()
await ds.invoke_startup()
actual = restrictions_allow_action(ds, restrictions, action, resource)
assert actual == expected
try:
await ds.invoke_startup()
actual = restrictions_allow_action(ds, restrictions, action, resource)
assert actual == expected
finally:
# Close databases first (while executor is still running)
for db in ds.databases.values():
db.close()
if hasattr(ds, "_internal_database"):
ds._internal_database.close()
# Then shut down executor
if ds.executor is not None:
ds.executor.shutdown(wait=True)
@pytest.mark.asyncio
@ -1524,28 +1554,36 @@ async def test_actor_restrictions_cannot_be_overridden_by_config():
}
ds = Datasette(config=config)
await ds.invoke_startup()
db = ds.add_memory_database("test_db")
await db.execute_write("create table t1 (id integer primary key)")
await db.execute_write("create table t2 (id integer primary key)")
try:
await ds.invoke_startup()
db = ds.add_memory_database("test_db")
await db.execute_write("create table t1 (id integer primary key)")
await db.execute_write("create table t2 (id integer primary key)")
# Actor restricted to ONLY t1 (not t2)
# Even though config allows t2, restrictions should deny it
actor = {"id": "user", "_r": {"r": {"test_db": {"t1": ["vt"]}}}}
# Actor restricted to ONLY t1 (not t2)
# Even though config allows t2, restrictions should deny it
actor = {"id": "user", "_r": {"r": {"test_db": {"t1": ["vt"]}}}}
# t1 should be allowed (in restrictions AND config allows)
result = await ds.allowed(
action="view-table", resource=TableResource("test_db", "t1"), actor=actor
)
assert result is True, "t1 should be allowed - in restriction allowlist"
# t1 should be allowed (in restrictions AND config allows)
result = await ds.allowed(
action="view-table", resource=TableResource("test_db", "t1"), actor=actor
)
assert result is True, "t1 should be allowed - in restriction allowlist"
# t2 should be DENIED (not in restrictions, even though config allows)
result = await ds.allowed(
action="view-table", resource=TableResource("test_db", "t2"), actor=actor
)
assert (
result is False
), "t2 should be denied - NOT in restriction allowlist, config cannot override"
# t2 should be DENIED (not in restrictions, even though config allows)
result = await ds.allowed(
action="view-table", resource=TableResource("test_db", "t2"), actor=actor
)
assert (
result is False
), "t2 should be denied - NOT in restriction allowlist, config cannot override"
finally:
if ds.executor is not None:
ds.executor.shutdown(wait=True)
for db_obj in ds.databases.values():
db_obj.close()
if hasattr(ds, "_internal_database"):
ds._internal_database.close()
@pytest.mark.asyncio
@ -1644,29 +1682,42 @@ async def test_permission_check_view_requires_debug_permission():
"""Test that /-/check requires permissions-debug permission"""
# Anonymous user should be denied
ds = Datasette()
response = await ds.client.get("/-/check.json?action=view-instance")
assert response.status_code == 403
assert "permissions-debug" in response.text
ds_with_root = None
try:
response = await ds.client.get("/-/check.json?action=view-instance")
assert response.status_code == 403
assert "permissions-debug" in response.text
# User without permissions-debug should be denied
response = await ds.client.get(
"/-/check.json?action=view-instance",
cookies={"ds_actor": ds.sign({"id": "user"}, "actor")},
)
assert response.status_code == 403
# User without permissions-debug should be denied
response = await ds.client.get(
"/-/check.json?action=view-instance",
cookies={"ds_actor": ds.sign({"id": "user"}, "actor")},
)
assert response.status_code == 403
# Root user should have access (root has all permissions)
ds_with_root = Datasette()
ds_with_root.root_enabled = True
root_token = ds_with_root.create_token("root")
response = await ds_with_root.client.get(
"/-/check.json?action=view-instance",
headers={"Authorization": f"Bearer {root_token}"},
)
assert response.status_code == 200
data = response.json()
assert data["action"] == "view-instance"
assert data["allowed"] is True
# Root user should have access (root has all permissions)
ds_with_root = Datasette()
ds_with_root.root_enabled = True
root_token = ds_with_root.create_token("root")
response = await ds_with_root.client.get(
"/-/check.json?action=view-instance",
headers={"Authorization": f"Bearer {root_token}"},
)
assert response.status_code == 200
data = response.json()
assert data["action"] == "view-instance"
assert data["allowed"] is True
finally:
for ds_obj in [ds, ds_with_root]:
if ds_obj is not None:
# Close databases first (while executor is still running)
for db in ds_obj.databases.values():
db.close()
if hasattr(ds_obj, "_internal_database"):
ds_obj._internal_database.close()
# Then shut down executor
if ds_obj.executor is not None:
ds_obj.executor.shutdown(wait=True)
@pytest.mark.asyncio
@ -1686,29 +1737,37 @@ async def test_root_allow_block_with_table_restricted_actor():
"allow": {"id": "admin"}, # Root-level allow block
}
)
await ds.invoke_startup()
db = ds.add_memory_database("mydb")
await db.execute_write("create table t1 (id integer primary key)")
await ds.client.get("/") # Trigger catalog refresh
try:
await ds.invoke_startup()
db = ds.add_memory_database("mydb")
await db.execute_write("create table t1 (id integer primary key)")
await ds.client.get("/") # Trigger catalog refresh
# Actor with table-level restrictions only (not global)
actor = {"id": "user", "_r": {"r": {"mydb": {"t1": ["view-table"]}}}}
# Actor with table-level restrictions only (not global)
actor = {"id": "user", "_r": {"r": {"mydb": {"t1": ["view-table"]}}}}
# The root-level allow: {id: admin} should be processed and deny this user
# because they're not "admin", even though they have table restrictions
result = await ds.allowed(
action="view-table",
resource=TableResource("mydb", "t1"),
actor=actor,
)
# Should be False because root allow: {id: admin} denies non-admin users
assert result is False
# The root-level allow: {id: admin} should be processed and deny this user
# because they're not "admin", even though they have table restrictions
result = await ds.allowed(
action="view-table",
resource=TableResource("mydb", "t1"),
actor=actor,
)
# Should be False because root allow: {id: admin} denies non-admin users
assert result is False
# But admin with same restrictions should be allowed
admin_actor = {"id": "admin", "_r": {"r": {"mydb": {"t1": ["view-table"]}}}}
result = await ds.allowed(
action="view-table",
resource=TableResource("mydb", "t1"),
actor=admin_actor,
)
assert result is True
# But admin with same restrictions should be allowed
admin_actor = {"id": "admin", "_r": {"r": {"mydb": {"t1": ["view-table"]}}}}
result = await ds.allowed(
action="view-table",
resource=TableResource("mydb", "t1"),
actor=admin_actor,
)
assert result is True
finally:
if ds.executor is not None:
ds.executor.shutdown(wait=True)
for db_obj in ds.databases.values():
db_obj.close()
if hasattr(ds, "_internal_database"):
ds._internal_database.close()