/
test_exponentiallydecayingsample.py
69 lines (52 loc) · 2.35 KB
/
test_exponentiallydecayingsample.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
import random
from unittest import TestCase
from txstatsd.stats.exponentiallydecayingsample import (
ExponentiallyDecayingSample)
class TestExponentiallyDecayingSample(TestCase):
def test_100_out_of_1000_elements(self):
population = [i for i in range(0, 100)]
sample = ExponentiallyDecayingSample(1000, 0.99)
for i in population:
sample.update(i)
self.assertEqual(sample.size(), 100, 'Should have 100 elements')
self.assertEqual(len(sample.get_values()), 100,
'Should have 100 elements')
self.assertEqual(
len(set(sample.get_values()).difference(set(population))), 0,
'Should only have elements from the population')
def test_100_out_of_10_elements(self):
population = [i for i in range(0, 10)]
sample = ExponentiallyDecayingSample(100, 0.99)
for i in population:
sample.update(i)
self.assertEqual(sample.size(), 10)
self.assertEqual(len(sample.get_values()), 10,
'Should have 10 elements')
self.assertEqual(
len(set(sample.get_values()).difference(set(population))), 0,
'Should only have elements from the population')
def test_heavily_biased_100_out_of_1000_elements(self):
population = [i for i in range(0, 100)]
sample = ExponentiallyDecayingSample(1000, 0.01)
for i in population:
sample.update(i)
self.assertEqual(sample.size(), 100, 'Should have 100 elements')
self.assertEqual(len(sample.get_values()), 100,
'Should have 100 elements')
self.assertEqual(
len(set(sample.get_values()).difference(set(population))), 0,
'Should only have elements from the population')
def test_ewma_sample_load(self):
_time = [10000]
def wtime():
return _time[0]
sample = ExponentiallyDecayingSample(100, 0.99, wall_time=wtime)
sample.RESCALE_THRESHOLD = 100
sample.clear()
for i in xrange(10000000):
sample.update(random.normalvariate(0, 10))
_time[0] += 1
self.assertEqual(sample.size(), 100)
self.assertEqual(len(sample.get_values()), 100,
'Should have 100 elements')
test_ewma_sample_load.skip = "takes too long to run"