100 lines
1.9 KiB
Python
100 lines
1.9 KiB
Python
import math
|
|
import random
|
|
from hyperloglog import HyperLogLog
|
|
|
|
|
|
def test_empty():
|
|
h = HyperLogLog(p=10)
|
|
assert h.count() == 0.0
|
|
|
|
|
|
def test_exact_small():
|
|
h = HyperLogLog(p=12)
|
|
n = 1000
|
|
for i in range(n):
|
|
h.add(i)
|
|
est = h.count()
|
|
assert abs(est - n) / n < 0.05, (est, n)
|
|
|
|
|
|
def test_large():
|
|
h = HyperLogLog(p=14)
|
|
n = 100_000
|
|
for i in range(n):
|
|
h.add(i)
|
|
est = h.count()
|
|
rel = abs(est - n) / n
|
|
assert rel < 0.02, (est, n, rel)
|
|
|
|
|
|
def test_duplicates():
|
|
h = HyperLogLog(p=12)
|
|
for i in range(5000):
|
|
h.add(i % 1000)
|
|
est = h.count()
|
|
assert abs(est - 1000) / 1000 < 0.05, est
|
|
|
|
|
|
def test_strings():
|
|
h = HyperLogLog(p=12)
|
|
words = [f"item-{i}" for i in range(2000)]
|
|
for w in words:
|
|
h.add(w)
|
|
est = h.count()
|
|
assert abs(est - 2000) / 2000 < 0.05, est
|
|
|
|
|
|
def test_merge():
|
|
a = HyperLogLog(p=12)
|
|
b = HyperLogLog(p=12)
|
|
for i in range(0, 1000, 2):
|
|
a.add(i)
|
|
for i in range(1, 1000, 2):
|
|
b.add(i)
|
|
m = a.merge(b)
|
|
est = m.count()
|
|
assert abs(est - 1000) / 1000 < 0.05, est
|
|
|
|
|
|
def test_merge_precision_mismatch():
|
|
try:
|
|
HyperLogLog(p=10).merge(HyperLogLog(p=12))
|
|
except ValueError:
|
|
return
|
|
raise AssertionError("expected ValueError")
|
|
|
|
|
|
def test_extend():
|
|
h = HyperLogLog(p=10)
|
|
h.extend(range(500))
|
|
assert abs(h.count() - 500) / 500 < 0.06
|
|
|
|
|
|
def test_len():
|
|
h = HyperLogLog(p=10)
|
|
for i in range(300):
|
|
h.add(i)
|
|
assert isinstance(len(h), int)
|
|
|
|
|
|
def test_determinism():
|
|
a = HyperLogLog(p=12)
|
|
b = HyperLogLog(p=12)
|
|
for i in range(1000):
|
|
a.add(i); b.add(i)
|
|
assert a.count() == b.count()
|
|
|
|
|
|
def test_invalid_p():
|
|
for p in (3, 17):
|
|
try:
|
|
HyperLogLog(p=p)
|
|
except ValueError:
|
|
continue
|
|
raise AssertionError("expected ValueError")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import pytest
|
|
raise SystemExit(pytest.main([__file__, "-q"]))
|