from __future__ import annotations
from sqlite3 import Row
from datetime import datetime
-from typing import Any
+from typing import Any, Set
from plomtask.db import DatabaseConnection
from plomtask.exceptions import NotFoundException, BadFormatException
if s.parent_step_id == node_id]
for child in explicit_children:
node['steps'][child.id_] = make_node(child)
+ node['seen'] = node_id in seen_step_ids
+ seen_step_ids.add(node_id)
for id_, step in node['steps'].items():
walk_steps(id_, step)
steps: dict[int, dict[str, object]] = {}
+ seen_step_ids: Set[int] = set()
if external_owner is None:
external_owner = self
for step in [s for s in self.explicit_steps
p_1.add_step(self.db_conn, None, p_2.id_, None)
p_1_dict: dict[int, dict[str, Any]] = {1: {
'process': p_2, 'parent_id': None,
- 'is_explicit': True, 'steps': {}
+ 'is_explicit': True, 'steps': {}, 'seen': False
}}
self.assertEqual(p_1.get_steps(self.db_conn, None), p_1_dict)
s_b = p_1.add_step(self.db_conn, None, p_3.id_, None)
p_1_dict[2] = {
'process': p_3, 'parent_id': None,
- 'is_explicit': True, 'steps': {}
+ 'is_explicit': True, 'steps': {}, 'seen': False
}
self.assertEqual(p_1.get_steps(self.db_conn, None), p_1_dict)
s_c = p_2.add_step(self.db_conn, None, p_3.id_, None)
assert s_c.id_ is not None
p_1_dict[1]['steps'] = {3: {
'process': p_3, 'parent_id': None,
- 'is_explicit': False, 'steps': {}
+ 'is_explicit': False, 'steps': {}, 'seen': False
}}
self.assertEqual(p_1.get_steps(self.db_conn, None), p_1_dict)
p_1.add_step(self.db_conn, None, p_2.id_, s_b.id_)
p_1_dict[2]['steps'][4] = {
- 'process': p_2, 'parent_id': s_b.id_,
+ 'process': p_2, 'parent_id': s_b.id_, 'seen': False,
'is_explicit': True, 'steps': {3: {
'process': p_3, 'parent_id': None,
- 'is_explicit': False, 'steps': {}
+ 'is_explicit': False, 'steps': {}, 'seen': True
}}}
self.assertEqual(p_1.get_steps(self.db_conn, None), p_1_dict)
p_1.add_step(self.db_conn, None, p_3.id_, 999)
p_1_dict[5] = {
'process': p_3, 'parent_id': None,
- 'is_explicit': True, 'steps': {}
+ 'is_explicit': True, 'steps': {}, 'seen': False
}
self.assertEqual(p_1.get_steps(self.db_conn, None), p_1_dict)
p_1.add_step(self.db_conn, None, p_3.id_, 3)
p_1_dict[6] = {
'process': p_3, 'parent_id': None,
- 'is_explicit': True, 'steps': {}
+ 'is_explicit': True, 'steps': {}, 'seen': False
}
self.assertEqual(p_1.get_steps(self.db_conn, None), p_1_dict)