Django硒测试后仍在使用的数据库

此错误消息。。。

django.db.utils.OperationalError: database "test_example" is being accessed by other users
DETAIL:  There is 1 other session using the database.

……暗示使用数据库存在现有会话,而新会话不能访问数据库。

关于 Django公司 版本 , 数据库 类型 和 版本 随着 硒 , 铬河 和 铬 版本可以帮助我们更好地调试这个问题。

但是,你需要从 硒 观点如下:

• 在下一次运行测试时启动新会话时,需要删除代码行 cls.webdriver.close() 作为下一行代码 cls.webdriver.quit() 将足以终止现有会话。根据最佳实践,您应该调用 quit() 方法中的 tearDown() {} . 调用 退出()) DELETE 通过发送当前浏览会话 “退出” 命令 {“flags”:[“eForceQuit”]} 最后把 得到 请求打开 /关闭 EndPoint . 下面是一个例子:

  1503397488598   webdriver::server   DEBUG   -> DELETE /session/8e457516-3335-4d3b-9140-53fb52aa8b74 
  1503397488607   geckodriver::marionette TRACE   -> 37:[0,4,"quit",{"flags":["eForceQuit"]}]
  1503397488821   webdriver::server   DEBUG   -> GET /shutdown
  

• 所以调用 退出()) 方法 Web Browser 会议和 WebDriver 实例被完全杀死。因此,您不必包含任何额外的步骤,这将是一个开销。

您可以在中找到详细的讨论 Selenium : How to stop geckodriver process impacting PC memory, without calling driver.quit()?

• 当前生成 Web应用程序 正在逐渐走向动态渲染 HTML DOM 因此要与 DOM Tree , ImplicitWait 不再有效,你需要使用 WebDriverWait 相反。在这一点上,值得一提的是,混淆。 Implicit Wait 和 Explicit Wait 可能导致 unpredictable wait times

• 所以你需要:

  • 移除 implicitly_wait(10) :

    cls.webdriver.implicitly_wait(10)
    

  • 诱导 网络驱动器 与元素交互时:

    WebDriverWait(driver, 20).until(EC.element_to_be_clickable((By.ID, "id_username"))).send_keys(BACH_EMAIL)
    self.webdriver.find_element_by_id("id_password").send_keys(PASSWORD)
    self.webdriver.find_element_by_id("id_password").send_keys(Keys.RETURN)
    WebDriverWait(driver, 20).until(EC.element_to_be_clickable((By.LINK_TEXT, "Users"))).click()
    

现在,根据讨论 Persistent connections not closed by LiveServerTestCase, preventing dropping test databases 这一问题在 Djangov1.6版 固定的。主要问题是:

每当PostgreSQL连接被标记为持久连接时( CONN_MAX_AGE = None )执行LiveServerTestCase时,服务器线程的连接从未关闭,导致无法删除测试数据库。

这正是你看到的原因:

select * from pg_stat_activity where datname = 'test_example';

100123  test_example    29892   16393   pupeno  ""  ::1     61967   2018-11-15 17:28:19.552431      2018-11-15 17:28:19.562398  2018-11-15 17:28:19.564623          idle            SELECT "core_user"."id", "core_user"."password", "core_user"."last_login", "core_user"."is_superuser", "core_user"."email", "core_user"."is_staff", "core_user"."is_active", "core_user"."date_joined" FROM "core_user" WHERE "core_user"."id" = 1
100123  test_example    33028   16393   pupeno  ""  ::1     61930   2018-11-15 17:28:18.792466      2018-11-15 17:28:18.843383  2018-11-15 17:28:18.851828          idle            SELECT "django_admin_log"."id", "django_admin_log"."action_time", "django_admin_log"."user_id", "django_admin_log"."content_type_id", "django_admin_log"."object_id", "django_admin_log"."object_repr", "django_admin_log"."action_flag", "django_admin_log"."change_message", "core_user"."id", "core_user"."password", "core_user"."last_login", "core_user"."is_superuser", "core_user"."email", "core_user"."is_staff", "core_user"."is_active", "core_user"."date_joined", "django_content_type"."id", "django_content_type"."app_label", "django_content_type"."model" FROM "django_admin_log" INNER JOIN "core_user" ON ("django_admin_log"."user_id" = "core_user"."id") LEFT OUTER JOIN "django_content_type" ON ("django_admin_log"."content_type_id" = "django_content_type"."id") WHERE "django_admin_log"."user_id" = 1 ORDER BY "django_admin_log"."action_time" DESC  LIMIT 10
100123  test_example    14128   16393   pupeno  ""  ::1     61988   2018-11-15 17:28:19.767225      2018-11-15 17:28:19.776150  2018-11-15 17:28:19.776479          idle            SELECT "core_firm"."id", "core_firm"."name", "core_firm"."host_name" FROM "core_firm" WHERE "core_firm"."id" = 1
100123  test_example    9604    16393   pupeno  ""  ::1     61960   2018-11-15 17:28:19.469197      2018-11-15 17:28:19.478775  2018-11-15 17:28:19.478788          idle            COMMIT

