publisher_client = 
topic = 'projects/python-docs-samples-tests/topics/subscription-test-topic-3.8-7faada6fd3a64b79b055298584530098'
subscription_async = 'projects/python-docs-samples-tests/subscriptions/subscription-test-subscription-async-3.8-7faada6fd3a64b79b055298584530098'
capsys = <_pytest.capture.CaptureFixture object at 0x7f142c3a9070>
def test_receive_with_blocking_shutdown(
    publisher_client, topic, subscription_async, capsys
):
    _publish_messages(publisher_client, topic, message_num=3)

    subscriber.receive_messages_with_blocking_shutdown(
        PROJECT_ID, SUBSCRIPTION_ASYNC, timeout=5.0
    )

    out, _ = capsys.readouterr()
    out_lines = out.splitlines()

    msg_received_lines = [
        i for i, line in enumerate(out_lines)
        if re.search(r".*received.*message.*", line, flags=re.IGNORECASE)
    ]
    msg_done_lines = [
        i for i, line in enumerate(out_lines)
        if re.search(r".*done processing.*message.*", line, flags=re.IGNORECASE)
    ]
    stream_canceled_lines = [
        i for i, line in enumerate(out_lines)
        if re.search(r".*streaming pull future canceled.*", line, flags=re.IGNORECASE)
    ]
    shutdown_done_waiting_lines = [
        i for i, line in enumerate(out_lines)
        if re.search(r".*done waiting.*stream shutdown.*", line, flags=re.IGNORECASE)
    ]

    assert "Listening" in out
    assert subscription_async in out

    assert len(stream_canceled_lines) == 1
    assert len(shutdown_done_waiting_lines) == 1


  assert len(msg_received_lines) == 3


E       assert 4 == 3

E        +  where 4 = len([2, 3, 4, 5])
subscriber_test.py:433: AssertionError