test: 将测试中获取 Event 的代码统一

tests/github/event.py

@@ -0,0 +1,41 @@
+from pathlib import Path
+from typing import TypeVar
+
+from nonebot.adapters.github import (
+    Adapter,
+    Event,
+    IssueCommentCreated,
+    IssuesOpened,
+    PullRequestClosed,
+    PullRequestReviewSubmitted,
+)
+
+T = TypeVar("T", bound=Event)
+
+# 事件类型对应的事件名称和事件文件名
+EVENT_INFO = {
+    IssuesOpened: ("issues", "issue-open"),
+    IssueCommentCreated: ("issue_comment", "issue-comment"),
+    PullRequestClosed: ("pull_request", "pr-close"),
+    PullRequestReviewSubmitted: (
+        "pull_request_review",
+        "pull_request_review_submitted",
+    ),
+}
+
+
+def get_mock_event(event_type: type[T], filename: str = "", id: str = "1") -> T:
+    """通过事件类型获取事件对象"""
+
+    if event_type not in EVENT_INFO:
+        raise ValueError(f"Unknown event type: {event_type}")
+
+    event_name, event_filename = EVENT_INFO[event_type]
+    if filename:
+        event_filename = filename
+
+    event_path = Path(__file__).parent / "events" / f"{event_filename}.json"
+    event = Adapter.payload_to_event(id, event_name, event_path.read_bytes())
+
+    assert isinstance(event, event_type)
+    return event

tests/github/publish/process/test_auto_merge.py

@@ -1,9 +1,8 @@
-from pathlib import Path
-
-from nonebot.adapters.github import Adapter, PullRequestReviewSubmitted
+from nonebot.adapters.github import PullRequestReviewSubmitted
 from nonebug import App
 from pytest_mock import MockerFixture
 
+from tests.github.event import get_mock_event
 from tests.github.utils import get_github_bot
 
 
@@ -21,15 +20,7 @@ async def test_auto_merge(app: App, mocker: MockerFixture, mock_installation) ->
 
     async with app.test_matcher() as ctx:
         adapter, bot = get_github_bot(ctx)
