1
from landscape.lib.persist import Persist
2
from landscape.accumulate import Accumulator, accumulate
3
from landscape.tests.helpers import LandscapeTest
6
class AccumulateTest(LandscapeTest):
7
"""Test for the accumulate function that implements accumulation logic."""
9
def test_accumulate(self):
15
accumulated_value, step_data = accumulate(0, 0, 5, 4, 5)
16
self.assertEquals(accumulated_value, 0)
17
self.assertEquals(step_data, (5, 4))
19
def test_accumulate_non_zero_accumulated_value(self):
22
--|--+--+--+--+--|--+--+--+--+--|--
25
accumulated_value, step_data = accumulate(7, 8, 13, 3, 5)
26
self.assertEquals(accumulated_value, 9)
27
self.assertEquals(step_data, (10, float((2*4) + (3*3))/5))
29
def test_accumulate_skipped_step(self):
32
--|--+--+--+--+--|--+--+--+--+--|--+--+--+--+--|--
35
accumulated_value, step_data = accumulate(0, 0, 12, 4, 5)
36
self.assertEquals(accumulated_value, 8)
37
self.assertEquals(step_data, None)
39
def test_accumulate_within_step(self):
45
accumulated_value, step_data = accumulate(0, 0, 2, 4, 5)
46
self.assertEquals(accumulated_value, 8)
47
self.assertEquals(step_data, None)
49
def test_accumulate_within_step_with_nonzero_start_accumulated_value(self):
55
accumulated_value, step_data = accumulate(2, 6, 4, 4, 5)
56
self.assertEquals(accumulated_value, 14)
57
self.assertEquals(step_data, None)
59
def test_accumulate_with_first_value_on_step_boundary(self):
65
accumulated_value, step_data = accumulate(0, 0, 0, 14, 5)
66
self.assertEquals(accumulated_value, 0)
67
self.assertEquals(step_data, None)
70
class AccumulatorTest(LandscapeTest):
71
"""Tests for the Accumulator plugin helper class."""
73
def test_accumulate(self):
80
accumulate = Accumulator(persist, 5)
82
self.assertEquals(persist.get("key"), None)
83
step_data = accumulate(5, 4, "key")
84
self.assertEquals(step_data, (5, 4))
85
self.assertEquals(persist.get("key"), (5, 0))
87
def test_accumulate_non_zero_accumulated_value(self):
90
--|--+--+--+--+--|--+--+--+--+--|--
94
accumulate = Accumulator(persist, 5)
96
# Persist data that would have been stored when
97
# accumulate(7, 4, "key") was called.
98
persist.set("key", (7, 8))
99
step_data = accumulate(13, 3, "key")
100
self.assertEquals(step_data, (10, float((2*4) + (3*3))/5))
101
self.assertEquals(persist.get("key"), (13, 9))
103
def test_accumulate_skipped_step(self):
106
--|--+--+--+--+--|--+--+--+--+--|--+--+--+--+--|--
110
accumulate = Accumulator(persist, 5)
112
self.assertEquals(persist.get("key"), None)
113
step_data = accumulate(12, 4, "key")
114
self.assertEquals(step_data, None)
115
self.assertEquals(persist.get("key"), (12, 8))
117
def test_accumulate_within_step(self):
124
accumulate = Accumulator(persist, 5)
126
self.assertEquals(persist.get("key"), None)
127
step_data = accumulate(2, 4, "key")
128
self.assertEquals(step_data, None)
129
self.assertEquals(persist.get("key"), (2, 8))
131
def test_accumulate_within_step_with_nonzero_start_accumulated_value(self):
138
accumulate = Accumulator(persist, 5)
140
# Persist data that would have been stored when
141
# accumulate(2, 3, "key") was called.
142
persist.set("key", (2, 6))
143
step_data = accumulate(4, 4, "key")
144
self.assertEquals(step_data, None)
145
self.assertEquals(persist.get("key"), (4, 14))
147
def test_accumulate_with_first_value_on_step_boundary(self):
154
accumulate = Accumulator(persist, 5)
156
self.assertEquals(persist.get("key"), None)
157
step_data = accumulate(0, 14, "key")
158
self.assertEquals(step_data, None)
159
self.assertEquals(persist.get("key"), (0, 0))