add spaces, leading underscore to oracle checks

Expand the test suite from #8708 which unfortunately did
not exercise the bound parameter codepaths completely.

Continued fixes for Oracle fix :ticket:`8708` released in 1.4.43 where
bound parameter names that start with underscores, which are disallowed by
Oracle, were still not being properly escaped in all circumstances.

Fixes: #8708
Change-Id: Ic389c09bd7c53b773e5de35f1a18ef20769b92a7
(cherry picked from commit 2886412438de072b4925818ac746e56a2067bee3)

diff --git a/doc/build/changelog/unreleased_14/8708.rst b/doc/build/changelog/unreleased_14/8708.rst
new file mode 100644 (file)
index 0000000..61dcbf6
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/8708.rst
@@ -0,0 +1,9 @@
+.. change::
+    :tags: bug, oracle
+    :tickets: 8708
+    :versions: 2.0.0b4
+
+    Continued fixes for Oracle fix :ticket:`8708` released in 1.4.43 where
+    bound parameter names that start with underscores, which are disallowed by
+    Oracle, were still not being properly escaped in all circumstances.
+

diff --git a/lib/sqlalchemy/dialects/oracle/cx_oracle.py b/lib/sqlalchemy/dialects/oracle/cx_oracle.py
index fe18d1310b02f0888a754fa1cfd44a43878e3cbb..acdf4ded28abff12ac838531ec63e92749da6f8b 100644 (file)
--- a/lib/sqlalchemy/dialects/oracle/cx_oracle.py
+++ b/lib/sqlalchemy/dialects/oracle/cx_oracle.py
@@ -472,13 +472,13 @@ from ...sql import expression
 from ...util import compat
 
 
-_ORACLE_BIND_TRANSLATE_RE = re.compile(r"[%\(\):\[\]\.\/\?]")
+_ORACLE_BIND_TRANSLATE_RE = re.compile(r"[%\(\):\[\]\.\/\? ]")
 
 # Oracle bind names can't start with digits or underscores.
 # currently we rely upon Oracle-specific quoting of bind names in most cases.
 # however for expanding params, the escape chars are used.
 # see #8708
-_ORACLE_BIND_TRANSLATE_CHARS = dict(zip("%():[]./?", "PAZCCCCCCC"))
+_ORACLE_BIND_TRANSLATE_CHARS = dict(zip("%():[]./? ", "PAZCCCCCCCC"))
 
 
 class _OracleInteger(sqltypes.Integer):
@@ -740,11 +740,11 @@ class OracleCompiler_cx_oracle(OracleCompiler):
                     lambda m: _ORACLE_BIND_TRANSLATE_CHARS[m.group(0)],
                     name,
                 )
-                if new_name[0].isdigit():
+                if new_name[0].isdigit() or new_name[0] == "_":
                     new_name = "D" + new_name
                 kw["escaped_from"] = name
                 name = new_name
-            elif name[0].isdigit():
+            elif name[0].isdigit() or name[0] == "_":
                 new_name = "D" + name
                 kw["escaped_from"] = name
                 name = new_name

diff --git a/lib/sqlalchemy/testing/suite/test_dialect.py b/lib/sqlalchemy/testing/suite/test_dialect.py
index 54acc7ec4b9ccfdf836f1afbe61fe8e9ac30ffc9..99947bbe4f536c201e3892635917c10b265a02d5 100644 (file)
--- a/lib/sqlalchemy/testing/suite/test_dialect.py
+++ b/lib/sqlalchemy/testing/suite/test_dialect.py
@@ -317,7 +317,7 @@ class FutureWeCanSetDefaultSchemaWEventsTest(
 class DifficultParametersTest(fixtures.TestBase):
     __backend__ = True
 
-    @testing.combinations(
+    tough_parameters = testing.combinations(
         ("boring",),
         ("per cent",),
         ("per % cent",),
@@ -328,14 +328,26 @@ class DifficultParametersTest(fixtures.TestBase):
         ("_starts_with_underscore",),
         ("dot.s",),
         ("more :: %colons%",),
+        ("_name",),
+        ("___name",),
+        ("[BracketsAndCase]",),
+        ("42numbers",),
+        ("percent%signs",),
+        ("has spaces",),
         ("/slashes/",),
         ("more/slashes",),
         ("q?marks",),
         ("1param",),
         ("1col:on",),
-        argnames="name",
+        argnames="paramname",
     )
-    def test_round_trip(self, name, connection, metadata):
+
+    @tough_parameters
+    def test_round_trip_same_named_column(
+        self, paramname, connection, metadata
+    ):
+        name = paramname
+
         t = Table(
             "t",
             metadata,
@@ -368,3 +380,51 @@ class DifficultParametersTest(fixtures.TestBase):
         )
 
         row = connection.execute(stmt).first()
+
+    @testing.fixture
+    def multirow_fixture(self, metadata, connection):
+        mytable = Table(
+            "mytable",
+            metadata,
+            Column("myid", Integer),
+            Column("name", String(50)),
+            Column("desc", String(50)),
+        )
+
+        mytable.create(connection)
+
+        connection.execute(
+            mytable.insert(),
+            [
+                {"myid": 1, "name": "a", "desc": "a_desc"},
+                {"myid": 2, "name": "b", "desc": "b_desc"},
+                {"myid": 3, "name": "c", "desc": "c_desc"},
+                {"myid": 4, "name": "d", "desc": "d_desc"},
+            ],
+        )
+        yield mytable
+
+    @tough_parameters
+    def test_standalone_bindparam_escape(
+        self, paramname, connection, multirow_fixture
+    ):
+        tbl1 = multirow_fixture
+        stmt = select(tbl1.c.myid).where(
+            tbl1.c.name == bindparam(paramname, value="x")
+        )
+        res = connection.scalar(stmt, {paramname: "c"})
+        eq_(res, 3)
+
+    @tough_parameters
+    def test_standalone_bindparam_escape_expanding(
+        self, paramname, connection, multirow_fixture
+    ):
+        tbl1 = multirow_fixture
+        stmt = (
+            select(tbl1.c.myid)
+            .where(tbl1.c.name.in_(bindparam(paramname, value=["a", "b"])))
+            .order_by(tbl1.c.myid)
+        )
+
+        res = connection.scalars(stmt, {paramname: ["d", "a"]}).all()
+        eq_(res, [1, 4])