updates

coned_rtu/tests/coned_test.py

@@ -0,0 +1,28 @@
+from datetime import datetime, timedelta, timezone
+import unittest
+
+from coned_rtu import Reading, json_to_readings
+
+tz = timezone(timedelta(hours=-4))
+
+
+class TestConed(unittest.TestCase):
+    def test_json_to_readings(self):
+        with open("tests/usage.json") as f:
+            j = f.read()
+        got = json_to_readings(j)
+        want = [
+            Reading(
+                datetime(2021, 8, 29, 0, 30, 0, tzinfo=tz),
+                datetime(2021, 8, 29, 0, 45, 0, tzinfo=tz),
+                "wh",
+                110.5,
+            ),
+            Reading(
+                datetime(2021, 8, 29, 0, 45, 0, tzinfo=tz),
+                datetime(2021, 8, 29, 1, 0, 0, tzinfo=tz),
+                "wh",
+                121.5,
+            ),
+        ]
+        self.assertEqual(got, want)

coned_rtu/tests/reading_test.py

@@ -0,0 +1,109 @@
+from datetime import datetime, timedelta, timezone
+import unittest
+
+from coned_rtu import Reading
+
+
+class TestReading(unittest.TestCase):
+    def setUp(self):
+        self.sometime = datetime(2010, 12, 25, 14, 30, 0, tzinfo=timezone.utc)
+        self.later = datetime(2010, 12, 26, 14, 30, 0, tzinfo=timezone.utc)
+
+    def test_start_time_must_be_before_end_time(self):
+        # start and end time the same
+        with self.assertRaises(ValueError):
+            Reading(self.sometime, self.sometime, "wh", 1)
+
+        # start time after end time
+        with self.assertRaises(ValueError):
+            Reading(self.later, self.sometime, "wh", 1)
+
+    def test_enforces_units(self):
+        # only wh or kwh supported
+        with self.assertRaises(ValueError):
+            Reading(self.sometime, self.later, "mwh", 1)
+
+        # incorrect capitalization is okay
+        Reading(self.sometime, self.later, "Wh", 1)
+
+    def test_duration(self):
+        r = Reading(self.sometime, self.later, "wh", 1)
+        self.assertEqual(r.duration(), timedelta(hours=24))
+
+    def test_overlaps(self):
+        t1 = self.sometime
+        t2 = t1 + timedelta(hours=1)
+        t3 = t2 + timedelta(hours=1)
+        t4 = t3 + timedelta(hours=1)
+
+        # one within the other
+        a = Reading(t1, t4, "wh", 1)
+        b = Reading(t2, t3, "wh", 1)
+        self.assertTrue(a.overlaps(b))
+        self.assertTrue(b.overlaps(a))
+
+        # one ends during the other
+        a = Reading(t1, t3, "wh", 1)
+        b = Reading(t2, t4, "wh", 1)
+        self.assertTrue(a.overlaps(b))
+        self.assertTrue(b.overlaps(a))
+
+        # same start, one is shorter than the other
+        a = Reading(t1, t3, "wh", 1)
+        b = Reading(t1, t2, "wh", 1)
+        self.assertTrue(a.overlaps(b))
+        self.assertTrue(b.overlaps(a))
+
+        # same end, one is shorter than the other
+        a = Reading(t1, t3, "wh", 1)
+        b = Reading(t2, t3, "wh", 1)
+        self.assertTrue(a.overlaps(b))
+        self.assertTrue(b.overlaps(a))
+
+        # not overlapping
+        a = Reading(t1, t2, "wh", 1)
+        b = Reading(t2, t3, "wh", 1)
+        self.assertFalse(a.overlaps(b))
+        self.assertFalse(b.overlaps(a))
+
+    def test_hash_bucket(self):
+        a = Reading(self.sometime, self.later, "wh", 1)
+        b = Reading(self.sometime, self.later, "wh", 2)
+        c = Reading(
+            self.sometime + timedelta(hours=24),
+            self.later + timedelta(hours=24),
+            "wh",
+            1,
+        )
+        self.assertEqual(a.hash_bucket(), b.hash_bucket())
+        self.assertNotEqual(a.hash_bucket(), c.hash_bucket())
+
+    def test_equality(self):
+        r = Reading(self.sometime, self.later, "wh", 1)
+
+        # times mismatch
+        bad_time = self.sometime + timedelta(seconds=10)
+
+        other = Reading(bad_time, self.later, "wh", 1)
+        self.assertNotEqual(r, other)
+
+        other = Reading(self.sometime, bad_time, "wh", 1)
+        self.assertNotEqual(r, other)
+
+        # energy mismatch
+        other = Reading(self.sometime, self.later, "wh", 1.0000001)
+        self.assertNotEqual(r, other)
+
+        # fully equal
+        other = Reading(self.sometime, self.later, "wh", 1)
+        self.assertEqual(r, other)
+
+    def test_hashing(self):
+        # unequal
+        a = Reading(self.sometime, self.later, "wh", 1)
+        b = Reading(self.sometime, self.later, "wh", 2)
+        self.assertNotEqual(hash(a), hash(b))
+
+        # equal
+        b = Reading(self.sometime, self.later, "wh", 1)
+        self.assertEqual(hash(a), hash(b))

coned_rtu/tests/store_test.py

@@ -0,0 +1,52 @@
+from datetime import datetime, timedelta, timezone
+import unittest
+
+from coned_rtu import Reading, BucketList, OverlappingReadingError
+
+
+class TestStore(unittest.TestCase):
+    def setUp(self):
+        self.sometime = datetime(2010, 12, 25, 14, 30, 0, tzinfo=timezone.utc)
+        self.later = datetime(2010, 12, 26, 14, 30, 0, tzinfo=timezone.utc)
+        self.hour = timedelta(hours=1)
+        self.day = timedelta(hours=24)
+
+    def test(self):
+        bl = BucketList()
+        t1 = self.sometime
+        t2 = t1 + self.hour
+        t3 = t2 + self.hour
+        t4 = t3 + self.day
+
+        b = Reading(t2, t3, "wh", 1)
+        bl.insert(b)
+
+        a = Reading(t1, t2, "wh", 1)
+        bl.insert(a)
+
+        # inserting an overlapping reading is an error
+        bad = Reading(t2, t2 + timedelta(minutes=30), "wh", 1)
+        with self.assertRaises(OverlappingReadingError):
+            bl.insert(bad)
+
+        # inserting an identical reading results in no change
+        bl.insert(a)
+
+        # inserted readings are sorted
+        got = bl.get_bucket(a.hash_bucket())
+        want = [a, b]
+        self.assertEqual(got, want)
+
+        # to_list returns a list in sorted order
+        c = Reading(t3, t4, "wh", 3)
+        bl.insert(c)
+        got = bl.to_list()
+        want = [a, b, c]
+        self.assertEqual(got, want)
+
+    def test_get_bucket_makes_bucket_if_not_exist(self):
+        bl = BucketList()
+        day = "2021-08-30"
+        _ = bl.get_bucket(day)
+        self.assertEqual(len(bl._dict), 1)
+        self.assertEqual(list(bl._dict.keys())[0], day)