From e8c7bf1be8713b2e8fedf4e03e1cf984e0a6912b Mon Sep 17 00:00:00 2001 From: Thor Whalen <1906276+thorwhalen@users.noreply.github.com> Date: Tue, 22 Sep 2026 14:30:03 +0000 Subject: [PATCH 1/2] fix(base,sql_base): bind keys as parameters, allowlist raw-SQL table names SqlBaseKvStore._mk_column_filter built the WHERE clause of __setitem__ and __delitem__ by formatting keys (and, for mapping keys, column names) into text(). A key containing a quote broke the write, and a key could change which rows the statement touched. - The key filter is now built from column expressions (table.c[col] == value), so values are bound parameters and column names can only be columns the table has (clear KeyError otherwise). This is how __getitem__ already matched keys, so reads and writes now agree. Empty mapping keys raise ValueError. - The legacy raw-SQL paths in sql_base (SqlTableRowsCollection, iter_rows) validate table names against an identifier allowlist and coerce LIMIT/OFFSET to integers. - Regression tests write, read back, overwrite and delete keys holding quotes, semicolons and --, and check no other row changes. Closes #6 Co-Authored-By: Claude Opus 5 --- sqldol/base.py | 42 ++++-- sqldol/sql_base.py | 45 +++++- sqldol/tests/test_sql_injection.py | 226 +++++++++++++++++++++++++++++ 3 files changed, 299 insertions(+), 14 deletions(-) create mode 100644 sqldol/tests/test_sql_injection.py diff --git a/sqldol/base.py b/sqldol/base.py index e02b50a..263a6bf 100644 --- a/sqldol/base.py +++ b/sqldol/base.py @@ -10,10 +10,10 @@ insert, exists, update, - text, func, Engine, Column, + and_, ) from sqlalchemy import Table, Column, MetaData @@ -291,19 +291,35 @@ def __getitem__(self, key): # TODO: Needs to be made compliant with the "Base" strategy (see SqlBaseKvReader) # For example, perhaps values are not dicts, but lists of rows class SqlBaseKvStore(SqlBaseKvReader, MutableMapping): - def _mk_column_filter(self, key): - if isinstance(key, str): - return text(f"{self.key_columns} = '{key}'") - elif isinstance(key, int): # the key is a tuple of columns - return text(f"{self.key_columns} = {key}") - elif isinstance(key, dict): - return text(" AND ".join(f"{col} = '{val}'" for col, val in key.items())) - else: - return text( - " AND ".join( - f"{col} = '{val}'" for col, val in zip(self.key_columns, key) - ) + def _key_column(self, column_name): + """The ``Column`` object named ``column_name``, or an informative ``KeyError``. + + Going through ``self.table.c`` means a column name is never spliced into SQL + text: only names the table actually has can be used, and SQLAlchemy quotes + them as the dialect requires. + """ + try: + return self.table.c[column_name] + except KeyError: + msg = ( + f"{column_name!r} is not a column of table {self.table_name!r}. " + f"Its columns are {self._column_names}." ) + raise KeyError(msg) from None + + def _mk_column_filter(self, key): + """The ``WHERE`` clause selecting the rows of ``key``. + + A ``Mapping`` key is a ``{column_name: value, ...}`` conjunction; any other key + is the value of the key column. Values are always bound parameters, never + interpolated into SQL text, so keys containing quotes, semicolons or ``--`` + are matched literally -- the same way ``__getitem__`` matches them. + """ + if isinstance(key, Mapping): + if not key: + raise ValueError("A mapping key must name at least one column") + return and_(*(self._key_column(col) == val for col, val in key.items())) + return self._key_column(self.key_columns) == key def __setitem__(self, key, value): filter = self._mk_column_filter(key) diff --git a/sqldol/sql_base.py b/sqldol/sql_base.py index b465818..380a1d4 100644 --- a/sqldol/sql_base.py +++ b/sqldol/sql_base.py @@ -2,7 +2,9 @@ sql with a simple (dict-like or list-like) interface """ +import re from functools import partial +from operator import index import sqlalchemy from sqlalchemy import create_engine, Column, String, Table @@ -18,6 +20,40 @@ DFLT_SQL_PORT = 1433 DFLT_SQL_HOST = "localhost" +SQL_IDENTIFIER_PATTERN = re.compile( + r"[A-Za-z_][A-Za-z0-9_$]*(\.[A-Za-z_][A-Za-z0-9_$]*)?" +) +"""What a table name may look like where it is written into raw SQL text: a plain +identifier, optionally qualified by a schema (``schema.table``).""" + + +def validate_sql_identifier(name, *, pattern=SQL_IDENTIFIER_PATTERN): + """Return ``name`` if it is safe to write into raw SQL text, else raise ``ValueError``. + + The raw-SQL paths of this module cannot bind a table name as a parameter (no SQL + dialect allows that), and they may be handed a plain DB-API connection that has no + quoting helper, so they only accept names matching an allowlist. + + >>> validate_sql_identifier("my_table") + 'my_table' + >>> validate_sql_identifier("my_schema.my_table") + 'my_schema.my_table' + >>> validate_sql_identifier("t; SELECT 1") # doctest: +ELLIPSIS + Traceback (most recent call last): + ... + ValueError: Not a valid SQL table name: 't; SELECT 1'. ... + """ + if not isinstance(name, str) or not pattern.fullmatch(name): + msg = ( + f"Not a valid SQL table name: {name!r}. " + f"Expected letters, digits, '_' or '$' (not starting with a digit), " + f"optionally qualified as 'schema.table'. " + f"Names that need quoting are not supported here: use the SQLAlchemy-based " + f"stores (sqldol.base, sqldol.stores), which quote identifiers themselves." + ) + raise ValueError(msg) + return name + # TODO: decorator to automatically retry (once) if the connection times out # --> from sqlalchemy.exc import OperationalError @@ -44,7 +80,9 @@ class SqlTableRowsCollection(Collection): def __init__(self, connection, table_name, batch_size=2000, limit=int(1e16)): self.connection = connection - self.table_name = table_name + # Note: table_name is written into raw SQL text (see the _tmpl_* templates), + # so it has to pass the identifier allowlist. + self.table_name = validate_sql_identifier(table_name) self.iter_rows = partial( iter_rows, connection=connection, @@ -398,7 +436,12 @@ def __len__(self): def iter_rows(connection, table_name, batch_size=1000, offset=0, limit=int(1e12)): """Iterate the over the rows of a table. The limit argument is mostly there to avoid an infinite loop, but can also be used to get ranges. + + ``table_name`` must pass :func:`validate_sql_identifier`, and ``batch_size``, + ``offset`` and ``limit`` must be integers, because they are written into the SQL text. """ + table_name = validate_sql_identifier(table_name) + batch_size, offset, limit = index(batch_size), index(offset), index(limit) stop_offset = limit + offset # to the range act like like sql limit i = 0 for offset in range(offset, stop_offset, batch_size): diff --git a/sqldol/tests/test_sql_injection.py b/sqldol/tests/test_sql_injection.py new file mode 100644 index 0000000..80ec0b7 --- /dev/null +++ b/sqldol/tests/test_sql_injection.py @@ -0,0 +1,226 @@ +"""Regression tests: keys and table names are never spliced into SQL text. + +``SqlBaseKvStore`` (and so ``SqlDictStore``) used to build the ``WHERE`` clause of +``__setitem__`` / ``__delitem__`` by string formatting, so a key containing a quote +broke the query and a key could rewrite it. The legacy raw-SQL paths of +``sqldol.sql_base`` wrote table names into SQL text unchecked. + +Every key here holds SQL metacharacters (quotes, semicolons, ``--``). Each test +writes it, reads it back, and checks that no other row was touched. + +Uses in-memory SQLite only -- no external service, no credentials. +""" + +import sqlite3 + +import pytest +from sqlalchemy import create_engine, text + +from sqldol.base import SqlBaseKvStore +from sqldol.sql_base import SqlTableRowsCollection, iter_rows, validate_sql_identifier +from sqldol.stores import SqlDictStore + +TABLE_NAME = "t" +KEY_COLUMN = "k" +ORIGINAL_ROWS = {"a": "1", "b": "2"} + +TRICKY_KEYS = [ + "it's", + 'say "hi"', + "a; DROP TABLE t; --", + "x' OR '1'='1", + "x' OR '1'='1' --", + "k -- comment", + "';", + "--", + "\\'", +] + + +@pytest.fixture +def engine(): + """An in-memory SQLite engine holding ``t(k, v)`` with rows a/1 and b/2.""" + engine = create_engine("sqlite:///:memory:") + with engine.connect() as connection: + connection.execute(text(f"CREATE TABLE {TABLE_NAME} (k TEXT, v TEXT)")) + connection.execute( + text(f"INSERT INTO {TABLE_NAME} VALUES ('a', '1'), ('b', '2')") + ) + connection.commit() + return engine + + +def _all_rows(engine): + with engine.connect() as connection: + rows = connection.execute(text(f"SELECT k, v FROM {TABLE_NAME}")).fetchall() + return sorted(map(tuple, rows)) + + +def _original_rows(): + return sorted(ORIGINAL_ROWS.items()) + + +@pytest.mark.parametrize("key", TRICKY_KEYS) +def test_tricky_key_insert_then_read_back(engine, key): + store = SqlDictStore(engine, TABLE_NAME, key_columns=KEY_COLUMN) + + store[key] = {"k": key, "v": "new"} + + assert store[key] == {"k": key, "v": "new"} + assert _all_rows(engine) == sorted(_original_rows() + [(key, "new")]) + + +@pytest.mark.parametrize("key", TRICKY_KEYS) +def test_tricky_key_overwrite_updates_only_that_row(engine, key): + store = SqlDictStore(engine, TABLE_NAME, key_columns=KEY_COLUMN) + store[key] = {"k": key, "v": "first"} + + store[key] = {"k": key, "v": "second"} + + assert store[key] == {"k": key, "v": "second"} + assert _all_rows(engine) == sorted(_original_rows() + [(key, "second")]) + + +@pytest.mark.parametrize("key", TRICKY_KEYS) +def test_deleting_an_absent_tricky_key_touches_no_row(engine, key): + """With interpolation, ``x' OR '1'='1`` as a key emptied the whole table.""" + store = SqlDictStore(engine, TABLE_NAME, key_columns=KEY_COLUMN) + + del store[key] + + assert _all_rows(engine) == _original_rows() + + +@pytest.mark.parametrize("key", TRICKY_KEYS) +def test_deleting_a_present_tricky_key_removes_only_that_row(engine, key): + store = SqlDictStore(engine, TABLE_NAME, key_columns=KEY_COLUMN) + store[key] = {"k": key, "v": "new"} + + del store[key] + + assert store[key] is None + assert _all_rows(engine) == _original_rows() + + +@pytest.mark.parametrize("key", TRICKY_KEYS) +def test_tricky_values_in_a_mapping_key_are_bound(engine, key): + store = SqlBaseKvStore(engine, TABLE_NAME, key_columns=KEY_COLUMN) + store[key] = {"k": key, "v": key} + + store[{"k": key, "v": key}] = {"v": "updated"} + assert _all_rows(engine) == sorted(_original_rows() + [(key, "updated")]) + + del store[{"k": key, "v": "updated"}] + assert _all_rows(engine) == _original_rows() + + +def test_normal_keys_behave_as_before(engine): + """Guard for existing callers: plain str and int keys, and dict keys.""" + store = SqlDictStore(engine, TABLE_NAME, key_columns=KEY_COLUMN) + + store["a"] = {"k": "a", "v": "10"} + store["c"] = {"k": "c", "v": "3"} + assert store["a"] == {"k": "a", "v": "10"} + assert store["c"] == {"k": "c", "v": "3"} + + base = SqlBaseKvStore(engine, TABLE_NAME, key_columns=KEY_COLUMN) + del base[{"k": "c"}] + del store["b"] + assert _all_rows(engine) == [("a", "10")] + + +def test_int_keys_on_an_integer_key_column(engine): + with engine.connect() as connection: + connection.execute(text("CREATE TABLE n (id INTEGER, v TEXT)")) + connection.execute(text("INSERT INTO n VALUES (1, 'one'), (2, 'two')")) + connection.commit() + store = SqlDictStore(engine, "n", key_columns="id") + + store[1] = {"id": 1, "v": "uno"} + store[3] = {"id": 3, "v": "tres"} + del store[2] + + assert store[1] == {"id": 1, "v": "uno"} + assert sorted(store) == [1, 3] + + +def test_mapping_key_of_ints_on_integer_columns(engine): + """The shape of a known dependent's call: ``del store[{"app_id": 1, "user_id": 7}]`` + on an integer-keyed permission table, with ``value_columns`` naming a subset.""" + with engine.connect() as connection: + connection.execute(text("CREATE TABLE perm (app_id INTEGER, user_id INTEGER)")) + connection.execute(text("INSERT INTO perm VALUES (1, 7), (1, 8), (2, 7)")) + connection.commit() + store = SqlDictStore( + engine, "perm", key_columns="app_id", value_columns=["user_id", "app_id"] + ) + + store[3] = {"app_id": 3, "user_id": 9} + del store[{"app_id": 1, "user_id": 7}] + + with engine.connect() as connection: + rows = connection.execute(text("SELECT app_id, user_id FROM perm")).fetchall() + assert sorted(map(tuple, rows)) == [(1, 8), (2, 7), (3, 9)] + assert store[3] == {"user_id": 9, "app_id": 3} + + +def test_mapping_key_with_an_unknown_column_is_rejected(engine): + store = SqlBaseKvStore(engine, TABLE_NAME, key_columns=KEY_COLUMN) + with pytest.raises(KeyError, match="not a column"): + del store[{"k = 'a' OR 1=1 --": "a"}] + assert _all_rows(engine) == _original_rows() + + +def test_empty_mapping_key_is_rejected(engine): + store = SqlBaseKvStore(engine, TABLE_NAME, key_columns=KEY_COLUMN) + with pytest.raises(ValueError): + del store[{}] + assert _all_rows(engine) == _original_rows() + + +# Legacy raw-SQL paths (sqldol.sql_base) ------------------------------------------ + + +@pytest.mark.parametrize("name", ["t", "_t2", "my_schema.t", "t$1"]) +def test_valid_identifiers_pass(name): + assert validate_sql_identifier(name) == name + + +@pytest.mark.parametrize( + "name", + ["t; DROP TABLE t", "t --", "t'", 't"', "1t", "", "a.b.c", "t t", None, 3], +) +def test_invalid_identifiers_are_rejected(name): + with pytest.raises(ValueError, match="Not a valid SQL table name"): + validate_sql_identifier(name) + + +@pytest.fixture +def dbapi_connection(): + """A raw ``sqlite3`` connection: the kind these legacy paths can still execute on.""" + connection = sqlite3.connect(":memory:") + connection.execute(f"CREATE TABLE {TABLE_NAME} (k TEXT, v TEXT)") + connection.execute(f"INSERT INTO {TABLE_NAME} VALUES ('a', '1'), ('b', '2')") + connection.commit() + yield connection + connection.close() + + +def test_raw_sql_collection_rejects_a_table_name_carrying_sql(dbapi_connection): + with pytest.raises(ValueError): + SqlTableRowsCollection(dbapi_connection, "t; DELETE FROM t; --") + rows = dbapi_connection.execute(f"SELECT k, v FROM {TABLE_NAME}").fetchall() + assert sorted(rows) == _original_rows() + + +def test_raw_sql_paths_still_read_a_plain_table(dbapi_connection): + SqlTableRowsCollection(dbapi_connection, TABLE_NAME) # accepted + # Note: a single bounded batch, because on a DB-API cursor whose SELECT rowcount + # is -1 (sqlite3), iter_rows does not detect the end of the table by itself. + rows = iter_rows(dbapi_connection, TABLE_NAME, batch_size=10, limit=10) + assert sorted(rows) == _original_rows() + + +def test_iter_rows_rejects_non_integer_limits(dbapi_connection): + with pytest.raises(TypeError): + list(iter_rows(dbapi_connection, TABLE_NAME, limit="1; DELETE FROM t")) From 02cbfe29c93b9cf70c40df39c37e7a1bde3a8a0b Mon Sep 17 00:00:00 2001 From: Thor Whalen <1906276+thorwhalen@users.noreply.github.com> Date: Tue, 22 Sep 2026 14:33:19 +0000 Subject: [PATCH 2/2] fix(base,sql_base): address refute review of the injection fix - Unknown column in a mapping key raises ValueError, not KeyError, so a typo isn't read as "key absent"; a case-insensitive unique match resolves, as unquoted SQL names did before. - Tuple/list keys raise a clear TypeError (single key column only). - Identifier allowlist accepts Unicode letters and digit-leading names (e.g. MySQL's 2020_sales), still rejecting all-digit names and anything that can end an identifier. - Drop two vacuous test assertions; add iter_rows table-name test. Co-Authored-By: Claude Opus 5 --- sqldol/base.py | 42 ++++++++++++++++++--------- sqldol/sql_base.py | 11 ++++--- sqldol/tests/test_sql_injection.py | 46 ++++++++++++++++++++++++------ 3 files changed, 74 insertions(+), 25 deletions(-) diff --git a/sqldol/base.py b/sqldol/base.py index 263a6bf..a372e19 100644 --- a/sqldol/base.py +++ b/sqldol/base.py @@ -292,33 +292,49 @@ def __getitem__(self, key): # For example, perhaps values are not dicts, but lists of rows class SqlBaseKvStore(SqlBaseKvReader, MutableMapping): def _key_column(self, column_name): - """The ``Column`` object named ``column_name``, or an informative ``KeyError``. + """The ``Column`` object named ``column_name``, or an informative ``ValueError``. Going through ``self.table.c`` means a column name is never spliced into SQL text: only names the table actually has can be used, and SQLAlchemy quotes - them as the dialect requires. + them as the dialect requires. As with unquoted names in SQL, a name that + matches no column exactly but exactly one column case-insensitively + resolves to that column. + + An unknown column is a ``ValueError``, not a ``KeyError``, so that a typo is + not mistaken for "that key is absent" by callers catching ``KeyError``. """ - try: + if column_name in self.table.c: return self.table.c[column_name] - except KeyError: - msg = ( - f"{column_name!r} is not a column of table {self.table_name!r}. " - f"Its columns are {self._column_names}." - ) - raise KeyError(msg) from None + if isinstance(column_name, str): + matches = [c for c in self.table.c if c.name.lower() == column_name.lower()] + if len(matches) == 1: + return matches[0] + msg = ( + f"{column_name!r} is not a column of table {self.table_name!r}. " + f"Its columns are {self._column_names}." + ) + raise ValueError(msg) def _mk_column_filter(self, key): """The ``WHERE`` clause selecting the rows of ``key``. A ``Mapping`` key is a ``{column_name: value, ...}`` conjunction; any other key - is the value of the key column. Values are always bound parameters, never - interpolated into SQL text, so keys containing quotes, semicolons or ``--`` - are matched literally -- the same way ``__getitem__`` matches them. + is the value of the (single) key column. Values are always bound parameters, + never interpolated into SQL text, so keys containing quotes, semicolons or + ``--`` are matched literally. For non-mapping keys this is the same + comparison ``__getitem__`` makes. """ if isinstance(key, Mapping): if not key: raise ValueError("A mapping key must name at least one column") return and_(*(self._key_column(col) == val for col, val in key.items())) + if isinstance(key, (tuple, list)): + msg = ( + f"Only a single key column ({self.key_columns!r}) is supported, so a " + f"key can't be a {type(key).__name__}. To match several columns, use " + f"a mapping key: {{column_name: value, ...}}." + ) + raise TypeError(msg) return self._key_column(self.key_columns) == key def __setitem__(self, key, value): @@ -343,7 +359,7 @@ def __setitem__(self, key, value): # TODO: Should we return something useful? - def __delitem__(self, key: str) -> None: + def __delitem__(self, key) -> None: filter = self._mk_column_filter(key) query = delete(self.table).where(filter) diff --git a/sqldol/sql_base.py b/sqldol/sql_base.py index 380a1d4..5c79979 100644 --- a/sqldol/sql_base.py +++ b/sqldol/sql_base.py @@ -20,11 +20,14 @@ DFLT_SQL_PORT = 1433 DFLT_SQL_HOST = "localhost" +_SQL_IDENTIFIER_PART = r"(?=[\w$]*[^\W\d])[\w$]+" # word chars or $, not all digits SQL_IDENTIFIER_PATTERN = re.compile( - r"[A-Za-z_][A-Za-z0-9_$]*(\.[A-Za-z_][A-Za-z0-9_$]*)?" + rf"{_SQL_IDENTIFIER_PART}(\.{_SQL_IDENTIFIER_PART})?" ) -"""What a table name may look like where it is written into raw SQL text: a plain -identifier, optionally qualified by a schema (``schema.table``).""" +"""What a table name may look like where it is written into raw SQL text: letters +(Unicode included), digits, ``_`` or ``$``, not all digits (MySQL allows e.g. +``2020_sales``), optionally qualified by a schema (``schema.table``). None of these +characters can end an identifier or start a new SQL token.""" def validate_sql_identifier(name, *, pattern=SQL_IDENTIFIER_PATTERN): @@ -46,7 +49,7 @@ def validate_sql_identifier(name, *, pattern=SQL_IDENTIFIER_PATTERN): if not isinstance(name, str) or not pattern.fullmatch(name): msg = ( f"Not a valid SQL table name: {name!r}. " - f"Expected letters, digits, '_' or '$' (not starting with a digit), " + f"Expected letters, digits, '_' or '$' (not all digits), " f"optionally qualified as 'schema.table'. " f"Names that need quoting are not supported here: use the SQLAlchemy-based " f"stores (sqldol.base, sqldol.stores), which quote identifiers themselves." diff --git a/sqldol/tests/test_sql_injection.py b/sqldol/tests/test_sql_injection.py index 80ec0b7..4061dcc 100644 --- a/sqldol/tests/test_sql_injection.py +++ b/sqldol/tests/test_sql_injection.py @@ -166,7 +166,7 @@ def test_mapping_key_of_ints_on_integer_columns(engine): def test_mapping_key_with_an_unknown_column_is_rejected(engine): store = SqlBaseKvStore(engine, TABLE_NAME, key_columns=KEY_COLUMN) - with pytest.raises(KeyError, match="not a column"): + with pytest.raises(ValueError, match="not a column"): del store[{"k = 'a' OR 1=1 --": "a"}] assert _all_rows(engine) == _original_rows() @@ -178,17 +178,47 @@ def test_empty_mapping_key_is_rejected(engine): assert _all_rows(engine) == _original_rows() +def test_column_name_case_resolves_like_unquoted_sql(engine): + """Unquoted SQL names are case-insensitive, and the old text() filter relied on it.""" + store = SqlBaseKvStore(engine, TABLE_NAME, key_columns=KEY_COLUMN) + del store[{"K": "a"}] + assert _all_rows(engine) == [("b", "2")] + + +@pytest.mark.parametrize("key", [("a", "1"), ["a"]]) +def test_sequence_keys_are_rejected(engine, key): + store = SqlBaseKvStore(engine, TABLE_NAME, key_columns=KEY_COLUMN) + with pytest.raises(TypeError, match="mapping key"): + del store[key] + assert _all_rows(engine) == _original_rows() + + # Legacy raw-SQL paths (sqldol.sql_base) ------------------------------------------ -@pytest.mark.parametrize("name", ["t", "_t2", "my_schema.t", "t$1"]) +@pytest.mark.parametrize( + "name", ["t", "_t2", "my_schema.t", "t$1", "2020_sales", "café", "s.2020_t"] +) def test_valid_identifiers_pass(name): assert validate_sql_identifier(name) == name @pytest.mark.parametrize( "name", - ["t; DROP TABLE t", "t --", "t'", 't"', "1t", "", "a.b.c", "t t", None, 3], + [ + "t; DROP TABLE t", + "t --", + "t'", + 't"', + "t`", + "123", + "", + "a.b.c", + "t t", + "t\n", + None, + 3, + ], ) def test_invalid_identifiers_are_rejected(name): with pytest.raises(ValueError, match="Not a valid SQL table name"): @@ -209,6 +239,11 @@ def dbapi_connection(): def test_raw_sql_collection_rejects_a_table_name_carrying_sql(dbapi_connection): with pytest.raises(ValueError): SqlTableRowsCollection(dbapi_connection, "t; DELETE FROM t; --") + + +def test_iter_rows_rejects_a_table_name_carrying_sql(dbapi_connection): + with pytest.raises(ValueError): + list(iter_rows(dbapi_connection, "t; DELETE FROM t; --", limit=10)) rows = dbapi_connection.execute(f"SELECT k, v FROM {TABLE_NAME}").fetchall() assert sorted(rows) == _original_rows() @@ -219,8 +254,3 @@ def test_raw_sql_paths_still_read_a_plain_table(dbapi_connection): # is -1 (sqlite3), iter_rows does not detect the end of the table by itself. rows = iter_rows(dbapi_connection, TABLE_NAME, batch_size=10, limit=10) assert sorted(rows) == _original_rows() - - -def test_iter_rows_rejects_non_integer_limits(dbapi_connection): - with pytest.raises(TypeError): - list(iter_rows(dbapi_connection, TABLE_NAME, limit="1; DELETE FROM t"))