Use higher resolution for VersionedAttribute.history timestamps, avoid conflicts...

[plomtask] / tests / versioned_attributes.py

diff --git a/tests/versioned_attributes.py b/tests/versioned_attributes.py
new file mode 100644 (file)
index 0000000..df6fc46
--- /dev/null
+++ b/tests/versioned_attributes.py
@@ -0,0 +1,125 @@
+""""Test Versioned Attributes in the abstract."""
+from unittest import TestCase
+from tests.utils import TestCaseWithDB
+from time import sleep
+from datetime import datetime
+from plomtask.misc import VersionedAttribute, TIMESTAMP_FMT
+from plomtask.db import BaseModel 
+
+SQL_TEST_TABLE = '''
+CREATE TABLE versioned_tests (
+  parent INTEGER NOT NULL,
+  timestamp TEXT NOT NULL,
+  value TEXT NOT NULL,
+  PRIMARY KEY (parent, timestamp)
+);
+'''
+class TestParentType(BaseModel[int]):
+    pass
+
+
+class TestsSansDB(TestCase):
+    """Tests not requiring DB setup."""
+    
+    def test_VersionedAttribute_set(self) -> None:
+        """Test .set() behaves as expected."""
+        # check value gets set even if already is the default
+        attr = VersionedAttribute(None, '', 'A')
+        attr.set('A')
+        self.assertEqual(list(attr.history.values()), ['A'])
+        # check same value does not get set twice in a row,
+        # and that not even its timestamp get updated
+        timestamp = list(attr.history.keys())[0]
+        attr.set('A')
+        self.assertEqual(list(attr.history.values()), ['A'])
+        self.assertEqual(list(attr.history.keys())[0], timestamp)
+        # check that different value _will_ be set/added
+        attr.set('B')
+        self.assertEqual(sorted(attr.history.values()), ['A', 'B'])
+        # check that a previously used value can be set if not most recent 
+        attr.set('A')
+        self.assertEqual(sorted(attr.history.values()), ['A', 'A', 'B'])
+        # again check for same value not being set twice in a row, even for
+        # later items
+        attr.set('D')
+        self.assertEqual(sorted(attr.history.values()), ['A', 'A', 'B', 'D'])
+        attr.set('D')
+        self.assertEqual(sorted(attr.history.values()), ['A', 'A', 'B', 'D'])
+    
+    def test_VersionedAttribute_newest(self) -> None:
+        """Test .newest returns newest element, or default on empty."""
+        attr = VersionedAttribute(None, '', 'A')
+        self.assertEqual(attr.newest, 'A')
+        attr.set('B')
+        self.assertEqual(attr.newest, 'B')
+        attr.set('C')
+    
+    def test_VersionedAttribute_at(self) -> None:
+        """Test .at() returns values nearest to queried time, or default."""
+        # check .at() return default on empty history
+        attr = VersionedAttribute(None, '', 'A')
+        timestamp_A = datetime.now().strftime(TIMESTAMP_FMT)
+        self.assertEqual(attr.at(timestamp_A), 'A')
+        # check value exactly at timestamp returned 
+        attr.set('B')
+        timestamp_B = list(attr.history.keys())[0]
+        self.assertEqual(attr.at(timestamp_B), 'B')
+        # check earliest value returned if exists, rather than default 
+        self.assertEqual(attr.at(timestamp_A), 'B')
+        # check reverts to previous value for timestamps not indexed 
+        sleep(0.00001)
+        timestamp_between = datetime.now().strftime(TIMESTAMP_FMT)
+        sleep(0.00001)
+        attr.set('C')
+        timestamp_C = sorted(attr.history.keys())[-1]
+        self.assertEqual(attr.at(timestamp_C), 'C')
+        self.assertEqual(attr.at(timestamp_between), 'B')
+        sleep(0.00001)
+        timestamp_after_C = datetime.now().strftime(TIMESTAMP_FMT)
+        self.assertEqual(attr.at(timestamp_after_C), 'C')
+
+
+class TestsWithDB(TestCaseWithDB):
+    """Module tests requiring DB setup."""
+
+    def setUp(self) -> None:
+        super().setUp()
+        self.db_conn.exec(SQL_TEST_TABLE)
+
+    def test_VersionedAttribute_save(self) -> None:
+        """Test .save() to write to DB."""
+        test_parent = TestParentType(1)
+        attr = VersionedAttribute(test_parent, 'versioned_tests', 'A')
+        # check mere .set() calls do not by themselves reflect in the DB
+        attr.set('B')
+        self.assertEqual([],
+                         self.db_conn.row_where('versioned_tests', 'parent', 1))
+        # check .save() makes history appear in DB 
+        attr.save(self.db_conn)
+        vals_found = []
+        for row in self.db_conn.row_where('versioned_tests', 'parent', 1):
+            vals_found += [row[2]]
+        self.assertEqual(['B'], vals_found)
+        # check .save() also updates history in DB
+        attr.set('C')
+        attr.save(self.db_conn)
+        vals_found = []
+        for row in self.db_conn.row_where('versioned_tests', 'parent', 1):
+            vals_found += [row[2]]
+        self.assertEqual(['B', 'C'], sorted(vals_found))
+
+    def test_VersionedAttribute_history_from_row(self) -> None:
+        """"Test .history_from_row() properly interprets DB rows."""
+        test_parent = TestParentType(1)
+        attr = VersionedAttribute(test_parent, 'versioned_tests', 'A')
+        attr.set('B')
+        attr.set('C')
+        attr.save(self.db_conn)
+        loaded_attr = VersionedAttribute(test_parent, 'versioned_tests', 'A')
+        for row in self.db_conn.row_where('versioned_tests', 'parent', 1):
+            loaded_attr.history_from_row(row)
+        for timestamp, value in attr.history.items():
+            self.assertEqual(value, loaded_attr.history[timestamp])
+        self.assertEqual(len(attr.history.keys()),
+                         len(loaded_attr.history.keys()))
+