|
1
|
|
|
"""Test of.stats app.""" |
|
2
|
|
|
import os |
|
3
|
|
|
import unittest |
|
4
|
|
|
from tempfile import mkstemp |
|
5
|
|
|
from unittest.mock import patch # noqa (isort conflict) |
|
6
|
|
|
|
|
7
|
|
|
from napps.kytos.of_stats.settings import STATS_INTERVAL |
|
8
|
|
|
from napps.kytos.of_stats.stats import RRD |
|
9
|
|
|
|
|
10
|
|
|
|
|
11
|
|
|
class TestRRD(unittest.TestCase): |
|
12
|
|
|
"""Test RRD interpolation.""" |
|
13
|
|
|
|
|
14
|
|
|
@patch.object(RRD, 'get_rrd') |
|
15
|
|
|
def test_2_points_out_of_3(self, get_rrd_method): |
|
16
|
|
|
"""Without the middle point, interpolate first and last points. |
|
17
|
|
|
|
|
18
|
|
|
Suppose STATS_INTERVAL is 10: |
|
19
|
|
|
- At 1234567800 (first), rx = tx = 10; |
|
20
|
|
|
- At 1234567810 (second), rx = tx = unknown; |
|
21
|
|
|
- At 1234567820 (third), rx = tx = 30. |
|
22
|
|
|
|
|
23
|
|
|
Test if the average is 1 at 1234567810 (second). |
|
24
|
|
|
""" |
|
25
|
|
|
# Mock will return a temporary file for rrd file. |
|
26
|
|
|
file, rrd_file = mkstemp('.rrd') |
|
27
|
|
|
os.close(file) |
|
28
|
|
|
get_rrd_method.return_value = rrd_file |
|
29
|
|
|
|
|
30
|
|
|
start = 1234567800 - STATS_INTERVAL # can't update using start time |
|
31
|
|
|
rrd = RRD('test', ('rx', 'tx')) |
|
32
|
|
|
rrd.create_rrd(rrd_file, tstamp=start) |
|
33
|
|
|
|
|
34
|
|
|
def update_rrd(multiplier): |
|
35
|
|
|
"""Update rrd.""" |
|
36
|
|
|
rx_value = tx_value = multiplier * STATS_INTERVAL |
|
37
|
|
|
tstamp = start + rx_value |
|
38
|
|
|
# any value for index is OK |
|
39
|
|
|
rrd.update([None], tstamp, rx=rx_value, tx=tx_value) |
|
40
|
|
|
|
|
41
|
|
|
update_rrd(1) |
|
42
|
|
|
update_rrd(3) |
|
43
|
|
|
|
|
44
|
|
|
second = start + 2 * STATS_INTERVAL |
|
45
|
|
|
# Interval between x and y excludes y |
|
46
|
|
|
tstamps, _, rows = rrd.fetch([None], start=second, end=second) |
|
47
|
|
|
|
|
48
|
|
|
os.unlink(rrd_file) |
|
49
|
|
|
|
|
50
|
|
|
self.assertEqual(second, list(tstamps)[0]) |
|
51
|
|
|
self.assertEqual((1.0, 1.0), rows[0]) |
|
52
|
|
|
|
|
53
|
|
|
def test_non_existent_rrd(self): |
|
54
|
|
|
"""Test fetch_latest with non existent rrd.""" |
|
55
|
|
|
# obj = path_mock.return_value.exists.return_value = False |
|
56
|
|
|
# was not being used |
|
57
|
|
|
rrd = RRD('app_folder', ['data_source']) |
|
58
|
|
|
row = rrd.fetch_latest('index') |
|
59
|
|
|
self.assertEqual(row, {}) |
|
60
|
|
|
|
kytos /
of_stats