-        event_path = (
-            Path(__file__).parent.parent.parent
-            / "events"
-            / "pull_request_review_submitted.json"
-        )
-        event = adapter.payload_to_event(
-            "1", "pull_request_review", event_path.read_bytes()
-        )
-        assert isinstance(event, PullRequestReviewSubmitted)
+        event = get_mock_event(PullRequestReviewSubmitted)
 
         ctx.should_call_api(
             "rest.apps.async_get_repo_installation",
@@ -94,15 +85,7 @@ async def test_auto_merge_need_rebase(
 
     async with app.test_matcher() as ctx:
         adapter, bot = get_github_bot(ctx)
-        event_path = (
-            Path(__file__).parent.parent.parent
-            / "events"
-            / "pull_request_review_submitted.json"
-        )
-        event = Adapter.payload_to_event(
-            "1", "pull_request_review", event_path.read_bytes()
-        )
-        assert isinstance(event, PullRequestReviewSubmitted)
+        event = get_mock_event(PullRequestReviewSubmitted)
 
         ctx.should_call_api(
             "rest.apps.async_get_repo_installation",
@@ -162,15 +145,7 @@ async def test_auto_merge_not_publish(app: App, mocker: MockerFixture) -> None:
 
     async with app.test_matcher() as ctx:
         adapter, bot = get_github_bot(ctx)
-        event_path = (
-            Path(__file__).parent.parent.parent
-            / "events"
-            / "pull_request_review_submitted.json"
-        )
-        event = Adapter.payload_to_event(
-            "1", "pull_request_review", event_path.read_bytes()
-        )
-        assert isinstance(event, PullRequestReviewSubmitted)
+        event = get_mock_event(PullRequestReviewSubmitted)
         event.payload.pull_request.labels = []
 
         ctx.receive_event(bot, event)
@@ -192,17 +167,8 @@ async def test_auto_merge_not_member(app: App, mocker: MockerFixture) -> None:
 
     async with app.test_matcher() as ctx:
         adapter, bot = get_github_bot(ctx)
-        event_path = (
-            Path(__file__).parent.parent.parent
-            / "events"
-            / "pull_request_review_submitted.json"
-        )
-        event = Adapter.payload_to_event(
-            "1", "pull_request_review", event_path.read_bytes()
-        )
-        assert isinstance(event, PullRequestReviewSubmitted)
+        event = get_mock_event(PullRequestReviewSubmitted)
         event.payload.review.author_association = "CONTRIBUTOR"
-
         ctx.receive_event(bot, event)
 
     # 测试 git 命令
@@ -222,15 +188,7 @@ async def test_auto_merge_not_approve(app: App, mocker: MockerFixture) -> None:
 
     async with app.test_matcher() as ctx:
         adapter, bot = get_github_bot(ctx)
-        event_path = (
-            Path(__file__).parent.parent.parent
-            / "events"
-            / "pull_request_review_submitted.json"
-        )
-        event = Adapter.payload_to_event(
-            "1", "pull_request_review", event_path.read_bytes()
-        )
-        assert isinstance(event, PullRequestReviewSubmitted)
+        event = get_mock_event(PullRequestReviewSubmitted)
         event.payload.review.state = "commented"
 
         ctx.receive_event(bot, event)

tests/github/publish/process/test_publish_check.py

@@ -5,11 +5,12 @@
 from githubkit import Response
 from githubkit.exception import RequestFailed
 from inline_snapshot import snapshot
-from nonebot.adapters.github import Adapter, IssueCommentCreated, IssuesOpened
+from nonebot.adapters.github import IssueCommentCreated, IssuesOpened
 from nonebug import App
 from pytest_mock import MockerFixture
 from respx import MockRouter
 
+from tests.github.event import get_mock_event
 from tests.github.utils import (
     MockBody,
     MockIssue,
@@ -61,9 +62,7 @@ async def test_bot_process_publish_check(
 
     async with app.test_matcher() as ctx:
         adapter, bot = get_github_bot(ctx)
-        event_path = Path(__file__).parent.parent.parent / "events" / "issue-open.json"
-        event = Adapter.payload_to_event("1", "issues", event_path.read_bytes())
-        assert isinstance(event, IssuesOpened)
+        event = get_mock_event(IssuesOpened)
 
         ctx.should_call_api(
             "rest.apps.async_get_repo_installation",
@@ -252,10 +251,7 @@ async def test_adapter_process_publish_check(
 
     async with app.test_matcher() as ctx:
         adapter, bot = get_github_bot(ctx)
-        event_path = Path(__file__).parent.parent.parent / "events" / "issue-open.json"
-        event = Adapter.payload_to_event("1", "issues", event_path.read_bytes())
-
-        assert isinstance(event, IssuesOpened)
+        event = get_mock_event(IssuesOpened)
         event.payload.issue.labels = get_issue_labels(["Adapter"])
 
         ctx.should_call_api(
@@ -465,9 +461,7 @@ async def test_edit_title(
 
     async with app.test_matcher() as ctx:
         adapter, bot = get_github_bot(ctx)
-        event_path = Path(__file__).parent.parent.parent / "events" / "issue-open.json"
-        event = Adapter.payload_to_event("1", "issues", event_path.read_bytes())
-        assert isinstance(event, IssuesOpened)
+        event = get_mock_event(IssuesOpened)
 
         ctx.should_call_api(
             "rest.apps.async_get_repo_installation",
@@ -685,9 +679,7 @@ async def test_edit_title_too_long(
 
     async with app.test_matcher() as ctx:
         adapter, bot = get_github_bot(ctx)
-        event_path = Path(__file__).parent.parent.parent / "events" / "issue-open.json"
-        event = Adapter.payload_to_event("1", "issues", event_path.read_bytes())
-        assert isinstance(event, IssuesOpened)
+        event = get_mock_event(IssuesOpened)
 
         ctx.should_call_api(
             "rest.apps.async_get_repo_installation",
@@ -816,9 +808,7 @@ async def test_process_publish_check_not_pass(
 
     async with app.test_matcher() as ctx:
         adapter, bot = get_github_bot(ctx)
-        event_path = Path(__file__).parent.parent.parent / "events" / "issue-open.json"
-        event = Adapter.payload_to_event("1", "issues", event_path.read_bytes())
-        assert isinstance(event, IssuesOpened)
+        event = get_mock_event(IssuesOpened)
 
         ctx.should_call_api(
             "rest.apps.async_get_repo_installation",
@@ -919,9 +909,7 @@ async def test_comment_at_pull_request(
 
     async with app.test_matcher() as ctx:
         adapter, bot = get_github_bot(ctx)
-        event_path = Path(__file__).parent.parent.parent / "events" / "pr-comment.json"
-        event = Adapter.payload_to_event("1", "issue_comment", event_path.read_bytes())
-        assert isinstance(event, IssueCommentCreated)
+        event = get_mock_event(IssueCommentCreated, "pr-comment")
 
         ctx.receive_event(bot, event)
 
@@ -946,9 +934,7 @@ async def test_issue_state_closed(
 
     async with app.test_matcher() as ctx:
         adapter, bot = get_github_bot(ctx)
-        event_path = Path(__file__).parent.parent.parent / "events" / "issue-open.json"
-        event = Adapter.payload_to_event("1", "issues", event_path.read_bytes())
-        assert isinstance(event, IssuesOpened)
+        event = get_mock_event(IssuesOpened)
 
         ctx.should_call_api(
             "rest.apps.async_get_repo_installation",
@@ -993,9 +979,7 @@ async def test_not_publish_issue(
 
     async with app.test_matcher() as ctx:
         adapter, bot = get_github_bot(ctx)
-        event_path = Path(__file__).parent.parent.parent / "events" / "issue-open.json"
-        event = Adapter.payload_to_event("1", "issues", event_path.read_bytes())
-        assert isinstance(event, IssuesOpened)
+        event = get_mock_event(IssuesOpened)
         event.payload.issue.labels = []
 
         ctx.receive_event(bot, event)
@@ -1013,11 +997,7 @@ async def test_comment_by_self(
 
     async with app.test_matcher() as ctx:
         adapter, bot = get_github_bot(ctx)
-        event_path = (
-            Path(__file__).parent.parent.parent / "events" / "issue-comment-bot.json"
-        )
-        event = Adapter.payload_to_event("1", "issue_comment", event_path.read_bytes())
-        assert isinstance(event, IssueCommentCreated)
+        event = get_mock_event(IssueCommentCreated, "issue-comment-bot")
 
         ctx.receive_event(bot, event)
 
@@ -1064,11 +1044,7 @@ async def test_skip_plugin_check(
 
     async with app.test_matcher() as ctx:
         adapter, bot = get_github_bot(ctx)
-        event_path = (
-            Path(__file__).parent.parent.parent / "events" / "issue-comment-skip.json"
-        )
-        event = Adapter.payload_to_event("1", "issue_comment", event_path.read_bytes())
-        assert isinstance(event, IssueCommentCreated)
+        event = get_mock_event(IssueCommentCreated, "issue-comment-skip")
 
         ctx.should_call_api(
             "rest.apps.async_get_repo_installation",
@@ -1268,9 +1244,7 @@ async def test_convert_pull_request_to_draft(
 
     async with app.test_matcher() as ctx:
         adapter, bot = get_github_bot(ctx)
-        event_path = Path(__file__).parent.parent.parent / "events" / "issue-open.json"
-        event = Adapter.payload_to_event("1", "issues", event_path.read_bytes())
-        assert isinstance(event, IssuesOpened)
+        event = get_mock_event(IssuesOpened)
 
         ctx.should_call_api(
             "rest.apps.async_get_repo_installation",
@@ -1411,9 +1385,7 @@ async def test_process_publish_check_ready_for_review(
 
     async with app.test_matcher() as ctx:
         adapter, bot = get_github_bot(ctx)
-        event_path = Path(__file__).parent.parent.parent / "events" / "issue-open.json"
-        event = Adapter.payload_to_event("1", "issues", event_path.read_bytes())
-        assert isinstance(event, IssuesOpened)
+        event = get_mock_event(IssuesOpened)
 
         ctx.should_call_api(
             "rest.apps.async_get_repo_installation",

tests/github/publish/process/test_publish_pull_request.py

@@ -1,11 +1,10 @@
-from pathlib import Path
-
 from inline_snapshot import snapshot
-from nonebot.adapters.github import Adapter, PullRequestClosed
+from nonebot.adapters.github import PullRequestClosed
 from nonebug import App
 from pytest_mock import MockerFixture
 from respx import MockRouter
 
+from tests.github.event import get_mock_event
 from tests.github.utils import (
     MockBody,
     MockIssue,
@@ -59,9 +58,7 @@ async def test_process_pull_request(
 
     async with app.test_matcher() as ctx:
         adapter, bot = get_github_bot(ctx)
-        event_path = Path(__file__).parent.parent.parent / "events" / "pr-close.json"
-        event = Adapter.payload_to_event("1", "pull_request", event_path.read_bytes())
-        assert isinstance(event, PullRequestClosed)
+        event = get_mock_event(PullRequestClosed)
         event.payload.pull_request.merged = True
 
         ctx.should_call_api(
@@ -173,8 +170,6 @@ async def test_process_pull_request(
 async def test_process_pull_request_not_merged(
     app: App, mocker: MockerFixture, mock_installation
 ) -> None:
-    event_path = Path(__file__).parent.parent.parent / "events" / "pr-close.json"
-
     mock_subprocess_run = mocker.patch("subprocess.run")
 
     mock_issue = MockIssue().as_mock(mocker)
@@ -184,7 +179,7 @@ async def test_process_pull_request_not_merged(
 
     async with app.test_matcher() as ctx:
         adapter, bot = get_github_bot(ctx)
-        event = adapter.payload_to_event("1", "pull_request", event_path.read_bytes())
+        event = get_mock_event(PullRequestClosed)
         assert isinstance(event, PullRequestClosed)
 
         ctx.should_call_api(
@@ -235,8 +230,6 @@ async def test_process_pull_request_skip_plugin_test(
     """跳过测试的插件合并时的情况"""
     from src.providers.validation import PublishType
 
-    event_path = Path(__file__).parent.parent.parent / "events" / "pr-close.json"
-
     mock_time = mocker.patch("src.providers.models.datetime")
     mock_now = mocker.MagicMock()
     mock_now.isoformat.return_value = "2023-09-01T00:00:00+00:00Z"
@@ -262,8 +255,7 @@ async def test_process_pull_request_skip_plugin_test(
 
     async with app.test_matcher() as ctx:
         adapter, bot = get_github_bot(ctx)
-        event = Adapter.payload_to_event("1", "pull_request", event_path.read_bytes())
-        assert isinstance(event, PullRequestClosed)
+        event = get_mock_event(PullRequestClosed)
         event.payload.pull_request.merged = True
 
         ctx.should_call_api(
@@ -373,14 +365,11 @@ async def test_process_pull_request_skip_plugin_test(
 
 async def test_not_publish(app: App, mocker: MockerFixture) -> None:
     """测试与发布无关的拉取请求"""
-    event_path = Path(__file__).parent.parent.parent / "events" / "pr-close.json"
-
     mock_subprocess_run = mocker.patch("subprocess.run")
 
     async with app.test_matcher() as ctx:
         adapter, bot = get_github_bot(ctx)
-        event = Adapter.payload_to_event("1", "pull_request", event_path.read_bytes())
-        assert isinstance(event, PullRequestClosed)
+        event = get_mock_event(PullRequestClosed)
         event.payload.pull_request.labels = []
 
         ctx.receive_event(bot, event)
@@ -393,14 +382,12 @@ async def test_extract_issue_number_from_ref_failed(
     app: App, mocker: MockerFixture
 ) -> None:
     """测试从分支名中提取议题号失败"""
-    event_path = Path(__file__).parent.parent.parent / "events" / "pr-close.json"
 
     mock_subprocess_run = mocker.patch("subprocess.run")
 
     async with app.test_matcher() as ctx:
         adapter, bot = get_github_bot(ctx)
-        event = Adapter.payload_to_event("1", "pull_request", event_path.read_bytes())
-        assert isinstance(event, PullRequestClosed)
+        event = get_mock_event(PullRequestClosed)
         event.payload.pull_request.head.ref = "1"
 
         ctx.receive_event(bot, event)