Function test_connection_banner

tests/integration/test_connection_banner.py:214–282

Test that the connection banner is displayed when the websocket drops. Args: connection_banner: AppHarness instance. redis: Redis instance used by the app, or None if not using Redis.

(connection_banner: AppHarness, redis: Redis | None)

@pytest.mark.asyncio
async def test_connection_banner(connection_banner: AppHarness, redis: Redis | None):
    """Test that the connection banner is displayed when the websocket drops.

    Args:
        connection_banner: AppHarness instance.
        redis: Redis instance used by the app, or None if not using Redis.
    """
    assert connection_banner.app_instance is not None
    assert connection_banner.backend is not None
    driver = connection_banner.frontend()

    token = _assert_token(connection_banner, driver)
    AppHarness.expect(lambda: not has_error_modal(driver))

    # Check that the token association was established.
    app_token_manager = connection_banner.token_manager()
    assert token in app_token_manager.token_to_sid
    sid_before = app_token_manager.token_to_sid[token]
    if redis is not None:
        assert isinstance(app_token_manager, RedisTokenManager)
        assert await redis.get(app_token_manager._get_redis_key(token)) == pickle.dumps(
            SocketRecord(instance_id=app_token_manager.instance_id, sid=sid_before)
        )

    delay_button = driver.find_element(By.ID, "delay")
    increment_button = driver.find_element(By.ID, "increment")
    counter_element = driver.find_element(By.ID, "counter")

    # Increment the counter
    increment_button.click()
    assert connection_banner.poll_for_value(counter_element, exp_not_equal="0") == "1"

    # Start a long event before blocking the network, to mark event_processing=true
    delay_button.click()

    with browser_offline(driver):
        # Error modal should now be displayed
        AppHarness.expect(lambda: has_error_modal(driver))

        # The token association should be removed once the websocket closes on the server.
        assert connection_banner._poll_for(
            lambda: token not in app_token_manager.token_to_sid
        )
        if redis is not None:
            assert isinstance(app_token_manager, RedisTokenManager)
            assert await redis.get(app_token_manager._get_redis_key(token)) is None

        # Increment the counter while disconnected
        increment_button.click()
        assert (
            connection_banner.poll_for_value(counter_element, exp_not_equal="0") == "1"
        )

    # Banner should be gone now (network restored on context manager exit)
    AppHarness.expect(lambda: not has_error_modal(driver))

    # After reconnecting, the token association should be re-established.
    app_token_manager = connection_banner.token_manager()