From 27d16b1c2d18327b615bb5b5a12043a95c7d3493 Mon Sep 17 00:00:00 2001
From: MigraDiff Agent <agent@migradiff.dev>
Date: Sat, 8 Aug 2026 13:50:34 -0700
Subject: [PATCH] fix: wrap raw SQL in text() in test_history.py for SQLAlchemy
 2.x
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Three s.execute() calls passed raw strings directly instead of
wrapping them in sqlalchemy.text(). SQLAlchemy 1.4 (installed locally)
only warns about this; SQLAlchemy 2.x — which CI resolves to since
pyproject.toml pins sqlalchemy to "*" — raises ArgumentError and fails
the test. Verified against SQLAlchemy 2.0.51 in an isolated venv:
these 3 tests plus the other 2 new test files (32 tests total) and
the full 342-test suite all pass.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
---
 tests/test_history.py | 17 +++++++++++------
 1 file changed, 11 insertions(+), 6 deletions(-)

diff --git a/tests/test_history.py b/tests/test_history.py
index e18c5af..1942082 100644
--- a/tests/test_history.py
+++ b/tests/test_history.py
@@ -2,6 +2,7 @@
 
 import io
 
+from sqlalchemy import text
 from sqlbag import S, temporary_database
 
 
@@ -58,16 +59,20 @@ def test_table_creation_idempotent(self):
 
                 # Verify table exists by querying it
                 rows = s.execute(
-                    "SELECT table_name FROM information_schema.tables "
-                    "WHERE table_name = 'migradiff_history'"
+                    text(
+                        "SELECT table_name FROM information_schema.tables "
+                        "WHERE table_name = 'migradiff_history'"
+                    )
                 ).fetchall()
                 assert len(rows) == 1
 
                 # Verify columns
                 cols = s.execute(
-                    "SELECT column_name, data_type FROM information_schema.columns "
-                    "WHERE table_name = 'migradiff_history' "
-                    "ORDER BY ordinal_position"
+                    text(
+                        "SELECT column_name, data_type FROM information_schema.columns "
+                        "WHERE table_name = 'migradiff_history' "
+                        "ORDER BY ordinal_position"
+                    )
                 ).fetchall()
                 col_names = [c[0] for c in cols]
                 assert "id" in col_names
@@ -95,7 +100,7 @@ def test_default_rollback_status(self):
                     forward_sql="SELECT 1;",
                 )
                 row = s.execute(
-                    "SELECT rollback_status FROM migradiff_history"
+                    text("SELECT rollback_status FROM migradiff_history")
                 ).fetchone()
                 assert row[0] == "not_attempted"