test: twins turn a failed query into query_error(), never their own error text (#1339)

test/pytest/TESTS.md

@@ -2282,6 +2282,8 @@ parameter is `cur` is excluded for the same reason `wrote` is.
 | `test_a_hardcoded_sentinel_survives_the_same_rewrite` | a sentinel no producer minted — the corpus writes three, one from inside SQL — is still refused after a rewrite | a matcher reading a rewritable global would stop seeing them |
 | `test_the_ordering_premise_refuses_a_failed_reading` | `ordering_observable` refuses a failed reading on either side | with the old shared constant two failed readings were identical and it went RED; unique sentinels differ, so it passed and greenlit every ordered assertion resting on it |
 | `test_a_legitimate_comparison_is_untouched` | equal text, rows, sets and numbers still pass | a refusal that also refuses real data is not a refusal |
+| `test_the_minted_failure_scan_sees_the_shape` | the scan finds an except handler returning `f'error: {exc}'`, and not `query_error()`, a SQLSTATE, or the shape inside a string literal | a scan that sees nothing would make the corpus arm below vacuous |
+| `test_no_test_mints_its_own_failure_text` | no except handler under test/pytest builds an "error..." string from the error's message (#1525 review) | a helper that mints its own failure text is not the producer's prefix, so no refusal fires and two failures with the same message compare equal; on main it finds 32 sites in 22 files |
 
 **Each refusal is proved load-bearing.** Removing the one call from each assertion,
 one at a time, with `__pycache__` cleared between runs:

test/pytest/expected_tests.txt

@@ -292,7 +292,10 @@
 # in test_compare_to_bash.py: one test, by collection over NO_CLUSTER.
 # guard_tests 439 -> 440 for #1339's test_a_premise_in_name_only_is_still_caught in
 # test_loop_coverage_premise.py: one test, by collection over NO_CLUSTER.
-guard_tests 440
+# guard_tests 440 -> 442 for #1339's test_the_minted_failure_scan_sees_the_shape and
+# test_no_test_mints_its_own_failure_text in test_failed_query_sentinel.py: two tests,
+# by collection over NO_CLUSTER.
+guard_tests 442
 
 # The complement: tests that need the driver and a throwaway cluster. Until #1016 these ran
 # in no CI job at all -- a quarter of the corpus, green when somebody ran them by hand and

test/pytest/test_advisory_lock_class.py

@@ -30,6 +30,8 @@
 """
 import psycopg
 
+from pgc_vacuity import query_error
+
 # The two field4 values an application can produce, from lockfuncs.c.
 SQL_REACHABLE = (1, 2)
 PROBE_KEY = 900001
@@ -213,7 +215,7 @@ def test_a_user_advisory_lock_on_that_tag_does_not_block_an_insert(
         except psycopg.errors.QueryCanceled:
             verdict = "BLOCKED by the user lock"
         except psycopg.Error as exc:
-            verdict = f"ERROR: {str(exc).splitlines()[0]}"
+            verdict = query_error(str(exc).splitlines()[0])
         finally:
             with pgc_conn.cursor() as cur:
                 cur.execute("RESET statement_timeout")
@@ -366,7 +368,7 @@ def test_the_row_identity_lock_cannot_be_reached_from_sql(pgc_cluster, pgc_conn,
         except psycopg.errors.QueryCanceled:
             verdict = "BLOCKED by the user lock"
         except psycopg.Error as exc:
-            verdict = f"ERROR: {str(exc).splitlines()[0]}"
+            verdict = query_error(str(exc).splitlines()[0])
         finally:
             with pgc_conn.cursor() as cur:
                 cur.execute("RESET statement_timeout")
@@ -407,7 +409,7 @@ def write_second():
                     sc.execute("INSERT INTO bw VALUES (2)")
                 result["verdict"] = "ok"
             except psycopg.Error as exc:
-                result["verdict"] = f"ERROR: {str(exc).splitlines()[0]}"
+                result["verdict"] = query_error(str(exc).splitlines()[0])
 
         import threading
         import time

test/pytest/test_failed_query_sentinel.py

@@ -23,7 +23,9 @@
 accepts one at all; it is NOT safe against a helper that compares by hand, which is
 why the producer exists and why new code should use it.
 """
+import ast
 import inspect
+import pathlib
 
 import pgc_vacuity
 from pgc_vacuity import QUERY_ERROR, VacuityError, query_error
@@ -398,3 +400,65 @@ def test_a_legitimate_comparison_is_untouched(expect):
     e.row_set([(1,), (2,)], [(2,), (1,)], "a set comparison still ignores order")
     e.num(7, 7, "equal numbers still pass")
     expect.num(1, 1, "four honest comparisons passed through the new refusal")
+
+
+# A HELPER THAT MINTS ITS OWN FAILURE TEXT REOPENS THE MODE (#1525 review). Seventeen
+# twins caught a failed query and returned f"error: {message}": not the producer's
+# prefix, so no refusal fires, and two failures with the same message compare equal.
+# Measured on test_native_skip.py's copy: with the backend aborting on one LIKE, the
+# check on that query PASSED, as did two more read on the dead connection afterwards,
+# and the run named no red at all.
+#
+# Parsed, not grepped, for the reason the layer's other scans give: a string literal
+# that merely contains the shape is not an except handler.
+def _minted_failure_text(source):
+    """-> line numbers where an except handler builds a string starting "error" from
+    the error's message. A SQLSTATE alone is a typed field, which a test compares
+    with the code it expects (test_rls_query_paths.py's "ERROR: 0A000"), so an
+    f-string whose every interpolation is a `.sqlstate` is not this shape."""
+    def starts_error(node):
+        if isinstance(node, ast.Constant) and isinstance(node.value, str):
+            return node.value.lower().startswith("error")
+        if isinstance(node, ast.JoinedStr) and node.values:
+            return starts_error(node.values[0]) and not all(
+                isinstance(v.value, ast.Attribute) and v.value.attr == "sqlstate"
+                for v in node.values if isinstance(v, ast.FormattedValue))
+        return False
+    tree = ast.parse(source)
+    # an f-string's literal parts are judged with the f-string, not alone
+    parts = {id(v) for j in ast.walk(tree) if isinstance(j, ast.JoinedStr) for v in j.values}
+    hits = []
+    for h in ast.walk(tree):
+        if isinstance(h, ast.ExceptHandler):
+            hits += [n.lineno for stmt in h.body for n in ast.walk(stmt)
+                     if id(n) not in parts and starts_error(n)]
+    return sorted(set(hits))
+
+
+def test_the_minted_failure_scan_sees_the_shape(expect):
+    """The scan's own control: it finds the shape, and not the producer's form."""
+    bad = ("def f(c):\n    try:\n        return c()\n"
+           "    except Exception as exc:\n        return f'error: {exc}'\n")
+    good = bad.replace("f'error: {exc}'", "query_error(str(exc))")
+    code = bad.replace("f'error: {exc}'", "f'error {exc.sqlstate}'")
+    in_a_string = "SRC = " + repr(bad) + "\n"
+    expect.text(" ".join(str(_minted_failure_text(x)) for x in (bad, good, code, in_a_string)),
+                "[5] [] [] []", "the scan finds a minted failure string, and not "
+                "query_error(), a SQLSTATE, or the shape inside a literal")
+
+
+def test_no_test_mints_its_own_failure_text(expect):
+    """A failed query becomes a value through query_error(), so it is refused, never
+    compared. The corpus is every test file and helper under test/pytest/.
+
+    The scan keys on the spelling the tree used, a handler value starting "error"
+    (any case, f-string, `+`, `.format` or `%`). A handler that builds other text from
+    the exception (`f"failed: {exc}"`, a bare `str(exc)`) is not seen: most such
+    handlers are deliberate refusal capture, compared with a fixed message or
+    SQLSTATE, and the mode is only live when that text reaches a two-sided
+    comparison (#1527 review)."""
+    here = pathlib.Path(__file__).resolve().parent
+    hits = [f"{p.name}:{n}" for p in sorted(here.glob("*.py"))
+            for n in _minted_failure_text(p.read_text())]
+    expect.text(" ".join(hits) or "none", "none",
+                "no except handler turns a failed query into its own error text")

test/pytest/test_latest_reading.py

@@ -25,6 +25,8 @@
 import psycopg
 import pytest
 
+from pgc_vacuity import query_error
+
 DOC = pathlib.Path(__file__).resolve().parent.parent.parent / "docs" / "how-to.md"
 SECTION = "## Find the latest reading per device"
 DEVICES = 25
@@ -74,7 +76,7 @@ def _try(fn, *args):
     try:
         return fn(*args)
     except psycopg.Error as exc:
-        return f"error: {exc.diag.message_primary}"
+        return query_error(exc.diag.message_primary)
 
 
 @pytest.fixture(scope="module")

test/pytest/test_maintenance_crash.py

@@ -21,6 +21,8 @@
 
 import psycopg
 
+from pgc_vacuity import query_error
+
 ROWS = ("SELECT count(*) || '/' || md5(string_agg(id || ':' || v || ':' || w, ',' ORDER BY id)) FROM mc")
 GROUPS = ("SELECT count(*) || '/' || coalesce(sum(row_count), 0) FROM pgcolumnar.row_group "
           "WHERE storage_id = pgcolumnar.get_storage_id('mc')")
@@ -33,7 +35,7 @@ def _one(dsn, sql, setup=()):
                 c.execute(s)
             return c.execute(sql).fetchone()[0]
     except psycopg.Error as exc:
-        return f"ERROR: {str(exc).splitlines()[0]}"
+        return query_error(str(exc).splitlines()[0])
 
 
 BY_INDEX = ("SET enable_seqscan = off", "SET enable_bitmapscan = off", "SET pgcolumnar.enable_custom_scan = off")

test/pytest/test_maintenance_rollback.py

@@ -16,6 +16,8 @@
 """
 import psycopg
 
+from pgc_vacuity import query_error
+
 ROWS = ("SELECT count(*) || '/' || md5(string_agg(id || ':' || v || ':' || w, ',' ORDER BY id)) "
         "FROM mr")
 GROUPS = ("SELECT count(*) || '/' || coalesce(sum(row_count), 0) FROM pgcolumnar.row_group "
@@ -29,7 +31,7 @@ def _one(conn, sql):
             cur.execute(sql)
             return cur.fetchone()[0]
     except psycopg.Error as exc:
-        return f"ERROR: {str(exc).splitlines()[0]}"
+        return query_error(str(exc).splitlines()[0])
 
 
 BY_INDEX_SETS = ("enable_seqscan", "enable_bitmapscan", "pgcolumnar.enable_custom_scan")

test/pytest/test_native_agg.py

@@ -29,6 +29,8 @@
 import psycopg
 import pytest
 
+from pgc_vacuity import query_error
+
 ROWS = 300000
 
 
@@ -48,7 +50,7 @@ def _try(fn, *args, **kw):
     try:
         return fn(*args, **kw)
     except psycopg.Error as exc:
-        return f"error: {exc.diag.message_primary}"
+        return query_error(exc.diag.message_primary)
 
 
 def _blocks(conn, table, sql):

test/pytest/test_native_bloom.py

@@ -11,6 +11,8 @@
 import psycopg
 import pytest
 
+from pgc_vacuity import query_error
+
 HASH = ("SELECT coalesce(md5(string_agg(t, chr(10) ORDER BY t)), $e$EMPTY$e$) "
         "FROM (SELECT _row::text AS t FROM ({}) _row) _s")
 # a well-spread tag over a large domain: each 2048-row group's [min,max] spans almost
@@ -23,7 +25,7 @@ def _one(c, sql):
     try:
         return str(c.execute(sql).fetchone()[0])
     except psycopg.Error as exc:
-        return f"error: {exc.diag.message_primary}"
+        return query_error(exc.diag.message_primary)
 
 
 def _skipped(c, sql, bloom):

test/pytest/test_native_fetch_bigcap.py

@@ -12,6 +12,8 @@
 import psycopg
 import pytest
 
+from pgc_vacuity import query_error
+
 CAP = 32 * 1024 * 1024
 FORCE_INDEX = ("max_parallel_workers_per_gather=0", "enable_seqscan=off", "enable_bitmapscan=off",
                "pgcolumnar.enable_index_fetch_penalty=off", "pgcolumnar.enable_custom_scan=off")
@@ -34,7 +36,7 @@ def _one(c, sql):
     try:
         return c.execute(sql).fetchone()[0]
     except psycopg.Error as exc:
-        return f"error: {exc.diag.message_primary}"
+        return query_error(exc.diag.message_primary)
 
 
 def _index_scan(c, sql):

test/pytest/test_native_fetch_position.py

@@ -18,6 +18,8 @@
 import psycopg
 import pytest
 
+from pgc_vacuity import query_error
+
 # the shell reads PGC_FETCHPOS_ROWS, and so does this
 ROWS = int(os.environ.get("PGC_FETCHPOS_ROWS", "120000"))
 SRC = pathlib.Path(__file__).resolve().parents[2] / "src"
@@ -50,7 +52,7 @@ def _rows_or_error(c, sql):
     try:
         return _one(c, sql)
     except psycopg.Error as exc:
-        return f"error: {exc.diag.message_primary}"
+        return query_error(exc.diag.message_primary)
 
 
 def _median_ms(c, sql):

test/pytest/test_native_gap.py

@@ -12,6 +12,8 @@
 import psycopg
 import pytest
 
+from pgc_vacuity import query_error
+
 HASH = ("SELECT coalesce(md5(string_agg(t, chr(10) ORDER BY t)), $e$EMPTY$e$) "
         "FROM (SELECT _row::text AS t FROM ({}) _row) _s")
 IDX_SETS = ("SET enable_seqscan = off", "SET enable_bitmapscan = off", "SET pgcolumnar.enable_custom_scan = off")
@@ -22,7 +24,7 @@ def _one(c, sql):
     try:
         return str(c.execute(sql).fetchone()[0])
     except psycopg.Error as exc:
-        return f"error: {exc.diag.message_primary}"
+        return query_error(exc.diag.message_primary)
 
 
 def _size(c):

test/pytest/test_native_groupagg.py

@@ -45,6 +45,8 @@
 import json
 import psycopg
 
+from pgc_vacuity import query_error
+
 ROWS = 20000
 MARKER = "Columnar Vectorized Group Keys"
 
@@ -801,7 +803,7 @@ def _groups_or_error(conn, cap, sql):
     try:
         return str(len(_rows(conn, sql)))
     except psycopg.Error as exc:
-        return f"ERROR: {exc.diag.message_primary}"
+        return query_error(exc.diag.message_primary)
     finally:
         _gate_reset(conn)
 

test/pytest/test_native_groupagg_batch.py

@@ -15,6 +15,8 @@
 import psycopg
 import pytest
 
+from pgc_vacuity import query_error
+
 ANALYZE = "EXPLAIN (ANALYZE, TIMING OFF, COSTS OFF, SUMMARY OFF)"
 PLAIN = "EXPLAIN (COSTS OFF)"
 MARKER = "Columnar Vectorized Group Keys"
@@ -48,7 +50,7 @@ def _rows(c, sql):
     try:
         return sorted(tuple("\0" if v is None else str(v) for v in r) for r in c.execute(sql))
     except psycopg.Error as exc:
-        return [(f"error: {exc.diag.message_primary}",)]
+        return [(query_error(exc.diag.message_primary),)]
 
 
 def _one(c, sql):
@@ -60,7 +62,7 @@ def _text(c, sql):
     try:
         return "\n".join(str(r[0]) for r in c.execute(sql))
     except psycopg.Error as exc:
-        return f"error: {exc.diag.message_primary}"
+        return query_error(exc.diag.message_primary)
 
 
 def _groupvec(c, sql):

test/pytest/test_native_index.py

@@ -10,6 +10,8 @@
 """
 import psycopg
 
+from pgc_vacuity import query_error
+
 FORCE_INDEX = ("SET enable_seqscan=off", "SET enable_bitmapscan=off", "SET enable_indexonlyscan=off",
                "SET pgcolumnar.enable_custom_scan=off")
 
@@ -27,7 +29,7 @@ def _forced(cur, sql):
             cur.execute(s)
         return _one(cur, sql)
     except psycopg.Error as exc:
-        return f"error: {exc.diag.message_primary}"
+        return query_error(exc.diag.message_primary)
     finally:
         for g in ("enable_seqscan", "enable_bitmapscan", "enable_indexonlyscan", "pgcolumnar.enable_custom_scan"):
             cur.execute(f"RESET {g}")

test/pytest/test_native_ios.py

@@ -14,6 +14,8 @@
 import psycopg
 import pytest
 
+from pgc_vacuity import query_error
+
 HASH = ("SELECT coalesce(md5(string_agg(t, chr(10) ORDER BY t)), $e$EMPTY$e$) "
         "FROM (SELECT _row::text AS t FROM ({}) _row) _s")
 RANGE = "SELECT id FROM {} WHERE id BETWEEN 1000 AND 5000"
@@ -23,7 +25,7 @@ def _one(c, sql):
     try:
         return str(c.execute(sql).fetchone()[0])
     except psycopg.Error as exc:
-        return f"error: {exc.diag.message_primary}"
+        return query_error(exc.diag.message_primary)
 
 
 def _hash(c, sql):

test/pytest/test_native_parquet_flba.py

@@ -12,6 +12,8 @@
 
 import psycopg
 
+from pgc_vacuity import query_error
+
 HASH = ("SELECT coalesce(md5(string_agg(t, chr(10) ORDER BY t)), $e$EMPTY$e$) "
         "FROM (SELECT _row::text AS t FROM ({}) _row) _s")
 SHAPE = "t(id int, u uuid, d numeric(12,4), dn numeric(12,4), fb bytea)"
@@ -23,15 +25,15 @@ def _hash(cur, sql):
         cur.execute(HASH.format(sql))
         return cur.fetchone()[0]
     except psycopg.Error as exc:
-        return f"error: {exc.diag.message_primary} [{sql}]"
+        return query_error(f"{exc.diag.message_primary} [{sql}]")
 
 
 def _one(cur, sql):
     try:
         cur.execute(sql)
         return str(cur.fetchone()[0])
     except psycopg.Error as exc:
-        return f"error: {exc.diag.message_primary}"
+        return query_error(exc.diag.message_primary)
 
 
 def _patch_scale(path):