Skip to content
20 changes: 20 additions & 0 deletions .changelog/030.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
name: __ROOT__
ts: 2026-02-25 07:05:40.310869+00:00
type: fix
author: Espen Albert
changelog_message: 'fix(git-ops): truncate PR body to stay within GitHub''s 65536-char limit'
message: 'fix(git-ops): truncate PR body to stay within GitHub''s 65536-char limit'
short_sha: c30f60
---
name: remote_branch_has_same_content
ts: 2026-02-25 07:29:42.955395+00:00
type: keep_private
full_path: _internal.git_ops.remote_branch_has_same_content
---
name: __ROOT__
ts: 2026-02-25 07:29:47.509062+00:00
type: fix
author: Espen Albert
changelog_message: 'fix: skip force-push when remote branch has same content'
message: 'fix: skip force-push when remote branch has same content'
short_sha: 84c7f3
6 changes: 4 additions & 2 deletions path_sync/_internal/cmd_copy.py
Original file line number Diff line number Diff line change
Expand Up @@ -607,8 +607,10 @@ def _push_and_pr(
if not prompt_utils.prompt_confirm(f"Push {dest.name} to origin?", opts.no_prompt):
return None

git_ops.push_branch(repo, copy_branch, force=True)
typer.echo(f" Pushed: {copy_branch} (force)", err=True)
if git_ops.push_branch(repo, copy_branch, force=True):
typer.echo(f" Pushed: {copy_branch} (force)", err=True)
else:
typer.echo(f" Skipped push for {copy_branch}: content unchanged on remote", err=True)

if opts.no_pr or not prompt_utils.prompt_confirm(f"Create PR for {dest.name}?", opts.no_prompt):
return None
Expand Down
7 changes: 5 additions & 2 deletions path_sync/_internal/cmd_dep_update.py
Original file line number Diff line number Diff line change
Expand Up @@ -201,9 +201,12 @@ def _create_prs(config: DepConfig, results: list[RepoResult], opts: DepUpdateOpt
continue

repo = git_ops.get_repo(result.repo_path)
git_ops.push_branch(repo, config.pr.branch, force=True)

body = _build_pr_body(result.log_content, result.failures)

if not git_ops.push_branch(repo, config.pr.branch, force=True):
git_ops.update_pr_body(result.repo_path, config.pr.branch, body)
continue

pr_url = git_ops.create_or_update_pr(
result.repo_path,
config.pr.branch,
Expand Down
48 changes: 48 additions & 0 deletions path_sync/_internal/cmd_dep_update_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,9 @@
from path_sync._internal import verify as verify_module
from path_sync._internal.cmd_dep_update import (
DepUpdateOptions,
RepoResult,
Status,
_create_prs,
_process_single_repo,
_run_updates,
_update_and_validate,
Expand Down Expand Up @@ -197,3 +199,49 @@ def test_update_and_validate_keeps_pr_when_config_flag_set(
assert not results
git_ops.has_open_pr.assert_not_called()
git_ops.close_pr.assert_not_called()


# --- _create_prs tests ---

CREATE_PRS_MODULE = _create_prs.__module__


def test_create_prs_skips_push_when_content_unchanged(config: DepConfig, tmp_path: Path):
result = RepoResult(
dest=Destination(name="test-repo", dest_path_relative="code/test-repo", default_branch="main"),
repo_path=tmp_path,
status=Status.PASSED,
log_content="some output",
)
opts = DepUpdateOptions()

with patch(f"{CREATE_PRS_MODULE}.git_ops") as git_ops:
git_ops.get_repo.return_value = MagicMock()
git_ops.push_branch.return_value = False

pr_refs = _create_prs(config, [result], opts)

git_ops.create_or_update_pr.assert_not_called()
git_ops.update_pr_body.assert_called_once()
assert not pr_refs


def test_create_prs_pushes_when_content_changed(config: DepConfig, tmp_path: Path):
result = RepoResult(
dest=Destination(name="test-repo", dest_path_relative="code/test-repo", default_branch="main"),
repo_path=tmp_path,
status=Status.PASSED,
log_content="some output",
)
opts = DepUpdateOptions()

with patch(f"{CREATE_PRS_MODULE}.git_ops") as git_ops:
git_ops.get_repo.return_value = MagicMock()
git_ops.push_branch.return_value = True
git_ops.create_or_update_pr.return_value = "https://github.com/test/pr/1"

pr_refs = _create_prs(config, [result], opts)

git_ops.push_branch.assert_called_once_with(git_ops.get_repo.return_value, "chore/deps", force=True)
git_ops.create_or_update_pr.assert_called_once()
assert len(pr_refs) == 1
35 changes: 33 additions & 2 deletions path_sync/_internal/git_ops.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import logging
import os
import re
import subprocess
from contextlib import suppress
from pathlib import Path
Expand All @@ -10,6 +11,20 @@

logger = logging.getLogger(__name__)

GH_PR_BODY_MAX_CHARS = 64536 # real limit is 65536, but we leave some buffer
_TRUNCATION_NOTICE = "\n\n... (truncated, output too long for PR body)"
_CODE_FENCE_RE = re.compile(r"^`{3,}", re.MULTILINE)


def _truncate_body(body: str) -> str:
if len(body) <= GH_PR_BODY_MAX_CHARS:
return body
cut = GH_PR_BODY_MAX_CHARS - len(_TRUNCATION_NOTICE)
truncated = body[:cut]
fence_count = len(_CODE_FENCE_RE.findall(truncated))
suffix = "\n```" + _TRUNCATION_NOTICE if fence_count % 2 else _TRUNCATION_NOTICE
return truncated[: GH_PR_BODY_MAX_CHARS - len(suffix)] + suffix


def _auth_url(url: str) -> str:
"""Inject GH_TOKEN into HTTPS URL for authentication."""
Expand Down Expand Up @@ -145,10 +160,24 @@ def _ensure_git_user(repo: Repo) -> None:
repo.config_writer().set_value("user", "email", "path-sync[bot]@users.noreply.github.com").release()


def push_branch(repo: Repo, branch: str, force: bool = True) -> None:
def remote_branch_has_same_content(repo: Repo, branch: str) -> bool:
"""Check if origin/{branch} has identical file content (tree) as local {branch}."""
with suppress(GitCommandError):
local_tree = repo.git.rev_parse(f"{branch}^{{tree}}")
remote_tree = repo.git.rev_parse(f"origin/{branch}^{{tree}}")
return local_tree == remote_tree
return False


def push_branch(repo: Repo, branch: str, force: bool = True) -> bool:
"""Push branch to origin. Returns False if skipped (content unchanged on remote)."""
if force and remote_branch_has_same_content(repo, branch):
logger.info(f"Skipping push for {branch}: content unchanged on remote")
return False
logger.info(f"Pushing {branch}" + (" (force)" if force else ""))
args = ["--force", "-u", "origin", branch] if force else ["-u", "origin", branch]
repo.git.push(*args)
return True


def _get_repo_full_name(repo_path: Path) -> str | None:
Expand Down Expand Up @@ -189,6 +218,7 @@ def update_pr_body(repo_path: Path, branch: str, body: str) -> bool:
if not pr_number:
return False

body = _truncate_body(body)
cmd = [
"gh",
"api",
Expand Down Expand Up @@ -243,8 +273,9 @@ def create_or_update_pr(
reviewers: list[str] | None = None,
assignees: list[str] | None = None,
) -> str:
body = _truncate_body(body) if body else ""
cmd = ["gh", "pr", "create", "--head", branch, "--title", title]
cmd.extend(["--body", body or ""])
cmd.extend(["--body", body])
if labels:
cmd.extend(["--label", ",".join(labels)])
if reviewers:
Expand Down
133 changes: 133 additions & 0 deletions path_sync/_internal/git_ops_test.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,133 @@
from __future__ import annotations

from pathlib import Path

from git import Repo

from path_sync._internal.git_ops import (
GH_PR_BODY_MAX_CHARS,
_truncate_body,
push_branch,
remote_branch_has_same_content,
)


def test_truncate_body_short_unchanged():
body = "short body"
assert _truncate_body(body) == body


def test_truncate_body_at_limit_unchanged():
body = "x" * GH_PR_BODY_MAX_CHARS
assert _truncate_body(body) == body


def test_truncate_body_over_limit():
body = "x" * (GH_PR_BODY_MAX_CHARS + 500)
result = _truncate_body(body)
assert len(result) == GH_PR_BODY_MAX_CHARS
assert result.endswith("... (truncated, output too long for PR body)")


def test_truncate_body_closes_open_code_fence():
prefix = "## Command Output\n\n```\n"
filler = "log line\n" * 10_000
body = prefix + filler
assert len(body) > GH_PR_BODY_MAX_CHARS

result = _truncate_body(body)
assert len(result) <= GH_PR_BODY_MAX_CHARS
assert "\n```\n" in result[len(prefix) :]
assert result.endswith("... (truncated, output too long for PR body)")


def test_truncate_body_no_extra_fence_when_closed():
prefix = "## Output\n\n```\nsome log\n```\n\nmore text\n"
filler = "x" * GH_PR_BODY_MAX_CHARS
body = prefix + filler

result = _truncate_body(body)
assert len(result) == GH_PR_BODY_MAX_CHARS
assert "```\n\n... (truncated" not in result


def _init_repo_with_remote(tmp_path: Path) -> tuple[Repo, Repo]:
"""Create a bare 'remote' and a clone that points to it."""
bare_path = tmp_path / "remote.git"
bare = Repo.init(bare_path, bare=True)
bare.git.symbolic_ref("HEAD", "refs/heads/main")

clone_path = tmp_path / "clone"
clone = Repo.clone_from(str(bare_path), str(clone_path))
(clone_path / "file.txt").write_text("initial")
clone.index.add(["file.txt"])
clone.index.commit("initial")
clone.git.push("-u", "origin", "main")
return bare, clone


def test_same_content_returns_true(tmp_path: Path):
_, clone = _init_repo_with_remote(tmp_path)
clone.git.checkout("-b", "feature")
(Path(clone.working_dir) / "file.txt").write_text("updated")
clone.index.add(["file.txt"])
clone.index.commit("first")
clone.git.push("-u", "origin", "feature")

# New commit with same tree (content unchanged)
clone.git.commit("--allow-empty", "-m", "second")

assert remote_branch_has_same_content(clone, "feature")


def test_different_content_returns_false(tmp_path: Path):
_, clone = _init_repo_with_remote(tmp_path)
clone.git.checkout("-b", "feature")
(Path(clone.working_dir) / "file.txt").write_text("v1")
clone.index.add(["file.txt"])
clone.index.commit("first")
clone.git.push("-u", "origin", "feature")

(Path(clone.working_dir) / "file.txt").write_text("v2")
clone.index.add(["file.txt"])
clone.index.commit("second")

assert not remote_branch_has_same_content(clone, "feature")


def test_no_remote_branch_returns_false(tmp_path: Path):
_, clone = _init_repo_with_remote(tmp_path)
clone.git.checkout("-b", "new-branch")
(Path(clone.working_dir) / "file.txt").write_text("changed")
clone.index.add(["file.txt"])
clone.index.commit("first")

assert not remote_branch_has_same_content(clone, "new-branch")


def test_push_branch_skips_when_content_unchanged(tmp_path: Path):
_, clone = _init_repo_with_remote(tmp_path)
clone.git.checkout("-b", "feature")
(Path(clone.working_dir) / "file.txt").write_text("updated")
clone.index.add(["file.txt"])
clone.index.commit("first")
clone.git.push("-u", "origin", "feature")

clone.git.commit("--allow-empty", "-m", "empty commit")

assert not push_branch(clone, "feature", force=True)


def test_push_branch_pushes_when_content_differs(tmp_path: Path):
_, clone = _init_repo_with_remote(tmp_path)
clone.git.checkout("-b", "feature")
(Path(clone.working_dir) / "file.txt").write_text("v1")
clone.index.add(["file.txt"])
clone.index.commit("first")
clone.git.push("-u", "origin", "feature")

(Path(clone.working_dir) / "file.txt").write_text("v2")
clone.index.add(["file.txt"])
clone.index.commit("second")

assert push_branch(clone, "feature", force=True)