Add task to show added / removed invoke tasks

tasks/diff.py

@@ -3,18 +3,21 @@
 """
 
 import datetime
+import json
 import os
 import tempfile
+from datetime import timedelta
 
 from invoke import task
 from invoke.exceptions import Exit
 
 from tasks.build_tags import get_default_build_tags
 from tasks.flavor import AgentFlavor
 from tasks.go import GOARCH_MAPPING, GOOS_MAPPING
-from tasks.libs.common.color import color_message
+from tasks.libs.common.color import Color, color_message
 from tasks.libs.common.datadog_api import create_count, send_metrics
 from tasks.libs.common.git import check_uncommitted_changes, get_commit_sha, get_current_branch
+from tasks.libs.common.worktree import agent_context
 from tasks.release import _get_release_json_value
 
 BINARIES: dict[str, dict] = {
@@ -230,3 +233,61 @@ def patch_summary(diff):
         elif line.startswith("-"):
             remove_count += 1
     return add_count, remove_count
+
+
+@task
+def invoke_tasks(ctx, diff_date: str | None = None):
+    """Shows the added / removed invoke tasks since diff_date with their description.
+
+    Args:
+        diff_date: The date to compare the tasks to ('YYYY-MM-DD' format). Will be the last 30 days if not provided.
+    """
+
+    def get_tasks() -> dict[str, str]:
+        tasks = json.loads(ctx.run('invoke --list -F json', hide=True).stdout)
+
+        def get_tasks_rec(collection, prefix='', res=None):
+            res = res or {}
+
+            if isinstance(collection, dict):
+                newpref = prefix + collection['name']
+
+                for task in collection['tasks']:
+                    res[newpref + '.' + task['name']] = task['help']
+
+                for subtask in collection['collections']:
+                    get_tasks_rec(subtask, newpref + '.', res)
+
+            return res
+
+        # Remove 'tasks.' prefix
+        return {name.removeprefix(tasks['name'] + '.'): desc for name, desc in get_tasks_rec(tasks).items()}
+
+    if not diff_date:
+        diff_date = (datetime.datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d')
+    else:
+        try:
+            datetime.datetime.strptime(diff_date, '%Y-%m-%d')
+        except ValueError as e:
+            raise Exit('Invalid date format. Please use the format "YYYY-MM-DD".') from e
+
+    old_commit = ctx.run(f"git rev-list -n 1 --before='{diff_date} 23:59' HEAD", hide=True).stdout.strip()
+    assert old_commit, f"No commit found before {diff_date}"
+
+    with agent_context(ctx, commit=old_commit):
+        old_tasks = get_tasks()
+    current_tasks = get_tasks()
+
+    all_tasks = set(old_tasks.keys()).union(current_tasks.keys())
+    removed_tasks = {task for task in all_tasks if task not in current_tasks}
+    added_tasks = {task for task in all_tasks if task not in old_tasks}
+
+    print(f'* {color_message("Removed tasks", Color.BOLD)}:')
+    print('\n'.join(sorted(f'- {name}' for name in removed_tasks)))
+
+    print(f'\n* {color_message("Added tasks", Color.BOLD)}:')
+    for name, description in sorted((name, current_tasks[name]) for name in added_tasks):
+        line = '+ ' + name
+        if description:
+            line += ': ' + description
+        print(line)

tasks/libs/common/worktree.py

@@ -17,7 +17,7 @@
 LOCAL_DIRECTORY = Path.cwd().resolve()
 
 
-def init_env(ctx, branch: str | None = None):
+def init_env(ctx, branch: str | None = None, commit: str | None = None):
     """Will prepare the environment for commands applying to a worktree.
 
     To be used before each worktree section.
@@ -53,6 +53,12 @@ def init_env(ctx, branch: str | None = None):
         if not os.environ.get("AGENT_WORKTREE_NO_PULL"):
             ctx.run(f"git -C '{WORKTREE_DIRECTORY}' pull", hide=True)
 
+    if commit:
+        if not os.environ.get("AGENT_WORKTREE_NO_PULL"):
+            ctx.run(f"git -C '{WORKTREE_DIRECTORY}' fetch", hide=True)
+
+        ctx.run(f"git -C '{WORKTREE_DIRECTORY}' checkout '{commit}'", hide=True)
+
 
 def remove_env(ctx):
     """Will remove the environment for commands applying to a worktree."""
@@ -66,14 +72,14 @@ def is_worktree():
     return Path.cwd().resolve() == WORKTREE_DIRECTORY.resolve()
 
 
-def enter_env(ctx, branch: str | None, skip_checkout=False):
+def enter_env(ctx, branch: str | None, skip_checkout=False, commit: str | None = None):
     """Enters the worktree environment."""
 
-    if not branch:
+    if not (branch or commit):
         assert skip_checkout, 'skip_checkout must be set to True if branch is None'
 
     if not skip_checkout:
-        init_env(ctx, branch)
+        init_env(ctx, branch, commit=commit)
     else:
         assert WORKTREE_DIRECTORY.is_dir(), "Worktree directory is not present and skip_checkout is set to True"
 
@@ -92,12 +98,13 @@ def exit_env():
 
 
 @contextmanager
-def agent_context(ctx, branch: str | None, skip_checkout=False):
+def agent_context(ctx, branch: str | None = None, skip_checkout=False, commit: str | None = None):
     """Applies code to the worktree environment if the branch is not None.
 
     Args:
         branch: The branch to switch to. If None, will enter the worktree environment without switching branch (ensures that skip_checkout is True).
         skip_checkout: If True, the branch will not be checked out (no pull will be performed too).
+        commit: The commit to checkout. Is used instead of branch if provided.
 
     Usage:
         > with agent_context(ctx, branch):
@@ -111,7 +118,7 @@ def agent_context(ctx, branch: str | None, skip_checkout=False):
 
     try:
         # Enter
-        enter_env(ctx, branch, skip_checkout=skip_checkout)
+        enter_env(ctx, branch, skip_checkout=skip_checkout, commit=commit)
 
         yield
     finally: