Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 6 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

feat: add mirror_refs.sh for bidirectional ref syncing

Browse Source 
- mirror_refs.sh: additive branch+tag mirroring between dadeschools (HTTPS)
  and prgs (SSH:2222). Dry-run default, --apply to execute, --force for
  diverged branches. Uses bare repo cache for isolation.
- test_mirror_refs.py: flag parsing, safety defaults, brace-delimited refspec
  validation, and local bare-repo integration tests (FF detection, branch/tag
  comparison).
- README.md: document mirror_refs.sh, test suite, and multi-instance auth.
This commit is contained in:
Jason Walker
2026-06-21 18:08:53 -04:00
parent c4c9993039
commit dd6f1308c1
3 changed files with 620 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+47 -6
View File
@@ -11,20 +11,23 @@ A collection of Python and Bash scripts to automate interactions with Gitea inst
## Authentication
These scripts securely extract tokens from the macOS keychain to avoid hardcoding secrets.
Scripts extract credentials from the macOS keychain automatically — no tokens on the command line.
Credentials are retrieved via `git credential fill` for the target host. Ensure you have
logged in via Git over HTTPS at least once so the keychain caches your credentials.
- **dadeschools** — HTTPS via `git credential fill` (SSH:2222 is flaky)
- **prgs** — SSH via `ssh://git@gitea-ssh.prgs.cc:2222` (SSH is reliable here)
Ensure you've logged in via Git over HTTPS at least once so the keychain caches your credentials.
## Available Scripts
| Script | Description |
|---------------------|-----------------------------------------------------|
| `create_issue.py` | Create an issue (`--remote`, `--title`, `--body`) |
|---------------------|--------------------------------------------------------------------|
| `create_issue.py` | Create an issue (`--remote`, `--title`, `--body`, `--body-file`) |
| `create_pr.py` | Open a Pull Request (`--remote`, `--title`, `--head`, `--base`) |
| `close_issue.sh` | Close a specific issue (dadeschools only) |
| `mark_issue.sh` | Claim/release an issue via `status:in-progress` label |
| `manage_labels.py` | Create label set and apply label mappings (`--dry` to preview) |
| `mirror_refs.sh` | Mirror branches + tags between dadeschools ⇄ prgs |
### Quick Examples
@@ -32,6 +35,9 @@ logged in via Git over HTTPS at least once so the keychain caches your credentia
# Create an issue
./create_issue.py --title "Fix PDF output" --body "Blank on Safari"
# Create an issue on the prgs instance
./create_issue.py --remote prgs --title "Add tests" --body-file description.md
# Create a PR
./create_pr.py --title "feat: add validation" --head feat/validation --body "Closes #12"
@@ -43,6 +49,41 @@ logged in via Git over HTTPS at least once so the keychain caches your credentia
# Release when done
./mark_issue.sh 10 done
# Mirror refs (dry-run by default)
./mirror_refs.sh
# Actually push the refs
./mirror_refs.sh --apply
```
Use `--help` on any Python script for full usage details.
Use `--help` on any Python script or shell script for full usage details.
## Mirror Refs
`mirror_refs.sh` keeps branches and tags in sync between dadeschools and prgs:
- **Additive only** — never deletes branches or tags
- **Dry-run by default** — pass `--apply` to actually push
- **Divergence protection** — shared branches that have diverged are skipped with a warning; pass `--force` to override
- Uses a bare repo cache in `/tmp/gitea-mirror-Timesheet` for isolation
- Won't auto-close or merge anything — just ref mirroring
## Tests
Run the full test suite:
```bash
python3 -m pytest tests/ -v
```
| Test file | Covers |
|--------------------------|---------------------------------------------------------|
| `test_create_issue.py` | Arg parsing, remote resolution, payload, auth, errors |
| `test_create_pr.py` | Arg parsing, remote resolution, payload, auth, errors |
| `test_credentials.py` | `get_credentials()` parsing edge cases |
| `test_manage_labels.py` | Label create/skip, dry run, mapping, constant validation|
| `test_shell_scripts.py` | `close_issue.sh` + `mark_issue.sh` arg validation |
| `test_mirror_refs.py` | Flags, safety defaults, local integration tests |
All tests mock network and keychain access — no real API calls are made.
mirror_refs.sh
Executable
+299
View File
@@ -0,0 +1,299 @@
#!/usr/bin/env bash
# mirror_refs.sh — Additive ref mirroring between two Gitea instances.
#
# Mirrors branches and tags between dadeschools and prgs Timesheet repos.
# Safe by default: dry-run, additive-only, divergence-skipping.
#
# Usage:
# ./mirror_refs.sh # dry-run (default) — shows what would happen
# ./mirror_refs.sh --apply # actually push
# ./mirror_refs.sh --apply --force # also force-push diverged branches
#
# Auth:
# dadeschools: HTTPS via `git credential fill` (SSH:2222 flaky)
# prgs: SSH via ssh://git@gitea-ssh.prgs.cc:2222 (SSH ok)
#
# Safety:
# - Additive only: never deletes branches/tags.
# - Divergence protection: if a shared branch has diverged (no FF path),
# skip + warn unless --force is passed.
# - Won't auto-close/merge anything — just ref mirroring.
set -euo pipefail
# ── Configuration ─────────────────────────────────────────────────────────────
DS_HOST="gitea.dadeschools.net"
DS_ORG="Contractor"
DS_REPO="Timesheet"
PRGS_SSH_URL="ssh://git@gitea-ssh.prgs.cc:2222/Scaled-Tech-Consulting/Timesheet.git"
CACHE_DIR="/tmp/gitea-mirror-${DS_REPO}"
# ── Flag parsing ──────────────────────────────────────────────────────────────
DRY_RUN=true
FORCE=false
for arg in "$@"; do
case "$arg" in
--apply) DRY_RUN=false ;;
--force) FORCE=true ;;
--help|-h)
sed -n '2,/^$/{ s/^# //; s/^#$//; p; }' "$0"
exit 0
;;
*)
echo "Unknown flag: $arg (try --help)" >&2
exit 1
;;
esac
done
if $DRY_RUN; then
echo "═══ DRY RUN (pass --apply to execute) ═══"
echo ""
fi
# ── Helpers ───────────────────────────────────────────────────────────────────
log() { printf " %s\n" "$*"; }
ok() { printf " ✓ %s\n" "$*"; }
warn() { printf " ⚠ %s\n" "$*" >&2; }
err() { printf " ✗ %s\n" "$*" >&2; }
# Build an authenticated HTTPS URL for dadeschools using the macOS keychain.
get_ds_url() {
local creds user pass
creds=$(printf "host=%s\nprotocol=https\n\n" "$DS_HOST" | git credential fill 2>/dev/null) || true
user=$(printf '%s\n' "$creds" | sed -n 's/^username=//p')
pass=$(printf '%s\n' "$creds" | sed -n 's/^password=//p')
if [[ -z "$user" || -z "$pass" ]]; then
err "Could not get credentials for $DS_HOST"
err "Ensure you've logged in via HTTPS at least once (git credential fill)."
exit 1
fi
echo "https://${user}:${pass}@${DS_HOST}/${DS_ORG}/${DS_REPO}.git"
}
# Run a git command, or print it if dry-run.
run_or_dry() {
if $DRY_RUN; then
log "[dry-run] $*"
else
"$@"
fi
}
# ── Set up bare mirror cache ─────────────────────────────────────────────────
setup_cache() {
if [[ ! -d "$CACHE_DIR" ]]; then
log "Creating mirror cache at $CACHE_DIR"
git init --bare "$CACHE_DIR" --quiet
fi
local ds_url
ds_url=$(get_ds_url)
# Set/update remotes
git -C "$CACHE_DIR" remote remove ds 2>/dev/null || true
git -C "$CACHE_DIR" remote remove prgs 2>/dev/null || true
git -C "$CACHE_DIR" remote add ds "$ds_url"
git -C "$CACHE_DIR" remote add prgs "$PRGS_SSH_URL"
}
# ── Fetch ─────────────────────────────────────────────────────────────────────
fetch_both() {
echo "── Fetching dadeschools ──"
git -C "$CACHE_DIR" fetch ds --tags --prune --quiet 2>&1 | sed 's/^/ /'
ok "dadeschools fetched"
echo "── Fetching prgs ──"
git -C "$CACHE_DIR" fetch prgs --tags --prune --quiet 2>&1 | sed 's/^/ /'
ok "prgs fetched"
echo ""
}
# ── Branch mirroring ──────────────────────────────────────────────────────────
mirror_branches() {
echo "── Branch comparison ──"
# Collect branch names from each remote (strip "refs/remotes/<remote>/")
local -A ds_branches prgs_branches
local ref
while IFS= read -r ref; do
local name="${ref#refs/remotes/ds/}"
[[ "$name" == "HEAD" ]] && continue
ds_branches["$name"]=$(git -C "$CACHE_DIR" rev-parse "$ref")
done < <(git -C "$CACHE_DIR" for-each-ref --format='%(refname)' refs/remotes/ds/)
while IFS= read -r ref; do
local name="${ref#refs/remotes/prgs/}"
[[ "$name" == "HEAD" ]] && continue
prgs_branches["$name"]=$(git -C "$CACHE_DIR" rev-parse "$ref")
done < <(git -C "$CACHE_DIR" for-each-ref --format='%(refname)' refs/remotes/prgs/)
local total_ds=${#ds_branches[@]}
local total_prgs=${#prgs_branches[@]}
log "dadeschools: $total_ds branches, prgs: $total_prgs branches"
local pushed=0 skipped=0
# All unique branch names
local -A all_branches
for b in "${!ds_branches[@]}"; do all_branches["$b"]=1; done
for b in "${!prgs_branches[@]}"; do all_branches["$b"]=1; done
for b in $(printf '%s\n' "${!all_branches[@]}" | sort); do
local ds_sha="${ds_branches[$b]:-}"
local prgs_sha="${prgs_branches[$b]:-}"
if [[ -n "$ds_sha" && -z "$prgs_sha" ]]; then
# Only on dadeschools → push to prgs
ok "$b → prgs (new)"
run_or_dry git -C "$CACHE_DIR" push prgs "refs/remotes/ds/${b}:refs/heads/${b}"
((pushed++)) || true
elif [[ -z "$ds_sha" && -n "$prgs_sha" ]]; then
# Only on prgs → push to dadeschools
ok "$b → dadeschools (new)"
run_or_dry git -C "$CACHE_DIR" push ds "refs/remotes/prgs/${b}:refs/heads/${b}"
((pushed++)) || true
elif [[ "$ds_sha" == "$prgs_sha" ]]; then
# In sync — nothing to do
log "$b ✔ in sync (${ds_sha:0:8})"
else
# Both have it, different SHAs — check fast-forward
if git -C "$CACHE_DIR" merge-base --is-ancestor "$ds_sha" "$prgs_sha" 2>/dev/null; then
# dadeschools is behind prgs — push prgs → ds
ok "$b → dadeschools (ff: ${ds_sha:0:8}..${prgs_sha:0:8})"
run_or_dry git -C "$CACHE_DIR" push ds "refs/remotes/prgs/${b}:refs/heads/${b}"
((pushed++)) || true
elif git -C "$CACHE_DIR" merge-base --is-ancestor "$prgs_sha" "$ds_sha" 2>/dev/null; then
# prgs is behind dadeschools — push ds → prgs
ok "$b → prgs (ff: ${prgs_sha:0:8}..${ds_sha:0:8})"
run_or_dry git -C "$CACHE_DIR" push prgs "refs/remotes/ds/${b}:refs/heads/${b}"
((pushed++)) || true
else
# Diverged
if $FORCE; then
warn "$b DIVERGED — force-pushing dadeschools → prgs"
run_or_dry git -C "$CACHE_DIR" push prgs --force "refs/remotes/ds/${b}:refs/heads/${b}"
((pushed++)) || true
else
warn "$b DIVERGED (ds: ${ds_sha:0:8}, prgs: ${prgs_sha:0:8}) — skipping (use --force)"
((skipped++)) || true
fi
fi
fi
done
echo ""
log "Branches: $pushed pushed, $skipped skipped"
echo ""
}
# ── Tag mirroring ─────────────────────────────────────────────────────────────
mirror_tags() {
echo "── Tag comparison ──"
local -A ds_tags prgs_tags
local ref
while IFS= read -r ref; do
[[ -z "$ref" ]] && continue
local name="${ref#refs/tags/}"
# Dereference annotated tags to the underlying commit
ds_tags["$name"]=$(git -C "$CACHE_DIR" rev-parse "${ref}^{}" 2>/dev/null || git -C "$CACHE_DIR" rev-parse "$ref")
done < <(git -C "$CACHE_DIR" for-each-ref --format='%(refname)' "refs/tags/" 2>/dev/null)
# Tags are global in git (not per-remote), so after fetching both with --tags,
# they're merged into refs/tags/. We need to check which remote actually has
# each tag by checking if the tagged commit is reachable from that remote's refs.
# Simpler approach: fetch tags explicitly per remote and compare via ls-remote.
local -A ds_remote_tags prgs_remote_tags
while IFS=$'\t' read -r sha ref; do
[[ -z "$ref" ]] && continue
local name="${ref#refs/tags/}"
name="${name%\^\{\}}" # strip ^{} suffix from dereferenced tags
ds_remote_tags["$name"]="$sha"
done < <(git -C "$CACHE_DIR" ls-remote --tags ds 2>/dev/null)
while IFS=$'\t' read -r sha ref; do
[[ -z "$ref" ]] && continue
local name="${ref#refs/tags/}"
name="${name%\^\{\}}"
prgs_remote_tags["$name"]="$sha"
done < <(git -C "$CACHE_DIR" ls-remote --tags prgs 2>/dev/null)
local total_ds=${#ds_remote_tags[@]}
local total_prgs=${#prgs_remote_tags[@]}
log "dadeschools: $total_ds tags, prgs: $total_prgs tags"
local pushed=0 skipped=0
# Combine all tag names
local -A all_tags
for t in "${!ds_remote_tags[@]}"; do all_tags["$t"]=1; done
for t in "${!prgs_remote_tags[@]}"; do all_tags["$t"]=1; done
for t in $(printf '%s\n' "${!all_tags[@]}" | sort -V); do
local ds_sha="${ds_remote_tags[$t]:-}"
local prgs_sha="${prgs_remote_tags[$t]:-}"
if [[ -n "$ds_sha" && -z "$prgs_sha" ]]; then
ok "$t → prgs (new tag)"
run_or_dry git -C "$CACHE_DIR" push prgs "refs/tags/${t}:refs/tags/${t}"
((pushed++)) || true
elif [[ -z "$ds_sha" && -n "$prgs_sha" ]]; then
ok "$t → dadeschools (new tag)"
run_or_dry git -C "$CACHE_DIR" push ds "refs/tags/${t}:refs/tags/${t}"
((pushed++)) || true
elif [[ "$ds_sha" == "$prgs_sha" ]]; then
log "$t ✔ in sync"
else
warn "$t MISMATCH (ds: ${ds_sha:0:8}, prgs: ${prgs_sha:0:8}) — tags shouldn't diverge, skipping"
((skipped++)) || true
fi
done
echo ""
log "Tags: $pushed pushed, $skipped skipped"
echo ""
}
# ── Main ──────────────────────────────────────────────────────────────────────
main() {
echo ""
echo "╔══════════════════════════════════════════════╗"
echo "║ Gitea Ref Mirror: dadeschools ⇄ prgs ║"
echo "╚══════════════════════════════════════════════╝"
echo ""
setup_cache
fetch_both
mirror_branches
mirror_tags
if $DRY_RUN; then
echo "═══ DRY RUN complete. Pass --apply to execute. ═══"
else
echo "═══ Mirror complete. ═══"
fi
}
main
tests/test_mirror_refs.py
+269
View File
@@ -0,0 +1,269 @@
"""Tests for mirror_refs.sh.
These test the script's argument handling and dry-run output. Actual mirroring
is not tested because it requires real git remotes with credentials.
For the mirroring logic, we create a local test harness with two bare "remote"
repos and verify the script's branching/tagging comparison logic indirectly.
"""
import os
import subprocess
import tempfile
import unittest
REPO_ROOT = str(__import__("pathlib").Path(__file__).resolve().parent.parent)
SCRIPT = os.path.join(REPO_ROOT, "mirror_refs.sh")
def _run(args=None, env_override=None):
"""Run mirror_refs.sh with given args, return (rc, stdout, stderr)."""
cmd = [SCRIPT]
if args:
cmd.extend(args)
env = os.environ.copy()
if env_override:
env.update(env_override)
result = subprocess.run(
cmd, capture_output=True, text=True, env=env, timeout=30,
)
return result.returncode, result.stdout, result.stderr
# ---------------------------------------------------------------------------
# Flag parsing
# ---------------------------------------------------------------------------
class TestFlagParsing(unittest.TestCase):
"""Verify --help, unknown flags, and flag combinations."""
def test_help_flag(self):
rc, stdout, stderr = _run(["--help"])
self.assertEqual(rc, 0)
self.assertIn("mirror_refs.sh", stdout)
self.assertIn("--apply", stdout)
self.assertIn("--force", stdout)
def test_h_flag(self):
rc, stdout, stderr = _run(["-h"])
self.assertEqual(rc, 0)
self.assertIn("--apply", stdout)
def test_unknown_flag_fails(self):
rc, stdout, stderr = _run(["--bogus"])
self.assertNotEqual(rc, 0)
self.assertIn("Unknown flag", stderr)
# ---------------------------------------------------------------------------
# Dry-run behavior
# ---------------------------------------------------------------------------
class TestDryRunBanner(unittest.TestCase):
"""The script should clearly indicate dry-run mode."""
def test_dry_run_banner_shown_by_default(self):
"""Without --apply, the script prints a DRY RUN banner.
This test will fail if credentials are unavailable (expected in CI).
We just check the banner appears before the credential check.
"""
rc, stdout, stderr = _run()
# The dry-run banner is printed before any remote operations
combined = stdout + stderr
self.assertTrue(
"DRY RUN" in combined or "credential" in combined.lower(),
"Script should either show DRY RUN banner or fail at credentials"
)
# ---------------------------------------------------------------------------
# Script structure
# ---------------------------------------------------------------------------
class TestScriptStructure(unittest.TestCase):
"""Verify the script has the expected structure and safety defaults."""
def test_shebang(self):
with open(SCRIPT, "r") as f:
first_line = f.readline().strip()
self.assertEqual(first_line, "#!/usr/bin/env bash")
def test_set_euo_pipefail(self):
with open(SCRIPT, "r") as f:
content = f.read()
self.assertIn("set -euo pipefail", content)
def test_dry_run_is_default(self):
"""Verify DRY_RUN=true is the default in the script source."""
with open(SCRIPT, "r") as f:
content = f.read()
self.assertIn("DRY_RUN=true", content)
def test_force_is_off_by_default(self):
with open(SCRIPT, "r") as f:
content = f.read()
self.assertIn("FORCE=false", content)
def test_no_force_push_without_flag(self):
"""The script should never force-push unless FORCE is true."""
with open(SCRIPT, "r") as f:
content = f.read()
# All --force pushes should be gated by $FORCE
self.assertIn("if $FORCE", content)
def test_uses_brace_delimited_refspecs(self):
"""Verify refspecs use ${b} not $b to avoid zsh :r issues."""
with open(SCRIPT, "r") as f:
content = f.read()
# The push commands should use ${...} brace-delimited variable refs
# in refspec strings to prevent zsh's :r modifier from eating the colon
lines_with_push_refspec = [
line for line in content.splitlines()
if "refs/heads/" in line and "push" in line
]
for line in lines_with_push_refspec:
# Should not contain bare $b: or $t: patterns (without braces)
self.assertNotRegex(
line, r'\$[a-z]:',
f"Unbraced variable before colon in refspec: {line.strip()}"
)
def test_additive_only_no_delete(self):
"""Script should never use --delete or push :refs/... (delete refspec)."""
with open(SCRIPT, "r") as f:
content = f.read()
self.assertNotIn("--delete", content)
# A delete refspec looks like ":refs/..." (colon with no left side)
for line in content.splitlines():
stripped = line.strip()
if stripped.startswith("#"):
continue
self.assertNotRegex(
stripped, r'push\s.*\s":refs/',
f"Found delete refspec: {stripped}"
)
# ---------------------------------------------------------------------------
# Local mirror logic (integration test with local bare repos)
# ---------------------------------------------------------------------------
class TestLocalMirrorLogic(unittest.TestCase):
"""Test the core mirroring logic using local bare repos as stand-ins.
This creates two bare repos ("ds" and "prgs"), puts some branches/tags
on them, and then runs a simplified version of the comparison logic
to validate the algorithm.
"""
def setUp(self):
self.tmpdir = tempfile.mkdtemp(prefix="mirror-test-")
self.ds_repo = os.path.join(self.tmpdir, "ds.git")
self.prgs_repo = os.path.join(self.tmpdir, "prgs.git")
self.work_repo = os.path.join(self.tmpdir, "work")
# Create two bare repos
subprocess.run(["git", "init", "--bare", self.ds_repo], capture_output=True)
subprocess.run(["git", "init", "--bare", self.prgs_repo], capture_output=True)
# Create a working repo to push some initial content
subprocess.run(["git", "init", self.work_repo], capture_output=True)
subprocess.run(
["git", "-C", self.work_repo, "commit", "--allow-empty", "-m", "init"],
capture_output=True,
env={**os.environ, "GIT_AUTHOR_NAME": "test", "GIT_AUTHOR_EMAIL": "t@t",
"GIT_COMMITTER_NAME": "test", "GIT_COMMITTER_EMAIL": "t@t"},
)
def tearDown(self):
import shutil
shutil.rmtree(self.tmpdir, ignore_errors=True)
def _git(self, *args, cwd=None):
env = {
**os.environ,
"GIT_AUTHOR_NAME": "test", "GIT_AUTHOR_EMAIL": "t@t",
"GIT_COMMITTER_NAME": "test", "GIT_COMMITTER_EMAIL": "t@t",
}
result = subprocess.run(
["git"] + list(args), capture_output=True, text=True,
cwd=cwd or self.work_repo, env=env,
)
return result.stdout.strip()
def test_branch_only_on_ds_detected(self):
"""A branch only on ds should be identified as needing push to prgs."""
# Push main to both, then a feature branch only to ds
self._git("remote", "add", "ds", self.ds_repo)
self._git("remote", "add", "prgs", self.prgs_repo)
self._git("push", "ds", "HEAD:refs/heads/main")
self._git("push", "prgs", "HEAD:refs/heads/main")
self._git("push", "ds", "HEAD:refs/heads/feat/only-ds")
# Fetch both into a bare mirror
mirror = os.path.join(self.tmpdir, "mirror.git")
subprocess.run(["git", "init", "--bare", mirror], capture_output=True)
self._git("remote", "add", "ds", self.ds_repo, cwd=mirror)
self._git("remote", "add", "prgs", self.prgs_repo, cwd=mirror)
self._git("fetch", "ds", cwd=mirror)
self._git("fetch", "prgs", cwd=mirror)
# Check: ds has feat/only-ds, prgs does not
ds_refs = self._git("for-each-ref", "--format=%(refname)",
"refs/remotes/ds/", cwd=mirror)
prgs_refs = self._git("for-each-ref", "--format=%(refname)",
"refs/remotes/prgs/", cwd=mirror)
self.assertIn("feat/only-ds", ds_refs)
self.assertNotIn("feat/only-ds", prgs_refs)
def test_tag_only_on_one_side_detected(self):
"""A tag only on ds should be identified as needing push to prgs."""
self._git("remote", "add", "ds", self.ds_repo)
self._git("remote", "add", "prgs", self.prgs_repo)
self._git("push", "ds", "HEAD:refs/heads/main")
self._git("push", "prgs", "HEAD:refs/heads/main")
self._git("tag", "v1.0.0")
self._git("push", "ds", "v1.0.0")
# ls-remote to check tags
ds_tags = self._git("ls-remote", "--tags", self.ds_repo)
prgs_tags = self._git("ls-remote", "--tags", self.prgs_repo)
self.assertIn("v1.0.0", ds_tags)
self.assertNotIn("v1.0.0", prgs_tags)
def test_in_sync_branches_need_no_push(self):
"""Branches at the same commit on both sides need no action."""
self._git("remote", "add", "ds", self.ds_repo)
self._git("remote", "add", "prgs", self.prgs_repo)
self._git("push", "ds", "HEAD:refs/heads/main")
self._git("push", "prgs", "HEAD:refs/heads/main")
ds_sha = self._git("ls-remote", self.ds_repo, "refs/heads/main").split()[0]
prgs_sha = self._git("ls-remote", self.prgs_repo, "refs/heads/main").split()[0]
self.assertEqual(ds_sha, prgs_sha)
def test_fast_forward_detected(self):
"""When one side is ahead (FF), it should be pushable."""
self._git("remote", "add", "ds", self.ds_repo)
self._git("remote", "add", "prgs", self.prgs_repo)
self._git("push", "ds", "HEAD:refs/heads/main")
self._git("push", "prgs", "HEAD:refs/heads/main")
# Make a new commit and push only to ds
self._git("commit", "--allow-empty", "-m", "advance")
self._git("push", "ds", "HEAD:refs/heads/main")
ds_sha = self._git("ls-remote", self.ds_repo, "refs/heads/main").split()[0]
prgs_sha = self._git("ls-remote", self.prgs_repo, "refs/heads/main").split()[0]
self.assertNotEqual(ds_sha, prgs_sha)
# prgs_sha should be ancestor of ds_sha (fast-forward)
rc = subprocess.run(
["git", "-C", self.work_repo, "merge-base", "--is-ancestor",
prgs_sha, ds_sha],
capture_output=True,
).returncode
self.assertEqual(rc, 0, "prgs commit should be ancestor of ds commit (FF)")
if __name__ == "__main__":
unittest.main()