diff --git a/packages/google-cloud-spanner/tests/mockserver_tests/test_dbapi_ddl.py b/packages/google-cloud-spanner/tests/mockserver_tests/test_dbapi_ddl.py new file mode 100644 index 000000000000..5834a1295bfd --- /dev/null +++ b/packages/google-cloud-spanner/tests/mockserver_tests/test_dbapi_ddl.py @@ -0,0 +1,239 @@ +# Copyright 2026 Google LLC All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from google.cloud.spanner_dbapi.connection import Connection +from google.cloud.spanner_dbapi.exceptions import ProgrammingError +from google.cloud.spanner_v1 import ExecuteBatchDmlRequest, ExecuteSqlRequest +from tests.mockserver_tests.mock_server_test_base import ( + MockServerTestBase, + add_select1_result, + add_update_count, +) + + +class TestDbapiDdl(MockServerTestBase): + def test_ddl_autocommit(self): + connection = Connection(self.instance, self.database) + connection.autocommit = True + ddl_statement = ( + "CREATE TABLE Singers (" + "SingerId INT64 NOT NULL, " + "Name STRING(1024)" + ") PRIMARY KEY (SingerId)" + ) + + try: + with connection.cursor() as cursor: + cursor.execute(ddl_statement) + + self.assertEqual(1, len(self.database_admin_service.requests)) + request = self.database_admin_service.requests[0] + self.assertEqual(self.database.name, request.database) + self.assertEqual([ddl_statement], request.statements) + finally: + connection.close() + + def test_ddl_buffered_until_commit(self): + connection = Connection(self.instance, self.database) + connection.autocommit = False + ddl_statement = ( + "CREATE TABLE Singers (" + "SingerId INT64 NOT NULL, " + "Name STRING(1024)" + ") PRIMARY KEY (SingerId)" + ) + + try: + with connection.cursor() as cursor: + cursor.execute(ddl_statement) + + self.assertEqual(0, len(self.database_admin_service.requests)) + connection.commit() + + self.assertEqual(1, len(self.database_admin_service.requests)) + request = self.database_admin_service.requests[0] + self.assertEqual(self.database.name, request.database) + self.assertEqual([ddl_statement], request.statements) + finally: + connection.close() + + def test_ddl_multiple_statements_buffered_until_commit(self): + connection = Connection(self.instance, self.database) + connection.autocommit = False + statement1 = "CREATE TABLE Table1 (Id INT64) PRIMARY KEY (Id)" + statement2 = "CREATE TABLE Table2 (Id INT64) PRIMARY KEY (Id)" + + try: + with connection.cursor() as cursor: + cursor.execute(statement1) + cursor.execute(statement2) + + self.assertEqual(0, len(self.database_admin_service.requests)) + connection.commit() + + self.assertEqual(1, len(self.database_admin_service.requests)) + request = self.database_admin_service.requests[0] + self.assertEqual(self.database.name, request.database) + self.assertEqual([statement1, statement2], request.statements) + finally: + connection.close() + + def test_ddl_semicolon_separated_statements_autocommit(self): + connection = Connection(self.instance, self.database) + connection.autocommit = True + statement1 = "CREATE TABLE Table1 (Id INT64) PRIMARY KEY (Id)" + statement2 = "CREATE TABLE Table2 (Id INT64) PRIMARY KEY (Id)" + + try: + with connection.cursor() as cursor: + cursor.execute(f"{statement1}; {statement2}") + + self.assertEqual(1, len(self.database_admin_service.requests)) + request = self.database_admin_service.requests[0] + self.assertEqual(self.database.name, request.database) + self.assertEqual([statement1, statement2], request.statements) + finally: + connection.close() + + def test_ddl_flushed_before_subsequent_execute(self): + dml_statement = "INSERT INTO Singers (SingerId, Name) VALUES (1, 'first-name')" + add_update_count(dml_statement, 1) + + connection = Connection(self.instance, self.database) + connection.autocommit = False + ddl_statement = ( + "CREATE TABLE Singers (" + "SingerId INT64 NOT NULL, " + "Name STRING(1024)" + ") PRIMARY KEY (SingerId)" + ) + + try: + with connection.cursor() as cursor: + cursor.execute(ddl_statement) + self.assertEqual(0, len(self.database_admin_service.requests)) + + cursor.execute(dml_statement) + self.assertEqual(1, len(self.database_admin_service.requests)) + request = self.database_admin_service.requests[0] + self.assertEqual(self.database.name, request.database) + self.assertEqual([ddl_statement], request.statements) + self.assertEqual(1, cursor.rowcount) + + execute_requests = [ + req + for req in self.spanner_service.requests + if isinstance(req, ExecuteSqlRequest) + ] + self.assertEqual(1, len(execute_requests)) + self.assertEqual(dml_statement, execute_requests[0].sql) + + connection.commit() + finally: + connection.close() + + def test_ddl_flushed_before_subsequent_executemany(self): + dml_statement = "INSERT INTO Singers (SingerId, Name) VALUES (1, 'first-name')" + add_update_count(dml_statement, 1) + + connection = Connection(self.instance, self.database) + connection.autocommit = False + ddl_statement = ( + "CREATE TABLE Singers (" + "SingerId INT64 NOT NULL, " + "Name STRING(1024)" + ") PRIMARY KEY (SingerId)" + ) + + try: + with connection.cursor() as cursor: + cursor.execute(ddl_statement) + self.assertEqual(0, len(self.database_admin_service.requests)) + + cursor.executemany(dml_statement, [()]) + self.assertEqual(1, len(self.database_admin_service.requests)) + request = self.database_admin_service.requests[0] + self.assertEqual(self.database.name, request.database) + self.assertEqual([ddl_statement], request.statements) + self.assertEqual(1, cursor.rowcount) + + batch_requests = [ + req + for req in self.spanner_service.requests + if isinstance(req, ExecuteBatchDmlRequest) + ] + self.assertEqual(1, len(batch_requests)) + self.assertEqual(dml_statement, batch_requests[0].statements[0].sql) + + connection.commit() + finally: + connection.close() + + def test_ddl_flushed_before_subsequent_query(self): + add_select1_result() + + connection = Connection(self.instance, self.database) + connection.autocommit = False + ddl_statement = ( + "CREATE TABLE Singers (" + "SingerId INT64 NOT NULL, " + "Name STRING(1024)" + ") PRIMARY KEY (SingerId)" + ) + + try: + with connection.cursor() as cursor: + cursor.execute(ddl_statement) + self.assertEqual(0, len(self.database_admin_service.requests)) + + cursor.execute("SELECT 1") + self.assertEqual(1, len(self.database_admin_service.requests)) + request = self.database_admin_service.requests[0] + self.assertEqual(self.database.name, request.database) + self.assertEqual([ddl_statement], request.statements) + + connection.commit() + finally: + connection.close() + + def test_ddl_uncommitted_on_close(self): + connection = Connection(self.instance, self.database) + connection.autocommit = False + ddl_statement = ( + "CREATE TABLE Singers (" + "SingerId INT64 NOT NULL, " + "Name STRING(1024)" + ") PRIMARY KEY (SingerId)" + ) + + with connection.cursor() as cursor: + cursor.execute(ddl_statement) + + connection.close() + self.assertEqual(0, len(self.database_admin_service.requests)) + + def test_ddl_executemany_rejected_client_side(self): + connection = Connection(self.instance, self.database) + ddl_statement = "CREATE TABLE Singers (SingerId INT64) PRIMARY KEY (SingerId)" + + try: + for autocommit in (False, True): + connection.autocommit = autocommit + with connection.cursor() as cursor: + with self.assertRaises(ProgrammingError): + cursor.executemany(ddl_statement, []) + + self.assertEqual(0, len(self.database_admin_service.requests)) + finally: + connection.close() diff --git a/packages/google-cloud-spanner/tests/system/test_dbapi.py b/packages/google-cloud-spanner/tests/system/test_dbapi.py index cc0666153d1b..8bb6a3966961 100644 --- a/packages/google-cloud-spanner/tests/system/test_dbapi.py +++ b/packages/google-cloud-spanner/tests/system/test_dbapi.py @@ -1169,247 +1169,24 @@ def test_execute_many(self): assert res[0] == 1 - @pytest.mark.noautofixt - def test_DDL_autocommit(self, shared_instance, dbapi_database): - """Check that DDLs in autocommit mode are immediately executed.""" - - try: - conn = Connection(shared_instance, dbapi_database) - conn.autocommit = True - - cur = conn.cursor() - cur.execute( - """ - CREATE TABLE Singers ( - SingerId INT64 NOT NULL, - Name STRING(1024), - ) PRIMARY KEY (SingerId) - """ - ) - conn.close() - - # if previous DDL wasn't committed, the next DROP TABLE - # statement will fail with a ProgrammingError - conn = Connection(shared_instance, dbapi_database) - cur = conn.cursor() - - cur.execute("DROP TABLE Singers") - conn.commit() - finally: - # Delete table - table = dbapi_database.table("Singers") - if table.exists(): - op = dbapi_database.update_ddl(["DROP TABLE Singers"]) - op.result() - - def test_ddl_execute_autocommit_true(self, dbapi_database): - """Check that DDL statement in autocommit mode results in successful - DDL statement execution for execute method.""" - # Cleanup if table exists from previous failed run - table = dbapi_database.table("DdlExecuteAutocommit") - if table.exists(): - dbapi_database.update_ddl(["DROP TABLE DdlExecuteAutocommit"]).result() - - try: - self._conn.autocommit = True - self._cursor.execute( - """ - CREATE TABLE DdlExecuteAutocommit ( - SingerId INT64 NOT NULL, - Name STRING(1024), - ) PRIMARY KEY (SingerId) - """ - ) - assert table.exists() is True - finally: - if table.exists(): - dbapi_database.update_ddl(["DROP TABLE DdlExecuteAutocommit"]).result() - - def test_ddl_executemany_autocommit_true(self, dbapi_database): - """Check that DDL statement in autocommit mode results in exception for - executemany method .""" - # Cleanup if table exists from previous failed run - table = dbapi_database.table("DdlExecuteManyAutocommit") - if table.exists(): - dbapi_database.update_ddl(["DROP TABLE DdlExecuteManyAutocommit"]).result() - - try: - self._conn.autocommit = True - with pytest.raises(ProgrammingError): - self._cursor.executemany( - """ - CREATE TABLE DdlExecuteManyAutocommit ( - SingerId INT64 NOT NULL, - Name STRING(1024), - ) PRIMARY KEY (SingerId) - """, - [], - ) - assert table.exists() is False - finally: - if table.exists(): - dbapi_database.update_ddl( - ["DROP TABLE DdlExecuteManyAutocommit"] - ).result() - - def test_ddl_executemany_autocommit_false(self, dbapi_database): - """Check that DDL statement in non-autocommit mode results in exception for - executemany method .""" - # Cleanup if table exists from previous failed run - table = dbapi_database.table("DdlExecuteManyAutocommit") - if table.exists(): - dbapi_database.update_ddl(["DROP TABLE DdlExecuteManyAutocommit"]).result() - - try: - with pytest.raises(ProgrammingError): - self._cursor.executemany( - """ - CREATE TABLE DdlExecuteManyAutocommit ( - SingerId INT64 NOT NULL, - Name STRING(1024), - ) PRIMARY KEY (SingerId) - """, - [], - ) - assert table.exists() is False - finally: - if table.exists(): - dbapi_database.update_ddl( - ["DROP TABLE DdlExecuteManyAutocommit"] - ).result() - - def test_ddl_execute(self, dbapi_database): - """Check that DDL statement followed by non-DDL execute statement in - non autocommit mode results in successful DDL statement execution.""" - table_name = "DdlExecute" - # Cleanup if table exists from previous failed run - table = dbapi_database.table(table_name) - if table.exists(): - dbapi_database.update_ddl([f"DROP TABLE {table_name}"]).result() - - try: - want_row = ( - 1, - "first-name", - ) - self._cursor.execute( - f""" - CREATE TABLE {table_name} ( - SingerId INT64 NOT NULL, - Name STRING(1024), - ) PRIMARY KEY (SingerId) - """ - ) - # DDL execute in non-autocommit mode should not be visible until next statement or commit? - # Actually dbapi DDL is usually immediate. - - self._cursor.execute( - f""" - INSERT INTO {table_name} (SingerId, Name) - VALUES (1, "first-name") - """ - ) - assert table.exists() is True - self._conn.commit() - - # read the resulting data from the database - self._cursor.execute(f"SELECT * FROM {table_name}") - got_rows = self._cursor.fetchall() - - assert got_rows == [want_row] - finally: - if table.exists(): - try: - dbapi_database.update_ddl([f"DROP TABLE {table_name}"]).result() - except Exception: - pass - - def test_ddl_executemany(self, dbapi_database): - """Check that DDL statement followed by non-DDL executemany statement in - non autocommit mode results in successful DDL statement execution.""" - table_name = "DdlExecuteMany" - # Cleanup if table exists from previous failed run - table = dbapi_database.table(table_name) - if table.exists(): - dbapi_database.update_ddl([f"DROP TABLE {table_name}"]).result() - - try: - want_row = ( - 1, - "first-name", - ) - self._cursor.execute( - f""" - CREATE TABLE {table_name} ( - SingerId INT64 NOT NULL, - Name STRING(1024), - ) PRIMARY KEY (SingerId) - """ - ) - - self._cursor.executemany( - f""" - INSERT INTO {table_name} (SingerId, Name) - VALUES (%s, %s) - """, - [want_row], - ) - assert table.exists() is True - self._conn.commit() - - # read the resulting data from the database - self._cursor.execute(f"SELECT * FROM {table_name}") - got_rows = self._cursor.fetchall() - - assert got_rows == [want_row] - finally: - if table.exists(): - try: - dbapi_database.update_ddl([f"DROP TABLE {table_name}"]).result() - except Exception: - pass - @pytest.mark.skipif(_helpers.USE_EMULATOR, reason="Emulator does not support json.") - def test_autocommit_with_json_data(self, dbapi_database): - """ - Check that DDLs in autocommit mode are immediately - executed for json fields. - """ + def test_autocommit_with_json_data(self): + """Check that json fields in autocommit mode are inserted and queried.""" try: self._conn.autocommit = True self._cursor.execute( - """ - CREATE TABLE JsonDetails ( - DataId INT64 NOT NULL, - Details JSON, - ) PRIMARY KEY (DataId) - """ + "INSERT INTO all_types (col_int64, col_json) VALUES (%s, %s)", + (123, JsonObject({"name": "Jakob", "age": "26"})), ) - - # Insert data to table self._cursor.execute( - sql="INSERT INTO JsonDetails (DataId, Details) VALUES (%s, %s)", - args=(123, JsonObject({"name": "Jakob", "age": "26"})), + "SELECT col_int64, col_json FROM all_types WHERE col_int64 = 123" ) - - # Read back the data. - self._cursor.execute("""select * from JsonDetails;""") got_rows = self._cursor.fetchall() - - # Assert the response assert len(got_rows) == 1 assert got_rows[0][0] == 123 assert got_rows[0][1] == {"age": "26", "name": "Jakob"} - - # Drop the table - self._cursor.execute("DROP TABLE JsonDetails") - self._conn.commit() finally: - # Delete table - table = dbapi_database.table("JsonDetails") - if table.exists(): - op = dbapi_database.update_ddl(["DROP TABLE JsonDetails"]) - op.result() + self._cursor.execute("DELETE FROM all_types WHERE col_int64 = 123") @pytest.mark.skipif(_helpers.USE_EMULATOR, reason="Emulator does not support json.") def test_json_array(self): @@ -1429,38 +1206,6 @@ def test_json_array(self): finally: self._cursor.execute("DELETE FROM all_types WHERE col_int64 = 1") - @pytest.mark.noautofixt - def test_DDL_commit(self, shared_instance, dbapi_database): - """Check that DDLs in commit mode are executed on calling `commit()`.""" - try: - conn = Connection(shared_instance, dbapi_database) - cur = conn.cursor() - - cur.execute( - """ - CREATE TABLE Singers ( - SingerId INT64 NOT NULL, - Name STRING(1024), - ) PRIMARY KEY (SingerId) - """ - ) - conn.commit() - conn.close() - - # if previous DDL wasn't committed, the next DROP TABLE - # statement will fail with a ProgrammingError - conn = Connection(shared_instance, dbapi_database) - cur = conn.cursor() - - cur.execute("DROP TABLE Singers") - conn.commit() - finally: - # Delete table - table = dbapi_database.table("Singers") - if table.exists(): - op = dbapi_database.update_ddl(["DROP TABLE Singers"]) - op.result() - def test_ping(self): """Check connection validation method.""" self._conn.validate()