Move logic of reading the event payload file to settings.py

Author: ewjoachim

coverage_comment/activity.py

@@ -16,21 +16,20 @@ class ActivityNotFound(Exception):
 def find_activity(
     event_name: str,
     is_default_branch: bool,
-    event_action: str | None = None,
+    event_type: str,
+    is_pr_merged: bool,
 ) -> str:
     """Find the activity to perform based on the event type and payload."""
     if event_name == "workflow_run":
         return "post_comment"
 
     if (
         (event_name == "push" and is_default_branch)
-        or (
-            event_name == "pull_request"
-            and event_action == "merged"
-            and is_default_branch
-        )
         or event_name == "schedule"
+        or (event_name == "pull_request" and event_type == "closed")
     ):
+        if event_name == "pull_request" and event_type == "closed" and not is_pr_merged:
+            raise ActivityNotFound
         return "save_coverage_data_files"
 
     if event_name not in {"pull_request", "push"}:

coverage_comment/main.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import functools
-import json
 import logging
 import os
 import sys
@@ -70,24 +69,16 @@ def action(
     log.debug(f"Operating on {config.GITHUB_REF}")
     gh = github_client.GitHub(session=github_session)
     event_name = config.GITHUB_EVENT_NAME
-    event_path = config.GITHUB_EVENT_PATH
-    event_action = None
-
-    if event_path and os.path.exists(event_path):
-        with open(event_path) as event_file:
-            event_payload = json.load(event_file)
-        is_merged_pr_action = event_payload.get("pull_request", {}).get("merged", False)
-        if is_merged_pr_action:
-            event_action = "merged"
 
     repo_info = github.get_repository_info(
         github=gh, repository=config.GITHUB_REPOSITORY
     )
     try:
         activity = activity_module.find_activity(
             event_name=event_name,
-            event_action=event_action,
             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(

coverage_comment/settings.py

@@ -2,7 +2,9 @@
 
 import dataclasses
 import decimal
+import functools
 import inspect
+import json
 import pathlib
 from collections.abc import MutableMapping
 from typing import Any
@@ -139,6 +141,23 @@ def GITHUB_BRANCH_NAME(self) -> str | None:
             return self.GITHUB_REF.split("/", 2)[2]
         return None
 
+    @functools.cached_property
+    def GITHUB_EVENT_PAYLOAD(self) -> dict:
+        if not self.GITHUB_EVENT_PATH:
+            return {}
+        return json.loads(self.GITHUB_EVENT_PATH.read_text())
+
+    @property
+    def GITHUB_EVENT_TYPE(self) -> str | None:
+        return self.GITHUB_EVENT_PAYLOAD.get("action")
+
+    @property
+    def IS_PR_MERGED(self) -> bool:
+        try:
+            return self.GITHUB_EVENT_PAYLOAD["pull_request"]["merged"]
+        except KeyError:
+            return False
+
     @property
     def FINAL_COMMENT_FILENAME(self):
         filename = self.COMMENT_FILENAME

tests/conftest.py

@@ -7,6 +7,7 @@
 import os
 import pathlib
 import zipfile
+from collections.abc import Callable
 
 import httpx
 import pytest
@@ -47,7 +48,7 @@ def _(**kwargs):
 
 
 @pytest.fixture
-def pull_request_config(base_config):
+def pull_request_config(base_config) -> Callable[..., settings.Config]:
     def _(**kwargs):
         defaults = {
             # GitHub stuff
@@ -250,6 +251,14 @@ def summary_file(tmp_path):
     return file
 
 
+@pytest.fixture
+def pull_request_event_payload(tmp_path):
+    file = tmp_path / "event.json"
+    file.touch()
+
+    return file
+
+
 _is_failed = []
 
 

tests/integration/test_main.py

@@ -12,11 +12,9 @@
 
 
 @pytest.fixture
-def in_integration_env(integration_env, integration_dir):
-    curdir = os.getcwd()
-    os.chdir(integration_dir)
+def in_integration_env(integration_env, integration_dir, monkeypatch):
+    monkeypatch.chdir(integration_dir)
     yield integration_dir
-    os.chdir(curdir)
 
 
 @pytest.fixture
@@ -673,6 +671,55 @@ def test_action__push__default_branch(
     assert "Missing" in summary_content
 
 
+def test_action__pull_request_closed_merged(
+    pull_request_config,
+    session,
+    in_integration_env,
+    get_logs,
+    git,
+    summary_file,
+    pull_request_event_payload,
+):
+    session.register("GET", "/repos/py-cov-action/foobar")(
+        json={"default_branch": "main", "visibility": "public"}
+    )
+    session.register(
+        "GET",
+        "https://img.shields.io/static/v1?label=Coverage&message=77%25&color=orange",
+    )(text="<this is a svg badge>")
+
+    git.register("git branch --show-current")(stdout="foo")
+    git.register("git reset --hard")()
+    git.register("git fetch origin python-coverage-comment-action-data")()
+    git.register("git switch python-coverage-comment-action-data")()
+    git.register("git add endpoint.json")()
+    git.register("git add data.json")()
+    git.register("git add badge.svg")()
+    git.register("git add htmlcov")()
+    git.register("git add README.md")()
+    git.register("git diff --staged --exit-code")(exit_code=1)
+    git.register("git commit --message Update coverage data")()
+    git.register("git push origin python-coverage-comment-action-data")()
+    git.register("git switch foo")()
+
+    pull_request_event_payload.write_text(
+        """{"action": "closed", "pull_request": {"merged": true}}"""
+    )
+
+    result = main.action(
+        config=pull_request_config(
+            GITHUB_STEP_SUMMARY=summary_file,
+            GITHUB_EVENT_PATH=pull_request_event_payload,
+        ),
+        github_session=session,
+        http_session=session,
+        git=git,
+    )
+    assert result == 0
+
+    assert get_logs("INFO", "Saving coverage files")
+
+
 def test_action__push__default_branch__private(
     push_config, session, in_integration_env, get_logs, git
 ):

tests/unit/test_activity.py

@@ -6,26 +6,44 @@
 
 
 @pytest.mark.parametrize(
-    "event_name, event_action, is_default_branch, expected_activity",
+    "event_name, is_default_branch, event_type, is_pr_merged, expected_activity",
     [
-        ("workflow_run", None, True, "post_comment"),
-        ("push", None, True, "save_coverage_data_files"),
-        ("push", None, False, "process_pr"),
-        ("pull_request", "merged", True, "save_coverage_data_files"),
-        ("pull_request", None, True, "process_pr"),
-        ("pull_request", None, False, "process_pr"),
-        ("schedule", None, False, "save_coverage_data_files"),
+        ("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"),
     ],
 )
-def test_find_activity(event_name, event_action, is_default_branch, expected_activity):
+def test_find_activity(
+    event_name, is_default_branch, event_type, is_pr_merged, expected_activity
+):
     result = activity.find_activity(
         event_name=event_name,
-        event_action=event_action,
         is_default_branch=is_default_branch,
+        event_type=event_type,
+        is_pr_merged=is_pr_merged,
     )
     assert result == expected_activity
 
 
 def test_find_activity_not_found():
     with pytest.raises(activity.ActivityNotFound):
-        activity.find_activity(event_name="not_found", is_default_branch=False)
+        activity.find_activity(
+            event_name="not_found",
+            is_default_branch=False,
+            event_type="not_found",
+            is_pr_merged=False,
+        )
+
+
+def test_find_activity_pr_closed_not_merged():
+    with pytest.raises(activity.ActivityNotFound):
+        activity.find_activity(
+            event_name="pull_request",
+            is_default_branch=False,
+            event_type="closed",
+            is_pr_merged=False,
+        )

tests/unit/test_main.py

@@ -3,6 +3,7 @@
 import os
 
 import httpx
+import pytest
 
 from coverage_comment import main, settings, subprocess
 
@@ -14,7 +15,6 @@ def test_main(mocker, get_logs):
     # We could also accept not to test this function but if we've come this
     # far and have 98% coverage, we can as well have 100%.
 
-    exit = mocker.patch("sys.exit")
     action = mocker.patch("coverage_comment.main.action")
 
     os.environ.update(
@@ -26,11 +26,14 @@ def test_main(mocker, get_logs):
             "GITHUB_BASE_REF": "",
             "GITHUB_EVENT_NAME": "push",
             "GITHUB_STEP_SUMMARY": "step_summary",
+            "GITHUB_EVENT_PATH": "foo/bar",
         }
     )
-    main.main()
 
-    exit.assert_called_with(action.return_value)
+    with pytest.raises(SystemExit) as exc_data:
+        main.main()
+
+    assert exc_data.value.code == action.return_value
     kwargs = action.call_args_list[0].kwargs
     assert isinstance(kwargs["config"], settings.Config)
     assert isinstance(kwargs["git"], subprocess.Git)

tests/unit/test_settings.py

@@ -1,7 +1,9 @@
 from __future__ import annotations
 
 import decimal
+import json
 import pathlib
+from collections.abc import Callable
 
 import pytest
 
@@ -104,7 +106,7 @@ def test_config__verbose_deprecated(get_logs):
 
 
 @pytest.fixture
-def config():
+def config() -> Callable[..., settings.Config]:
     defaults = {
         "GITHUB_BASE_REF": "master",
         "GITHUB_TOKEN": "foo",
@@ -122,7 +124,7 @@ def config():
         "MERGE_COVERAGE_FILES": True,
     }
 
-    def _(**kwargs):
+    def _(**kwargs) -> settings.Config:
         return settings.Config(**(defaults | kwargs))
 
     return _
@@ -196,3 +198,22 @@ def test_final_coverage_data_branch(config):
         SUBPROJECT_ID="bar",
     )
     assert config_obj.FINAL_COVERAGE_DATA_BRANCH == "foo-bar"
+
+
+@pytest.mark.parametrize("merged", [True, False])
+def test_is_pr_merged(tmp_path, config, merged):
+    path = tmp_path / "event.json"
+    path.write_text(
+        json.dumps({"action": "closed", "pull_request": {"merged": merged}})
+    )
+    config_obj = config(GITHUB_EVENT_PATH=path)
+
+    assert config_obj.IS_PR_MERGED is merged
+
+
+def test_is_pr_merged__other_payload(tmp_path, config):
+    path = tmp_path / "event.json"
+    path.write_text(json.dumps({"action": "other"}))
+    config_obj = config(GITHUB_EVENT_PATH=path)
+
+    assert config_obj.IS_PR_MERGED is False