text: from unittest.mock import MagicMock, patch
  
  from django.db import DEFAULT_DB_ALIAS, connection, connections
  from django.db.backends.base.base import BaseDatabaseWrapper
  from django.test import SimpleTestCase, TestCase, skipUnlessDBFeature
  
  from ..models import Square
  
  
  class DatabaseWrapperTests(SimpleTestCase):
      def test_repr(self):
          conn = connections[DEFAULT_DB_ALIAS]
          self.assertEqual(
              repr(conn),
              f"<DatabaseWrapper vendor={connection.vendor!r} alias='default'>",
          )
  
      def test_initialization_class_attributes(self):
          """
          The "initialization" class attributes like client_class and
          creation_class should be set on the class and reflected in the
          corresponding instance attributes of the instantiated backend.
          """
          conn = connections[DEFAULT_DB_ALIAS]
          conn_class = type(conn)
          attr_names = [
              ("client_class", "client"),
              ("creation_class", "creation"),
              ("features_class", "features"),
              ("introspection_class", "introspection"),
              ("ops_class", "ops"),
              ("validation_class", "validation"),
          ]
          for class_attr_name, instance_attr_name in attr_names:
              class_attr_value = getattr(conn_class, class_attr_name)
              self.assertIsNotNone(class_attr_value)
              instance_attr_value = getattr(conn, instance_attr_name)
              self.assertIsInstance(instance_attr_value, class_attr_value)
  
      def test_initialization_display_name(self):
          self.assertEqual(BaseDatabaseWrapper.display_name, "unknown")
          self.assertNotEqual(connection.display_name, "unknown")
  
      def test_get_database_version(self):
          with patch.object(BaseDatabaseWrapper, "__init__", return_value=None):
              msg = (
                  "subclasses of BaseDatabaseWrapper may require a "
                  "get_database_version() method."
              )
              with self.assertRaisesMessage(NotImplementedError, msg):
                  BaseDatabaseWrapper().get_database_version()
  
      def test_check_database_version_supported_with_none_as_database_version(self):
          with patch.object(connection.features, "minimum_database_version", None):
              connection.check_database_version_supported()
  
  
  class ExecuteWrapperTests(TestCase):
      @staticmethod
      def call_execute(connection, params=None):
          ret_val = "1" if params is None else "%s"
          sql = "SELECT " + ret_val + connection.features.bare_select_suffix
          with connection.cursor() as cursor:
              cursor.execute(sql, params)
  
      def call_executemany(self, connection, params=None):
          # executemany() must use an update query. Make sure it does nothing
          # by putting a false condition in the WHERE clause.
          sql = "DELETE FROM {} WHERE 0=1 AND 0=%s".format(Square._meta.db_table)
          if params is None:
              params = [(i,) for i in range(3)]
          with connection.cursor() as cursor:
              cursor.executemany(sql, params)
  
      @staticmethod
      def mock_wrapper():
          return MagicMock(side_effect=lambda execute, *args: execute(*args))
  
      def test_wrapper_invoked(self):
          wrapper = self.mock_wrapper()
          with connection.execute_wrapper(wrapper):
              self.call_execute(connection)
          self.assertTrue(wrapper.called)
          (_, sql, params, many, context), _ = wrapper.call_args
          self.assertIn("SELECT", sql)
          self.assertIsNone(params)
          self.assertIs(many, False)
          self.assertEqual(context["connection"], connection)
  
      def test_wrapper_invoked_many(self):
          wrapper = self.mock_wrapper()
          with connection.execute_wrapper(wrapper):
              self.call_executemany(connection)
          self.assertTrue(wrapper.called)
          (_, sql, param_list, many, context), _ = wrapper.call_args
          self.assertIn("DELETE", sql)
          self.assertIsInstance(param_list, (list, tuple))
          self.assertIs(many, True)
          self.assertEqual(context["connection"], connection)
  
      def test_database_queried(self):
          wrapper = self.mock_wrapper()
          with connection.execute_wrapper(wrapper):
              with connection.cursor() as cursor:
                  sql = "SELECT 17" + connection.features.bare_select_suffix
                  cursor.execute(sql)
                  seventeen = cursor.fetchall()
                  self.assertEqual(list(seventeen), [(17,)])
              self.call_executemany(connection)
  
      def test_nested_wrapper_invoked(self):
          outer_wrapper = self.mock_wrapper()
          inner_wrapper = self.mock_wrapper()
          with connection.execute_wrapper(outer_wrapper), connection.execute_wrapper(
              inner_wrapper
          ):
              self.call_execute(connection)
              self.assertEqual(inner_wrapper.call_count, 1)
              self.call_executemany(connection)
              self.assertEqual(inner_wrapper.call_count, 2)
  
      def test_outer_wrapper_blocks(self):
          def blocker(*args):
              pass
  
          wrapper = self.mock_wrapper()
          c = connection  # This alias shortens the next line.
          with c.execute_wrapper(wrapper), c.execute_wrapper(blocker), c.execute_wrapper(
              wrapper
          ):
              with c.cursor() as cursor:
                  cursor.execute("The database never sees this")
                  self.assertEqual(wrapper.call_count, 1)
                  cursor.executemany("The database never sees this %s", [("either",)])
                  self.assertEqual(wrapper.call_count, 2)
  
      def test_wrapper_gets_sql(self):
          wrapper = self.mock_wrapper()
          sql = "SELECT 'aloha'" + connection.features.bare_select_suffix
          with connection.execute_wrapper(wrapper), connection.cursor() as cursor:
              cursor.execute(sql)
          (_, reported_sql, _, _, _), _ = wrapper.call_args
          self.assertEqual(reported_sql, sql)
  
      def test_wrapper_connection_specific(self):
          wrapper = self.mock_wrapper()
          with connections["other"].execute_wrapper(wrapper):
              self.assertEqual(connections["other"].execute_wrappers, [wrapper])
              self.call_execute(connection)
          self.assertFalse(wrapper.called)
          self.assertEqual(connection.execute_wrappers, [])
          self.assertEqual(connections["other"].execute_wrappers, [])
  
  
  class ConnectionHealthChecksTests(SimpleTestCase):
      databases = {"default"}
  
      def setUp(self):
          # All test cases here need newly configured and created connections.
          # Use the default db connection for convenience.
          connection.close()
          self.addCleanup(connection.close)
  
      def patch_settings_dict(self, conn_health_checks):
          self.settings_dict_patcher = patch.dict(
              connection.settings_dict,
              {
                  **connection.settings_dict,
                  "CONN_MAX_AGE": None,
                  "CONN_HEALTH_CHECKS": conn_health_checks,
              },
          )
          self.settings_dict_patcher.start()
          self.addCleanup(self.settings_dict_patcher.stop)
  
      def run_query(self):
          with connection.cursor() as cursor:
              cursor.execute("SELECT 42" + connection.features.bare_select_suffix)
  
      @skipUnlessDBFeature("test_db_allows_multiple_connections")
      def test_health_checks_enabled(self):
          self.patch_settings_dict(conn_health_checks=True)
          self.assertIsNone(connection.connection)
          # Newly created connections are considered healthy without performing
          # the health check.
          with patch.object(connection, "is_usable", side_effect=AssertionError):
              self.run_query()
  
          old_connection = connection.connection
          # Simulate request_finished.
          connection.close_if_unusable_or_obsolete()
          self.assertIs(old_connection, connection.connection)
  
          # Simulate connection health check failing.
          with patch.object(
              connection, "is_usable", return_value=False
          ) as mocked_is_usable:
              self.run_query()
              new_connection = connection.connection
              # A new connection is established.
              self.assertIsNot(new_connection, old_connection)
              # Only one health check per "request" is performed, so the next
              # query will carry on even if the health check fails. Next query
              # succeeds because the real connection is healthy and only the
              # health check failure is mocked.
              self.run_query()
              self.assertIs(new_connection, connection.connection)
          self.assertEqual(mocked_is_usable.call_count, 1)
  
          # Simulate request_finished.
          connection.close_if_unusable_or_obsolete()
          # The underlying connection is being reused further with health checks
          # succeeding.
          self.run_query()
          self.run_query()
          self.assertIs(new_connection, connection.connection)
  
      @skipUnlessDBFeature("test_db_allows_multiple_connections")
      def test_health_checks_enabled_errors_occurred(self):
          self.patch_settings_dict(conn_health_checks=True)
          self.assertIsNone(connection.connection)
          # Newly created connections are considered healthy without performing
          # the health check.
          with patch.object(connection, "is_usable", side_effect=AssertionError):
              self.run_query()
  
          old_connection = connection.connection
          # Simulate errors_occurred.
          connection.errors_occurred = True
          # Simulate request_started (the connection is healthy).
          connection.close_if_unusable_or_obsolete()
          # Persistent connections are enabled.
          self.assertIs(old_connection, connection.connection)
          # No additional health checks after the one in
          # close_if_unusable_or_obsolete() are executed during this "request"
          # when running queries.
          with patch.object(connection, "is_usable", side_effect=AssertionError):
              self.run_query()
  
      @skipUnlessDBFeature("test_db_allows_multiple_connections")
      def test_health_checks_disabled(self):
          self.patch_settings_dict(conn_health_checks=False)
          self.assertIsNone(connection.connection)
          # Newly created connections are considered healthy without performing
          # the health check.
          with patch.object(connection, "is_usable", side_effect=AssertionError):
              self.run_query()
  
          old_connection = connection.connection
          # Simulate request_finished.
          connection.close_if_unusable_or_obsolete()
          # Persistent connections are enabled (connection is not).
          self.assertIs(old_connection, connection.connection)
          # Health checks are not performed.
          with patch.object(connection, "is_usable", side_effect=AssertionError):
              self.run_query()
              # Health check wasn't performed and the connection is unchanged.
              self.assertIs(old_connection, connection.connection)
              self.run_query()
              # The connection is unchanged after the next query either during
              # the current "request".
              self.assertIs(old_connection, connection.connection)
  
      @skipUnlessDBFeature("test_db_allows_multiple_connections")
      def test_set_autocommit_health_checks_enabled(self):
          self.patch_settings_dict(conn_health_checks=True)
          self.assertIsNone(connection.connection)
          # Newly created connections are considered healthy without performing
          # the health check.
          with patch.object(connection, "is_usable", side_effect=AssertionError):
              # Simulate outermost atomic block: changing autocommit for
              # a connection.
              connection.set_autocommit(False)
              self.run_query()
              connection.commit()
              connection.set_autocommit(True)
  
          old_connection = connection.connection
          # Simulate request_finished.
          connection.close_if_unusable_or_obsolete()
          # Persistent connections are enabled.
          self.assertIs(old_connection, connection.connection)
  
          # Simulate connection health check failing.
          with patch.object(
              connection, "is_usable", return_value=False
          ) as mocked_is_usable:
              # Simulate outermost atomic block: changing autocommit for
              # a connection.
              connection.set_autocommit(False)
              new_connection = connection.connection
              self.assertIsNot(new_connection, old_connection)
              # Only one health check per "request" is performed, so a query will
              # carry on even if the health check fails. This query succeeds
              # because the real connection is healthy and only the health check
              # failure is mocked.
              self.run_query()
              connection.commit()
              connection.set_autocommit(True)
              # The connection is unchanged.
              self.assertIs(new_connection, connection.connection)
          self.assertEqual(mocked_is_usable.call_count, 1)
  
          # Simulate request_finished.
          connection.close_if_unusable_or_obsolete()
          # The underlying connection is being reused further with health checks
          # succeeding.
          connection.set_autocommit(False)
          self.run_query()
          connection.commit()
          connection.set_autocommit(True)
          self.assertIs(new_connection, connection.connection)
  
  
  class MultiDatabaseTests(TestCase):
      databases = {"default", "other"}
  
      def test_multi_database_init_connection_state_called_once(self):
          for db in self.databases:
              with self.subTest(database=db):
                  with patch.object(connections[db], "commit", return_value=None):
                      with patch.object(
                          connections[db],
                          "check_database_version_supported",
                      ) as mocked_check_database_version_supported:
                          connections[db].init_connection_state()
                          after_first_calls = len(
                              mocked_check_database_version_supported.mock_calls
                          )
                          connections[db].init_connection_state()
                          self.assertEqual(
                              len(mocked_check_database_version_supported.mock_calls),
                              after_first_calls,
                          )
  