据进一步观察,即使 CONN_MAX_AGE=None ,之后 LiveServerTestCase.tearDownClass() ,查询PostgreSQL的 pg_stat_activity 在状态中显示延迟连接 虚度 (这是上一个测试在您的案例中创建的连接)。所以很明显,当线终止时,当平卧针位于:

• LiveServerThread(threading.Thread) 在测试运行时控制运行实时http服务器的线程:

  class LiveServerThread(threading.Thread):
  
      def __init__(self, host, static_handler, connections_override=None):
          self.host = host
          self.port = None
          self.is_ready = threading.Event()
          self.error = None
          self.static_handler = static_handler
          self.connections_override = connections_override
          super(LiveServerThread, self).__init__()
  
      def run(self):
          """
          Sets up the live server and databases, and then loops over handling
          http requests.
          """
          if self.connections_override:
              # Override this thread's database connections with the ones
              # provided by the main thread.
              for alias, conn in self.connections_override.items():
                  connections[alias] = conn
          try:
              # Create the handler for serving static and media files
              handler = self.static_handler(_MediaFilesHandler(WSGIHandler()))
              self.httpd = self._create_server(0)
              self.port = self.httpd.server_address[1]
              self.httpd.set_app(handler)
              self.is_ready.set()
              self.httpd.serve_forever()
          except Exception as e:
              self.error = e
              self.is_ready.set()
  
      def _create_server(self, port):
          return WSGIServer((self.host, port), QuietWSGIRequestHandler, allow_reuse_address=False)
  
      def terminate(self):
          if hasattr(self, 'httpd'):
              # Stop the WSGI server
              self.httpd.shutdown()
              self.httpd.server_close()
  

• LiveServerTestCase(TransactionTestCase) 基本上和 TransactionTestCase 同时在一个单独的线程中启动一个实时http服务器,以便测试可以使用另一个测试框架 硒 ,而不是内置的虚拟客户端:

  class LiveServerTestCase(TransactionTestCase):
  
      host = 'localhost'
      static_handler = _StaticFilesHandler
  
      @classproperty
      def live_server_url(cls):
          return 'http://%s:%s' % (cls.host, cls.server_thread.port)
  
      @classmethod
      def setUpClass(cls):
          super(LiveServerTestCase, cls).setUpClass()
          connections_override = {}
          for conn in connections.all():
              # If using in-memory sqlite databases, pass the connections to
              # the server thread.
              if conn.vendor == 'sqlite' and conn.is_in_memory_db(conn.settings_dict['NAME']):
                  # Explicitly enable thread-shareability for this connection
                  conn.allow_thread_sharing = True
                  connections_override[conn.alias] = conn
  
      cls._live_server_modified_settings = modify_settings(
          ALLOWED_HOSTS={'append': cls.host},
      )
      cls._live_server_modified_settings.enable()
      cls.server_thread = cls._create_server_thread(connections_override)
      cls.server_thread.daemon = True
      cls.server_thread.start()
  
      # Wait for the live server to be ready
      cls.server_thread.is_ready.wait()
      if cls.server_thread.error:
          # Clean up behind ourselves, since tearDownClass won't get called in
          # case of errors.
          cls._tearDownClassInternal()
          raise cls.server_thread.error
  
      @classmethod
      def _create_server_thread(cls, connections_override):
          return LiveServerThread(
              cls.host,
              cls.static_handler,
              connections_override=connections_override,
      )
  
      @classmethod
      def _tearDownClassInternal(cls):
          # There may not be a 'server_thread' attribute if setUpClass() for some
          # reasons has raised an exception.
          if hasattr(cls, 'server_thread'):
              # Terminate the live server's thread
              cls.server_thread.terminate()
              cls.server_thread.join()
  
          # Restore sqlite in-memory database connections' non-shareability
          for conn in connections.all():
              if conn.vendor == 'sqlite' and conn.is_in_memory_db(conn.settings_dict['NAME']):
                  conn.allow_thread_sharing = False
  
      @classmethod
      def tearDownClass(cls):
          cls._tearDownClassInternal()
          cls._live_server_modified_settings.disable()
          super(LiveServerTestCase, cls).tearDownClass()
  

