Add support for workflow_dispatch events

.gitignore

@@ -1,2 +1 @@
 __pycache__
-.vscode/*

.vscode/extensions.json

@@ -0,0 +1,8 @@
+{
+  "recommendations": [
+    "ms-python.python",
+    "ms-python.black-formatter",
+    "tamasfe.even-better-toml",
+    "esbenp.prettier-vscode"
+  ]
+}

.vscode/settings.json

@@ -0,0 +1,8 @@
+{
+  "[python]": {
+    "editor.defaultFormatter": "ms-python.black-formatter"
+  },
+  "editor.defaultFormatter": "esbenp.prettier-vscode",
+  "editor.formatOnSave": true,
+  "editor.trimAutoWhitespace": true
+}

neo/neo.py

@@ -10,10 +10,17 @@
 import re
 from urllib.parse import quote_plus
 
-from common import env_default, hdict, strtobool
+try:
+    from .common import env_default, hdict, strtobool
+except ImportError:
+    from common import env_default, hdict, strtobool
 
 
-def update_matches(files, include_regex, old_matches=defaultdict(set), ):
+def update_matches(
+    files,
+    include_regex,
+    old_matches=defaultdict(set),
+):
     """
     The update_matches function takes a list of files and their statuses,
     and returns a dictionary mapping the job matrix keys to sets of statuses.
@@ -25,7 +32,7 @@ def update_matches(files, include_regex, old_matches=defaultdict(set), ):
     :return: A dictionary of dictionaries
     """
     matches = defaultdict(set)
-    for (filename, status) in files:
+    for filename, status in files:
         match = include_regex.match(filename)
         if match:
             if match.groupdict():
@@ -90,10 +97,10 @@ def generate_matrix(
             for path, _, files in os.walk(default_dir)
             for f in files
         ]
-        matches = update_matches( default_files, include_regex, matches)
+        matches = update_matches(default_files, include_regex, matches)
     # mark matrix entries with a status if all its matches have the same status
     status_matrix = []
-    for (groups, statuses) in matches.items():
+    for groups, statuses in matches.items():
         groups["reason"] = statuses.pop() if len(statuses) == 1 else "updated"
         status_matrix.append(groups)
 
@@ -114,6 +121,18 @@ def main(
 
     if default_patterns is None:
         default_patterns = []
+
+    # Check if this is a workflow_dispatch or schedule event
+    github_event_name = os.getenv("GITHUB_EVENT_NAME", None)
+
+    # For workflow_dispatch and schedule events, use default behavior (list all matched directories)
+    if github_event_name in ["workflow_dispatch", "schedule"]:
+        logging.info(
+            f"{github_event_name} event detected, using default behavior to list all matched directories"
+        )
+        # Pass empty files list, but force defaults=True to trigger directory listing behavior
+        return generate_matrix([], include_regex, True, default_patterns)
+
     with requests.session() as session:
         session.hooks = {
             "response": lambda resp, *resp_args, **kwargs: resp.raise_for_status()
@@ -159,18 +178,44 @@ def github_webhook_ref(dest: str, option_strings: list):
             github_event = json.load(fp)
             if github_event_name == "pull_request":
                 return argparse.Action(
-                    default=github_event["pull_request"]["head"]["sha"]
-                    if is_github_head_ref
-                    else github_event["pull_request"]["base"]["sha"],
+                    default=(
+                        github_event["pull_request"]["head"]["sha"]
+                        if is_github_head_ref
+                        else github_event["pull_request"]["base"]["sha"]
+                    ),
                     required=False,
                     dest=dest,
                     option_strings=option_strings,
                 )
             elif github_event_name == "push":
                 return argparse.Action(
-                    default=github_event["after"]
-                    if is_github_head_ref
-                    else github_event["before"],
+                    default=(
+                        github_event["after"]
+                        if is_github_head_ref
+                        else github_event["before"]
+                    ),
+                    required=False,
+                    dest=dest,
+                    option_strings=option_strings,
+                )
+            elif github_event_name == "workflow_dispatch":
+                # For workflow_dispatch, we use the default branch ref
+                # since this is a manual trigger without specific commit comparison
+                return argparse.Action(
+                    default=github_event.get("ref", "refs/heads/main").replace(
+                        "refs/heads/", ""
+                    ),
+                    required=False,
+                    dest=dest,
+                    option_strings=option_strings,
+                )
+            elif github_event_name == "schedule":
+                # For scheduled events, we use the default branch ref
+                # since this is a time-based trigger without specific commit comparison
+                return argparse.Action(
+                    default=github_event.get("ref", "refs/heads/main").replace(
+                        "refs/heads/", ""
+                    ),
                     required=False,
                     dest=dest,
                     option_strings=option_strings,
@@ -237,9 +282,11 @@ def set_github_actions_output(generated_matrix: list) -> None:
     args = vars(parser.parse_args())
 
     logging.basicConfig(
-        level=logging.DEBUG
-        if os.getenv("NEO_LOG_LEVEL", "INFO") == "DEBUG"
-        else logging.INFO
+        level=(
+            logging.DEBUG
+            if os.getenv("NEO_LOG_LEVEL", "INFO") == "DEBUG"
+            else logging.INFO
+        )
     )
 
     matrix = main(**args)

neo/tests.py

@@ -37,8 +37,8 @@ def test_no_changes_with_default_pattern(self):
                     ],
                 ),
                 [
-                    {'environment': 'live', 'reason': 'default'},
-                    {'environment': 'staging', 'reason': 'default'}
+                    {"environment": "live", "reason": "default"},
+                    {"environment": "staging", "reason": "default"},
                 ],
             )
 
@@ -58,8 +58,8 @@ def test_changes_with_default_pattern(self):
                     ],
                 ),
                 [
-                    {'environment': 'live', 'reason': 'default'},
-                    {'environment': 'staging', 'reason': 'modified'}
+                    {"environment": "live", "reason": "default"},
+                    {"environment": "staging", "reason": "modified"},
                 ],
             )
 
@@ -203,6 +203,80 @@ def test_github_outputs(self):
         self.assertIn(f"matrix={expected_matrix_output}", output)
         self.assertIn(f"matrix-length=3", output)
 
+    def test_workflow_dispatch_behavior(self):
+        """Test that workflow_dispatch events use default behavior to list all matched directories"""
+        with tempfile.TemporaryDirectory() as d:
+            # Create test directory structure
+            staging_dir = os.path.join(d, "staging")
+            live_dir = os.path.join(d, "live")
+            os.makedirs(staging_dir)
+            os.makedirs(live_dir)
+            Path(os.path.join(staging_dir, "config.yaml")).touch()
+            Path(os.path.join(live_dir, "config.yaml")).touch()
+
+            # Simulate workflow_dispatch event by setting environment variable
+            original_event = os.getenv("GITHUB_EVENT_NAME")
+            os.environ["GITHUB_EVENT_NAME"] = "workflow_dispatch"
+
+            try:
+                # Test that workflow_dispatch triggers default behavior (listing all files)
+                result = neo.generate_matrix(
+                    files=[],  # Empty files list to simulate workflow_dispatch
+                    include_regex="(?P<environment>staging|live)",
+                    defaults=True,  # This should be forced for workflow_dispatch
+                    default_patterns=[],
+                    default_dir=d,  # Use our test directory
+                )
+
+                # Should return entries for both staging and live environments
+                environments = [entry.get("environment") for entry in result]
+                self.assertIn("staging", environments)
+                self.assertIn("live", environments)
+
+            finally:
+                # Restore original environment
+                if original_event is None:
+                    os.environ.pop("GITHUB_EVENT_NAME", None)
+                else:
+                    os.environ["GITHUB_EVENT_NAME"] = original_event
+
+    def test_schedule_behavior(self):
+        """Test that schedule events use default behavior to list all matched directories"""
+        with tempfile.TemporaryDirectory() as d:
+            # Create test directory structure
+            staging_dir = os.path.join(d, "staging")
+            live_dir = os.path.join(d, "live")
+            os.makedirs(staging_dir)
+            os.makedirs(live_dir)
+            Path(os.path.join(staging_dir, "config.yaml")).touch()
+            Path(os.path.join(live_dir, "config.yaml")).touch()
+
+            # Simulate schedule event by setting environment variable
+            original_event = os.getenv("GITHUB_EVENT_NAME")
+            os.environ["GITHUB_EVENT_NAME"] = "schedule"
+
+            try:
+                # Test that schedule triggers default behavior (listing all files)
+                result = neo.generate_matrix(
+                    files=[],  # Empty files list to simulate schedule
+                    include_regex="(?P<environment>staging|live)",
+                    defaults=True,  # This should be forced for schedule
+                    default_patterns=[],
+                    default_dir=d,  # Use our test directory
+                )
+
+                # Should return entries for both staging and live environments
+                environments = [entry.get("environment") for entry in result]
+                self.assertIn("staging", environments)
+                self.assertIn("live", environments)
+
+            finally:
+                # Restore original environment
+                if original_event is None:
+                    os.environ.pop("GITHUB_EVENT_NAME", None)
+                else:
+                    os.environ["GITHUB_EVENT_NAME"] = original_event
+
 
 class IntegrationTest(unittest.TestCase):
     empty_repo_commit_sha = "6b5794416e6750d16fb126a04eadb681349e6947"