to_save = ['is_active']
to_save_versioned = ['title', 'description']
to_search = ['title.newest', 'description.newest']
+ can_create_by_id = True
def __init__(self, id_: int | None, is_active: bool = False) -> None:
super().__init__(id_)
Checks for Todos and Processes that depend on Condition, prohibits
deletion if found.
"""
- if self.id_ is None:
- raise HandledException('cannot remove unsaved item')
- for item in ('process', 'todo'):
- for attr in ('conditions', 'blockers', 'enables', 'disables'):
- table_name = f'{item}_{attr}'
- for _ in db_conn.row_where(table_name, 'condition', self.id_):
- raise HandledException('cannot remove Condition in use')
+ if self.id_ is not None:
+ for item in ('process', 'todo'):
+ for attr in ('conditions', 'blockers', 'enables', 'disables'):
+ table_name = f'{item}_{attr}'
+ for _ in db_conn.row_where(table_name, 'condition',
+ self.id_):
+ msg = 'cannot remove Condition in use'
+ raise HandledException(msg)
super().remove(db_conn)
"""Individual days defined by their dates."""
table_name = 'days'
to_save = ['comment']
+ can_create_by_id = True
def __init__(self, date: str, comment: str = '') -> None:
id_ = valid_date(date)
return day
@classmethod
- def by_id(cls,
- db_conn: DatabaseConnection, id_: str | None,
- create: bool = False,
- ) -> Day:
+ def by_id(cls, db_conn: DatabaseConnection, id_: str | None) -> Day:
"""Extend BaseModel.by_id checking for new/lost .todos."""
- day = super().by_id(db_conn, id_, create)
+ day = super().by_id(db_conn, id_)
assert day.id_ is not None
if day.id_ in Todo.days_to_update:
Todo.days_to_update.remove(day.id_)
id_: None | BaseModelId
cache_: dict[BaseModelId, Self]
to_search: list[str] = []
+ can_create_by_id = False
_exists = True
def __init__(self, id_: BaseModelId | None) -> None:
@classmethod
def by_id(cls, db_conn: DatabaseConnection,
- id_: BaseModelId | None,
- # pylint: disable=unused-argument
- create: bool = False) -> Self:
+ id_: BaseModelId | None
+ ) -> Self:
"""Retrieve by id_, on failure throw NotFoundException.
First try to get from cls.cache_, only then check DB; if found,
put into cache.
-
- If create=True, make anew (but do not cache yet).
"""
obj = None
if id_ is not None:
break
if obj:
return obj
- if create:
+ raise NotFoundException(f'found no object of ID {id_}')
+
+ @classmethod
+ def by_id_or_create(cls, db_conn: DatabaseConnection,
+ id_: BaseModelId | None
+ ) -> Self:
+ """Wrapper around .by_id, creating (not caching/saving) if not find."""
+ if not cls.can_create_by_id:
+ raise HandledException('Class cannot .by_id_or_create.')
+ try:
+ return cls.by_id(db_conn, id_)
+ except NotFoundException:
obj = cls(id_)
return obj
- raise NotFoundException(f'found no object of ID {id_}')
@classmethod
def all(cls: type[BaseModelInstance],
def do_GET_day(self) -> dict[str, object]:
"""Show single Day of ?date=."""
date = self._params.get_str('date', date_in_n_days(0))
- day = Day.by_id(self.conn, date, create=True)
+ day = Day.by_id_or_create(self.conn, date)
make_type = self._params.get_str('make_type')
conditions_present = []
enablers_for = {}
def do_GET_condition(self) -> dict[str, object]:
"""Show Condition of ?id=."""
id_ = self._params.get_int_or_none('id')
- c = Condition.by_id(self.conn, id_, create=True)
+ c = Condition.by_id_or_create(self.conn, id_)
ps = Process.all(self.conn)
return {'condition': c, 'is_new': c.id_ is None,
'enabled_processes': [p for p in ps if c in p.conditions],
def do_GET_process(self) -> dict[str, object]:
"""Show Process of ?id=."""
id_ = self._params.get_int_or_none('id')
- process = Process.by_id(self.conn, id_, create=True)
+ process = Process.by_id_or_create(self.conn, id_)
title_64 = self._params.get_str('title_b64')
if title_64:
title = b64decode(title_64.encode()).decode()
def do_POST_day(self) -> str:
"""Update or insert Day of date and Todos mapped to it."""
date = self._params.get_str('date')
- day = Day.by_id(self.conn, date, create=True)
+ day = Day.by_id_or_create(self.conn, date)
day.comment = self._form_data.get_str('day_comment')
day.save(self.conn)
make_type = self._form_data.get_str('make_type')
process = Process.by_id(self.conn, id_)
process.remove(self.conn)
return '/processes'
- process = Process.by_id(self.conn, id_, create=True)
+ process = Process.by_id_or_create(self.conn, id_)
process.title.set(self._form_data.get_str('title'))
process.description.set(self._form_data.get_str('description'))
process.effort.set(self._form_data.get_float('effort'))
condition = Condition.by_id(self.conn, id_)
condition.remove(self.conn)
return '/conditions'
- condition = Condition.by_id(self.conn, id_, create=True)
+ condition = Condition.by_id_or_create(self.conn, id_)
condition.is_active = self._form_data.get_str('is_active') == 'True'
condition.title.set(self._form_data.get_str('title'))
condition.description.set(self._form_data.get_str('description'))
('process_step_suppressions', 'process',
'suppressed_steps', 0)]
to_search = ['title.newest', 'description.newest']
+ can_create_by_id = True
def __init__(self, id_: int | None, calendarize: bool = False) -> None:
BaseModel.__init__(self, id_)
self.check_versioned_from_table_row('title', str)
self.check_versioned_from_table_row('description', str)
- def test_Condition_by_id(self) -> None:
- """Test .by_id(), including creation."""
- self.check_by_id()
-
def test_Condition_versioned_attributes_singularity(self) -> None:
"""Test behavior of VersionedAttributes on saving (with .title)."""
self.check_versioned_singularity()
kwargs = {'date': self.default_ids[0], 'comment': 'foo'}
self.check_saving_and_caching(**kwargs)
- def test_Day_by_id(self) -> None:
- """Test .by_id()."""
- self.check_by_id()
-
def test_Day_by_date_range_filled(self) -> None:
"""Test Day.by_date_range_filled."""
date1, date2, date3 = self.default_ids
self.assertEqual(Day.by_date_range_filled(self.db_conn,
day5.date, day7.date),
[day5, day6, day7])
- self.check_storage([day1, day2, day3, day6])
+ self.check_identity_with_cache_and_db([day1, day2, day3, day6])
# check 'today' is interpreted as today's date
today = Day(date_in_n_days(0))
today.save(self.db_conn)
"""Tests against our HTTP server/handler (and database)."""
def test_do_GET(self) -> None:
- """Test / redirect, and unknown targets failing."""
+ """Test GET / redirect, and unknown targets failing."""
self.conn.request('GET', '/')
self.check_redirect('/day')
self.check_get('/foo', 404)
method(self.db_conn, [c1.id_, c2.id_])
self.assertEqual(getattr(p, target), [c1, c2])
- def test_Process_by_id(self) -> None:
- """Test .by_id(), including creation"""
- self.check_by_id()
-
def test_Process_versioned_attributes_singularity(self) -> None:
"""Test behavior of VersionedAttributes on saving (with .title)."""
self.check_versioned_singularity()
p1.set_steps(self.db_conn, [step])
step.remove(self.db_conn)
self.assertEqual(p1.explicit_steps, [])
- self.check_storage([])
+ self.check_identity_with_cache_and_db([])
class TestsWithServer(TestCaseWithServer):
attr = getattr(retrieved, attr_name)
self.assertEqual(sorted(attr.history.values()), vals)
- 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
"""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([])
+ self.check_identity_with_cache_and_db([])
# 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])
+ self.check_identity_with_cache_and_db([obj])
- 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))
+ # obj1.save(self.db_conn)
+ # self.check_identity_with_cache_and_db([obj1, obj2])
+
+ @_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:
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):
projects / plomtask / commitdiff