Function test_multi_upload

tests/test_wal_transfer.py:105–118

Model a case with upload concurrency.

()

def test_multi_upload():
    """Model a case with upload concurrency."""
    group = worker.WalTransferGroup(FakeWalUploader())
    segments = list(prepare_multi_upload_segments())

    # "Start" fake uploads
    for seg in segments:
        group.start(seg)

    group.join()

    # Check invariants on the non-explicit segments.
    for seg in segments:
        assert success(seg)


def test_simple_fail():