monitor/datastore_test.py

   1 #! /usr/bin/env python
   2 #
   3 # -*- coding: utf-8 -*-
   4
   5 """Comment
   6 """
   7
   8 __author__ = "rlazo.paz@gmail.com (Rodrigo Lazo)"
   9
  10 import unittest
  11 from datastore import DataEntry, DataPoint
  12 from operator import attrgetter, itemgetter
  13
  14
  15 class TestDataEntry(unittest.TestCase):
  16     def setUp(self):
  17         self.entry = DataEntry()
  18
  19     def test_invalid_single_inserts(self):
  20         self.assertFalse(self.entry.insert_point(-1209, 2),
  21                          "Negative timestamp accepted.")
  22         self.assertFalse(self.entry.insert_point(12.1, 34830923),
  23                          "Float timestamp accepted.")
  24         self.assertFalse(self.entry.insert_point("23", 34830923),
  25                          "Str timestamp accepted.")
  26         self.assertEqual(len(self.entry.get_all()), 0,
  27                          "Only invalid inserts tried, it must be empty.")
  28
  29     def test_invalid_batch_inserts(self):
  30         self.assertFalse(self.entry.insert_points([(-1209, 2)]),
  31                          "Negative timestamp accepted.")
  32         self.assertEqual(len(self.entry.get_all()), 0,
  33                          "Only invalid inserts tried, it must be empty.")
  34         data_points = ((232, 2), (-193, 3), ("st", 5), (182, 4), (2.1, 4))
  35         self.assertFalse(self.entry.insert_points(data_points),
  36                          "Invalid timestamps accepted.")
  37         self.assertEqual(len(self.entry.get_all()), 2,
  38                          "Only two valid entries submitted.")
  39
  40     def test_valid_single_insert(self):
  41         self.assertTrue(self.entry.insert_point(23, 393),
  42                         "Valid value rejected.")
  43         self.assertTrue(self.entry.insert_point(0, "no problemo"),
  44                         "Valid value rejected.")
  45         self.assertEqual(len(self.entry.get_all()), 2,
  46                          "Lenght doesn't match, only 2 insertions")
  47
  48     def test_valid_batch_inserts(self):
  49         values = ((19282, "x"), (00, 338), (5, 0.53))
  50         self.assertTrue(self.entry.insert_points(values),
  51                         "Valid values rejected.")
  52         self.assertEqual(len(self.entry.get_all()), len(values),
  53                          "Lenght doesn't match, 5 insertions")
  54
  55     def test_get_latest(self):
  56         values = ((2, "x"), (48281, "m"), (100000000, 2), (1, "sor"))
  57         latest = max(values, key=itemgetter(0))
  58         self.entry.insert_points(values)
  59         self.assertEqual(self.entry.get_latest().timestamp, latest[0],
  60                          "latest value must be %s:%s" % latest)
  61
  62     def test_points_sorted(self):
  63         values = ((2, "x"), (48281, "m"), (100000000, 2), (1, "sor"))
  64         self.entry.insert_points(values)
  65         self.assertTrue(is_sorted(self.entry.get_all(),
  66                                   func=attrgetter("timestamp")),
  67                         "Points are not sorted by timestamp.")
  68
  69     def test_get_since(self):
  70         values = ((10, None), (30, None), (50, None), (60, None), (70, None))
  71         points = [DataPoint(*i) for i in values]
  72
  73         self.entry.insert_points(values)
  74         result = self.entry.get_since(100000)
  75         self.assertFalse(result, "Timestamp in the future should return empty,"\
  76                          " got list of size %d" % len(result))
  77         result = self.entry.get_since(70.1)
  78         self.assertFalse(result, "Timestamp in the future should return empty,"\
  79                          " got list of size %d" % len(result))
  80
  81         result = self.entry.get_since(0)
  82         self.assertEquals(result, points,
  83                         "Timestamp 0 should return all points,"\
  84                          " got %s" % str(result))
  85
  86         self.assertEquals(self.entry.get_since(10), points)
  87         self.assertEquals(self.entry.get_since(11), points[1:])
  88         self.assertEquals(self.entry.get_since(29), points[1:])
  89         self.assertEquals(self.entry.get_since(30), points[1:])
  90         self.assertEquals(self.entry.get_since(40), points[2:])
  91         self.assertEquals(self.entry.get_since(70), points[4:])
  92
  93
  94 def is_sorted(container, func=None):
  95     if func:
  96         return all(func(container[i]) <= func(container[i+1])
  97                    for i in xrange(len(container)-1))
  98     return all(container[i] <= container[i+1]
  99                for i in xrange(len(container)-1))
 100
 101
 102 if __name__ == '__main__':
 103     unittest.main()