@@ -28,6 +28,7 @@ from patman.database import Database, Pcommit, SerVer
from patman import patchwork
from patman.series import Series
from patman import status
+from patman.workflow import Wtype
# Tag to use for Change IDs
@@ -1552,7 +1553,11 @@ class CseriesHelper:
if val in states:
state = val
state_str, pad = self._build_col(state, base_str=name)
- print(f"{state_str}{pad} {stats.rjust(6)} {desc}")
+ marker = ''
+ ts = self.db.workflow_get(Wtype.TODO, ser.idnum)
+ if ts and ts <= self.get_now().strftime('%Y-%m-%d %H:%M:%S'):
+ marker = ' [todo]'
+ print(f"{state_str}{pad} {stats.rjust(6)} {desc}{marker}")
def _series_max_version(self, idnum):
"""Find the latest version of a series
@@ -7,7 +7,6 @@
import asyncio
from collections import OrderedDict, defaultdict
-
import pygit2
from u_boot_pylib import cros_subprocess
@@ -4176,3 +4176,100 @@ Date: .*
'SELECT archived FROM workflow WHERE series_id = ?',
(ser.idnum,))
self.assertEqual(1, res.fetchone()[0])
+
+ def test_workflow_todo(self):
+ """Test setting and clearing a todo"""
+ cser = self.get_cser()
+ with terminal.capture():
+ cser.add('first', 'my description', allow_unmarked=True)
+
+ cser.fake_now = datetime(2025, 3, 1, 12, 0, 0)
+ ser = cser.get_series_by_name('first')
+
+ # Set a todo for 7 days
+ with terminal.capture() as (out, _):
+ wf.todo(cser,'first', 7)
+ self.assertIn('2025-03-08 12:00:00', out.getvalue())
+
+ # Check the DB entry
+ ts = cser.db.workflow_get('todo', ser.idnum)
+ self.assertEqual('2025-03-08 12:00:00', ts)
+
+ # Replacing the todo should work
+ with terminal.capture() as (out, _):
+ wf.todo(cser,'first', 14)
+ self.assertIn('2025-03-15 12:00:00', out.getvalue())
+ ts = cser.db.workflow_get('todo', ser.idnum)
+ self.assertEqual('2025-03-15 12:00:00', ts)
+
+ # Clear it
+ with terminal.capture() as (out, _):
+ wf.todo_clear(cser,'first')
+ self.assertIn('Todo cleared', out.getvalue())
+ self.assertIsNone(cser.db.workflow_get('todo', ser.idnum))
+
+ def test_workflow_todo_list(self):
+ """Test listing todos"""
+ cser = self.get_cser()
+ with terminal.capture():
+ cser.add('first', 'my description', allow_unmarked=True)
+ cser.add('second', 'board stuff', allow_unmarked=True)
+
+ cser.fake_now = datetime(2025, 3, 10, 12, 0, 0)
+
+ # Set todos: first is due, second is in the future
+ with terminal.capture():
+ wf.todo(cser,'first', 0)
+ wf.todo(cser,'second', 7)
+
+ # Default list shows only due entries
+ with terminal.capture() as (out, _):
+ wf.todo_list(cser,show_all=False)
+ lines = out.getvalue().splitlines()
+ self.assertEqual(3, len(lines))
+ self.assertIn('first', lines[2])
+ self.assertIn('today', lines[2])
+
+ # --all shows all entries
+ with terminal.capture() as (out, _):
+ wf.todo_list(cser,show_all=True)
+ lines = out.getvalue().splitlines()
+ self.assertEqual(4, len(lines))
+ self.assertIn('first', lines[2])
+ self.assertIn('today', lines[2])
+ self.assertIn('second', lines[3])
+ self.assertIn('in 7d', lines[3])
+
+ # No todos
+ with terminal.capture():
+ wf.todo_clear(cser,'first')
+ wf.todo_clear(cser,'second')
+ with terminal.capture() as (out, _):
+ wf.todo_list(cser,show_all=False)
+ self.assertIn('No todos due', out.getvalue())
+
+ def test_workflow_summary_marker(self):
+ """Test that [todo] shows in series summary"""
+ cser = self.get_cser()
+ with terminal.capture():
+ cser.add('first', 'my description', allow_unmarked=True)
+
+ cser.fake_now = datetime(2025, 3, 10, 12, 0, 0)
+
+ # Set a todo that is already due
+ with terminal.capture():
+ wf.todo(cser,'first', 0)
+
+ # Summary should show [todo]
+ with terminal.capture() as (out, _):
+ cser.summary(None)
+ self.assertIn('[todo]', out.getvalue())
+
+ # Set a todo in the future
+ with terminal.capture():
+ wf.todo(cser,'first', 14)
+
+ # Summary should NOT show [todo]
+ with terminal.capture() as (out, _):
+ cser.summary(None)
+ self.assertNotIn('[todo]', out.getvalue())
@@ -5,9 +5,72 @@
"""Workflow types and operations for patman series management"""
+from datetime import datetime, timedelta
import enum
class Wtype(enum.StrEnum):
"""Types of workflow entry"""
TODO = 'todo'
+
+
+def todo(cser, series, days):
+ """Mark a series as a todo item after a number of days
+
+ Args:
+ cser (CseriesHelper): Series helper with open database
+ series (str): Name of series to use, or None for current branch
+ days (int): Number of days from now to mark as due
+ """
+ ser = cser._parse_series(series)
+ cser.db.workflow_archive(Wtype.TODO, ser.idnum)
+ when = cser.get_now() + timedelta(days=days)
+ ts = when.strftime('%Y-%m-%d %H:%M:%S')
+ cser.db.workflow_add(Wtype.TODO, ser.idnum, ts)
+ cser.commit()
+ print(f"Series '{ser.name}' marked for todo on {ts}")
+
+
+def todo_clear(cser, series):
+ """Clear the todo marker for a series
+
+ Args:
+ cser (CseriesHelper): Series helper with open database
+ series (str): Name of series to use, or None for current branch
+ """
+ ser = cser._parse_series(series)
+ cser.db.workflow_archive(Wtype.TODO, ser.idnum)
+ cser.commit()
+ print(f"Todo cleared for series '{ser.name}'")
+
+
+def todo_list(cser, show_all):
+ """List series that are due (or scheduled) for attention
+
+ Args:
+ cser (CseriesHelper): Series helper with open database
+ show_all (bool): True to show all scheduled todos, not just
+ those that are due
+ """
+ now = cser.get_now().strftime('%Y-%m-%d %H:%M:%S')
+ before = None if show_all else now
+ entries = cser.db.workflow_get_by_type(Wtype.TODO, before=before)
+ if not entries:
+ if show_all:
+ print('No todos scheduled')
+ else:
+ print('No todos due')
+ return
+ print(f"{'Series':17} {'Due':>14} Description")
+ print(f"{'-' * 17} {'-' * 14} {'-' * 30}")
+ for _sid, name, desc, ts in entries:
+ when = datetime.strptime(ts, '%Y-%m-%d %H:%M:%S')
+ delta = when - cser.get_now()
+ days = delta.days
+ if days < 0:
+ due = f'{-days}d overdue'
+ elif days == 0:
+ due = 'today'
+ else:
+ due = f'in {days}d'
+ print(f"{name:17} {due:>14} {desc}")