chore(django-spanner): add unit test coverage

packages/django-google-spanner/.coveragerc

@@ -10,14 +10,13 @@
 branch = True
 source =
     django_spanner
-
 [paths]
 source =
     django_spanner
     */site-packages/django_spanner
 
 [report]
-fail_under = 68
+fail_under = 100
 show_missing = True
 exclude_lines =
     # Re-enable the standard pragma
@@ -26,6 +25,7 @@ exclude_lines =
     def __repr__
     # Ignore abstract methods
     raise NotImplementedError
+    if __name__ == .__main__.:
 omit =
   tests/*
   */tests/*

packages/django-google-spanner/tests/mockserver_tests/test_basics.py

@@ -132,4 +132,4 @@ class LocalSinger(models.Model):
         finally:
             for db, config in DATABASES.items():
                 if config["ENGINE"] == "django_spanner":
-                    config.pop("DISABLE_RANDOM_ID_GENERATION", None)
+                    config.pop("RANDOM_ID_GENERATION_ENABLED", None)

packages/django-google-spanner/tests/unit/django_spanner/test__opentelemetry_tracing.py

@@ -100,6 +100,12 @@ def test_trace_call(self):
             self.assertEqual(span.name, "CloudSpannerDjango.Test")
             self.assertEqual(span.status.status_code, StatusCode.OK)
 
+        def test_trace_call_no_extra_attributes(self):
+            with _opentelemetry_tracing.trace_call(
+                "CloudSpannerDjango.TestNoExtra", _make_connection()
+            ) as span:
+                self.assertIsNotNone(span)
+
         def test_trace_error(self):
             extra_attributes = {"db.instance": "database_name"}
 

packages/django-google-spanner/tests/unit/django_spanner/test_base.py

@@ -15,6 +15,8 @@ def setUp(self):
         import django_spanner.base
 
         django_spanner.base._SPANNER_CLIENT_CACHE = None
+        self.db_wrapper.connection = None
+        self.db_wrapper.settings_dict = dict(self.settings_dict)
         self.client_patcher = mock.patch("django_spanner.base.spanner.Client")
         self.mock_client = self.client_patcher.start()
         self.mock_client.return_value.instance.return_value.instance_id = (
@@ -44,15 +46,19 @@ def test_get_connection_params(self):
         self.assertEqual(params["option"], self.OPTIONS["option"])
 
     def test_get_new_connection(self):
-        self.db_wrapper.Database = mock_database = mock.MagicMock()
-        mock_database.connect = mock_connection = mock.MagicMock()
-        conn_params = {"test_param": "dummy"}
-        self.db_wrapper.get_new_connection(conn_params)
-        mock_connection.assert_called_once_with(
-            self.INSTANCE_ID,
-            client=mock_database.connect.call_args[1]["client"],
-            **conn_params,
-        )
+        orig_database = self.db_wrapper.Database
+        try:
+            self.db_wrapper.Database = mock_database = mock.MagicMock()
+            mock_database.connect = mock_connection = mock.MagicMock()
+            conn_params = {"test_param": "dummy"}
+            self.db_wrapper.get_new_connection(conn_params)
+            mock_connection.assert_called_once_with(
+                self.INSTANCE_ID,
+                client=mock_database.connect.call_args[1]["client"],
+                **conn_params,
+            )
+        finally:
+            self.db_wrapper.Database = orig_database
 
     def test_init_connection_state(self):
         from google.cloud.spanner_dbapi.connection import Connection
@@ -87,15 +93,28 @@ def test_is_usable(self):
         mock_connection.is_closed = False
         self.assertTrue(self.db_wrapper.is_usable())
 
-    def test_is_usable_with_error(self):
-        from google.cloud.spanner_dbapi.exceptions import Error
+    def test_allow_transactions_in_auto_commit(self):
+        self.assertTrue(self.db_wrapper.allow_transactions_in_auto_commit)
+        self.db_wrapper.settings_dict["ALLOW_TRANSACTIONS_IN_AUTO_COMMIT"] = False
+        self.assertFalse(self.db_wrapper.allow_transactions_in_auto_commit)
 
-        self.db_wrapper.connection = mock_connection = mock.MagicMock()
-        mock_connection.cursor = mock.MagicMock(side_effect=Error)
+    def test_is_usable_with_error(self):
+        mock_cursor = mock.MagicMock()
+        mock_cursor.execute.side_effect = self.db_wrapper.Database.Error("error")
+        mock_connection = mock.MagicMock(is_closed=False)
+        mock_connection.cursor.return_value = mock_cursor
+        self.db_wrapper.connection = mock_connection
         self.assertFalse(self.db_wrapper.is_usable())
 
     def test_start_transaction_under_autocommit(self):
+        mock_cursor = mock.MagicMock()
         self.db_wrapper.connection = mock_connection = mock.MagicMock()
-        mock_connection.cursor = mock_cursor = mock.MagicMock()
+        mock_connection.cursor.return_value = mock_cursor
+
         self.db_wrapper._start_transaction_under_autocommit()
-        mock_cursor.assert_called_once_with()
+        mock_cursor.execute.assert_called_once_with("BEGIN")
+
+        mock_cursor.reset_mock()
+        self.db_wrapper.settings_dict["ALLOW_TRANSACTIONS_IN_AUTO_COMMIT"] = False
+        self.db_wrapper._start_transaction_under_autocommit()
+        mock_cursor.execute.assert_called_once_with("SELECT 1")

packages/django-google-spanner/tests/unit/django_spanner/test_compiler.py

@@ -4,6 +4,7 @@
 # license that can be found in the LICENSE file or at
 # https://developers.google.com/open-source/licenses/bsd
 
+from unittest import mock
 from django.core.exceptions import EmptyResultSet
 from django.db.models.query import QuerySet
 from django.db.utils import DatabaseError
@@ -129,12 +130,12 @@ def test_get_combinator_sql_union_and_difference_query_together(self):
         self.assertEqual(
             sql_compiled,
             [
-                "SELECT tests_number.num AS num FROM tests_number WHERE "
-                + "tests_number.num <= %s UNION DISTINCT SELECT * FROM ("
+                "(SELECT tests_number.num AS num FROM tests_number WHERE "
+                + "tests_number.num <= %s) UNION DISTINCT (SELECT * FROM ("
                 + "SELECT tests_number.num AS num FROM tests_number WHERE "
                 + "tests_number.num >= %s EXCEPT DISTINCT "
                 + "SELECT tests_number.num AS num FROM tests_number "
-                + "WHERE tests_number.num = %s)"
+                + "WHERE tests_number.num = %s))"
             ],
         )
         self.assertEqual(params, [1, 8, 10])
@@ -173,3 +174,53 @@ def test_get_combinator_sql_empty_queryset_raises_exception(self):
         compiler = SQLCompiler(QuerySet().query, self.connection, "default")
         with self.assertRaises(EmptyResultSet):
             compiler.get_combinator_sql("union", False)
+
+    def test_get_combinator_sql_sliced_and_features(self):
+        qs1 = Number.objects.filter(num__lte=1)
+        qs2 = Number.objects.none()  # Empty result set
+        qs3 = Number.objects.filter(num__gte=8)
+        qs4 = qs1.union(qs3)
+
+        compiler = SQLCompiler(qs4.query, self.connection, "default")
+        compiler.connection.features.supports_slicing_ordering_in_compound = True
+        compiler.query.high_mark = 10
+        sql, params = compiler.get_combinator_sql("union", False)
+        self.assertTrue(len(sql) > 0)
+
+        # Test values_select set_values
+        qs_val = Number.objects.values("num")
+        qs_comb = qs_val.union(Number.objects.all())
+        comp_val = SQLCompiler(qs_comb.query, self.connection, "default")
+        sql2, params2 = comp_val.get_combinator_sql("union", True)
+        self.assertTrue(len(sql2) > 0)
+
+    def test_get_combinator_sql_edge_cases(self):
+        qs1, qs_empty = Number.objects.filter(num=1), Number.objects.none()
+
+        # Test valid combinators (union, difference, subquery)
+        for comp in [
+            SQLCompiler(qs1.union(qs_empty).query, self.connection, "default"),
+            SQLCompiler(qs1.difference(qs_empty).query, self.connection, "default"),
+        ]:
+            comp.query.subquery = True
+            self.assertTrue(len(comp.get_combinator_sql("union", False)[0]) > 0)
+
+        # Test EmptyResultSet exceptions (intersection on empty, union on empty parts)
+        qs_none1, qs_none2 = Number.objects.none(), Number.objects.none()
+        for comp in [
+            SQLCompiler(qs_empty.intersection(qs1).query, self.connection, "default"),
+            SQLCompiler(qs_none1.union(qs_none2).query, self.connection, "default"),
+        ]:
+            with self.assertRaises(EmptyResultSet):
+                comp.get_combinator_sql("union", False)
+
+        # supports_parentheses_in_compound False with subquery combinator
+        comp_sub = SQLCompiler(qs1.union(Number.objects.filter(num=2)).query, self.connection, "default")
+        mock_sub = mock.MagicMock(as_sql=mock.MagicMock(return_value=("SELECT 1", [])))
+        mock_sub.query.combinator = "union"
+        mock_sub.query.is_sliced = False
+        mock_sub.get_order_by.return_value = []
+        with mock.patch.object(self.connection.features, "supports_parentheses_in_compound", False):
+            with mock.patch.object(comp_sub.query.combined_queries[0], "get_compiler", return_value=mock_sub):
+                with mock.patch.object(comp_sub.query.combined_queries[1], "get_compiler", return_value=mock_sub):
+                    self.assertIn("SELECT * FROM", comp_sub.get_combinator_sql("union", False)[0][0])

packages/django-google-spanner/tests/unit/django_spanner/test_creation.py

@@ -0,0 +1,97 @@
+# Copyright 2026 Google LLC
+#
+# Use of this source code is governed by a BSD-style
+# license that can be found in the LICENSE file or at
+# https://developers.google.com/open-source/licenses/bsd
+
+import os
+from unittest import mock
+from django_spanner.creation import DatabaseCreation
+from tests.unit.django_spanner.simple_test import SpannerSimpleTestClass
+
+
+class TestCreation(SpannerSimpleTestClass):
+    def setUp(self):
+        super().setUp()
+        self.db_wrapper.settings_dict = dict(self.settings_dict)
+        self.db_wrapper.settings_dict["TEST"] = {"NAME": "test_db"}
+        self.creation = DatabaseCreation(self.db_wrapper)
+
+    def test_mark_skips(self):
+        with mock.patch("django.conf.settings.INSTALLED_APPS", ["django.contrib.contenttypes"]):
+            self.db_wrapper.features.skip_tests = (
+                "django.contrib.contenttypes.models.ContentType",
+            )
+            self.creation.mark_skips()
+
+    def test_create_test_db_user_cancel(self):
+        with mock.patch.object(self.creation, "_execute_create_test_db", side_effect=Exception("err")):
+            with mock.patch("builtins.input", return_value="no"):
+                with self.assertRaises(SystemExit) as cm:
+                    self.creation._create_test_db(verbosity=1, autoclobber=False, keepdb=False)
+                self.assertEqual(cm.exception.code, 1)
+
+    def test_create_test_db_recreate_error_exit(self):
+        with mock.patch.object(self.creation, "_execute_create_test_db", side_effect=Exception("err")):
+            with mock.patch.object(self.creation, "_destroy_test_db", side_effect=Exception("destroy_err")):
+                with self.assertRaises(SystemExit) as cm:
+                    self.creation._create_test_db(verbosity=1, autoclobber=True, keepdb=False)
+                self.assertEqual(cm.exception.code, 2)
+
+    def test_create_test_db(self):
+        with mock.patch.dict(os.environ, {"RUNNING_SPANNER_BACKEND_TESTS": "1"}):
+            with mock.patch.object(self.creation, "mark_skips") as mock_mark:
+                with mock.patch("django.db.backends.base.creation.BaseDatabaseCreation.create_test_db"):
+                    self.creation.create_test_db()
+                    mock_mark.assert_called_once()
+
+        # Without env var
+        env = dict(os.environ)
+        env.pop("RUNNING_SPANNER_BACKEND_TESTS", None)
+        with mock.patch.dict(os.environ, env, clear=True):
+            with mock.patch("django.db.backends.base.creation.BaseDatabaseCreation.create_test_db"):
+                self.creation.create_test_db()
+
+    def test_mark_skips_success_and_attribute_error(self):
+        class DummyTestCase:
+            def test_foo(self): pass
+        with mock.patch("django.conf.settings.INSTALLED_APPS", ["tests"]):
+            with mock.patch("django_spanner.creation.import_string", return_value=DummyTestCase):
+                with mock.patch.object(self.creation.connection.features, "skip_tests", {"tests.DummyTestCase.test_foo", "tests.DummyTestCase.test_non_existent"}):
+                    self.creation.mark_skips()
+                    self.assertTrue(hasattr(DummyTestCase.test_foo, "__unittest_skip__"))
+
+    def test_create_test_db_internal(self):
+        with mock.patch.object(self.creation, "_execute_create_test_db", side_effect=[Exception("db exists"), None]):
+            with mock.patch.object(self.creation, "_destroy_test_db") as mock_destroy:
+                with mock.patch.object(self.creation, "log") as mock_log:
+                    self.creation._create_test_db(verbosity=1, autoclobber=True, keepdb=False)
+                    mock_destroy.assert_called_once()
+                    mock_log.assert_called()
+
+        with mock.patch.object(self.creation, "_execute_create_test_db", side_effect=[Exception("db exists"), None]):
+            with mock.patch.object(self.creation, "_destroy_test_db"):
+                self.creation._create_test_db(verbosity=0, autoclobber=True, keepdb=False)
+
+    def test_create_test_db_internal_error_keepdb(self):
+        with mock.patch.object(self.creation, "_execute_create_test_db", side_effect=Exception("create error")):
+            dbname = self.creation._create_test_db(verbosity=1, autoclobber=True, keepdb=True)
+            self.assertEqual(dbname, "test_db")
+
+    def test_create_test_db_internal_error_recreate(self):
+        with mock.patch.object(self.creation, "_execute_create_test_db", side_effect=[Exception("err1"), None]):
+            with mock.patch.object(self.creation, "_destroy_test_db") as mock_destroy:
+                dbname = self.creation._create_test_db(verbosity=1, autoclobber=True, keepdb=False)
+                mock_destroy.assert_called_once()
+                self.assertEqual(dbname, "test_db")
+
+    def test_execute_create_and_destroy_test_db(self):
+        mock_instance = mock.MagicMock()
+        with mock.patch.object(type(self.db_wrapper), "instance", new_callable=mock.PropertyMock(return_value=mock_instance)):
+            self.creation._execute_create_test_db(None, {"dbname": "test_db"}, keepdb=False)
+            mock_instance.database.assert_called_with("test_db")
+            mock_instance.database.return_value.create.assert_called_once()
+
+            self.creation._destroy_test_db("test_db", verbosity=1)
+            mock_instance.database.assert_called_with("test_db")
+            mock_instance.database.return_value.drop.assert_called_once()

packages/django-google-spanner/tests/unit/django_spanner/test_features.py

@@ -0,0 +1,27 @@
+# Copyright 2026 Google LLC
+#
+# Use of this source code is governed by a BSD-style
+# license that can be found in the LICENSE file or at
+# https://developers.google.com/open-source/licenses/bsd
+
+import importlib
+import os
+from unittest import mock
+
+from tests.unit.django_spanner.simple_test import SpannerSimpleTestClass
+
+
+class TestFeatures(SpannerSimpleTestClass):
+    def test_features_emulator_and_env(self):
+        import django_spanner
+        import django_spanner.features
+
+        with mock.patch.dict(os.environ, {"RUNNING_SPANNER_BACKEND_TESTS": "1", "SPANNER_EMULATOR_HOST": "localhost:9010"}):
+            with mock.patch.object(django_spanner, "USE_EMULATOR", True):
+                importlib.reload(django_spanner.features)
+                feat = django_spanner.features.DatabaseFeatures(self.db_wrapper)
+                self.assertFalse(feat.supports_foreign_keys)
+                self.assertFalse(feat.supports_json_field)
+                self.assertTrue(any("test_loaddata" in test for test in feat.skip_tests))
+
+        importlib.reload(django_spanner.features)

packages/django-google-spanner/tests/unit/django_spanner/test_functions.py

@@ -234,3 +234,18 @@ def test_substr(self):
             + "name_prefix FROM tests_author",
         )
         self.assertEqual(params, (1, 5))
+
+    def test_chr(self):
+        from django.db.models.functions import Chr
+        q1 = Author.objects.values("num").annotate(chr_val=Chr("num"))
+        compiler = SQLCompiler(q1.query, self.connection, "default")
+        sql_query, params = compiler.query.as_sql(compiler, self.connection)
+        self.assertIn("CODE_POINTS_TO_STRING", sql_query)
+
+    def test_json_array(self):
+        from django.db.models.functions import JSONObject
+        from django_spanner.functions import JSONArray
+        q1 = Author.objects.values("num").annotate(json_arr=JSONArray("num"))
+        compiler = SQLCompiler(q1.query, self.connection, "default")
+        sql_query, params = compiler.query.as_sql(compiler, self.connection)
+        self.assertIn("TO_JSON_STRING", sql_query)