From 266eaddb7370fb9641c270daafd716f2f3402d16 Mon Sep 17 00:00:00 2001 From: Simon Willison Date: Wed, 16 Sep 2026 14:50:06 -0700 Subject: [PATCH] Fix remaining pytest warnings (#2928) - Close in-memory database connections, including reads opened on threads - Close completed and partial file upload file handles - Close SQLite connections and file handles owned by tests --- datasette/database.py | 26 +-- datasette/utils/multipart.py | 65 +++++-- tests/conftest.py | 62 ++++--- tests/fixtures.py | 17 +- tests/test_cli_serve_server.py | 4 +- tests/test_column_types.py | 2 + tests/test_internals_database.py | 14 ++ tests/test_multipart.py | 251 +++++++++++++++++++++------- tests/test_numeric_filter_values.py | 30 ++-- 9 files changed, 336 insertions(+), 135 deletions(-) diff --git a/datasette/database.py b/datasette/database.py index d444cbbf..024f62a4 100644 --- a/datasette/database.py +++ b/datasette/database.py @@ -94,8 +94,9 @@ 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 = [] + # Track file and memory connections, including reads on worker threads, + # so close() can release all of them from the calling thread. + self._all_connections = [] if not is_temp_disk: self.mode = mode @@ -146,9 +147,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: @@ -165,7 +169,7 @@ 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) if self.is_temp_disk and not self._wal_enabled: conn.execute("PRAGMA journal_mode=WAL") self._wal_enabled = True @@ -202,13 +206,13 @@ class Database: except Exception: # noqa: BLE001, S110 # Shutdown teardown - a failed pending write must not block close() pass - # Close anything still tracked in _all_file_connections - for connection in self._all_file_connections: + # Close anything still tracked in _all_connections + for connection in self._all_connections: try: connection.close() except Exception: # noqa: BLE001, S110 pass - self._all_file_connections = [] + self._all_connections = [] # Drop per-thread cached read connections we can reach try: delattr(connections, self._thread_local_id) @@ -324,9 +328,9 @@ 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 + # May already have been cleared by close(). pass if self.ds.executor is None: @@ -491,9 +495,9 @@ 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 + # May already have been cleared by close(). pass except Exception as e: # noqa: BLE001 # Write thread must survive any task failure or the database wedges diff --git a/datasette/utils/multipart.py b/datasette/utils/multipart.py index 13289b1c..182c7ab1 100644 --- a/datasette/utils/multipart.py +++ b/datasette/utils/multipart.py @@ -358,6 +358,13 @@ class MultipartParser: self.buffer.extend(chunk) self._process() + def close(self) -> None: + """Discard completed uploads and any file still being received.""" + if self.current_file is not None: + self.current_file.close() + self.current_file = None + self.form_data.close() + def _process(self) -> None: """Process buffered data.""" while True: @@ -577,6 +584,9 @@ class MultipartParser: def _finish_part(self) -> None: """Finalize current part and add to form data.""" if self.current_name is None: + if self.current_file is not None: + self.current_file.close() + self.current_file = None return if self.current_filename is not None: @@ -722,29 +732,50 @@ async def parse_form_data( batch_target = 64 * 1024 batch = bytearray() + async def run_parser(fn, *args): + # Cancellation must not close files while a worker is using them. + task = asyncio.create_task(asyncio.to_thread(fn, *args)) + try: + return await asyncio.shield(task) + except asyncio.CancelledError as cancelled: + try: + while not task.done(): + try: + await asyncio.shield(task) + except asyncio.CancelledError: + continue + task.result() + finally: + raise cancelled + async def flush_batch() -> None: if batch: data = bytes(batch) batch.clear() - await asyncio.to_thread(parser.feed, data) + await run_parser(parser.feed, data) - while True: - message = await receive() - message_type = message.get("type") - if message_type == "http.disconnect": - raise MultipartParseError("Client disconnected during request body") - if message_type is not None and message_type != "http.request": - continue - chunk = message.get("body", b"") - if chunk: - batch.extend(chunk) - if len(batch) >= batch_target: - await flush_batch() - if not message.get("more_body", False): - break + try: + while True: + message = await receive() + message_type = message.get("type") + if message_type == "http.disconnect": + raise MultipartParseError("Client disconnected during request body") + if message_type is not None and message_type != "http.request": + continue + chunk = message.get("body", b"") + if chunk: + batch.extend(chunk) + if len(batch) >= batch_target: + await flush_batch() + if not message.get("more_body", False): + break - await flush_batch() - return await asyncio.to_thread(parser.finalize) + await flush_batch() + return await run_parser(parser.finalize) + except BaseException: + # No FormData is returned to the caller to take ownership on failure. + await asyncio.to_thread(parser.close) + raise else: raise MultipartParseError( diff --git a/tests/conftest.py b/tests/conftest.py index 5ebcb14b..e90b5ab6 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -106,7 +106,10 @@ async def ds_client(): await db.execute_write_fn(prepare) await ds.invoke_startup() - return ds.client + try: + yield ds.client + finally: + ds.close() def pytest_report_header(config): @@ -266,12 +269,24 @@ def ds_localhost_http_server(): # 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() + try: + wait_until_responds("http://localhost:8041/", process=ds_proc) + yield ds_proc + finally: + stop_process(ds_proc) + + +def stop_process(proc): + try: + if proc.poll() is None: + proc.terminate() + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait() + finally: + proc.stdout.close() @pytest.fixture(scope="session") @@ -295,8 +310,24 @@ def ds_unix_domain_socket_server(tmp_path_factory): transport = httpx2.HTTPTransport(uds=uds) client = httpx2.Client(transport=transport) try: + # Probe with a socket we own: the HTTP transport can leak a socket + # when connect() fails before the UDS server has started listening. + start = time.monotonic() + while True: + try: + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as probe: + probe.settimeout(0.1) + probe.connect(uds) + break + except OSError: + if ds_proc.poll() is not None or time.monotonic() - start > 30: + raise + time.sleep(0.1) wait_until_responds( - "http://localhost/_memory.json", timeout=30.0, client=client + "http://localhost/_memory.json", + timeout=30.0, + client=client, + process=ds_proc, ) # Check it started successfully assert not ds_proc.poll(), ds_proc.stdout.read().decode("utf-8") @@ -304,12 +335,7 @@ def ds_unix_domain_socket_server(tmp_path_factory): 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() + stop_process(ds_proc) try: os.unlink(uds) except FileNotFoundError: @@ -372,13 +398,7 @@ def serve_with_plugins(tmp_path): 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() + stop_process(proc) # Import fixtures from fixtures.py to make them available diff --git a/tests/fixtures.py b/tests/fixtures.py index 83607c1a..f8710ad4 100644 --- a/tests/fixtures.py +++ b/tests/fixtures.py @@ -169,12 +169,10 @@ def make_app_client( template_dir=template_dir, crossdb=crossdb, ) - yield TestClient(ds) - # Close as many database connections as possible - # to try and avoid too many open files error - for db in ds.databases.values(): - if not db.is_memory: - db.close() + try: + yield TestClient(ds) + finally: + ds.close() @pytest.fixture(scope="session") @@ -186,9 +184,10 @@ def app_client(): @pytest.fixture(scope="session") def app_client_no_files(): ds = Datasette([]) - yield TestClient(ds) - for db in ds.databases.values(): - db.close() + try: + yield TestClient(ds) + finally: + ds.close() @pytest.fixture(scope="session") diff --git a/tests/test_cli_serve_server.py b/tests/test_cli_serve_server.py index e9686334..163d3fcc 100644 --- a/tests/test_cli_serve_server.py +++ b/tests/test_cli_serve_server.py @@ -24,8 +24,8 @@ 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 = httpx2.HTTPTransport(uds=uds) - client = httpx2.Client(transport=transport) - response = client.get("http://localhost/_memory.json") + with httpx2.Client(transport=transport) as client: + response = client.get("http://localhost/_memory.json") assert { "database": "_memory", "path": "/_memory", diff --git a/tests/test_column_types.py b/tests/test_column_types.py index 50c6daed..d8dfc627 100644 --- a/tests/test_column_types.py +++ b/tests/test_column_types.py @@ -31,6 +31,7 @@ def ds_ct(tmp_path_factory): "'https://example.com', '{\"key\": \"value\"}')" ) db.commit() + db.close() ds = Datasette( [db_path], config={ @@ -70,6 +71,7 @@ def ds_ct_editor_permission(tmp_path_factory): "'https://example.com', '{\"key\": \"value\"}')" ) db.commit() + db.close() ds = Datasette( [db_path], config={ diff --git a/tests/test_internals_database.py b/tests/test_internals_database.py index 4842d8b7..14a7ce11 100644 --- a/tests/test_internals_database.py +++ b/tests/test_internals_database.py @@ -1305,3 +1305,17 @@ async def test_database_close_is_idempotent(tmpdir): # Second call should be a no-op, not raise db.close() ds._internal_database.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("num_sql_threads", [0, 2]) +@pytest.mark.parametrize("named", [False, True]) +async def test_close_releases_memory_connections(num_sql_threads, named): + ds = Datasette(memory=True, settings={"num_sql_threads": num_sql_threads}) + db = ds.add_memory_database(uuid.uuid4().hex) if named else ds.get_database() + read_connection = await db.execute_fn(lambda conn: conn) + write_connection = await db.execute_write_fn(lambda conn: conn) + ds.close() + for conn in (read_connection, write_connection): + with pytest.raises(sqlite3.ProgrammingError, match="closed"): + conn.execute("select 1") diff --git a/tests/test_multipart.py b/tests/test_multipart.py index ab38bfb7..8cc91db3 100644 --- a/tests/test_multipart.py +++ b/tests/test_multipart.py @@ -4,14 +4,145 @@ Tests for request.form() multipart form data parsing. Uses TDD approach - these tests are written first, then implementation follows. """ +import asyncio import base64 import json +import threading from collections import namedtuple import pytest from multipart_form_data_conformance import get_tests_dir from datasette.utils.asgi import BadRequest, Request +from datasette.utils.multipart import MultipartParseError, parse_form_data + + +@pytest.fixture +def upload_files(monkeypatch): + from datasette.utils import multipart + + files = [] + original = multipart.tempfile.SpooledTemporaryFile + + def create_file(*args, **kwargs): + file = original(*args, **kwargs) + files.append(file) + return file + + monkeypatch.setattr(multipart.tempfile, "SpooledTemporaryFile", create_file) + yield files + for file in files: + file.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "failure", + [ + "file_limit", + "request_limit", + "truncated", + "disconnect", + "receive_error", + "cancel", + ], +) +async def test_failed_upload_closes_completed_and_partial_files(upload_files, failure): + # Complete one file and begin another, flushing the parser's 64 KiB batch. + body = ( + b'--boundary\r\nContent-Disposition: form-data; name="one"; filename="one"\r\n\r\n' + b'first\r\n--boundary\r\nContent-Disposition: form-data; name="two"; filename="two"\r\n\r\n' + + b"x" * (64 * 1024) + ) + calls = 0 + + async def receive(): + nonlocal calls + calls += 1 + if calls == 1: + return {"type": "http.request", "body": body, "more_body": True} + if failure == "disconnect": + return {"type": "http.disconnect"} + if failure == "receive_error": + raise OSError("receive failed") + if failure == "cancel": + raise asyncio.CancelledError + return {"type": "http.request", "body": b"x" * 100, "more_body": False} + + kwargs = {} + if failure == "file_limit": + kwargs["max_file_size"] = 1024 + if failure == "request_limit": + kwargs["max_request_size"] = len(body) + error = { + "receive_error": OSError, + "cancel": asyncio.CancelledError, + }.get(failure, MultipartParseError) + with pytest.raises(error): + await parse_form_data( + receive, "multipart/form-data; boundary=boundary", files=True, **kwargs + ) + assert len(upload_files) == 2 + assert all(file.closed for file in upload_files) + + +@pytest.mark.asyncio +async def test_unnamed_upload_is_closed(upload_files): + body = ( + b'--boundary\r\nContent-Disposition: form-data; filename="ignored"\r\n\r\n' + b"content\r\n--boundary--\r\n" + ) + form = await parse_form_data( + make_receive(body), "multipart/form-data; boundary=boundary", files=True + ) + assert len(form) == 0 + assert len(upload_files) == 1 + assert upload_files[0].closed + + +@pytest.mark.asyncio +@pytest.mark.parametrize("worker_error", [False, True]) +async def test_cancelled_upload_waits_for_worker( + upload_files, monkeypatch, worker_error +): + from datasette.utils.multipart import MultipartParser + + loop = asyncio.get_running_loop() + started = asyncio.Event() + release = threading.Event() + original_feed = MultipartParser.feed + + def blocking_feed(self, chunk): + original_feed(self, chunk) + loop.call_soon_threadsafe(started.set) + assert release.wait(5) + if worker_error: + raise MultipartParseError("worker failed") + + monkeypatch.setattr(MultipartParser, "feed", blocking_feed) + body = ( + b'--boundary\r\nContent-Disposition: form-data; name="file"; filename="file"\r\n\r\n' + + b"x" * (64 * 1024) + ) + task = asyncio.create_task( + parse_form_data( + make_receive(body), "multipart/form-data; boundary=boundary", files=True + ) + ) + try: + await asyncio.wait_for(started.wait(), timeout=5) + # A second cancellation must also leave cleanup waiting for the worker. + for _ in range(2): + task.cancel() + await asyncio.sleep(0) + assert not task.done() + assert len(upload_files) == 1 + assert not upload_files[0].closed + finally: + release.set() + with pytest.raises(asyncio.CancelledError): + await task + assert upload_files[0].closed def make_receive(body: bytes): @@ -1078,75 +1209,73 @@ async def test_conformance(test_spec, headers, body): await request.form(files=True) return - # Parse form data - form = await request.form(files=True) + async with await request.form(files=True) as form: + # Verify each expected part + for i, expected_part in enumerate(expected["parts"]): + name = expected_part["name"] - # Verify each expected part - for i, expected_part in enumerate(expected["parts"]): - name = expected_part["name"] + # Get value(s) for this name + values = form.getlist(name) - # Get value(s) for this name - values = form.getlist(name) + # Find the value at the correct index for this name + # (handles multiple values with same name) + same_name_count = sum(1 for p in expected["parts"][:i] if p["name"] == name) - # Find the value at the correct index for this name - # (handles multiple values with same name) - same_name_count = sum(1 for p in expected["parts"][:i] if p["name"] == name) + if same_name_count >= len(values): + pytest.fail( + f"Expected part {name} at index {same_name_count} but only {len(values)} found" + ) - if same_name_count >= len(values): - pytest.fail( - f"Expected part {name} at index {same_name_count} but only {len(values)} found" + value = values[same_name_count] + + # Determine expected content + if "body_base64" in expected_part: + expected_content = base64.b64decode(expected_part["body_base64"]) + elif "body_text" in expected_part: + expected_content = expected_part["body_text"].encode("utf-8") + else: + expected_content = None + + # Check for file vs field + # A part is a file if it has a filename OR filename_star + is_file = ( + expected_part.get("filename") is not None + or expected_part.get("filename_star") is not None ) - value = values[same_name_count] + if is_file: + # It's a file + assert hasattr(value, "filename"), f"Expected file for {name}" - # Determine expected content - if "body_base64" in expected_part: - expected_content = base64.b64decode(expected_part["body_base64"]) - elif "body_text" in expected_part: - expected_content = expected_part["body_text"].encode("utf-8") - else: - expected_content = None + # Check filename - use filename_star if present, else filename + expected_filename = expected_part.get( + "filename_star" + ) or expected_part.get("filename") + if expected_filename: + assert ( + value.filename == expected_filename + ), f"Filename mismatch: expected {expected_filename!r}, got {value.filename!r}" - # Check for file vs field - # A part is a file if it has a filename OR filename_star - is_file = ( - expected_part.get("filename") is not None - or expected_part.get("filename_star") is not None - ) + if expected_part.get("content_type"): + assert value.content_type == expected_part["content_type"] - if is_file: - # It's a file - assert hasattr(value, "filename"), f"Expected file for {name}" - - # Check filename - use filename_star if present, else filename - expected_filename = expected_part.get("filename_star") or expected_part.get( - "filename" - ) - if expected_filename: + content = await value.read() assert ( - value.filename == expected_filename - ), f"Filename mismatch: expected {expected_filename!r}, got {value.filename!r}" + len(content) == expected_part["body_size"] + ), f"Size mismatch: expected {expected_part['body_size']}, got {len(content)}" + if expected_content is not None: + assert content == expected_content + else: + # It's a text field + if hasattr(value, "filename"): + pytest.fail(f"Expected text field for {name}, got file") - if expected_part.get("content_type"): - assert value.content_type == expected_part["content_type"] - - content = await value.read() - assert ( - len(content) == expected_part["body_size"] - ), f"Size mismatch: expected {expected_part['body_size']}, got {len(content)}" - if expected_content is not None: - assert content == expected_content - else: - # It's a text field - if hasattr(value, "filename"): - pytest.fail(f"Expected text field for {name}, got file") - - if expected_content is not None: - # For text fields, value is a string - try: - expected_text = expected_content.decode("utf-8") - except UnicodeDecodeError: - expected_text = expected_content.decode("latin-1") - assert ( - value == expected_text - ), f"Value mismatch: expected {expected_text!r}, got {value!r}" + if expected_content is not None: + # For text fields, value is a string + try: + expected_text = expected_content.decode("utf-8") + except UnicodeDecodeError: + expected_text = expected_content.decode("latin-1") + assert ( + value == expected_text + ), f"Value mismatch: expected {expected_text!r}, got {value!r}" diff --git a/tests/test_numeric_filter_values.py b/tests/test_numeric_filter_values.py index 69c5cf7e..1fceeabe 100644 --- a/tests/test_numeric_filter_values.py +++ b/tests/test_numeric_filter_values.py @@ -1,4 +1,5 @@ import sqlite3 +from contextlib import closing import pytest @@ -27,19 +28,20 @@ def test_numeric_filter_parameters(value, expected): def test_numeric_filter_parameters_against_calculated_view(): - conn = sqlite3.connect(":memory:") - conn.execute("create table searchable(pk integer)") - conn.executemany("insert into searchable(pk) values (?)", [(0,), (1,), (2,)]) - conn.execute( - "create view calculated as " - "select pk + 1 as pk_plus_one, pk / 2.0 as score from searchable" - ) + with closing(sqlite3.connect(":memory:")) as conn: + conn.execute("create table searchable(pk integer)") + conn.executemany("insert into searchable(pk) values (?)", [(0,), (1,), (2,)]) + conn.execute( + "create view calculated as " + "select pk + 1 as pk_plus_one, pk / 2.0 as score from searchable" + ) - sql_bits, params = Filters((("score__gt", "0.1"),)).build_where_clauses( - "calculated" - ) - rows = conn.execute( - "select score from calculated where {}".format(" and ".join(sql_bits)), params - ).fetchall() + sql_bits, params = Filters((("score__gt", "0.1"),)).build_where_clauses( + "calculated" + ) + rows = conn.execute( + "select score from calculated where {}".format(" and ".join(sql_bits)), + params, + ).fetchall() - assert rows == [(0.5,), (1.0,)] + assert rows == [(0.5,), (1.0,)]