plomlompom.com Git - plomtask/blob - tests/versioned_attributes.py

   1 """"Test Versioned Attributes in the abstract."""
   2 from unittest import TestCase
   3 from tests.utils import TestCaseWithDB
   4 from time import sleep
   5 from datetime import datetime
   6 from plomtask.misc 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 class TestParentType(BaseModel[int]):
  18     pass
  19
  20
  21 class TestsSansDB(TestCase):
  22     """Tests not requiring DB setup."""
  23
  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')
  28         attr.set('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]
  33         attr.set('A')
  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
  37         attr.set('B')
  38         self.assertEqual(sorted(attr.history.values()), ['A', 'B'])
  39         # check that a previously used value can be set if not most recent
  40         attr.set('A')
  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
  43         # later items
  44         attr.set('D')
  45         self.assertEqual(sorted(attr.history.values()), ['A', 'A', 'B', 'D'])
  46         attr.set('D')
  47         self.assertEqual(sorted(attr.history.values()), ['A', 'A', 'B', 'D'])
  48
  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')
  53         attr.set('B')
  54         self.assertEqual(attr.newest, 'B')
  55         attr.set('C')
  56
  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
  64         attr.set('B')
  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
  70         sleep(0.00001)
  71         timestamp_between = datetime.now().strftime(TIMESTAMP_FMT)
  72         sleep(0.00001)
  73         attr.set('C')
  74         timestamp_C = sorted(attr.history.keys())[-1]
  75         self.assertEqual(attr.at(timestamp_C), 'C')
  76         self.assertEqual(attr.at(timestamp_between), 'B')
  77         sleep(0.00001)
  78         timestamp_after_C = datetime.now().strftime(TIMESTAMP_FMT)
  79         self.assertEqual(attr.at(timestamp_after_C), 'C')
  80
  81
  82 class TestsWithDB(TestCaseWithDB):
  83     """Module tests requiring DB setup."""
  84
  85     def setUp(self) -> None:
  86         super().setUp()
  87         self.db_conn.exec(SQL_TEST_TABLE)
  88
  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
  94         attr.set('B')
  95         self.assertEqual([],
  96                          self.db_conn.row_where('versioned_tests', 'parent', 1))
  97         # check .save() makes history appear in DB
  98         attr.save(self.db_conn)
  99         vals_found = []
 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
 104         attr.set('C')
 105         attr.save(self.db_conn)
 106         vals_found = []
 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))
 110
 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')
 115         attr.set('B')
 116         attr.set('C')
 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()))
 125