2

我正在尝试为我的 django (3.2) 频道消费者编写一个测试用例来实现 websocket 通信。我已经设法让测试用例通过，但现在似乎在尝试清理测试数据库时测试用例拆卸失败。

System check identified no issues (0 silenced).
.
----------------------------------------------------------------------
Ran 1 test in 0.130s

OK
Destroying test database for alias 'default'...

...

psycopg2.errors.ObjectInUse: database "test_myproject" is being accessed by other users
DETAIL:  There is 1 other session using the database.

...
  File ".../python3.8/site-packages/django/core/management/commands/test.py", line 55, in handle
    failures = test_runner.run_tests(test_labels)
  File ".../python3.8/site-packages/django/test/runner.py", line 736, in run_tests
    self.teardown_databases(old_config)
  File ".../python3.8/site-packages/django/test/runner.py", line 674, in teardown_databases
    _teardown_databases(
  File ".../python3.8/site-packages/django/test/utils.py", line 313, in teardown_databases
    connection.creation.destroy_test_db(old_name, verbosity, keepdb)
  File ".../python3.8/site-packages/django/db/backends/base/creation.py", line 282, in destroy_test_db
    self._destroy_test_db(test_database_name, verbosity)
  File ".../python3.8/site-packages/django/db/backends/base/creation.py", line 298, in _destroy_test_db
    cursor.execute("DROP DATABASE %s"
  File ".../python3.8/contextlib.py", line 162, in __exit__
    raise RuntimeError("generator didn't stop after throw()")
RuntimeError: generator didn't stop after throw()

消费者是使用异步编写的，看来我在异步如何与测试用例集成方面遇到了麻烦。我还需要做什么才能让测试用例正常运行。

我创建并配置了一个夹具来将数据加载到数据库中，并怀疑这与那里的测试用例数据库交互有关......

我尝试使用标准 django“TestCase”代替“TransactionTestCase”，但显然配置的夹具没有按预期加载到数据库中。

class ConsumerTestCase(TransactionTestCase):
    fixtures = ("basic-data", )

    async def test_alertconsumer__receive__valid(self):
        # defined in fixture
        username = "test-username"
        alert_id = 1

        url = "/alerts/ws/"
        communicator = WebsocketCommunicator(AlertConsumer.as_asgi(), url)
        connected, subprotocol = await communicator.connect()
        assert connected

        message = {"username": "me", "alert_id": 1 }
        await communicator.send_to(text_data=json.dumps(message))
        response = await communicator.receive_from()
        expected = {
            "message": "ACCEPTED"
        }
        self.assertDictEqual(actual, expected)
4

1 回答 1

1

@希娜

不知道我在哪里找到了这个解决方法，但我最终创建了一个自定义测试运行器来执行数据库的拆卸。

from django.conf import settings
from django.db import connection
from django.test.runner import DiscoverRunner


class DatabaseConnectionCleanupTestRunner(DiscoverRunner):
    def teardown_databases(self, old_config, **kwargs):
        # Django's ORM isn't cleaning up the connections after it's done with them.
        # The query below kills all database connections before dropping the database.
        with connection.cursor() as cursor:
            cursor.execute(
                f"""SELECT
                pg_terminate_backend(pid) FROM pg_stat_activity WHERE
                pid <> pg_backend_pid() AND
                pg_stat_activity.datname =
                  '{settings.DATABASES["default"]["NAME"]}';"""
            )
            print(f"Killed {len(cursor.fetchall())} stale connections.")
        super().teardown_databases(old_config, **kwargs)

settings.py然后在文件中定义了testrunner 。

（我创建了一个“commons”应用程序并将其放置在DatabaseConnectionCleanupTestRunner那里）

TEST_RUNNER = "commons.tests.tests.DatabaseConnectionCleanupTestRunner"
于 2022-01-11T12:01:55.313 回答