Add support for custom event dispatching

README.md

@@ -59,6 +59,36 @@ These files include:
 
 See [an example](https://github.com/py-cov-action/python-coverage-comment-action-v3-example)
 
+### Determining the mode
+
+By default, the action will attempt to pick the appropriate mode based on the
+current branch, whether or not it's in a pull request, and if that pull request
+is open or closed. This frequently results in the correct action taking place,
+but is only a heuristic. If you need more precise control, you should specify
+the `ACTIVITY` parameter to directly choose the mode. It may be one of:
+
+- `process_pr`, to select [PR mode](#pr-mode)
+- `save_coverage_data_files`, to select [Default branch mode](#default-branch-mode)
+- `post_comment`, to select [Commenting on the PR on the `push` event](#commenting-on-the-pr-on-the-push-event)
+
+Combining this with [Github's Expressions]
+(https://docs.github.com/en/actions/reference/workflows-and-actions/expressions) you can
+build out the the custom handling needed. For example:
+
+```yaml
+      - name: Coverage comment
+        id: coverage_comment
+        uses: py-cov-action/python-coverage-comment-action@v3
+        with:
+          GITHUB_TOKEN: ${{ github.token }}
+          activity: "${{ github.event_name == 'push' && 'save_coverage_data_files' || 'process_pr' }}"
+
+        # or
+
+        with:
+          activity: "${{ (github.event_name == 'push' && github.ref_name == 'main') && 'save_coverage_data_files' || 'process_pr' }}"
+```
+
 ## Usage
 
 ### Setup
@@ -469,6 +499,10 @@ Usage may look like this
 
     # Deprecated, see https://docs.github.com/en/actions/monitoring-and-troubleshooting-workflows/enabling-debug-logging
     VERBOSE: false
+
+    # The specific activity that should be taken on this event, see
+    # [Determining the mode](#determining-the-mode) above.
+    ACTIVITY: ""
 ```
 
 ### Commenting on the PR on the `push` event

coverage_comment/activity.py

@@ -8,20 +8,38 @@
 
 from __future__ import annotations
 
+from enum import Enum
+
+
+class Activity(Enum):
+    PROCESS_PR = "process_pr"
+    POST_COMMENT = "post_comment"
+    SAVE_COVERAGE_DATA_FILES = "save_coverage_data_files"
+
 
 class ActivityNotFound(Exception):
     pass
 
 
+class ActivityConfigError(Exception):
+    pass
+
+
+def validate_activity(activity: str) -> Activity:
+    if activity not in [a.value for a in Activity]:
+        raise ActivityConfigError(f"Invalid activity: {activity}")
+    return Activity(activity)
+
+
 def find_activity(
     event_name: str,
     is_default_branch: bool,
     event_type: str | None,
     is_pr_merged: bool,
-) -> str:
+) -> Activity:
     """Find the activity to perform based on the event type and payload."""
     if event_name == "workflow_run":
-        return "post_comment"
+        return Activity.POST_COMMENT
 
     if (
         (event_name == "push" and is_default_branch)
@@ -31,9 +49,9 @@ def find_activity(
     ):
         if event_name == "pull_request" and event_type == "closed" and not is_pr_merged:
             raise ActivityNotFound
-        return "save_coverage_data_files"
+        return Activity.SAVE_COVERAGE_DATA_FILES
 
     if event_name not in {"pull_request", "push", "merge_group"}:
         raise ActivityNotFound
 
-    return "process_pr"
+    return Activity.PROCESS_PR

coverage_comment/main.py

@@ -74,38 +74,43 @@
         github=gh, repository=config.GITHUB_REPOSITORY
     )
     try:
-        activity = activity_module.find_activity(
-            event_name=event_name,
-            is_default_branch=repo_info.is_default_branch(ref=config.GITHUB_REF),
-            event_type=config.GITHUB_EVENT_TYPE,
-            is_pr_merged=config.IS_PR_MERGED,
-        )
+        if config.ACTIVITY:
+            activity = activity_module.validate_activity(config.ACTIVITY)
+        else:
+            activity = activity_module.find_activity(
+                event_name=event_name,
+                is_default_branch=repo_info.is_default_branch(ref=config.GITHUB_REF),
+                event_type=config.GITHUB_EVENT_TYPE,
+                is_pr_merged=config.IS_PR_MERGED,
+            )
     except activity_module.ActivityNotFound:
         log.error(
-            'This action has only been designed to work for "pull_request", "push", '
-            f'"workflow_run", "schedule" or "merge_group" actions, not "{event_name}". Because there '
-            "are security implications. If you have a different usecase, please open an issue, "
-            "we'll be glad to add compatibility."
+            "This action's default behavior is to determine the appropriate "
+            "mode based on the current branch, whether or not it's in a pull "
+            "request, and if that pull request is open or closed. This "
+            "frequently results in the correct action taking place, but is "
+            "only a heuristic. If you need more precise control, you should "
+            'specify the "ACTIVITY" parameter as described in the documentation.'
         )
         return 1
 
-    if activity == "save_coverage_data_files":
+    if activity == activity_module.Activity.SAVE_COVERAGE_DATA_FILES:
         return save_coverage_data_files(
             config=config,
             git=git,
             http_session=http_session,
             repo_info=repo_info,
         )
 
-    elif activity == "process_pr":
+    elif activity == activity_module.Activity.PROCESS_PR:
         return process_pr(
             config=config,
             gh=gh,
             repo_info=repo_info,
         )
 
     else:
-        # activity == "post_comment":
+        # activity == activity_module.Activity.POST_COMMENT:
         return post_comment(
             config=config,
             gh=gh,

coverage_comment/settings.py

@@ -66,6 +66,7 @@ class Config:
     ANNOTATION_TYPE: str = "warning"
     MAX_FILES_IN_COMMENT: int = 25
     USE_GH_PAGES_HTML_URL: bool = False
+    ACTIVITY: str | None = None
     VERBOSE: bool = False
     # Only for debugging, not exposed in the action:
     FORCE_WORKFLOW_RUN: bool = False

tests/integration/test_main.py

@@ -59,7 +59,7 @@ def test_action__invalid_event_name(session, push_config, in_integration_env, ge
     )
 
     assert result == 1
-    assert get_logs("ERROR", "This action has only been designed to work for")
+    assert get_logs("ERROR", "This action's default behavior is to determine")
 
 
 def get_expected_output(

tests/unit/test_activity.py

@@ -8,14 +8,20 @@
 @pytest.mark.parametrize(
     "event_name, is_default_branch, event_type, is_pr_merged, expected_activity",
     [
-        ("workflow_run", True, None, False, "post_comment"),
-        ("push", True, None, False, "save_coverage_data_files"),
-        ("push", False, None, False, "process_pr"),
-        ("pull_request", True, "closed", True, "save_coverage_data_files"),
-        ("pull_request", True, None, False, "process_pr"),
-        ("pull_request", False, None, False, "process_pr"),
-        ("schedule", False, None, False, "save_coverage_data_files"),
-        ("merge_group", False, None, False, "save_coverage_data_files"),
+        ("workflow_run", True, None, False, activity.Activity.POST_COMMENT),
+        ("push", True, None, False, activity.Activity.SAVE_COVERAGE_DATA_FILES),
+        ("push", False, None, False, activity.Activity.PROCESS_PR),
+        (
+            "pull_request",
+            True,
+            "closed",
+            True,
+            activity.Activity.SAVE_COVERAGE_DATA_FILES,
+        ),
+        ("pull_request", True, None, False, activity.Activity.PROCESS_PR),
+        ("pull_request", False, None, False, activity.Activity.PROCESS_PR),
+        ("schedule", False, None, False, activity.Activity.SAVE_COVERAGE_DATA_FILES),
+        ("merge_group", False, None, False, activity.Activity.SAVE_COVERAGE_DATA_FILES),
     ],
 )
 def test_find_activity(
@@ -48,3 +54,13 @@ def test_find_activity_pr_closed_not_merged():
             event_type="closed",
             is_pr_merged=False,
         )
+
+
+def test_validate_activity__invalid():
+    with pytest.raises(activity.ActivityConfigError):
+        activity.validate_activity("invalid")
+
+
+def test_validate_activity__valid():
+    result = activity.validate_activity("process_pr")
+    assert result == activity.Activity.PROCESS_PR