From: Christian Heller Date: Sun, 28 Apr 2024 22:46:46 +0000 (+0200) Subject: Disallow deletion of Processes in use. X-Git-Url: https://plomlompom.com/repos/%7B%7Bdb.prefix%7D%7D/static/%7B%7B%20web_path%20%7D%7D/%7B%7Bprefix%7D%7D/edit?a=commitdiff_plain;h=ef0f4f93ba6519e6ea3352bde20ab405cc906c69;p=plomtask Disallow deletion of Processes in use. --- diff --git a/plomtask/processes.py b/plomtask/processes.py index 0a9b95b..375a0be 100644 --- a/plomtask/processes.py +++ b/plomtask/processes.py @@ -6,7 +6,8 @@ from sqlite3 import Row from plomtask.db import DatabaseConnection, BaseModel from plomtask.misc import VersionedAttribute from plomtask.conditions import Condition, ConditionsRelations -from plomtask.exceptions import NotFoundException, BadFormatException +from plomtask.exceptions import (NotFoundException, BadFormatException, + HandledException) @dataclass @@ -172,6 +173,10 @@ class Process(BaseModel[int], ConditionsRelations): def remove(self, db_conn: DatabaseConnection) -> None: """Remove from DB, with dependencies.""" assert isinstance(self.id_, int) + for _ in db_conn.row_where('process_steps', 'step_process', self.id_): + raise HandledException('cannot remove Process in use') + for _ in db_conn.row_where('todos', 'process', self.id_): + raise HandledException('cannot remove Process in use') db_conn.delete_where('process_conditions', 'process', self.id_) db_conn.delete_where('process_enables', 'process', self.id_) db_conn.delete_where('process_disables', 'process', self.id_) diff --git a/tests/processes.py b/tests/processes.py index 2ac8cf4..c6f4926 100644 --- a/tests/processes.py +++ b/tests/processes.py @@ -3,6 +3,7 @@ from unittest import TestCase from tests.utils import TestCaseWithDB, TestCaseWithServer from plomtask.processes import Process, ProcessStep, ProcessStepsNode from plomtask.conditions import Condition +from plomtask.todos import Todo from plomtask.exceptions import NotFoundException, HandledException @@ -158,12 +159,14 @@ class TestsWithDB(TestCaseWithDB): self.assertEqual(self.proc1.title.history, p_loaded.title.history) def test_Process_removal(self) -> None: - """Test removal of Processes.""" + """Test removal of Processes and ProcessSteps.""" assert isinstance(self.proc3.id_, int) self.proc1.remove(self.db_conn) self.assertEqual({self.proc2.id_, self.proc3.id_}, set(p.id_ for p in Process.all(self.db_conn))) self.proc2.set_steps(self.db_conn, [(None, self.proc3.id_, None)]) + with self.assertRaises(HandledException): + self.proc3.remove(self.db_conn) self.proc2.explicit_steps[0].remove(self.db_conn) retrieved = Process.by_id(self.db_conn, self.proc2.id_) self.assertEqual(retrieved.explicit_steps, []) @@ -172,6 +175,12 @@ class TestsWithDB(TestCaseWithDB): self.proc2.remove(self.db_conn) with self.assertRaises(NotFoundException): ProcessStep.by_id(self.db_conn, step.id_) + todo = Todo(None, self.proc3, False, '2024-01-01') + todo.save(self.db_conn) + with self.assertRaises(HandledException): + self.proc3.remove(self.db_conn) + todo.remove(self.db_conn) + self.proc3.remove(self.db_conn) class TestsWithServer(TestCaseWithServer):