"""Rewriting :N placeholders without rewriting the SQL around them. We advertise ``paramstyle="numeric"`` to match Informix's ESQL/C convention, and the wire protocol takes ``?``, so placeholders get rewritten on the way out. That was ``re.sub(r":(\\d+)", "?", sql)``, which cannot see a string literal and therefore rewrote the inside of one:: UPDATE jobs SET url = 'http://host:8080/x' WHERE id = ? stored as 'http://host?/x' Any ``HH:MM`` time, any URL with a port, any aspect ratio, any ``key:value`` string. It wrote wrong data and said nothing. Nothing about it was opt-in either. The rewrite runs whenever a statement has parameters, whatever placeholder style the caller actually used, so writing ``?`` everywhere and never touching numeric style did not protect you. And it changed the placeholder *count* while ``num_qmarks`` was still computed from ``len(params)``, leaving driver and server disagreeing about how many binds exist. The lexical rules below are Informix's own, measured against 12.10, 14.10 and 15 rather than assumed from standard SQL. The one that matters most is the backslash: ``'a\\'b'`` is an *unterminated string* to Informix (``-282``), not an escaped quote. A scanner written to Postgres habits would desync on it and corrupt everything after. """ from __future__ import annotations import contextlib import pytest import informix_db from informix_db.cursors import _rewrite_numeric_to_qmark as rewrite from tests.conftest import ConnParams # --------------------------------------------------------------------------- # Substitution happens where it should # --------------------------------------------------------------------------- @pytest.mark.parametrize( ("sql", "expected"), [ ("SELECT * FROM t WHERE a = :1", "SELECT * FROM t WHERE a = ?"), ( "SELECT * FROM t WHERE a = :1 AND b = :2", "SELECT * FROM t WHERE a = ? AND b = ?", ), ("SELECT * FROM t WHERE a = :10", "SELECT * FROM t WHERE a = ?"), # Already-? SQL is returned untouched, and cheaply. ("SELECT * FROM t WHERE a = ?", "SELECT * FROM t WHERE a = ?"), ("", ""), ], ) def test_placeholders_are_rewritten(sql: str, expected: str) -> None: assert rewrite(sql) == expected # --------------------------------------------------------------------------- # ...and nowhere else # --------------------------------------------------------------------------- @pytest.mark.parametrize( ("label", "sql", "expected"), [ ( "url-with-port", "UPDATE jobs SET url = 'http://host:8080/x' WHERE id = :1", "UPDATE jobs SET url = 'http://host:8080/x' WHERE id = ?", ), ( "time-of-day", "INSERT INTO log VALUES ('started at 09:15:30', :1)", "INSERT INTO log VALUES ('started at 09:15:30', ?)", ), ( "aspect-ratio", "UPDATE t SET ratio = '16:9' WHERE id = :1", "UPDATE t SET ratio = '16:9' WHERE id = ?", ), ( "doubled-quote-escape", "SELECT 'it''s 10:30' FROM t WHERE a = :1", "SELECT 'it''s 10:30' FROM t WHERE a = ?", ), ( "delimited-identifier", 'SELECT "col:1" FROM t WHERE a = :1', 'SELECT "col:1" FROM t WHERE a = ?', ), ( "line-comment", "-- ticket :99\nSELECT * FROM t WHERE a = :1", "-- ticket :99\nSELECT * FROM t WHERE a = ?", ), ( "block-comment", "SELECT /* not :99 */ * FROM t WHERE a = :1", "SELECT /* not :99 */ * FROM t WHERE a = ?", ), ( "brace-comment", "SELECT { not :99 } * FROM t WHERE a = :1", "SELECT { not :99 } * FROM t WHERE a = ?", ), ( "cast-operator", "SELECT a::INT FROM t WHERE a = :1", "SELECT a::INT FROM t WHERE a = ?", ), ( "colon-not-a-placeholder", "SELECT a FROM t WHERE b = ':x' AND c = :1", "SELECT a FROM t WHERE b = ':x' AND c = ?", ), ( "literal-after-placeholder", "SELECT * FROM t WHERE a = :1 AND b = '10:30'", "SELECT * FROM t WHERE a = ? AND b = '10:30'", ), ], ) def test_quotes_and_comments_are_left_alone( label: str, sql: str, expected: str ) -> None: assert rewrite(sql) == expected, label def test_backslash_does_not_escape_a_quote() -> None: """Informix answers -282 for ``'a\\'b'``: the backslash is an ordinary character and the string is unterminated. A scanner that treated it as an escape would think it was still inside the literal and stop substituting, or worse, resume in the wrong place.""" sql = "SELECT 'a\\' FROM t WHERE x = :1" # The quote closes at the character after the backslash, so :1 is # outside the literal and gets substituted. assert rewrite(sql) == "SELECT 'a\\' FROM t WHERE x = ?" def test_unterminated_quote_substitutes_nothing_further() -> None: """Under-substituting leaves the server to reject SQL that was already malformed. Guessing would corrupt a literal.""" assert rewrite("SELECT 'oops :1 FROM t") == "SELECT 'oops :1 FROM t" def test_unterminated_block_comment_substitutes_nothing_further() -> None: assert rewrite("SELECT /* oops :1 FROM t") == "SELECT /* oops :1 FROM t" def test_block_comments_do_not_nest() -> None: """Measured: ``/* a /* b */ c */`` is a syntax error on all three servers, so the first ``*/`` closes the comment. Treating them as nesting would swallow live SQL.""" assert ( rewrite("SELECT /* a /* b */ :1 FROM t") == "SELECT /* a /* b */ ? FROM t" ) # --------------------------------------------------------------------------- # Against a real server # --------------------------------------------------------------------------- def _connect(conn_params: ConnParams) -> informix_db.Connection: return informix_db.connect( host=conn_params.host, port=conn_params.port, user=conn_params.user, password=conn_params.password, database=conn_params.database, server=conn_params.server, connect_timeout=10.0, read_timeout=25.0, autocommit=True, ) @pytest.mark.integration def test_colon_literals_round_trip(conn_params: ConnParams) -> None: """The bug as a user meets it: the value stored is not the value written. Both placeholder styles, because the rewrite ran for both.""" with _connect(conn_params) as conn: cur = conn.cursor() cur.execute("CREATE TEMP TABLE t_rw (id INT, s VARCHAR(60))") cur.execute( "INSERT INTO t_rw VALUES (?, 'http://host:8080/path')", (1,) ) cur.execute("INSERT INTO t_rw VALUES (:1, 'at 09:15:30')", (2,)) cur.execute("SELECT id, s FROM t_rw ORDER BY id") assert cur.fetchall() == [ (1, "http://host:8080/path"), (2, "at 09:15:30"), ] @pytest.mark.integration def test_colon_literals_round_trip_through_executemany( conn_params: ConnParams, ) -> None: """executemany rewrites unconditionally, so it had the bug too.""" with _connect(conn_params) as conn: cur = conn.cursor() cur.execute("CREATE TEMP TABLE t_rw2 (id INT, s VARCHAR(40))") cur.executemany( "INSERT INTO t_rw2 VALUES (?, '16:9')", [(1,), (2,)] ) cur.execute("SELECT DISTINCT s FROM t_rw2") assert cur.fetchall() == [("16:9",)] @pytest.mark.integration def test_comment_bearing_sql_still_binds(conn_params: ConnParams) -> None: """A leading comment reaches PREPARE now that classification asks the server. Make sure the rewriter agrees and doesn't eat a placeholder that follows one.""" with _connect(conn_params) as conn: cur = conn.cursor() cur.execute( "/* report: daily */ SELECT FIRST 1 tabid FROM systables " "WHERE tabid > :1", (0,), ) assert cur.fetchone() is not None @pytest.mark.integration def test_placeholder_count_matches_after_rewrite( conn_params: ConnParams, ) -> None: """The old regex could add placeholders the driver never counted, leaving num_qmarks (from len(params)) disagreeing with the SQL. A literal containing three colon-digit sequences is the shape that used to break it.""" with _connect(conn_params) as conn: cur = conn.cursor() cur.execute("CREATE TEMP TABLE t_rw3 (id INT, s VARCHAR(40))") with contextlib.suppress(Exception): cur.execute("DELETE FROM t_rw3") cur.execute( "INSERT INTO t_rw3 VALUES (:1, 'a:1 b:2 c:3')", (7,) ) cur.execute("SELECT id, s FROM t_rw3") assert cur.fetchone() == (7, "a:1 b:2 c:3")