在Django进行selenium测试后，数据库仍在使用中

Question

我有一个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 = "[e-mail]"
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

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

有什么想法是什么导致了这种情况，或者如何找到更多关于问题的信息？

Answer 1

检查活动连接以了解问题的原因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()

Answer 2

很久以前就有一个问题被报道过（bug#22424）。

您需要确保的一件事是CONN\u MAX\u AGE设置为0，而不是None

另外，你也可以在你的拆卸中使用下面这样的东西

from django.db import connections
from django.db.utils import OperationalError

@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

Answer 3

此错误消息。。。

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。网络驱动程序。关闭（）作为下一行代码cls。网络驱动程序。quit（）将足以终止现有会话。根据最佳实践，您应该在tearDown（）{}中调用quit（）方法。调用quit（）DELETE通过发送带有{“flags”：[“eForceQuit”]}的“quit”命令来删除当前浏览会话，最后在/shutdownendpoint上发送GET请求。下面是一个例子：

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浏览器会话和WebDriver实例将被完全杀死。因此，您不必合并任何额外的步骤，这将是一项开销。

您可以在Selenium中找到详细的讨论：如何停止gecko驱动程序进程影响PC内存，而不调用driver.quit（）？

>

>

隐式删除等待（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\u MAX\u 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_活动显示处于空闲状态的延迟连接（这是您案例中的上一个测试创建的连接）。因此，很明显，当Thread终止且支撑针处于以下位置时，空闲连接未闭合：

>

LiveServerThread（线程。线程），它控制在测试运行时运行实时超文本传输协议服务器的线程：

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服务器，以便测试可以使用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()

解决方案是仅关闭非重写连接，并从该请求/提交中合并。这些变化是：

>

在django/test/testcases中。py添加：

finally:
    connections.close_all() 

添加一个新文件测试/服务器/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.py中删除：

finally:
    TestCase.tearDownClass()

在测试/服务器/tests.py中添加：

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

步骤：

• 确保您已更新至Django软件包的最新发布版本
• 确保您使用的是最新发布的selenium v3版本。141.0.
• 确保您使用的是最新版本的Chrome v76和ChromeDriver 76.0

在通过Selenium和Python Django执行LiveServerTestCase时，您可以在django.db.utils.完整性错误中找到类似的讨论：FOREIGN KEY约束失败