时间戳

首页 > 解决方案 > 在 Django 中进行硒测试后，数据库仍在使用中

python - 在 Django 中进行硒测试后，数据库仍在使用中

问题描述

我有一个 Django 项目，我开始在其中编写 Selenium 测试。第一个看起来像这样：

from django.contrib.staticfiles.testing import StaticLiveServerTestCase
from selenium import webdriver
from selenium.webdriver.common.keys import Keys

from core.models import User
from example import settings

BACH_EMAIL = "johann.sebastian.bach@classics.com"
PASSWORD = "password"


class TestImportCRMData(StaticLiveServerTestCase):
    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        cls.webdriver = webdriver.Chrome()
        cls.webdriver.implicitly_wait(10)

    @classmethod
    def tearDownClass(cls):
        cls.webdriver.close()
        cls.webdriver.quit()
        super().tearDownClass()

    def setUp(self):
        self.admin = User.objects.create_superuser(email=BACH_EMAIL, password=PASSWORD)

    def test_admin_tool(self):
        self.webdriver.get(f"http://{settings.ADMIN_HOST}:{self.server_thread.port}/admin")

        self.webdriver.find_element_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)
        self.webdriver.find_element_by_link_text("Users").click()

当我运行它时，测试通过但仍然以这个错误结束：

Traceback (most recent call last):
  File "C:\Users\pupeno\Documents\Eligible\code\example\venv\lib\site-packages\django\db\backends\utils.py", line 83, in _execute
    return self.cursor.execute(sql)
psycopg2.OperationalError: database "test_example" is being accessed by other users
DETAIL:  There is 1 other session using the database.


The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "C:\Program Files\JetBrains\PyCharm 2018.2.4\helpers\pycharm\django_test_manage.py", line 168, in <module>
    utility.execute()
  File "C:\Program Files\JetBrains\PyCharm 2018.2.4\helpers\pycharm\django_test_manage.py", line 142, in execute
    _create_command().run_from_argv(self.argv)
  File "C:\Users\pupeno\Documents\Eligible\code\example\venv\lib\site-packages\django\core\management\commands\test.py", line 26, in run_from_argv
    super().run_from_argv(argv)
  File "C:\Users\pupeno\Documents\Eligible\code\example\venv\lib\site-packages\django\core\management\base.py", line 316, in run_from_argv
    self.execute(*args, **cmd_options)
  File "C:\Users\pupeno\Documents\Eligible\code\example\venv\lib\site-packages\django\core\management\base.py", line 353, in execute
    output = self.handle(*args, **options)
  File "C:\Program Files\JetBrains\PyCharm 2018.2.4\helpers\pycharm\django_test_manage.py", line 104, in handle
    failures = TestRunner(test_labels, **options)
  File "C:\Program Files\JetBrains\PyCharm 2018.2.4\helpers\pycharm\django_test_runner.py", line 255, in run_tests
    extra_tests=extra_tests, **options)
  File "C:\Program Files\JetBrains\PyCharm 2018.2.4\helpers\pycharm\django_test_runner.py", line 156, in run_tests
    return super(DjangoTeamcityTestRunner, self).run_tests(test_labels, extra_tests, **kwargs)
  File "C:\Users\pupeno\Documents\Eligible\code\example\venv\lib\site-packages\django\test\runner.py", line 607, in run_tests
    self.teardown_databases(old_config)
  File "C:\Users\pupeno\Documents\Eligible\code\example\venv\lib\site-packages\django\test\runner.py", line 580, in teardown_databases
    keepdb=self.keepdb,
  File "C:\Users\pupeno\Documents\Eligible\code\example\venv\lib\site-packages\django\test\utils.py", line 297, in teardown_databases
    connection.creation.destroy_test_db(old_name, verbosity, keepdb)
  File "C:\Users\pupeno\Documents\Eligible\code\example\venv\lib\site-packages\django\db\backends\base\creation.py", line 257, in destroy_test_db
    self._destroy_test_db(test_database_name, verbosity)
  File "C:\Users\pupeno\Documents\Eligible\code\example\venv\lib\site-packages\django\db\backends\base\creation.py", line 274, in _destroy_test_db
    % self.connection.ops.quote_name(test_database_name))
  File "C:\Users\pupeno\Documents\Eligible\code\example\venv\lib\site-packages\django\db\backends\utils.py", line 68, in execute
    return self._execute_with_wrappers(sql, params, many=False, executor=self._execute)
  File "C:\Users\pupeno\Documents\Eligible\code\example\venv\lib\site-packages\django\db\backends\utils.py", line 77, in _execute_with_wrappers
    return executor(sql, params, many, context)
  File "C:\Users\pupeno\Documents\Eligible\code\example\venv\lib\site-packages\django\db\backends\utils.py", line 85, in _execute
    return self.cursor.execute(sql, params)
  File "C:\Users\pupeno\Documents\Eligible\code\example\venv\lib\site-packages\django\db\utils.py", line 89, in __exit__
    raise dj_exc_value.with_traceback(traceback) from exc_value
  File "C:\Users\pupeno\Documents\Eligible\code\example\venv\lib\site-packages\django\db\backends\utils.py", line 83, in _execute
    return self.cursor.execute(sql)
