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