plomlompom.com Git - plomtask/blob - 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
  91     def test_VersionedAttribute_save(self) -> None:
  92         """Test .save() to write to DB."""
  93         test_parent = TestParentType(1)
  94         attr = VersionedAttribute(test_parent, 'versioned_tests', 'A')
  95         # check mere .set() calls do not by themselves reflect in the DB
  96         attr.set('B')
  97         self.assertEqual([],
  98                          self.db_conn.row_where('versioned_tests',
  99                                                 'parent', 1))
 100         # check .save() makes history appear in DB
 101         attr.save(self.db_conn)
 102         vals_found = []
 103         for row in self.db_conn.row_where('versioned_tests', 'parent', 1):
 104             vals_found += [row[2]]
 105         self.assertEqual(['B'], vals_found)
 106         # check .save() also updates history in DB
 107         attr.set('C')
 108         attr.save(self.db_conn)
 109         vals_found = []
 110         for row in self.db_conn.row_where('versioned_tests', 'parent', 1):
 111             vals_found += [row[2]]
 112         self.assertEqual(['B', 'C'], sorted(vals_found))
 113
 114     def test_VersionedAttribute_history_from_row(self) -> None:
 115         """"Test .history_from_row() properly interprets DB rows."""
 116         test_parent = TestParentType(1)
 117         attr = VersionedAttribute(test_parent, 'versioned_tests', 'A')
 118         attr.set('B')
 119         attr.set('C')
 120         attr.save(self.db_conn)
 121         loaded_attr = VersionedAttribute(test_parent, '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 attr.history.items():
 125             self.assertEqual(value, loaded_attr.history[timestamp])
 126         self.assertEqual(len(attr.history.keys()),
 127                          len(loaded_attr.history.keys()))