django.db.utils.OperationalError: database "test_example" is being accessed by other users
DETAIL:  There is 1 other session using the database.

问题当然是下一次运行测试，数据库仍然存在，所以在没有确认删除数据库的情况下，测试不会运行。

如果我注释掉最后一行：

self.webdriver.find_element_by_link_text("Users").click()

那么我没有得到这个错误。我猜只是因为没有建立数据库连接。有时是另外 1 个会话，有时最多 4 个。在 4 个会话的其中一种情况下，这些是正在运行的会话：

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

我一直试图找到这个问题的最小可重现示例，但到目前为止我还没有成功。

有什么想法可能导致此问题或如何了解更多有关问题的信息？

标签： pythondjangoseleniumselenium-webdriverliveservertestcase

解决方案

此错误消息...

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

...意味着存在使用数据库的现有会话，并且新会话无法访问数据库。

有关Django 版本、数据库 类型和版本以及 Selenium、ChromeDriver和Chrome版本的更多信息将帮助我们以更好的方式调试此问题。

但是，您需要从Selenium的角度处理以下几件事：

  • 当您在下一次运行测试时启动新会话时，您需要删除该行代码cls.webdriver.close()，因为下一行代码cls.webdriver.quit()足以终止现有会话。根据最佳实践，您应该quit()在tearDown() {}. 通过发送带有{"flags":["eForceQuit"]}的"quit"命令调用当前浏览会话，最后在/shutdown上发送GET请求。下面是一个例子：quit() DELETE 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

  • 所以在调用quit()方法时，Web Browser会话和WebDriver实例被完全杀死。因此，您不必包含任何额外的步骤，这将是开销。

您可以在 Selenium 中找到详细讨论：如何在不调用 driver.quit() 的情况下停止影响 PC 内存的 geckodriver 进程？

  • 当前构建的Web 应用程序正在逐渐向动态呈现的HTML DOM发展，因此要与DOM Tree中的元素进行交互，ImplicitWait不再那么有效，您需要使用WebDriverWait来代替。此时值得一提的是，混淆隐式等待和显式等待会导致无法预测的等待时间

  • 所以你需要：

    • 删除implicitly_wait(10)：

      cls.webdriver.implicitly_wait(10)

    • 在与元素交互时引入 WebDriverWait：

      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()

现在，根据LiveServerTestCase 未关闭的持久连接，防止删除测试数据库的讨论，在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, after 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)这基本上与在单独的线程中启动实时 http 服务器的作用相同，TransactionTestCase以便测试可以使用Selenium使用的另一个测试框架，而不是内置的虚拟客户端：

    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()

  • 补充tests/servers/tests.py：

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

解决方案

脚步：

  • 确保您已更新到Django包的最新发布版本。
  • 确保您使用的是最新发布的selenium v​​3.141.0版本。
  • 确保您使用的是最新发布的Chrome v76和ChromeDriver 76.0版本。

奥特罗

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

推荐阅读