Add support for custom event dispatching

dgolombek · dgolombek · commit 01882d1f07ed · 2025-10-26T21:54:58.000-04:00
### Description This addresses #615 by adding the new EVENTS_AS_PR, EVENTS_AS_COVERAGE, and EVENTS_AS_COMMENT parameters. When any of these are specified, they override the default event-to-activity heuristic in #find_activity ### Testing I've added unit tests for these, but I haven't added integration or end-to-end tests, I was unable to get the environment set up correctly.
diff --git a/README.md b/README.md
@@ -59,6 +59,16 @@ 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 `EVENTS_AS_PR`, `EVENTS_AS_COVERAGE`, and `EVENTS_AS_COMMENT` parameters as
+described below. If any of these parameters are specified, they will be used
+instead of the default heuristic.
+
 ## Usage
 
 ### Setup
@@ -469,6 +479,23 @@ Usage may look like this
 
     # Deprecated, see https://docs.github.com/en/actions/monitoring-and-troubleshooting-workflows/enabling-debug-logging
     VERBOSE: false
+
+    # List of Github Events which should be treated as Pull Requests,
+    # analyzing the .coverage file as described in [PR Mode](#pr-mode)
+    # above. Specified as a comma-separated list
+    EVENTS_AS_PR: ""
+
+    # List of Github Events which should extract the coverage rate and store
+    # information onto the dedicated coverage branch, as described in
+    # [Default Branch Mode](#default-branch-mode) above. Specified as a
+    # comma-separated list.
+    EVENTS_AS_COVERAGE: ""
+
+    # List of Github Events which should result solely in comments being posted
+    # to the pull request. See
+    # [Commenting on the PR on the `push` event](#commenting-on-the-pr-on-the-push-event)
+    # below. Specified as a comma-separated list.
+    EVENTS_AS_COMMENT: ""
 ```
 
 ### Commenting on the PR on the `push` event
diff --git a/coverage_comment/activity.py b/coverage_comment/activity.py
@@ -8,20 +8,48 @@
 
 from __future__ import annotations
 
+from enum import Enum
+
+from coverage_comment import settings
+
+
+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 find_activity_from_config(
+    event_name: str,
+    config: settings.Config,
+) -> Activity:
+    activities: list[Activity] = []
+    if event_name in config.EVENTS_AS_PR:
+        activities.append(Activity.PROCESS_PR)
+    if event_name in config.EVENTS_AS_COVERAGE:
+        activities.append(Activity.SAVE_COVERAGE_DATA_FILES)
+    if event_name in config.EVENTS_AS_COMMENT:
+        activities.append(Activity.POST_COMMENT)
+    if len(activities) > 1:
+        raise ActivityConfigError(f"Event {event_name} is specified in multiple EVENTS_AS_* variables: {', '.join([activity.value for activity in activities])}")
+    if len(activities) == 0:
+        raise ActivityNotFound(f"Event {event_name} is not specified in any of the EVENTS_AS_* variables")
+    return activities[0]
 
 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 +59,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
diff --git a/coverage_comment/main.py b/coverage_comment/main.py
@@ -74,38 +74,47 @@ def action(
         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.event_activities_specified:
+            activity = activity_module.find_activity_from_config(
+                event_name=event_name,
+                config=config,
+            )
+        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 "EVENTS_AS_PR", "EVENTS_AS_COVERAGE", and '
+            '"EVENTS_AS_COMMENT" parameters 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,
diff --git a/coverage_comment/settings.py b/coverage_comment/settings.py
@@ -66,6 +66,9 @@ class Config:
     ANNOTATION_TYPE: str = "warning"
     MAX_FILES_IN_COMMENT: int = 25
     USE_GH_PAGES_HTML_URL: bool = False
+    EVENTS_AS_PR: list[str] = dataclasses.field(default_factory=list)
+    EVENTS_AS_COMMENT: list[str] = dataclasses.field(default_factory=list)
+    EVENTS_AS_COVERAGE: list[str] = dataclasses.field(default_factory=list)
     VERBOSE: bool = False
     # Only for debugging, not exposed in the action:
     FORCE_WORKFLOW_RUN: bool = False
@@ -131,6 +134,18 @@ def clean_github_output(cls, value: str) -> pathlib.Path:
     def clean_github_event_path(cls, value: str) -> pathlib.Path:
         return pathlib.Path(value)
 
+    @classmethod
+    def clean_events_as_pr(cls, value: str) -> list[str]:
+        return [e.strip() for e in value.split(",") if e.strip()]
+
+    @classmethod
+    def clean_events_as_comment(cls, value: str) -> list[str]:
+        return [e.strip() for e in value.split(",") if e.strip()]
+
+    @classmethod
+    def clean_events_as_coverage(cls, value: str) -> list[str]:
+        return [e.strip() for e in value.split(",") if e.strip()]
+
     @property
     def GITHUB_PR_NUMBER(self) -> int | None:
         # "refs/pull/2/merge"
@@ -176,6 +191,10 @@ def FINAL_COVERAGE_DATA_BRANCH(self):
             f"-{self.SUBPROJECT_ID}" if self.SUBPROJECT_ID else ""
         )
 
+    @property
+    def event_activities_specified(self) -> bool:
+        return bool(self.EVENTS_AS_COVERAGE) or bool(self.EVENTS_AS_PR) or bool(self.EVENTS_AS_COMMENT)
+
     # We need to type environ as a MutableMapping because that's what
     # os.environ is, and just saying `dict[str, str]` is not enough to make
     # mypy happy
diff --git a/tests/integration/test_main.py b/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(
diff --git a/tests/unit/test_activity.py b/tests/unit/test_activity.py
@@ -3,19 +3,20 @@
 import pytest
 
 from coverage_comment import activity
+from coverage_comment.settings import Config
 
 
 @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 +49,61 @@ def test_find_activity_pr_closed_not_merged():
             event_type="closed",
             is_pr_merged=False,
         )
+
+BASE_ENV = {
+    'GITHUB_BASE_REF': "",
+    'GITHUB_TOKEN': "foo",
+    'GITHUB_REPOSITORY': "owner/repo",
+    'GITHUB_REF': "master",
+    'GITHUB_EVENT_NAME': "pull_request",
+    'GITHUB_PR_RUN_ID': 123,
+    'GITHUB_STEP_SUMMARY': "step_summary",
+}
+
+def test_find_activity_from_config__as_pr():
+    config = Config.from_environ(BASE_ENV | {
+        "EVENTS_AS_PR": "pull_request",
+        "EVENTS_AS_COMMENT": "workflow_dispatch,workflow_call",
+        "EVENTS_AS_COVERAGE": "push,merge_queue",
+    })
+    result = activity.find_activity_from_config(
+        event_name="pull_request",
+        config=config,
+    )
+    assert result == activity.Activity.PROCESS_PR
+
+def test_find_activity_from_config__as_comment():
+    config = Config.from_environ(BASE_ENV | {
+        "EVENTS_AS_PR": "pull_request",
+        "EVENTS_AS_COMMENT": "workflow_dispatch,workflow_call",
+        "EVENTS_AS_COVERAGE": "push,merge_queue",
+    })
+    result = activity.find_activity_from_config(
+        event_name="workflow_dispatch",
+        config=config,
+    )
+    assert result == activity.Activity.POST_COMMENT
+
+def test_find_activity_from_config__as_coverage():
+    config = Config.from_environ(BASE_ENV | {
+        "EVENTS_AS_PR": "pull_request",
+        "EVENTS_AS_COMMENT": "workflow_dispatch,workflow_call",
+        "EVENTS_AS_COVERAGE": "push,merge_queue",
+    })
+    result = activity.find_activity_from_config(
+        event_name="push",
+        config=config,
+    )
+    assert result == activity.Activity.SAVE_COVERAGE_DATA_FILES
+
+def test_find_activity_from_config__multiple_activities():
+    config = Config.from_environ(BASE_ENV | {
+        "EVENTS_AS_PR": "pull_request",
+        "EVENTS_AS_COMMENT": "workflow_dispatch,workflow_call",
+        "EVENTS_AS_COVERAGE": "pull_request",
+    })
+    with pytest.raises(activity.ActivityConfigError):
+        activity.find_activity_from_config(
+            event_name="pull_request",
+            config=config,
+        )
diff --git a/tests/unit/test_settings.py b/tests/unit/test_settings.py

Original file line number	Diff line number	Diff line change
`@@ -59,7 +59,7 @@ def test_action__invalid_event_name(session, push_config, in_integration_env, ge`
`59`	`59`	`)`
`60`	`60`
`61`	`61`	`assert result == 1`
`62`		`- assert get_logs("ERROR", "This action has only been designed to work for")`
	`62`	`+ assert get_logs("ERROR", "This action's default behavior is to determine")`
`63`	`63`
`64`	`64`
`65`	`65`	`def get_expected_output(`