X-Git-Url: https://plomlompom.com/repos/processes?a=blobdiff_plain;f=tests%2Futils.py;h=60157104624ac79a59581757dd58a0344b30da6b;hb=8f28c8c685fa91b9cbabb4b424da4091e52058cf;hp=9d3d11d9f841290fed50b03c8d33acea7f5248ac;hpb=15ba54fa4132e46804e20748fd80243ad187b98c;p=plomtask diff --git a/tests/utils.py b/tests/utils.py index 9d3d11d..6015710 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -69,26 +69,52 @@ class TestCaseWithDB(TestCase): f(self) return wrapper + def _load_from_db(self, id_: int | str) -> list[object]: + db_found: list[object] = [] + for row in self.db_conn.row_where(self.checked_class.table_name, + 'id', id_): + db_found += [self.checked_class.from_table_row(self.db_conn, + row)] + return db_found + @_within_checked_class - def test_saving_and_caching(self) -> None: - """Test storage and initialization of instances and attributes.""" - self.check_saving_and_caching(id_=1, **self.default_init_kwargs) - obj = self.checked_class(None, **self.default_init_kwargs) - obj.save(self.db_conn) - self.assertEqual(obj.id_, 2) + def test_saving_versioned(self) -> None: + """Test storage and initialization of versioned attributes.""" + def retrieve_attr_vals() -> list[object]: + attr_vals_saved: list[object] = [] + assert hasattr(retrieved, 'id_') + for row in self.db_conn.row_where(attr.table_name, 'parent', + retrieved.id_): + attr_vals_saved += [row[2]] + return attr_vals_saved for attr_name, type_ in self.test_versioneds.items(): - owner = self.checked_class(None) + # fail saving attributes on non-saved owner + owner = self.checked_class(None, **self.default_init_kwargs) vals: list[Any] = ['t1', 't2'] if type_ == str else [0.9, 1.1] attr = getattr(owner, attr_name) attr.set(vals[0]) attr.set(vals[1]) + with self.assertRaises(NotFoundException): + attr.save(self.db_conn) owner.save(self.db_conn) - retrieved = owner.__class__.by_id(self.db_conn, owner.id_) + # check stored attribute is as expected + retrieved = self._load_from_db(owner.id_)[0] attr = getattr(retrieved, attr_name) self.assertEqual(sorted(attr.history.values()), vals) + # check owner.save() created entries in attr table + attr_vals_saved = retrieve_attr_vals() + self.assertEqual(vals, attr_vals_saved) + # check setting new val to attr inconsequential to DB without save + attr.set(vals[0]) + attr_vals_saved = retrieve_attr_vals() + self.assertEqual(vals, attr_vals_saved) + # check save finally adds new val + attr.save(self.db_conn) + attr_vals_saved = retrieve_attr_vals() + self.assertEqual(vals + [vals[0]], attr_vals_saved) - def check_storage(self, content: list[Any]) -> None: - """Test cache and DB equal content.""" + def check_identity_with_cache_and_db(self, content: list[Any]) -> None: + """Test both cache and DB equal content.""" expected_cache = {} for item in content: expected_cache[item.id_] = item @@ -97,41 +123,74 @@ class TestCaseWithDB(TestCase): db_found: list[Any] = [] for item in content: assert isinstance(item.id_, type(self.default_ids[0])) - for row in self.db_conn.row_where(self.checked_class.table_name, - 'id', item.id_): - db_found += [self.checked_class.from_table_row(self.db_conn, - row)] + db_found += self._load_from_db(item.id_) hashes_db_found = [hash(x) for x in db_found] self.assertEqual(sorted(hashes_content), sorted(hashes_db_found)) - def check_saving_and_caching(self, **kwargs: Any) -> None: - """Test instance.save in its core without relations.""" - obj = self.checked_class(**kwargs) # pylint: disable=not-callable - # check object init itself doesn't store anything yet - self.check_storage([]) - # check saving sets core attributes properly - obj.save(self.db_conn) - for key, value in kwargs.items(): - self.assertEqual(getattr(obj, key), value) - # check saving stored properly in cache and DB - self.check_storage([obj]) + @_within_checked_class + def test_saving_and_caching(self) -> None: + """Test effects of .cache() and .save().""" + id1 = self.default_ids[0] + # check failure to cache without ID (if None-ID input possible) + if isinstance(id1, int): + obj0 = self.checked_class(None, **self.default_init_kwargs) + with self.assertRaises(HandledException): + obj0.cache() + # check mere object init itself doesn't even store in cache + obj1 = self.checked_class(id1, **self.default_init_kwargs) + self.assertEqual(self.checked_class.get_cache(), {}) + # check .cache() fills cache, but not DB + obj1.cache() + self.assertEqual(self.checked_class.get_cache(), {id1: obj1}) + db_found = self._load_from_db(id1) + self.assertEqual(db_found, []) + # check .save() sets ID (for int IDs), updates cache, and fills DB + # (expect ID to be set to id1, despite obj1 already having that as ID: + # it's generated by cursor.lastrowid on the DB table, and with obj1 + # not written there, obj2 should get it first!) + id_input = None if isinstance(id1, int) else id1 + obj2 = self.checked_class(id_input, **self.default_init_kwargs) + obj2.save(self.db_conn) + obj2_hash = hash(obj2) + self.assertEqual(self.checked_class.get_cache(), {id1: obj2}) + db_found += self._load_from_db(id1) + self.assertEqual([hash(o) for o in db_found], [obj2_hash]) + # check we cannot overwrite obj2 with obj1 despite its same ID, + # since it has disappeared now + with self.assertRaises(HandledException): + obj1.save(self.db_conn) - def check_by_id(self) -> None: - """Test .by_id(), including creation.""" + @_within_checked_class + def test_by_id(self) -> None: + """Test .by_id().""" + id1, id2, _ = self.default_ids # check failure if not yet saved - id1, id2 = self.default_ids[0], self.default_ids[1] - obj = self.checked_class(id1) # pylint: disable=not-callable + obj1 = self.checked_class(id1, **self.default_init_kwargs) with self.assertRaises(NotFoundException): self.checked_class.by_id(self.db_conn, id1) + # check identity of cached and retrieved + obj1.cache() + self.assertEqual(obj1, self.checked_class.by_id(self.db_conn, id1)) # check identity of saved and retrieved - obj.save(self.db_conn) - self.assertEqual(obj, self.checked_class.by_id(self.db_conn, id1)) - # check create=True acts like normal instantiation (sans saving) - by_id_created = self.checked_class.by_id(self.db_conn, id2, - create=True) - # pylint: disable=not-callable - self.assertEqual(self.checked_class(id2), by_id_created) - self.check_storage([obj]) + obj2 = self.checked_class(id2, **self.default_init_kwargs) + obj2.save(self.db_conn) + self.assertEqual(obj2, self.checked_class.by_id(self.db_conn, id2)) + + @_within_checked_class + def test_by_id_or_create(self) -> None: + """Test .by_id_or_create.""" + # check .by_id_or_create acts like normal instantiation (sans saving) + id_ = self.default_ids[0] + if not self.checked_class.can_create_by_id: + with self.assertRaises(HandledException): + self.checked_class.by_id_or_create(self.db_conn, id_) + # check .by_id_or_create fails if wrong class + else: + by_id_created = self.checked_class.by_id_or_create(self.db_conn, + id_) + with self.assertRaises(NotFoundException): + self.checked_class.by_id(self.db_conn, id_) + self.assertEqual(self.checked_class(id_), by_id_created) @_within_checked_class def test_from_table_row(self) -> None: @@ -230,7 +289,7 @@ class TestCaseWithDB(TestCase): obj.remove(self.db_conn) obj.save(self.db_conn) obj.remove(self.db_conn) - self.check_storage([]) + self.check_identity_with_cache_and_db([]) class TestCaseWithServer(TestCaseWithDB):