test: close the sql-lexing gaps found by mutation testing

Drops the doubled-quote branch in skip_quoted, which masked the same span
either way and so could not be covered, and orders the failure report before
the guidance text.
This commit is contained in:
Yucheng Zhu 2026-08-21 15:37:05 -07:00
parent 6c7dfbd249
commit 777eb8af10
2 changed files with 46 additions and 17 deletions

View file

@ -10,6 +10,7 @@ Flagged, per statement, by its leading keyword:
UPDATE rewrites every matching row, and `WHERE` does not bound the scan UPDATE rewrites every matching row, and `WHERE` does not bound the scan
DELETE same scan, and the dead tuples outlive the migration DELETE same scan, and the dead tuples outlive the migration
MERGE both of the above in one statement
INSERT only when it draws rows from a `SELECT`; `INSERT ... VALUES` is bounded INSERT only when it draws rows from a `SELECT`; `INSERT ... VALUES` is bounded
by the literal row list and passes by the literal row list and passes
WITH a CTE-led statement containing any of the above WITH a CTE-led statement containing any of the above
@ -176,15 +177,10 @@ def skip_block_comment(sql: str, start: int) -> int:
def skip_quoted(sql: str, start: int, quote: str) -> int: def skip_quoted(sql: str, start: int, quote: str) -> int:
index = start + 1 """One quoted run, up to and including its closing quote. A doubled quote needs no
while index < len(sql): special case: closing on the first and reopening on the second masks the same span."""
if sql[index] != quote: stop = sql.find(quote, start + 1)
index += 1 return len(sql) if stop == -1 else stop + 1
elif sql[index + 1 : index + 2] == quote:
index += 2
else:
return index + 1
return len(sql)
def strip_parens(statement: str) -> str: def strip_parens(statement: str) -> str:
@ -304,8 +300,8 @@ def main() -> int:
print(f"{name}: listed in GRANDFATHERED but no longer violates; remove it from the set") print(f"{name}: listed in GRANDFATHERED but no longer violates; remove it from the set")
if violations: if violations:
print(GUIDANCE, file=sys.stderr) print(f"\n{len(violations)} data-rewriting statement(s) in migrations.")
print(f"{len(violations)} data-rewriting statement(s) in migrations.", file=sys.stderr) print(GUIDANCE)
if violations or stale: if violations or stale:
return 1 return 1

View file

