publisher_client = 
topic = 'projects/python-docs-samples-tests/topics/subscription-test-topic-3.8-7caeeb931df34a1f9c660ec6c5a91847'
subscription_async = 'projects/python-docs-samples-tests/subscriptions/subscription-test-subscription-async-3.8-7caeeb931df34a1f9c660ec6c5a91847'
capsys = <_pytest.capture.CaptureFixture object at 0x7fddcc73b430>
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)
    ]

    try:
        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 8 == 3

E            +  where 8 = len([2, 3, 4, 5, 6, 7, ...])
subscriber_test.py:435: AssertionError