解决办法是 仅关闭非重写的连接 并由此而成立 pull request / commit . 变化如下:

• 在 django/test/testcases.py 添加:

  finally:
      connections.close_all() 
  

• 添加新文件 tests/servers/test_liveserverthread.py :

  from django.db import DEFAULT_DB_ALIAS, connections
  from django.test import LiveServerTestCase, TestCase
  
  
  class LiveServerThreadTest(TestCase):
  
      def run_live_server_thread(self, connections_override=None):
          thread = LiveServerTestCase._create_server_thread(connections_override)
          thread.daemon = True
          thread.start()
          thread.is_ready.wait()
          thread.terminate()
  
      def test_closes_connections(self):
          conn = connections[DEFAULT_DB_ALIAS]
          if conn.vendor == 'sqlite' and conn.is_in_memory_db():
              self.skipTest("the sqlite backend's close() method is a no-op when using an in-memory database")
          # Pass a connection to the thread to check they are being closed.
          connections_override = {DEFAULT_DB_ALIAS: conn}
  
          saved_sharing = conn.allow_thread_sharing
      try:
          conn.allow_thread_sharing = True
          self.assertTrue(conn.is_usable())
          self.run_live_server_thread(connections_override)
          self.assertFalse(conn.is_usable())
      finally:
          conn.allow_thread_sharing = saved_sharing
  

• 在 tests/servers/tests.py 删除:

  finally:
      TestCase.tearDownClass()
  

• 在 测试/服务器/测试.py 添加:

  finally:
      if hasattr(TestCase, 'server_thread'):
          TestCase.server_thread.terminate()
  

解决方案

步骤:

• 确保已更新到 Django公司 包裹。
• 确保使用的是 硒v3.141.0 .
• 确保使用的是 铬v76 和 铬驱动76.0 .

奥特罗

您可以在中找到类似的讨论 django.db.utils.IntegrityError: FOREIGN KEY constraint failed while executing LiveServerTestCases through Selenium and Python Django

很久以前也有报道说

https://code.djangoproject.com/ticket/22414

你要确定的是 CONN_MAX_AGE 设置为 0 而不是 None

另外,你也可以在你的泪道里使用下面的东西

@classmethod
def tearDownClass(cls):
    # Workaround for https://code.djangoproject.com/ticket/22414
    # Persistent connections not closed by LiveServerTestCase, preventing dropping test databases
    # https://github.com/cjerdonek/django/commit/b07fbca02688a0f8eb159f0dde132e7498aa40cc
    def close_sessions(conn):
        close_sessions_query = """
            SELECT pg_terminate_backend(pg_stat_activity.pid) FROM pg_stat_activity WHERE
                datname = current_database() AND
                pid <> pg_backend_pid();
        """
        with conn.cursor() as cursor:
            try:
                cursor.execute(close_sessions_query)
            except OperationalError:
                # We get kicked out after closing.
                pass

    for alias in connections:
        connections[alias].close()
        close_sessions(connections[alias])

    print "Forcefully closed database connections."

上面的代码来自下面的URL

https://github.com/cga-harvard/Hypermap-Registry/blob/cd4efad61f18194ddab2c662aa431aa21dec03f4/hypermap/tests/test_csw.py

检查活动连接以了解问题的原因 select * from pg_stat_activity;

您可以禁用扩展:

    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        options = webdriver.chrome.options.Options()
        options.add_argument("--disable-extensions") 
        cls.webdriver = webdriver.Chrome(chrome_options=options)
        cls.webdriver.implicitly_wait(10)

然后泪流满面:

    @classmethod
    def tearDownClass(cls):
        cls.webdriver.stop_client()
        cls.webdriver.quit()
        connections.close_all()
        super().tearDownClass()