]> git.ipfire.org Git - thirdparty/sqlalchemy/sqlalchemy.git/commitdiff
- add a test suite that ensures textual autocommit works
authorMike Bayer <mike_mp@zzzcomputing.com>
Mon, 13 Feb 2017 19:17:11 +0000 (14:17 -0500)
committerMike Bayer <mike_mp@zzzcomputing.com>
Mon, 13 Feb 2017 19:17:11 +0000 (14:17 -0500)
for correct expressions

Change-Id: I17d35169be914924828487abba05658dff380f2a

test/engine/test_execute.py

index 145877dd34c8fa7dd02cfde19ea4299cf39b70a0..32834f950b8b246211d14ff911a6f7d717d6261b 100644 (file)
@@ -2635,3 +2635,59 @@ class DialectEventTest(fixtures.TestBase):
         conn.connection.invalidate()
         conn = e.connect()
         eq_(conn.info['boom'], "one")
+
+
+class AutocommitTextTest(fixtures.TestBase):
+    __backend__ = True
+
+    def _test_keyword(self, keyword, expected=True):
+        dbapi = Mock(
+            connect=Mock(
+                return_value=Mock(
+                    cursor=Mock(
+                        return_value=Mock(
+                            description=()
+                        )
+                    )
+                )
+            )
+        )
+        engine = engines.testing_engine(
+            options={
+                "_initialize": False,
+                "pool_reset_on_return": None
+            })
+        engine.dialect.dbapi = dbapi
+        engine.execute("%s something table something" % keyword)
+        if expected:
+            eq_(
+                dbapi.connect().mock_calls,
+                [call.cursor(), call.commit()]
+            )
+        else:
+            eq_(
+                dbapi.connect().mock_calls,
+                [call.cursor()]
+            )
+
+    def test_update(self):
+        self._test_keyword("UPDATE")
+
+    def test_insert(self):
+        self._test_keyword("INSERT")
+
+    def test_delete(self):
+        self._test_keyword("DELETE")
+
+    def test_alter(self):
+        self._test_keyword("ALTER TABLE")
+
+    def test_create(self):
+        self._test_keyword("CREATE TABLE foobar")
+
+    def test_drop(self):
+        self._test_keyword("DROP TABLE foobar")
+
+    def test_select(self):
+        self._test_keyword("SELECT foo FROM table", False)
+