home · contact · privacy
More refactoring.
[plomtask] / tests / versioned_attributes.py
1 """"Test Versioned Attributes in the abstract."""
2 from unittest import TestCase
3 from time import sleep
4 from datetime import datetime
5 from tests.utils import TestCaseWithDB
6 from plomtask.versioned_attributes import VersionedAttribute, TIMESTAMP_FMT
7 from plomtask.db import BaseModel
8
9 SQL_TEST_TABLE = '''
10 CREATE TABLE versioned_tests (
11   parent INTEGER NOT NULL,
12   timestamp TEXT NOT NULL,
13   value TEXT NOT NULL,
14   PRIMARY KEY (parent, timestamp)
15 );
16 '''
17
18
19 class TestParentType(BaseModel[int]):
20     """Dummy abstracting whatever may use VersionedAttributes."""
21
22
23 class TestsSansDB(TestCase):
24     """Tests not requiring DB setup."""
25
26     def test_VersionedAttribute_set(self) -> None:
27         """Test .set() behaves as expected."""
28         # check value gets set even if already is the default
29         attr = VersionedAttribute(None, '', 'A')
30         attr.set('A')
31         self.assertEqual(list(attr.history.values()), ['A'])
32         # check same value does not get set twice in a row,
33         # and that not even its timestamp get updated
34         timestamp = list(attr.history.keys())[0]
35         attr.set('A')
36         self.assertEqual(list(attr.history.values()), ['A'])
37         self.assertEqual(list(attr.history.keys())[0], timestamp)
38         # check that different value _will_ be set/added
39         attr.set('B')
40         self.assertEqual(sorted(attr.history.values()), ['A', 'B'])
41         # check that a previously used value can be set if not most recent
42         attr.set('A')
43         self.assertEqual(sorted(attr.history.values()), ['A', 'A', 'B'])
44         # again check for same value not being set twice in a row, even for
45         # later items
46         attr.set('D')
47         self.assertEqual(sorted(attr.history.values()), ['A', 'A', 'B', 'D'])
48         attr.set('D')
49         self.assertEqual(sorted(attr.history.values()), ['A', 'A', 'B', 'D'])
50
51     def test_VersionedAttribute_newest(self) -> None:
52         """Test .newest returns newest element, or default on empty."""
53         attr = VersionedAttribute(None, '', 'A')
54         self.assertEqual(attr.newest, 'A')
55         attr.set('B')
56         self.assertEqual(attr.newest, 'B')
57         attr.set('C')
58
59     def test_VersionedAttribute_at(self) -> None:
60         """Test .at() returns values nearest to queried time, or default."""
61         # check .at() return default on empty history
62         attr = VersionedAttribute(None, '', 'A')
63         timestamp_a = datetime.now().strftime(TIMESTAMP_FMT)
64         self.assertEqual(attr.at(timestamp_a), 'A')
65         # check value exactly at timestamp returned
66         attr.set('B')
67         timestamp_b = list(attr.history.keys())[0]
68         self.assertEqual(attr.at(timestamp_b), 'B')
69         # check earliest value returned if exists, rather than default
70         self.assertEqual(attr.at(timestamp_a), 'B')
71         # check reverts to previous value for timestamps not indexed
72         sleep(0.00001)
73         timestamp_between = datetime.now().strftime(TIMESTAMP_FMT)
74         sleep(0.00001)
75         attr.set('C')
76         timestamp_c = sorted(attr.history.keys())[-1]
77         self.assertEqual(attr.at(timestamp_c), 'C')
78         self.assertEqual(attr.at(timestamp_between), 'B')
79         sleep(0.00001)
80         timestamp_after_c = datetime.now().strftime(TIMESTAMP_FMT)
81         self.assertEqual(attr.at(timestamp_after_c), 'C')
82
83
84 class TestsWithDB(TestCaseWithDB):
85     """Module tests requiring DB setup."""
86
87     def setUp(self) -> None:
88         super().setUp()
89         self.db_conn.exec(SQL_TEST_TABLE)
90         self.test_parent = TestParentType(1)
91         self.attr = VersionedAttribute(self.test_parent,
92                                        'versioned_tests', 'A')
93
94     def test_VersionedAttribute_save(self) -> None:
95         """Test .save() to write to DB."""
96         # check mere .set() calls do not by themselves reflect in the DB
97         self.attr.set('B')
98         self.assertEqual([],
99                          self.db_conn.row_where('versioned_tests',
100                                                 'parent', 1))
101         # check .save() makes history appear in DB
102         self.attr.save(self.db_conn)
103         vals_found = []
104         for row in self.db_conn.row_where('versioned_tests', 'parent', 1):
105             vals_found += [row[2]]
106         self.assertEqual(['B'], vals_found)
107         # check .save() also updates history in DB
108         self.attr.set('C')
109         self.attr.save(self.db_conn)
110         vals_found = []
111         for row in self.db_conn.row_where('versioned_tests', 'parent', 1):
112             vals_found += [row[2]]
113         self.assertEqual(['B', 'C'], sorted(vals_found))
114
115     def test_VersionedAttribute_history_from_row(self) -> None:
116         """"Test .history_from_row() properly interprets DB rows."""
117         self.attr.set('B')
118         self.attr.set('C')
119         self.attr.save(self.db_conn)
120         loaded_attr = VersionedAttribute(self.test_parent,
121                                          'versioned_tests', 'A')
122         for row in self.db_conn.row_where('versioned_tests', 'parent', 1):
123             loaded_attr.history_from_row(row)
124         for timestamp, value in self.attr.history.items():
125             self.assertEqual(value, loaded_attr.history[timestamp])
126         self.assertEqual(len(self.attr.history.keys()),
127                          len(loaded_attr.history.keys()))