test_zip.py 文件源码

def test_zip(n, n2, kwargs, expected):
    source = Stream()
    source2 = Stream()

    L = es.zip(source, source2, **kwargs).sink_to_list()
    s = list(to_event_model(
        [np.random.random((10, 10)) for _ in range(n)],
        output_info=[('pe1_image', {'dtype': 'array'})]
    ))
    s2 = list(to_event_model(
        [np.random.random((10, 10)) for _ in range(n2)],
        output_info=[('pe1_image', {'dtype': 'array'})]
    ))

    for _ in range(2):
        L.clear()
        for b in s2:
            source2.emit(b)
        for a in s:
            source.emit(a)
        assert_docs = set()
        for name, (l1, l2) in L:
            assert_docs.add(name)
            assert_raises(AssertionError, assert_equal, l1, l2)
        assert set(assert_docs) == {'start', 'descriptor', 'event', 'stop'}