Skip to content

Commit da2248b

Browse files
silehtsileht
authored
chore(scopes): split cli.py (#801)
1 parent d6fb099 commit da2248b

File tree

6 files changed

+279
-253
lines changed

6 files changed

+279
-253
lines changed

‎mergify_cli/ci/scopes/base_detector.py‎

Lines changed: 99 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,99 @@
1+
from __future__ import annotations
2+
3+
import json
4+
import os
5+
import pathlib
6+
import typing
7+
8+
import click
9+
import yaml
10+
11+
12+
class MergeQueuePullRequest(typing.TypedDict):
13+
number: int
14+
15+
16+
class MergeQueueBatchFailed(typing.TypedDict):
17+
draft_pr_number: int
18+
checked_pull_request: list[int]
19+
20+
21+
class MergeQueueMetadata(typing.TypedDict):
22+
checking_base_sha: str
23+
pull_requests: list[MergeQueuePullRequest]
24+
previous_failed_batches: list[MergeQueueBatchFailed]
25+
26+
27+
def _yaml_docs_from_fenced_blocks(body: str) -> MergeQueueMetadata | None:
28+
lines = []
29+
found = False
30+
for line in body.splitlines():
31+
if line.startswith("```yaml"):
32+
found = True
33+
elif found:
34+
if line.startswith("```"):
35+
break
36+
lines.append(line)
37+
if lines:
38+
return typing.cast("MergeQueueMetadata", yaml.safe_load("\n".join(lines)))
39+
return None
40+
41+
42+
def _detect_base_from_merge_queue_payload(ev: dict[str, typing.Any]) -> str | None:
43+
pr = ev.get("pull_request")
44+
if not isinstance(pr, dict):
45+
return None
46+
title = pr.get("title") or ""
47+
if not isinstance(title, str):
48+
return None
49+
if not title.startswith("merge-queue: "):
50+
return None
51+
body = pr.get("body") or ""
52+
content = _yaml_docs_from_fenced_blocks(body)
53+
if content:
54+
return content["checking_base_sha"]
55+
return None
56+
57+
58+
def _detect_base_from_event(ev: dict[str, typing.Any]) -> str | None:
59+
pr = ev.get("pull_request")
60+
if isinstance(pr, dict):
61+
sha = pr.get("base", {}).get("sha")
62+
if isinstance(sha, str) and sha:
63+
return sha
64+
return None
65+
66+
67+
def detect() -> str:
68+
event_path = os.environ.get("GITHUB_EVENT_PATH")
69+
event: dict[str, typing.Any] | None = None
70+
if event_path and pathlib.Path(event_path).is_file():
71+
try:
72+
with pathlib.Path(event_path).open("r", encoding="utf-8") as f:
73+
event = json.load(f)
74+
except FileNotFoundError:
75+
event = None
76+
77+
if event is not None:
78+
# 0) merge-queue PR override
79+
mq_sha = _detect_base_from_merge_queue_payload(event)
80+
if mq_sha:
81+
return mq_sha
82+
83+
# 1) standard event payload
84+
event_sha = _detect_base_from_event(event)
85+
if event_sha:
86+
return event_sha
87+
88+
# 2) base ref (e.g., PR target branch)
89+
base_ref = os.environ.get("GITHUB_BASE_REF")
90+
if base_ref:
91+
return base_ref
92+
93+
msg = (
94+
"Could not detect base SHA. Ensure checkout has sufficient history "
95+
"(e.g., actions/checkout with fetch-depth: 0) or provide GITHUB_EVENT_PATH / GITHUB_BASE_REF."
96+
)
97+
raise click.ClickException(
98+
msg,
99+
)

‎mergify_cli/ci/scopes/changed_files.py‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
from __future__ import annotations
2+
3+
import subprocess
4+
5+
import click
6+
7+
8+
def _run(cmd: list[str]) -> str:
9+
try:
10+
return subprocess.check_output(cmd, text=True, encoding="utf-8").strip()
11+
except subprocess.CalledProcessError as e:
12+
msg = f"Command failed: {' '.join(cmd)}\n{e}"
13+
raise click.ClickException(msg) from e
14+
15+
16+
def git_changed_files(base: str) -> list[str]:
17+
# Committed changes only between base_sha and HEAD.
18+
# Includes: Added (A), Copied (C), Modified (M), Renamed (R), Type-changed (T), Deleted (D)
19+
# Excludes: Unmerged (U), Unknown (X), Broken (B)
20+
out = _run(
21+
["git", "diff", "--name-only", "--diff-filter=ACMRTD", f"{base}...HEAD"],
22+
)
23+
return [line for line in out.splitlines() if line]

‎mergify_cli/ci/scopes/cli.py‎

Lines changed: 5 additions & 112 deletions
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,16 @@
11
from __future__ import annotations
22

3-
import json
43
import os
54
import pathlib
6-
import subprocess
75
import typing
86

97
import click
108
import pydantic
119
import yaml
1210

11+
from mergify_cli.ci.scopes import base_detector
12+
from mergify_cli.ci.scopes import changed_files
13+
1314

1415
if typing.TYPE_CHECKING:
1516
from collections import abc
@@ -55,114 +56,6 @@ def from_yaml(cls, path: str) -> Config:
5556
return cls.from_dict(data)
5657

5758

58-
def _run(cmd: list[str]) -> str:
59-
try:
60-
return subprocess.check_output(cmd, text=True, encoding="utf-8").strip()
61-
except subprocess.CalledProcessError as e:
62-
msg = f"Command failed: {' '.join(cmd)}\n{e}"
63-
raise click.ClickException(msg) from e
64-
65-
66-
class MergeQueuePullRequest(typing.TypedDict):
67-
number: int
68-
69-
70-
class MergeQueueBatchFailed(typing.TypedDict):
71-
draft_pr_number: int
72-
checked_pull_request: list[int]
73-
74-
75-
class MergeQueueMetadata(typing.TypedDict):
76-
checking_base_sha: str
77-
pull_requests: list[MergeQueuePullRequest]
78-
previous_failed_batches: list[MergeQueueBatchFailed]
79-
80-
81-
def _yaml_docs_from_fenced_blocks(body: str) -> MergeQueueMetadata | None:
82-
lines = []
83-
found = False
84-
for line in body.splitlines():
85-
if line.startswith("```yaml"):
86-
found = True
87-
elif found:
88-
if line.startswith("```"):
89-
break
90-
lines.append(line)
91-
if lines:
92-
return typing.cast("MergeQueueMetadata", yaml.safe_load("\n".join(lines)))
93-
return None
94-
95-
96-
def _detect_base_from_merge_queue_payload(ev: dict[str, typing.Any]) -> str | None:
97-
pr = ev.get("pull_request")
98-
if not isinstance(pr, dict):
99-
return None
100-
title = pr.get("title") or ""
101-
if not isinstance(title, str):
102-
return None
103-
if not title.startswith("merge-queue: "):
104-
return None
105-
body = pr.get("body") or ""
106-
content = _yaml_docs_from_fenced_blocks(body)
107-
if content:
108-
return content["checking_base_sha"]
109-
return None
110-
111-
112-
def _detect_base_from_event(ev: dict[str, typing.Any]) -> str | None:
113-
pr = ev.get("pull_request")
114-
if isinstance(pr, dict):
115-
sha = pr.get("base", {}).get("sha")
116-
if isinstance(sha, str) and sha:
117-
return sha
118-
return None
119-
120-
121-
def detect_base() -> str:
122-
event_path = os.environ.get("GITHUB_EVENT_PATH")
123-
event: dict[str, typing.Any] | None = None
124-
if event_path and pathlib.Path(event_path).is_file():
125-
try:
126-
with pathlib.Path(event_path).open("r", encoding="utf-8") as f:
127-
event = json.load(f)
128-
except FileNotFoundError:
129-
event = None
130-
131-
if event is not None:
132-
# 0) merge-queue PR override
133-
mq_sha = _detect_base_from_merge_queue_payload(event)
134-
if mq_sha:
135-
return mq_sha
136-
137-
# 1) standard event payload
138-
event_sha = _detect_base_from_event(event)
139-
if event_sha:
140-
return event_sha
141-
142-
# 2) base ref (e.g., PR target branch)
143-
base_ref = os.environ.get("GITHUB_BASE_REF")
144-
if base_ref:
145-
return base_ref
146-
147-
msg = (
148-
"Could not detect base SHA. Ensure checkout has sufficient history "
149-
"(e.g., actions/checkout with fetch-depth: 0) or provide GITHUB_EVENT_PATH / GITHUB_BASE_REF."
150-
)
151-
raise click.ClickException(
152-
msg,
153-
)
154-
155-
156-
def git_changed_files(base: str) -> list[str]:
157-
# Committed changes only between base_sha and HEAD.
158-
# Includes: Added (A), Copied (C), Modified (M), Renamed (R), Type-changed (T), Deleted (D)
159-
# Excludes: Unmerged (U), Unknown (X), Broken (B)
160-
out = _run(
161-
["git", "diff", "--name-only", "--diff-filter=ACMRTD", f"{base}...HEAD"],
162-
)
163-
return [line for line in out.splitlines() if line]
164-
165-
16659
def match_scopes(
16760
config: Config,
16861
files: abc.Iterable[str],
@@ -210,8 +103,8 @@ def maybe_write_github_outputs(
210103

211104
def detect(config_path: str) -> None:
212105
cfg = Config.from_yaml(config_path)
213-
base = detect_base()
214-
changed = git_changed_files(base)
106+
base = base_detector.detect()
107+
changed = changed_files.git_changed_files(base)
215108
scopes_hit, per_scope = match_scopes(cfg, changed)
216109

217110
click.echo(f"Base: {base}")

‎mergify_cli/tests/ci/scopes/test_base_detector.py‎

Lines changed: 108 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,108 @@
1+
import json
2+
import pathlib
3+
4+
import click
5+
import pytest
6+
7+
from mergify_cli.ci.scopes import base_detector
8+
9+
10+
def test_detect_base_github_base_ref(
11+
monkeypatch: pytest.MonkeyPatch,
12+
) -> None:
13+
monkeypatch.setenv("GITHUB_BASE_REF", "main")
14+
monkeypatch.delenv("GITHUB_EVENT_PATH", raising=False)
15+
16+
result = base_detector.detect()
17+
18+
assert result == "main"
19+
20+
21+
def test_detect_base_from_event_path(
22+
monkeypatch: pytest.MonkeyPatch,
23+
tmp_path: pathlib.Path,
24+
) -> None:
25+
event_data = {
26+
"pull_request": {
27+
"base": {"sha": "abc123"},
28+
},
29+
}
30+
event_file = tmp_path / "event.json"
31+
event_file.write_text(json.dumps(event_data))
32+
33+
monkeypatch.setenv("GITHUB_EVENT_PATH", str(event_file))
34+
monkeypatch.delenv("GITHUB_BASE_REF", raising=False)
35+
36+
result = base_detector.detect()
37+
38+
assert result == "abc123"
39+
40+
41+
def test_detect_base_merge_queue_override(
42+
monkeypatch: pytest.MonkeyPatch,
43+
tmp_path: pathlib.Path,
44+
) -> None:
45+
event_data = {
46+
"pull_request": {
47+
"title": "merge-queue: Merge group",
48+
"body": "```yaml\nchecking_base_sha: xyz789\n```",
49+
"base": {"sha": "abc123"},
50+
},
51+
}
52+
event_file = tmp_path / "event.json"
53+
event_file.write_text(json.dumps(event_data))
54+
55+
monkeypatch.setenv("GITHUB_EVENT_PATH", str(event_file))
56+
57+
result = base_detector.detect()
58+
59+
assert result == "xyz789"
60+
61+
62+
def test_detect_base_no_info(monkeypatch: pytest.MonkeyPatch) -> None:
63+
monkeypatch.delenv("GITHUB_EVENT_PATH", raising=False)
64+
monkeypatch.delenv("GITHUB_BASE_REF", raising=False)
65+
66+
with pytest.raises(click.ClickException, match="Could not detect base SHA"):
67+
base_detector.detect()
68+
69+
70+
def test_yaml_docs_from_fenced_blocks_valid() -> None:
71+
body = """Some text
72+
```yaml
73+
---
74+
checking_base_sha: xyz789
75+
pull_requests: [{"number": 1}]
76+
previous_failed_batches: []
77+
...
78+
```
79+
More text"""
80+
81+
result = base_detector._yaml_docs_from_fenced_blocks(body)
82+
83+
assert result == base_detector.MergeQueueMetadata(
84+
{
85+
"checking_base_sha": "xyz789",
86+
"pull_requests": [{"number": 1}],
87+
"previous_failed_batches": [],
88+
},
89+
)
90+
91+
92+
def test_yaml_docs_from_fenced_blocks_no_yaml() -> None:
93+
body = "No yaml here"
94+
95+
result = base_detector._yaml_docs_from_fenced_blocks(body)
96+
97+
assert result is None
98+
99+
100+
def test_yaml_docs_from_fenced_blocks_empty_yaml() -> None:
101+
body = """Some text
102+
```yaml
103+
```
104+
More text"""
105+
106+
result = base_detector._yaml_docs_from_fenced_blocks(body)
107+
108+
assert result is None

0 commit comments

Comments
 (0)