使用 pytest 测试 flask-socketio 服务器发出的事件时出错

Question

我正在为最近部署的网络应用构建测试套件。它是使用 flask-socketio 构建的，并使用 pytest 作为测试套件。

这里的大问题是缺少使用 flask-socketio 进行测试的文档。我发现了一些包含测试套件的项目： https://github.com/miguelgrinberg/Flask-SocketIO/blob/main/test_socketio.py https://github.com/miguelgrinberg/flack/blob/master/tests/tests.py

但其中 none 对服务器对消息的响应实施测试。

就我而言，我的服务器中有以下侦听器：

@socketio.on('getBettors', namespace='/coin')
def get_current_bettors_coin():
    """
    functionshould send current tickets to the client to build
    """
    game_name = "coinv1"
    coin_game = Game.objects(game_name=game_name)[0]
    current_time = get_utc_now_func()


    current_game = GameSeed.objects(
        game_id=coin_game.pk,
        start_time__lte=current_time,
        end_time__gt=current_time,
    )[0]
    
    current_game_tickets = GameTicket.objects(
        game_id=current_game.pk
    )

    list_bets = []
    for ticket in current_game_tickets:
        ticket_dict = {
            "user": User.objects(id=ticket.owner_id.pk)[0].username,
            "choice": str(ticket.user_choice),
            "value": str(ticket.bet_value),
        }
        list_bets.append(ticket_dict)
    emit('getBettors', {"bets": list_bets}, broadcast=True)

基本上听众搜索数据库以获取特定游戏中的所有当前投注者（这是一个投注游戏）

但是我无法测试发出的结果是否是预期的实际结果，因为在 flask-socketio 测试客户端中没有这样的论据：

conftest.py


@pytest.fixture(scope='module')
def create_flask_app():
    #drop all records in testDatabase before strting new test module
    db = connect(host=os.environ["MONGODB_SETTINGS_TEST"], alias="testConnect")
    for collection in db["testDatabase"].list_collection_names():
        db["testDatabase"].drop_collection(collection)
    db.close()
    
    # Create a test client using the Flask application configured for testing
    flask_app = create_app()
    return flask_app

@pytest.fixture(scope='module')
def test_client(create_flask_app):
    """
    Establish a test client for use within each test module
    """
    with create_flask_app.test_client() as testing_client:
        # Establish an application context
        with create_flask_app.app_context():
            yield testing_client  # this is where the testing happens!

@pytest.fixture(scope='module')
def data_test_coin_toss_game():
    """
    Populate DB with mock data for coin toss game tests
    """
    #drop all records in testDatabase before strting new test class
    db = connect(host=os.environ["MONGODB_SETTINGS_TEST"], alias="data_test_coin_toss_game")

    mock_user_1 = User(
        username = "data_test_coin_toss_game",
        email = "data_test_coin_toss_game@gmail.com",
        password = "sdgibgsdgsdg",
        user_type = 1,
    )
    mock_user_1.create()

    first_deposit = Transaction(
        owner = mock_user_1,
        transaction_type = "deposit funds",
        currency_code = "MockCoin",
        value = Decimal("1000.50"),
        reference = "firstdeposit",
        datetime = datetime.datetime.utcnow(),
    )
    first_deposit.save()

    coin_game = Game(
        game_name = "coinv1",
        client_fixed_seed = hashlib.sha256("a random seed for the client".encode()).hexdigest(),
    )
    coin_game.save()

    base_time = get_utc_now_func()
    game_seed = GameSeed(
        game_id = coin_game,
        nonce = 4,
        seed_crypto = hashlib.sha256("wow a secret seed".encode()).hexdigest(),
        seed_client = hashlib.sha256("a random seed for the client".encode()).hexdigest(),
        result = Decimal("0"),
        start_time = base_time,
        lock_time = base_time + datetime.timedelta(days=1),
        reveal_time = base_time + datetime.timedelta(days=2),
        end_time = base_time + datetime.timedelta(days=3),
    )
    game_seed.save()

    db.close()

测试函数：

def test_get_bettors_websocket(create_flask_app, test_client, data_test_coin_toss_game):
    """
    GIVEN a endpoint to retrieve all current bettors in a game
    WHEN a user communicates to that endpoint
    THEN check that the correct information is being sent out
    """
    client = socketio.test_client(create_flask_app)
    assert client.is_connected()

    received_return = False
    @client.on('getBettors', namespace='/coin')
    def process_request():
        print("AAA")
        global received_return
        received_return = True

    client.emit('getBettors', namespace='/coin')

    assert received_return

此时我只是想找出一种方法来检查从服务器发出的消息，而没有实际测试业务逻辑。

测试产生以下错误：

FAILED tests/functional/test_coin_game.py::test_get_bettors_websocket - AttributeError: 'SocketIOTestClient' object has no attribute 'on'

显然测试客户端不允许注册“on”监听器。有谁知道如何测试这样的应用程序？

***理论上我可以将服务器端侦听器的所有业务逻辑封装在一个不同的函数中，然后可以对其进行测试并让侦听器调用该函数。但这看起来草率、容易出错，而且对于其他事件侦听器，客户端消息的内容用作查询的参数，因此我需要按照我上面建议的方式进行测试

Answer 1

Flask-SocketIO 测试客户端不是真正的客户端，您不能注册事件处理程序。测试客户端记录服务器发出的任何内容，并允许您的测试对其进行检查。例如，Flask-SocketIO 自己的单元测试就是这样做的 here。

        received = client.get_received()
        self.assertEqual(len(received), 3)
        self.assertEqual(received[0]['args'], 'connected')
        self.assertEqual(received[1]['args'], '{}')
        self.assertEqual(received[2]['args'], '{}')

使用 pytest 测试 flask-socketio 服务器发出的事件时出错

Error testing flask-socketio server emitted events with pytest

python

flask

pytest

flask-socketio