630 lines
22 KiB
Python
630 lines
22 KiB
Python
"""Git→SVN sync direction.
|
||
|
||
Detects new Git commits (developer pushes) and translates them into SVN
|
||
commits, maintaining the mapping DB so that SVN→Git sync skips them.
|
||
"""
|
||
|
||
import logging
|
||
import os
|
||
import re
|
||
import subprocess
|
||
from pathlib import Path
|
||
from typing import Dict, List, Optional, Tuple
|
||
|
||
from .config import SVNConfig
|
||
from .mirror import Mirror
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class GitToSVNError(Exception):
|
||
pass
|
||
|
||
|
||
# ─── Git subprocess helper (no circular import) ────────────────
|
||
|
||
|
||
def _git(git_dir: str, *args: str, input: str = None, timeout: int = 120,
|
||
check: bool = True) -> str:
|
||
"""Run a git command in the bare canonical repo and return stdout."""
|
||
try:
|
||
r = subprocess.run(
|
||
["git", "--git-dir", git_dir, *args],
|
||
input=input,
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=timeout,
|
||
)
|
||
except FileNotFoundError:
|
||
raise GitToSVNError("git not found on PATH")
|
||
except subprocess.TimeoutExpired:
|
||
raise GitToSVNError(f"git command timed out: git {' '.join(args)}")
|
||
|
||
if check and r.returncode != 0:
|
||
raise GitToSVNError(
|
||
f"git command failed (exit {r.returncode}): "
|
||
f"git {' '.join(args)}\n{r.stderr.strip()}"
|
||
)
|
||
return r.stdout.strip() if r.returncode == 0 else ""
|
||
|
||
|
||
# ─── Git ref → SVN path helpers ────────────────────────────────
|
||
|
||
|
||
def _git_ref_to_svn_branch(git_ref: str, svn_cfg: SVNConfig) -> Optional[str]:
|
||
"""Reverse of :func:`branch_to_git_ref`.
|
||
|
||
``refs/heads/master`` → ``"trunk"``
|
||
``refs/heads/feat`` → ``"branches/feat"``
|
||
``refs/tags/v1`` → ``"tags/v1"``
|
||
"""
|
||
if git_ref == "refs/heads/master":
|
||
return svn_cfg.trunk
|
||
if git_ref.startswith("refs/heads/"):
|
||
branch_name = git_ref[len("refs/heads/"):]
|
||
return f"{svn_cfg.branches}/{branch_name}"
|
||
if git_ref.startswith("refs/tags/"):
|
||
tag_name = git_ref[len("refs/tags/"):]
|
||
return f"{svn_cfg.tags}/{tag_name}"
|
||
return None
|
||
|
||
|
||
def _git_ref_to_svn_url(git_ref: str, svn_url: str, svn_cfg: SVNConfig) -> Optional[str]:
|
||
"""Return the full SVN URL for a given Git ref."""
|
||
branch = _git_ref_to_svn_branch(git_ref, svn_cfg)
|
||
if branch is None:
|
||
return None
|
||
url = svn_url.rstrip("/") + "/" + branch
|
||
return url
|
||
|
||
|
||
def _wc_path(mirror: Mirror, svn_branch: str) -> Path:
|
||
"""Return the working-copy path for an SVN branch name.
|
||
|
||
Replaces ``/`` with ``_`` so branch names like ``branches/stable``
|
||
become a flat directory name.
|
||
"""
|
||
sanitized = svn_branch.replace("/", "_")
|
||
return mirror.svn_wc_dir / sanitized
|
||
|
||
|
||
# ─── SVN working copy management ───────────────────────────────
|
||
|
||
|
||
def _run_svn(args: list, input: bytes = None, timeout: int = 120,
|
||
wc: Optional[Path] = None,
|
||
auth_args: Optional[list] = None):
|
||
"""Run an svn command. If *wc* is given, run from that directory."""
|
||
if auth_args:
|
||
args = auth_args + args
|
||
args = ["--non-interactive"] + args
|
||
env = {**os.environ, "LC_ALL": "C.UTF-8"}
|
||
kwargs = dict(
|
||
capture_output=True,
|
||
timeout=timeout,
|
||
cwd=str(wc) if wc else None,
|
||
env=env,
|
||
)
|
||
if input is not None:
|
||
kwargs["input"] = input
|
||
else:
|
||
kwargs["stdin"] = subprocess.DEVNULL
|
||
try:
|
||
r = subprocess.run(["svn"] + args, **kwargs)
|
||
except FileNotFoundError:
|
||
raise GitToSVNError("svn CLI not found – is subversion installed?")
|
||
except subprocess.TimeoutExpired:
|
||
raise GitToSVNError(f"svn command timed out: svn {' '.join(args)}")
|
||
|
||
if r.returncode != 0:
|
||
msg = (r.stderr or r.stdout or b"").decode("utf-8", errors="replace").strip()
|
||
raise GitToSVNError(f"svn failed (exit {r.returncode}): {msg}")
|
||
return r
|
||
|
||
|
||
def _ensure_wc(mirror: Mirror, svn_branch: str):
|
||
"""Ensure the SVN working copy for *svn_branch* exists and is at HEAD."""
|
||
auth_args = mirror.config.svn.auth_args()
|
||
wc = _wc_path(mirror, svn_branch)
|
||
svn_url = mirror.config.svn.url.rstrip("/") + "/" + svn_branch
|
||
if not (wc / ".svn").exists():
|
||
logger.info("Checking out SVN WC for %s …", svn_branch)
|
||
wc.mkdir(parents=True, exist_ok=True)
|
||
try:
|
||
_run_svn(["checkout", "--ignore-externals", svn_url, str(wc)],
|
||
timeout=900, auth_args=auth_args)
|
||
except GitToSVNError as e:
|
||
_cleanup_failed_wc(wc)
|
||
if _is_svn_not_found(e):
|
||
_create_svn_branch(mirror, svn_branch, auth_args)
|
||
wc.mkdir(parents=True, exist_ok=True)
|
||
_run_svn(["checkout", "--ignore-externals", svn_url, str(wc)],
|
||
timeout=900, auth_args=auth_args)
|
||
else:
|
||
raise
|
||
else:
|
||
_try_revert(wc, auth_args)
|
||
try:
|
||
_run_svn(["update", "--ignore-externals"], timeout=600,
|
||
wc=wc, auth_args=auth_args)
|
||
except GitToSVNError as e:
|
||
if _is_with_cleanup_recoverable(e):
|
||
logger.warning("WC interrupted – running cleanup for %s", svn_branch)
|
||
_run_svn(["cleanup"], timeout=600, wc=wc, auth_args=auth_args)
|
||
_try_revert(wc, auth_args)
|
||
_run_svn(["update", "--ignore-externals"], timeout=600,
|
||
wc=wc, auth_args=auth_args)
|
||
else:
|
||
raise
|
||
return wc
|
||
|
||
|
||
_INTERRUPTED_MSGS = ["E155004", "E155037", "previous operation",
|
||
"run.*cleanup", "locked"]
|
||
|
||
|
||
def _is_with_cleanup_recoverable(e: GitToSVNError) -> bool:
|
||
msg = str(e).lower()
|
||
return any(re.search(p.lower(), msg) for p in _INTERRUPTED_MSGS)
|
||
|
||
|
||
def _try_revert(wc: Path, auth_args: Optional[list]):
|
||
try:
|
||
_run_svn(["revert", "-R", "."], timeout=600, wc=wc, auth_args=auth_args)
|
||
except GitToSVNError as e:
|
||
if _is_with_cleanup_recoverable(e):
|
||
logger.warning("WC interrupted during revert – running cleanup")
|
||
_run_svn(["cleanup"], timeout=600, wc=wc, auth_args=auth_args)
|
||
_run_svn(["revert", "-R", "."], timeout=600,
|
||
wc=wc, auth_args=auth_args)
|
||
else:
|
||
raise
|
||
|
||
|
||
def _cleanup_failed_wc(wc: Path):
|
||
"""Remove a partial/failed SVN working copy."""
|
||
import shutil
|
||
import time
|
||
for attempt in range(3):
|
||
try:
|
||
if wc.exists():
|
||
shutil.rmtree(wc)
|
||
return
|
||
except PermissionError:
|
||
time.sleep(1)
|
||
logger.warning("Could not fully remove failed WC at %s", wc)
|
||
|
||
|
||
def _is_svn_not_found(e: GitToSVNError) -> bool:
|
||
return "E170000" in str(e)
|
||
|
||
|
||
def _create_svn_branch(mirror: Mirror, svn_branch: str,
|
||
auth_args: Optional[list]):
|
||
"""Create an SVN branch by copying from trunk.
|
||
|
||
Called when a Git branch has no corresponding SVN path yet.
|
||
"""
|
||
root = mirror.config.svn.url.rstrip("/")
|
||
trunk_url = f"{root}/{mirror.config.svn.trunk}"
|
||
branch_url = f"{root}/{svn_branch}"
|
||
branch_name = svn_branch.rsplit("/", 1)[-1]
|
||
msg = f"Create branch {branch_name} from {mirror.config.svn.trunk}"
|
||
logger.info("Creating SVN branch: %s …", branch_url)
|
||
result = _run_svn(
|
||
["copy", trunk_url, branch_url, "-m", msg],
|
||
timeout=120, auth_args=auth_args,
|
||
)
|
||
new_rev = _parse_commit_revision(result)
|
||
# Restore author revprop (svn copy uses buildbot as committer)
|
||
try:
|
||
_run_svn(
|
||
["propset", "--revprop", "-r", str(new_rev),
|
||
"svn:author", mirror.config.svn.username or "buildbot",
|
||
root],
|
||
timeout=30, auth_args=auth_args,
|
||
)
|
||
except GitToSVNError:
|
||
logger.debug("Could not set svn:author for branch creation r%d", new_rev)
|
||
logger.info("Created SVN branch %s (r%d)", branch_url, new_rev)
|
||
|
||
|
||
# ─── Author mapping (reverse) ──────────────────────────────────
|
||
|
||
|
||
def _parse_git_author(author_line: str) -> Tuple[str, str]:
|
||
"""Split ``"Name <email>"`` into (name, email).
|
||
|
||
Returns (author_line, "") if the format doesn't match.
|
||
"""
|
||
if "<" in author_line and author_line.endswith(">"):
|
||
name = author_line[:author_line.index("<")].strip()
|
||
email = author_line[author_line.index("<") + 1:-1].strip()
|
||
return (name, email)
|
||
return (author_line, "")
|
||
|
||
|
||
def _git_author_to_svn_username(author_name: str, author_email: str,
|
||
authors: Dict[str, str]) -> str:
|
||
"""Try to map a Git author (name, email) back to an SVN username.
|
||
|
||
The ``authors`` dict maps SVN username → ``"Name <email>"``. We
|
||
reverse the lookup by checking both the name and the email.
|
||
"""
|
||
for svn_user, mapped in authors.items():
|
||
name, email = _parse_git_author(mapped)
|
||
if name == author_name or email == author_email:
|
||
return svn_user
|
||
# Fallback: try name as-is
|
||
return author_name
|
||
|
||
|
||
# ─── Git commit detection ──────────────────────────────────────
|
||
|
||
|
||
def _tracked_refs(git_dir: str, svn_cfg: SVNConfig) -> List[Tuple[str, str]]:
|
||
"""Return ``(git_ref, svn_branch)`` pairs for commits → SVN.
|
||
|
||
Only returns ``refs/heads/*`` — tags are handled separately by
|
||
:func:`_detect_new_git_tags`.
|
||
"""
|
||
refs: List[Tuple[str, str]] = []
|
||
out = _git(git_dir, "for-each-ref", "--format=%(refname)", "refs/heads/",
|
||
timeout=30, check=False)
|
||
for ref in out.splitlines():
|
||
ref = ref.strip()
|
||
if not ref:
|
||
continue
|
||
branch = _git_ref_to_svn_branch(ref, svn_cfg)
|
||
if branch is not None:
|
||
refs.append((ref, branch))
|
||
return refs
|
||
|
||
|
||
def detect_new_git_commits(mirror: Mirror) -> List[Tuple[str, str, str]]:
|
||
"""Return ``[(commit_hash, svn_branch, author_name), …]`` oldest-first.
|
||
|
||
Walks each tracked Git ref backwards from its tip and collects
|
||
commits that are NOT yet recorded in the mapping DB.
|
||
"""
|
||
git_dir = str(mirror.canonical_dir)
|
||
svn_cfg = mirror.config.svn
|
||
new: List[Tuple[str, str, str]] = []
|
||
|
||
for git_ref, svn_branch in _tracked_refs(git_dir, svn_cfg):
|
||
out = _git(git_dir, "rev-list", "--topo-order", "--no-merges",
|
||
git_ref, timeout=60, check=False)
|
||
if not out:
|
||
continue
|
||
|
||
commits = out.splitlines()
|
||
unmapped: List[str] = []
|
||
found_boundary = False
|
||
|
||
for ch in commits:
|
||
if not ch:
|
||
continue
|
||
if mirror.db.get_svn_revision(ch) is not None:
|
||
found_boundary = True
|
||
break
|
||
unmapped.append(ch)
|
||
|
||
if not found_boundary:
|
||
logger.debug("No SVN ancestor found for %s – skipping", git_ref)
|
||
continue
|
||
|
||
# Add oldest-first
|
||
for ch in reversed(unmapped):
|
||
author = _git(git_dir, "log", "--format=%aN <%aE>", "-1", ch,
|
||
timeout=30)
|
||
new.append((ch, svn_branch, author))
|
||
|
||
return new
|
||
|
||
|
||
# ─── Applying a single Git commit to SVN ────────────────────────
|
||
|
||
|
||
def _apply_diff_to_wc(mirror: Mirror, commit_hash: str, svn_branch: str, wc: Path):
|
||
"""Apply the file changes from *commit_hash* to the SVN working copy."""
|
||
git_dir = str(mirror.canonical_dir)
|
||
|
||
out = _git(git_dir, "diff-tree", "--no-commit-id", "-r", "--name-status",
|
||
"-z", commit_hash, timeout=60)
|
||
if not out:
|
||
logger.debug("Commit %s has no file changes", commit_hash[:8])
|
||
return
|
||
|
||
# Parse null-separated output: "A\0README\0M\0src/main.rs\0..."
|
||
# Status may include a score suffix, e.g. R100, C075 – extract the letter.
|
||
parts = out.split("\0")
|
||
i = 0
|
||
while i < len(parts):
|
||
line = parts[i].strip()
|
||
if not line:
|
||
i += 1
|
||
continue
|
||
status = line[0]
|
||
if status not in ("A", "C", "D", "M", "R", "T"):
|
||
i += 1
|
||
continue
|
||
i += 1
|
||
if i >= len(parts):
|
||
break
|
||
path = parts[i]
|
||
i += 1
|
||
if status == "D":
|
||
_wc_delete(wc, path)
|
||
elif status in ("A", "C", "M", "T"):
|
||
_wc_add_or_modify(wc, git_dir, commit_hash, path)
|
||
elif status == "R":
|
||
if i < len(parts):
|
||
dest = parts[i]
|
||
i += 1
|
||
_wc_rename(wc, path, dest)
|
||
|
||
|
||
def _wc_delete(wc: Path, path: str):
|
||
full = wc / path
|
||
if full.exists():
|
||
_run_svn(["delete", "--force", str(full)], wc=wc)
|
||
logger.debug(" D %s", path)
|
||
|
||
|
||
def _wc_add_or_modify(wc: Path, git_dir: str, commit_hash: str, path: str):
|
||
full = wc / path
|
||
parent = full.parent
|
||
if not parent.exists():
|
||
parent.mkdir(parents=True, exist_ok=True)
|
||
_run_svn(["add", "--parents", str(parent)], wc=wc)
|
||
|
||
is_new = not full.exists()
|
||
full.parent.mkdir(parents=True, exist_ok=True)
|
||
|
||
r = subprocess.run(
|
||
["git", "--git-dir", git_dir, "show", f"{commit_hash}:{path}"],
|
||
capture_output=True, timeout=60,
|
||
)
|
||
if r.returncode == 0:
|
||
full.write_bytes(r.stdout)
|
||
|
||
if is_new:
|
||
_run_svn(["add", "--parents", str(full)], wc=wc)
|
||
logger.debug(" M %s", path)
|
||
|
||
|
||
def _wc_rename(wc: Path, src: str, dest: str):
|
||
src_full = wc / src
|
||
dest_full = wc / dest
|
||
if src_full.exists():
|
||
dest_full.parent.mkdir(parents=True, exist_ok=True)
|
||
_run_svn(["move", str(src_full), str(dest_full)], wc=wc)
|
||
logger.debug(" R %s → %s", src, dest)
|
||
|
||
|
||
# ─── Parsing helpers ──────────────────────────────────────────
|
||
|
||
|
||
def _parse_commit_revision(result: subprocess.CompletedProcess) -> int:
|
||
"""Parse revision from ``svn commit`` output (on either stdout or stderr)."""
|
||
text = (result.stdout + b"\n" + result.stderr).decode("utf-8", errors="replace")
|
||
import re
|
||
m = re.search(r"Committed revision\s+(\d+)", text)
|
||
if m:
|
||
return int(m.group(1))
|
||
raise GitToSVNError(
|
||
f"Cannot determine committed revision from svn output:\n{text[:500]}"
|
||
)
|
||
|
||
|
||
# ─── Main entry point ──────────────────────────────────────────
|
||
|
||
|
||
def _detect_new_git_tags(mirror: Mirror) -> List[Tuple[str, str, int, str, str]]:
|
||
"""Find Git tags not yet pushed to SVN.
|
||
|
||
Returns ``[(tag_ref, svn_branch, source_svn_rev, author_line), …]``.
|
||
"""
|
||
git_dir = str(mirror.canonical_dir)
|
||
svn_cfg = mirror.config.svn
|
||
new: List[Tuple[str, str, int, str, str]] = []
|
||
|
||
out = _git(git_dir, "for-each-ref", "--format=%(refname)", "refs/tags/",
|
||
timeout=30, check=False)
|
||
for tag_ref in out.splitlines():
|
||
tag_ref = tag_ref.strip()
|
||
if not tag_ref:
|
||
continue
|
||
svn_branch = _git_ref_to_svn_branch(tag_ref, svn_cfg)
|
||
if svn_branch is None:
|
||
continue
|
||
# Already pushed to SVN?
|
||
if mirror.db.ref_exists(tag_ref):
|
||
continue
|
||
|
||
target = _git(git_dir, "rev-parse", tag_ref, timeout=30)
|
||
mapping = mirror.db.get_mapping_by_git_hash(target)
|
||
if mapping is None:
|
||
logger.debug(
|
||
"Tag %s target commit not in mapping DB – skipping", tag_ref,
|
||
)
|
||
continue
|
||
source_rev = mapping["svn_revision"]
|
||
source_branch = mapping["svn_branch"]
|
||
author = _git(git_dir, "log", "--format=%aN <%aE>", "-1", target,
|
||
timeout=30)
|
||
new.append((tag_ref, svn_branch, source_rev, source_branch, author))
|
||
|
||
return new
|
||
|
||
|
||
def sync_git_to_svn(mirror: Mirror) -> int:
|
||
"""Push all pending Git commits and tags to SVN (thread-safe).
|
||
|
||
Returns the number of SVN revisions created.
|
||
"""
|
||
with mirror.sync_lock():
|
||
return _sync_git_to_svn_impl(mirror)
|
||
|
||
|
||
def _sync_git_to_svn_impl(mirror: Mirror) -> int:
|
||
auth_args = mirror.config.svn.auth_args()
|
||
git_dir = str(mirror.canonical_dir)
|
||
svn_cfg = mirror.config.svn
|
||
count = 0
|
||
|
||
# ── 1. Push new commits on branches ───────────────────────
|
||
commits = detect_new_git_commits(mirror)
|
||
for commit_hash, svn_branch, author_line in commits:
|
||
wc = _ensure_wc(mirror, svn_branch)
|
||
|
||
logger.info(
|
||
"Pushing Git commit %s → SVN %s", commit_hash[:8], svn_branch,
|
||
)
|
||
|
||
try:
|
||
_apply_diff_to_wc(mirror, commit_hash, svn_branch, wc)
|
||
|
||
message = _git(git_dir, "log", "--format=%B", "-1", commit_hash,
|
||
timeout=30)
|
||
|
||
svn_username = author_line
|
||
name, email = _parse_git_author(author_line)
|
||
if name != author_line and email:
|
||
svn_username = _git_author_to_svn_username(
|
||
name, email, mirror.config.authors,
|
||
)
|
||
|
||
result = _run_svn(
|
||
["commit", "--file", "-"],
|
||
input=message.encode("utf-8"),
|
||
wc=wc, timeout=120, auth_args=auth_args,
|
||
)
|
||
try:
|
||
new_rev = _parse_commit_revision(result)
|
||
except GitToSVNError:
|
||
logger.warning(
|
||
"svn commit no-op for %s on %s – recording HEAD",
|
||
commit_hash[:8], svn_branch,
|
||
)
|
||
info = subprocess.run(
|
||
["svnversion", str(wc)],
|
||
capture_output=True, text=True, timeout=30,
|
||
env={**os.environ, "LC_ALL": "C"},
|
||
)
|
||
rev_text = (info.stdout or "").strip().split(":")[0]
|
||
try:
|
||
new_rev = int(rev_text)
|
||
except (ValueError, TypeError):
|
||
logger.warning(
|
||
"Cannot determine current SVN revision for %s – "
|
||
"skipping mapping",
|
||
svn_branch,
|
||
)
|
||
continue
|
||
|
||
# Restore the real author as svn:author revprop
|
||
try:
|
||
_run_svn(
|
||
["propset", "--revprop", "-r", str(new_rev),
|
||
"svn:author", svn_username],
|
||
timeout=30, auth_args=auth_args, wc=wc,
|
||
)
|
||
except Exception:
|
||
logger.warning(
|
||
"Could not set svn:author for r%d (pre-revprop-change hook?)",
|
||
new_rev,
|
||
)
|
||
|
||
mirror.db.record_mapping(
|
||
new_rev, commit_hash, svn_branch,
|
||
_git_ref_from_branch(svn_branch, svn_cfg),
|
||
source="git",
|
||
)
|
||
|
||
logger.info(
|
||
"Pushed Git %s → SVN r%d (%s)",
|
||
commit_hash[:8], new_rev, svn_branch,
|
||
)
|
||
count += 1
|
||
|
||
except Exception:
|
||
logger.exception(
|
||
"Failed to push Git commit %s to SVN branch %s",
|
||
commit_hash[:8], svn_branch,
|
||
)
|
||
raise
|
||
|
||
# ── 2. Push new tags ──────────────────────────────────────
|
||
svn_url = svn_cfg.url.rstrip("/")
|
||
for tag_ref, svn_branch, source_rev, source_branch, author_line \
|
||
in _detect_new_git_tags(mirror):
|
||
logger.info("Pushing Git tag %s → SVN %s", tag_ref, svn_branch)
|
||
|
||
try:
|
||
# Build source URL (the SVN path being tagged)
|
||
src_url = f"{svn_url}/{source_branch}"
|
||
dst_url = f"{svn_url}/{svn_branch}"
|
||
|
||
message = _git(git_dir, "log", "--format=%B", "-1", tag_ref,
|
||
timeout=30)
|
||
|
||
svn_username = author_line
|
||
name, email = _parse_git_author(author_line)
|
||
if name != author_line and email:
|
||
svn_username = _git_author_to_svn_username(
|
||
name, email, mirror.config.authors,
|
||
)
|
||
|
||
result = _run_svn(
|
||
["copy", "-r", str(source_rev), src_url, dst_url,
|
||
"-m", message],
|
||
timeout=120, auth_args=auth_args,
|
||
)
|
||
new_rev = _parse_commit_revision(result)
|
||
|
||
# Restore the real author as svn:author revprop
|
||
try:
|
||
svn_root = svn_cfg.url.rstrip("/")
|
||
_run_svn(
|
||
["propset", "--revprop", "-r", str(new_rev),
|
||
"svn:author", svn_username, svn_root],
|
||
timeout=30, auth_args=auth_args,
|
||
)
|
||
except Exception:
|
||
logger.warning(
|
||
"Could not set svn:author for r%d (pre-revprop-change hook?)",
|
||
new_rev,
|
||
)
|
||
|
||
# The commit hash stored for a tag is the target commit hash
|
||
target = _git(git_dir, "rev-parse", tag_ref, timeout=30)
|
||
mirror.db.record_mapping(
|
||
new_rev, target, svn_branch, tag_ref, source="git",
|
||
)
|
||
|
||
logger.info(
|
||
"Pushed Git tag %s → SVN r%d (%s)", tag_ref, new_rev,
|
||
svn_branch,
|
||
)
|
||
count += 1
|
||
|
||
except Exception:
|
||
logger.exception(
|
||
"Failed to push Git tag %s to SVN", tag_ref,
|
||
)
|
||
raise
|
||
|
||
return count
|
||
|
||
|
||
def _git_ref_from_branch(svn_branch: str, svn_cfg: SVNConfig) -> str:
|
||
"""Convert SVN branch name to Git ref (mirror of :func:`branch_to_git_ref`)."""
|
||
if svn_branch == svn_cfg.trunk:
|
||
return "refs/heads/master"
|
||
if svn_branch.startswith(svn_cfg.branches + "/"):
|
||
return "refs/heads/" + svn_branch[len(svn_cfg.branches) + 1:]
|
||
if svn_branch.startswith(svn_cfg.tags + "/"):
|
||
return "refs/tags/" + svn_branch[len(svn_cfg.tags) + 1:]
|
||
return "refs/heads/" + svn_branch
|