@ -132,10 +132,36 @@ class TestDollarQuotedBlocks:
) )
assert _keywords(tmp_path, sql) == () assert _keywords(tmp_path, sql) == ()
def test_guarded_update_is_flagged(self, tmp_path):
sql = (
"DO $$\n"
"BEGIN\n"
' IF EXISTS (SELECT 1 FROM "Foo") THEN\n'
' UPDATE "Foo" SET "a" = 1;\n'
" END IF;\n"
"END $$;"
)
assert _keywords(tmp_path, sql) == ("UPDATE",)
def test_guard_with_a_nested_call_still_flags_the_update(self, tmp_path):
sql = (
"DO $$\n"
"BEGIN\n"
' IF EXISTS (SELECT 1 FROM "Foo" WHERE lower("a") = \'x\' UNION SELECT 1) THEN\n'
' UPDATE "Foo" SET "a" = 1;\n'
" END IF;\n"
"END $$;"
)
assert _keywords(tmp_path, sql) == ("UPDATE",)
def test_tagged_dollar_quote_is_scanned(self, tmp_path): def test_tagged_dollar_quote_is_scanned(self, tmp_path):
sql = 'DO $body$\nBEGIN\n DELETE FROM "Foo";\nEND $body$;' sql = 'DO $body$\nBEGIN\n DELETE FROM "Foo";\nEND $body$;'
assert _keywords(tmp_path, sql) == ("DELETE",) assert _keywords(tmp_path, sql) == ("DELETE",)
def test_tagged_dollar_quote_holds_an_apostrophe(self, tmp_path):
sql = 'INSERT INTO "Foo" ("t") VALUES ($body$don\'t$body$);\nUPDATE "Bar" SET "b" = 1;'
assert _keywords(tmp_path, sql) == ("UPDATE",)
def test_semicolons_inside_do_block_do_not_split_outer_statements(self, tmp_path): def test_semicolons_inside_do_block_do_not_split_outer_statements(self, tmp_path):
sql = 'DO $$ BEGIN PERFORM 1; END $$;\nALTER TABLE "Foo" ADD COLUMN "b" TEXT;' sql = 'DO $$ BEGIN PERFORM 1; END $$;\nALTER TABLE "Foo" ADD COLUMN "b" TEXT;'
assert _keywords(tmp_path, sql) == () assert _keywords(tmp_path, sql) == ()
@ -143,7 +169,7 @@ class TestDollarQuotedBlocks:
class TestQuotingAndComments: class TestQuotingAndComments:
def test_update_inside_string_literal_passes(self, tmp_path): def test_update_inside_string_literal_passes(self, tmp_path):
sql = "ALTER TABLE \"Foo\" ADD COLUMN \"note\" TEXT NOT NULL DEFAULT 'UPDATE nothing';" sql = 'ALTER TABLE "Foo" ADD COLUMN "note" TEXT NOT NULL DEFAULT \'UPDATE nothing\';'
assert _keywords(tmp_path, sql) == () assert _keywords(tmp_path, sql) == ()
def test_escaped_quote_inside_string_does_not_leak(self, tmp_path): def test_escaped_quote_inside_string_does_not_leak(self, tmp_path):
@ -160,9 +186,20 @@ class TestQuotingAndComments:
sql = '/* outer /* UPDATE "Foo" SET "a" = 1; */ still comment */\nDROP TABLE "Bar";' sql = '/* outer /* UPDATE "Foo" SET "a" = 1; */ still comment */\nDROP TABLE "Bar";'
assert _keywords(tmp_path, sql) == () assert _keywords(tmp_path, sql) == ()
def test_nested_block_comment_masks_past_the_inner_close(self, tmp_path):
sql = '/* outer /* inner */ UPDATE "Foo" SET "a" = 1; */\nDROP TABLE "Bar";'
assert _keywords(tmp_path, sql) == ()
def test_update_inside_quoted_identifier_passes(self, tmp_path): def test_update_inside_quoted_identifier_passes(self, tmp_path):
assert _keywords(tmp_path, 'ALTER TABLE "UPDATE Foo" ADD COLUMN "b" TEXT;') == () assert _keywords(tmp_path, 'ALTER TABLE "UPDATE Foo" ADD COLUMN "b" TEXT;') == ()
def test_select_in_a_quoted_identifier_does_not_make_an_insert_a_rewrite(self, tmp_path):
assert _keywords(tmp_path, 'INSERT INTO "SELECT Foo" ("id") VALUES (\'a\');') == ()
def test_update_in_a_quoted_identifier_does_not_make_a_cte_a_rewrite(self, tmp_path):
sql = 'WITH batch AS (SELECT "id" FROM "UPDATE Foo") SELECT count(*) FROM batch;'
assert _keywords(tmp_path, sql) == ()
def test_positional_parameter_is_not_a_dollar_quote(self, tmp_path): def test_positional_parameter_is_not_a_dollar_quote(self, tmp_path):
sql = 'ALTER TABLE "Foo" ADD COLUMN "b" TEXT;\nUPDATE "Foo" SET "b" = $1;' sql = 'ALTER TABLE "Foo" ADD COLUMN "b" TEXT;\nUPDATE "Foo" SET "b" = $1;'
assert _keywords(tmp_path, sql) == ("UPDATE",) assert _keywords(tmp_path, sql) == ("UPDATE",)
@ -177,11 +214,7 @@ class TestEscapeHatch:
assert _keywords(tmp_path, '-- data-migration-ok:\nUPDATE "Foo" SET "a" = 1;') == ("UPDATE",) assert _keywords(tmp_path, '-- data-migration-ok:\nUPDATE "Foo" SET "a" = 1;') == ("UPDATE",)
def test_marker_exempts_only_its_own_statement(self, tmp_path): def test_marker_exempts_only_its_own_statement(self, tmp_path):
sql = ( sql = '-- data-migration-ok: bounded to in-flight jobs\nUPDATE "Foo" SET "a" = 1;\nUPDATE "Bar" SET "b" = 2;\n'
"-- data-migration-ok: bounded to in-flight jobs\n"
'UPDATE "Foo" SET "a" = 1;\n'
'UPDATE "Bar" SET "b" = 2;\n'
)
assert _keywords(tmp_path, sql) == ("UPDATE",) assert _keywords(tmp_path, sql) == ("UPDATE",)
assert _scan(tmp_path, sql)[0].line == 3 assert _scan(tmp_path, sql)[0].line == 3