test_space_saving.py 文件源码

def test_topk_invariants():
    s = SpaceSaving(capacity=5, dtype='f8')
    s.update(data_f8)
    for k in [0, 5]:
        top = s.topk(k)
        assert isinstance(top, np.ndarray)
        dtype = np.dtype([('item', 'f8'), ('count', 'i8'), ('error', 'i8')])
        assert top.dtype == dtype
        assert len(top) == k
        assert (np.diff(top['count']) <= 0).all()

        top2 = s.topk(k, astuples=True)
        assert len(top2) == k
        np.testing.assert_equal(top['item'], [i.item for i in top2])
        np.testing.assert_equal(top['count'], [i.count for i in top2])
        np.testing.assert_equal(top['error'], [i.error for i in top2])

    with pytest.raises(ValueError):
        s.topk(-1)