1 """"Test Versioned Attributes in the abstract."""
2 from unittest import TestCase
3 from tests.utils import TestCaseWithDB
5 from datetime import datetime
6 from plomtask.misc import VersionedAttribute, TIMESTAMP_FMT
7 from plomtask.db import BaseModel
10 CREATE TABLE versioned_tests (
11 parent INTEGER NOT NULL,
12 timestamp TEXT NOT NULL,
14 PRIMARY KEY (parent, timestamp)
17 class TestParentType(BaseModel[int]):
21 class TestsSansDB(TestCase):
22 """Tests not requiring DB setup."""
24 def test_VersionedAttribute_set(self) -> None:
25 """Test .set() behaves as expected."""
26 # check value gets set even if already is the default
27 attr = VersionedAttribute(None, '', 'A')
29 self.assertEqual(list(attr.history.values()), ['A'])
30 # check same value does not get set twice in a row,
31 # and that not even its timestamp get updated
32 timestamp = list(attr.history.keys())[0]
34 self.assertEqual(list(attr.history.values()), ['A'])
35 self.assertEqual(list(attr.history.keys())[0], timestamp)
36 # check that different value _will_ be set/added
38 self.assertEqual(sorted(attr.history.values()), ['A', 'B'])
39 # check that a previously used value can be set if not most recent
41 self.assertEqual(sorted(attr.history.values()), ['A', 'A', 'B'])
42 # again check for same value not being set twice in a row, even for
45 self.assertEqual(sorted(attr.history.values()), ['A', 'A', 'B', 'D'])
47 self.assertEqual(sorted(attr.history.values()), ['A', 'A', 'B', 'D'])
49 def test_VersionedAttribute_newest(self) -> None:
50 """Test .newest returns newest element, or default on empty."""
51 attr = VersionedAttribute(None, '', 'A')
52 self.assertEqual(attr.newest, 'A')
54 self.assertEqual(attr.newest, 'B')
57 def test_VersionedAttribute_at(self) -> None:
58 """Test .at() returns values nearest to queried time, or default."""
59 # check .at() return default on empty history
60 attr = VersionedAttribute(None, '', 'A')
61 timestamp_A = datetime.now().strftime(TIMESTAMP_FMT)
62 self.assertEqual(attr.at(timestamp_A), 'A')
63 # check value exactly at timestamp returned
65 timestamp_B = list(attr.history.keys())[0]
66 self.assertEqual(attr.at(timestamp_B), 'B')
67 # check earliest value returned if exists, rather than default
68 self.assertEqual(attr.at(timestamp_A), 'B')
69 # check reverts to previous value for timestamps not indexed
71 timestamp_between = datetime.now().strftime(TIMESTAMP_FMT)
74 timestamp_C = sorted(attr.history.keys())[-1]
75 self.assertEqual(attr.at(timestamp_C), 'C')
76 self.assertEqual(attr.at(timestamp_between), 'B')
78 timestamp_after_C = datetime.now().strftime(TIMESTAMP_FMT)
79 self.assertEqual(attr.at(timestamp_after_C), 'C')
82 class TestsWithDB(TestCaseWithDB):
83 """Module tests requiring DB setup."""
85 def setUp(self) -> None:
87 self.db_conn.exec(SQL_TEST_TABLE)
89 def test_VersionedAttribute_save(self) -> None:
90 """Test .save() to write to DB."""
91 test_parent = TestParentType(1)
92 attr = VersionedAttribute(test_parent, 'versioned_tests', 'A')
93 # check mere .set() calls do not by themselves reflect in the DB
96 self.db_conn.row_where('versioned_tests', 'parent', 1))
97 # check .save() makes history appear in DB
98 attr.save(self.db_conn)
100 for row in self.db_conn.row_where('versioned_tests', 'parent', 1):
101 vals_found += [row[2]]
102 self.assertEqual(['B'], vals_found)
103 # check .save() also updates history in DB
105 attr.save(self.db_conn)
107 for row in self.db_conn.row_where('versioned_tests', 'parent', 1):
108 vals_found += [row[2]]
109 self.assertEqual(['B', 'C'], sorted(vals_found))
111 def test_VersionedAttribute_history_from_row(self) -> None:
112 """"Test .history_from_row() properly interprets DB rows."""
113 test_parent = TestParentType(1)
114 attr = VersionedAttribute(test_parent, 'versioned_tests', 'A')
117 attr.save(self.db_conn)
118 loaded_attr = VersionedAttribute(test_parent, 'versioned_tests', 'A')
119 for row in self.db_conn.row_where('versioned_tests', 'parent', 1):
120 loaded_attr.history_from_row(row)
121 for timestamp, value in attr.history.items():
122 self.assertEqual(value, loaded_attr.history[timestamp])
123 self.assertEqual(len(attr.history.keys()),
124 len(loaded_attr.history.keys()))