andre/github-to-gitea-mirror
1
0
Fork 0
mirror of https://github.com/Andre-Bzk/github-to-gitea-mirror.git synced 2026-11-15 09:33:13 +01:00
Code Issues Packages Projects Releases Wiki Activity

refactor: replace hidden module state with explicit data flow

Browse Source 
The token expiry check read response headers that an earlier, unrelated
request had left behind in module state. Its correctness depended on call
order that nothing but a comment enforced, so inserting or reordering a
request would have silently checked the wrong response, or none at all.

The exit code had the mirrored problem: any function in the call tree could
raise a shared error count, so nothing at the top level revealed which steps
could contribute to a failing run.

Both now travel as return values. The ordering constraint becomes a data
dependency the interpreter enforces instead of a convention the next reader
has to know about.
This commit is contained in:
Andre
2026-08-08 11:51:56 +02:00
parent 6ff4da99e4
commit 074fd22eff
2 changed files with 84 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sync_mirrors.py
+77 -55
View File
@@ -18,6 +18,7 @@ import urllib.error
import urllib.parse import urllib.parse
import urllib.request import urllib.request
import uuid import uuid
from collections import namedtuple
from datetime import datetime, timezone from datetime import datetime, timezone
KNOWN_KEYS = ( KNOWN_KEYS = (
@@ -37,20 +38,13 @@ KNOWN_KEYS = (
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
CONFIG_PATH = os.environ.get("MIRROR_ENV", os.path.join(SCRIPT_DIR, "mirror.env")) CONFIG_PATH = os.environ.get("MIRROR_ENV", os.path.join(SCRIPT_DIR, "mirror.env"))
errors = 0 Response = namedtuple("Response", "status body headers")
last_headers = {}
def log(level, msg): def log(level, msg):
print("%s [%s] %s" % (datetime.now().strftime("%Y-%m-%d %H:%M:%S"), level, msg), flush=True) print("%s [%s] %s" % (datetime.now().strftime("%Y-%m-%d %H:%M:%S"), level, msg), flush=True)
def fail(msg):
global errors
errors += 1
log("ERROR", msg)
def load_config(): def load_config():
cfg = { cfg = {
"GITHUB_USER": "", "GITHUB_USER": "",
@@ -91,7 +85,7 @@ def truthy(value):
def api(url, token=None, scheme="token", method="GET", payload=None): def api(url, token=None, scheme="token", method="GET", payload=None):
"""Return (status, parsed_body). Raises urllib.error.HTTPError on >=400.""" """Return a Response. Raises urllib.error.HTTPError on >=400."""
data = None data = None
headers = { headers = {
"Accept": "application/json", "Accept": "application/json",
@@ -106,9 +100,11 @@ def api(url, token=None, scheme="token", method="GET", payload=None):
req = urllib.request.Request(url, data=data, headers=headers, method=method) req = urllib.request.Request(url, data=data, headers=headers, method=method)
with urllib.request.urlopen(req, timeout=180) as resp: with urllib.request.urlopen(req, timeout=180) as resp:
body = resp.read().decode("utf-8") body = resp.read().decode("utf-8")
last_headers.clear() return Response(
last_headers.update({k.lower(): v for k, v in resp.headers.items()}) resp.status,
return resp.status, (json.loads(body) if body.strip() else None) json.loads(body) if body.strip() else None,
{k.lower(): v for k, v in resp.headers.items()},
)
def path_part(value): def path_part(value):
@@ -215,9 +211,15 @@ def upload_release_asset(cfg, repo_name, release_id, asset, source, size):
def github_repos(cfg): def github_repos(cfg):
"""All repos owned by the user; private ones included when a token is set.""" """Return (repos, headers of the last GitHub response).
The headers travel with the result because the token expiry check needs
them; keeping them here makes that dependency visible at the call site.
Private repos are included when a token is set.
"""
token = cfg["GITHUB_TOKEN"] token = cfg["GITHUB_TOKEN"]
repos = [] repos = []
headers = {}
page = 1 page = 1
while True: while True:
if token: if token:
@@ -226,7 +228,9 @@ def github_repos(cfg):
else: else:
url = ("https://api.github.com/users/%s/repos" url = ("https://api.github.com/users/%s/repos"
"?per_page=100&type=owner&page=%d" % (cfg["GITHUB_USER"], page)) "?per_page=100&type=owner&page=%d" % (cfg["GITHUB_USER"], page))
_, batch = api(url, token=token, scheme="Bearer") resp = api(url, token=token, scheme="Bearer")
headers = resp.headers
batch = resp.body
if not batch: if not batch:
break break
repos.extend(batch) repos.extend(batch)
@@ -238,17 +242,18 @@ def github_repos(cfg):
# affiliation=owner still returns repos owned by orgs the user belongs to # affiliation=owner still returns repos owned by orgs the user belongs to
repos = [r for r in repos repos = [r for r in repos
if r["owner"]["login"].lower() == cfg["GITHUB_USER"].lower()] if r["owner"]["login"].lower() == cfg["GITHUB_USER"].lower()]
return repos return repos, headers
def warn_token_expiry(): def warn_token_expiry(headers):
"""Fine-grained PATs always expire; a silent expiry would stall the mirror. """Fine-grained PATs always expire; a silent expiry would stall the mirror.
GitHub reports the date in a response header, so every run can check it. GitHub reports the date in a response header, so every run can check it.
Returns the number of errors found, so an expired token can fail the run.
""" """
raw = last_headers.get("github-authentication-token-expiration") raw = headers.get("github-authentication-token-expiration")
if not raw: if not raw:
return # classic token without expiry, or no token at all return 0 # classic token without expiry, or no token at all
text = raw.strip().replace(" UTC", "").replace("Z", "").replace("T", " ") text = raw.strip().replace(" UTC", "").replace("Z", "").replace("T", " ")
parsed = None parsed = None
for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M:%S %z", "%Y-%m-%d"): for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M:%S %z", "%Y-%m-%d"):
@@ -258,17 +263,19 @@ def warn_token_expiry():
except ValueError: except ValueError:
continue continue
if parsed is None: if parsed is None:
return return 0
now_utc = datetime.now(timezone.utc).replace(tzinfo=None) now_utc = datetime.now(timezone.utc).replace(tzinfo=None)
days = (parsed.replace(tzinfo=None) - now_utc).days days = (parsed.replace(tzinfo=None) - now_utc).days
if days < 0: if days < 0:
fail("GITHUB_TOKEN expired on %s - renew it" % parsed.date()) log("ERROR", "GITHUB_TOKEN expired on %s - renew it" % parsed.date())
elif days <= 14: return 1
if days <= 14:
log("WARN", "GITHUB_TOKEN expires in %d day(s), on %s - renew it" log("WARN", "GITHUB_TOKEN expires in %d day(s), on %s - renew it"
% (days, parsed.date())) % (days, parsed.date()))
else: else:
log("INFO", "GITHUB_TOKEN valid until %s (%d days)" % (parsed.date(), days)) log("INFO", "GITHUB_TOKEN valid until %s (%d days)" % (parsed.date(), days))
return 0
def duration_seconds(value): def duration_seconds(value):
@@ -285,9 +292,8 @@ def duration_seconds(value):
def gitea_repo(cfg, name): def gitea_repo(cfg, name):
try: try:
_, repo = api("%s/api/v1/repos/%s/%s" % (cfg["GITEA_URL"], cfg["GITEA_OWNER"], name), return api("%s/api/v1/repos/%s/%s" % (cfg["GITEA_URL"], cfg["GITEA_OWNER"], name),
token=cfg["GITEA_TOKEN"]) token=cfg["GITEA_TOKEN"]).body
return repo
except urllib.error.HTTPError as exc: except urllib.error.HTTPError as exc:
if exc.code == 404: if exc.code == 404:
return None return None
@@ -347,7 +353,7 @@ def github_releases(cfg, repo_name):
while True: while True:
url = ("https://api.github.com/repos/%s/%s/releases?per_page=100&page=%d" url = ("https://api.github.com/repos/%s/%s/releases?per_page=100&page=%d"
% (path_part(cfg["GITHUB_USER"]), path_part(repo_name), page)) % (path_part(cfg["GITHUB_USER"]), path_part(repo_name), page))
_, batch = api(url, token=cfg["GITHUB_TOKEN"], scheme="Bearer") batch = api(url, token=cfg["GITHUB_TOKEN"], scheme="Bearer").body
if not batch: if not batch:
break break
releases.extend(batch) releases.extend(batch)
@@ -364,7 +370,7 @@ def github_release_assets(cfg, repo_name, release_id):
url = ("https://api.github.com/repos/%s/%s/releases/%s/assets" url = ("https://api.github.com/repos/%s/%s/releases/%s/assets"
"?per_page=100&page=%d" "?per_page=100&page=%d"
% (path_part(cfg["GITHUB_USER"]), path_part(repo_name), release_id, page)) % (path_part(cfg["GITHUB_USER"]), path_part(repo_name), release_id, page))
_, batch = api(url, token=cfg["GITHUB_TOKEN"], scheme="Bearer") batch = api(url, token=cfg["GITHUB_TOKEN"], scheme="Bearer").body
if not batch: if not batch:
break break
assets.extend(batch) assets.extend(batch)
@@ -381,7 +387,7 @@ def gitea_releases(cfg, repo_name):
url = ("%s/api/v1/repos/%s/%s/releases?limit=50&page=%d" url = ("%s/api/v1/repos/%s/%s/releases?limit=50&page=%d"
% (cfg["GITEA_URL"], path_part(cfg["GITEA_OWNER"]), % (cfg["GITEA_URL"], path_part(cfg["GITEA_OWNER"]),
path_part(repo_name), page)) path_part(repo_name), page))
_, batch = api(url, token=cfg["GITEA_TOKEN"]) batch = api(url, token=cfg["GITEA_TOKEN"]).body
if not batch: if not batch:
break break
releases.extend(batch) releases.extend(batch)
@@ -398,7 +404,7 @@ def gitea_tags(cfg, repo_name):
url = ("%s/api/v1/repos/%s/%s/tags?limit=50&page=%d" url = ("%s/api/v1/repos/%s/%s/tags?limit=50&page=%d"
% (cfg["GITEA_URL"], path_part(cfg["GITEA_OWNER"]), % (cfg["GITEA_URL"], path_part(cfg["GITEA_OWNER"]),
path_part(repo_name), page)) path_part(repo_name), page))
_, batch = api(url, token=cfg["GITEA_TOKEN"]) batch = api(url, token=cfg["GITEA_TOKEN"]).body
if not batch: if not batch:
break break
tags.update(tag["name"] for tag in batch) tags.update(tag["name"] for tag in batch)
@@ -448,7 +454,8 @@ def release_update_payload(source):
def sync_release_assets(cfg, repo_name, source_release, target_release, dry_run): def sync_release_assets(cfg, repo_name, source_release, target_release, dry_run):
copied = preserved = 0 """Return (copied, preserved, failed) counts for this release's assets."""
copied = preserved = failed = 0
source_assets = github_release_assets(cfg, repo_name, source_release["id"]) source_assets = github_release_assets(cfg, repo_name, source_release["id"])
target_assets = {asset["name"]: asset for asset in target_release.get("assets", [])} target_assets = {asset["name"]: asset for asset in target_release.get("assets", [])}
@@ -482,19 +489,22 @@ def sync_release_assets(cfg, repo_name, source_release, target_release, dry_run)
% (repo_name, source_release["tag_name"], name, size)) % (repo_name, source_release["tag_name"], name, size))
except urllib.error.HTTPError as exc: except urllib.error.HTTPError as exc:
detail = exc.read().decode("utf-8", "replace")[:300] detail = exc.read().decode("utf-8", "replace")[:300]
fail("%s release %s asset %s failed (HTTP %s): %s" log("ERROR", "%s release %s asset %s failed (HTTP %s): %s"
% (repo_name, source_release["tag_name"], name, exc.code, detail)) % (repo_name, source_release["tag_name"], name, exc.code, detail))
failed += 1
except Exception as exc: except Exception as exc:
fail("%s release %s asset %s failed: %s" log("ERROR", "%s release %s asset %s failed: %s"
% (repo_name, source_release["tag_name"], name, exc)) % (repo_name, source_release["tag_name"], name, exc))
failed += 1
finally: finally:
if tmp is not None: if tmp is not None:
tmp.close() tmp.close()
return copied, preserved return copied, preserved, failed
def sync_releases(cfg, repo_name, dry_run): def sync_releases(cfg, repo_name, dry_run):
stats = {"created": 0, "updated": 0, "assets": 0, "preserved": 0, "waiting": 0} stats = {"created": 0, "updated": 0, "assets": 0, "preserved": 0,
"waiting": 0, "errors": 0}
source_releases = github_releases(cfg, repo_name) source_releases = github_releases(cfg, repo_name)
target_releases = gitea_releases(cfg, repo_name) target_releases = gitea_releases(cfg, repo_name)
by_tag = {release["tag_name"]: release for release in target_releases} by_tag = {release["tag_name"]: release for release in target_releases}
@@ -529,38 +539,41 @@ def sync_releases(cfg, repo_name, dry_run):
target = {"id": source["id"], "tag_name": tag, "assets": []} target = {"id": source["id"], "tag_name": tag, "assets": []}
log("DRY", "would create %s release %s" % (repo_name, tag)) log("DRY", "would create %s release %s" % (repo_name, tag))
else: else:
_, target = api( target = api(
"%s/api/v1/repos/%s/%s/releases" "%s/api/v1/repos/%s/%s/releases"
% (cfg["GITEA_URL"], path_part(cfg["GITEA_OWNER"]), % (cfg["GITEA_URL"], path_part(cfg["GITEA_OWNER"]),
path_part(repo_name)), path_part(repo_name)),
token=cfg["GITEA_TOKEN"], method="POST", token=cfg["GITEA_TOKEN"], method="POST",
payload=release_payload(source)) payload=release_payload(source)).body
log("OK", "created %s release %s" % (repo_name, tag)) log("OK", "created %s release %s" % (repo_name, tag))
stats["created"] += 1 stats["created"] += 1
elif release_needs_update(source, target): elif release_needs_update(source, target):
if dry_run: if dry_run:
log("DRY", "would update %s release %s" % (repo_name, tag)) log("DRY", "would update %s release %s" % (repo_name, tag))
else: else:
_, target = api( target = api(
"%s/api/v1/repos/%s/%s/releases/%s" "%s/api/v1/repos/%s/%s/releases/%s"
% (cfg["GITEA_URL"], path_part(cfg["GITEA_OWNER"]), % (cfg["GITEA_URL"], path_part(cfg["GITEA_OWNER"]),
path_part(repo_name), target["id"]), path_part(repo_name), target["id"]),
token=cfg["GITEA_TOKEN"], method="PATCH", token=cfg["GITEA_TOKEN"], method="PATCH",
payload=release_update_payload(source)) payload=release_update_payload(source)).body
log("OK", "updated %s release %s" % (repo_name, tag)) log("OK", "updated %s release %s" % (repo_name, tag))
stats["updated"] += 1 stats["updated"] += 1
if truthy(cfg["SYNC_RELEASE_ASSETS"]): if truthy(cfg["SYNC_RELEASE_ASSETS"]):
copied, preserved = sync_release_assets( copied, preserved, failed = sync_release_assets(
cfg, repo_name, source, target, dry_run) cfg, repo_name, source, target, dry_run)
stats["assets"] += copied stats["assets"] += copied
stats["preserved"] += preserved stats["preserved"] += preserved
stats["errors"] += failed
except urllib.error.HTTPError as exc: except urllib.error.HTTPError as exc:
detail = exc.read().decode("utf-8", "replace")[:300] detail = exc.read().decode("utf-8", "replace")[:300]
fail("%s release %s failed (HTTP %s): %s" log("ERROR", "%s release %s failed (HTTP %s): %s"
% (repo_name, tag, exc.code, detail)) % (repo_name, tag, exc.code, detail))
stats["errors"] += 1
except Exception as exc: except Exception as exc:
fail("%s release %s failed: %s" % (repo_name, tag, exc)) log("ERROR", "%s release %s failed: %s" % (repo_name, tag, exc))
stats["errors"] += 1
return stats return stats
@@ -578,7 +591,7 @@ def main():
log("WARN", "GITHUB_TOKEN is empty - private repositories will be skipped") log("WARN", "GITHUB_TOKEN is empty - private repositories will be skipped")
try: try:
repos = github_repos(cfg) repos, github_headers = github_repos(cfg)
except urllib.error.HTTPError as exc: except urllib.error.HTTPError as exc:
log("ERROR", "GitHub API %s: %s" % (exc.code, exc.read().decode("utf-8", "replace")[:300])) log("ERROR", "GitHub API %s: %s" % (exc.code, exc.read().decode("utf-8", "replace")[:300]))
return 2 return 2
@@ -586,7 +599,7 @@ def main():
log("ERROR", "GitHub API unreachable: %s" % exc) log("ERROR", "GitHub API unreachable: %s" % exc)
return 2 return 2
warn_token_expiry() # must run before any Gitea call overwrites the headers errors = warn_token_expiry(github_headers)
if not truthy(cfg["INCLUDE_FORKS"]): if not truthy(cfg["INCLUDE_FORKS"]):
repos = [r for r in repos if not r.get("fork")] repos = [r for r in repos if not r.get("fork")]
@@ -614,7 +627,8 @@ def main():
try: try:
existing = gitea_repo(cfg, name) existing = gitea_repo(cfg, name)
except Exception as exc: except Exception as exc:
fail("%s: cannot query Gitea: %s" % (name, exc)) log("ERROR", "%s: cannot query Gitea: %s" % (name, exc))
errors += 1
continue continue
if existing is None: if existing is None:
@@ -625,9 +639,11 @@ def main():
mirror_created = True mirror_created = True
except urllib.error.HTTPError as exc: except urllib.error.HTTPError as exc:
detail = exc.read().decode("utf-8", "replace")[:300] detail = exc.read().decode("utf-8", "replace")[:300]
fail("%s: migrate failed (HTTP %s): %s" % (name, exc.code, detail)) log("ERROR", "%s: migrate failed (HTTP %s): %s" % (name, exc.code, detail))
errors += 1
except Exception as exc: except Exception as exc:
fail("%s: migrate failed: %s" % (name, exc)) log("ERROR", "%s: migrate failed: %s" % (name, exc))
errors += 1
if mirror_created and truthy(cfg["SYNC_RELEASES"]): if mirror_created and truthy(cfg["SYNC_RELEASES"]):
try: try:
@@ -640,8 +656,10 @@ def main():
stats = sync_releases(cfg, name, dry_run) stats = sync_releases(cfg, name, dry_run)
for key in release_totals: for key in release_totals:
release_totals[key] += stats[key] release_totals[key] += stats[key]
errors += stats["errors"]
except Exception as exc: except Exception as exc:
fail("%s: could not synchronize releases: %s" % (name, exc)) log("ERROR", "%s: could not synchronize releases: %s" % (name, exc))
errors += 1
continue continue
skipped += 1 skipped += 1
@@ -651,26 +669,30 @@ def main():
try: try:
align_interval(cfg, name, existing, dry_run) align_interval(cfg, name, existing, dry_run)
except Exception as exc: except Exception as exc:
fail("%s: could not update mirror interval: %s" % (name, exc)) log("ERROR", "%s: could not update mirror interval: %s" % (name, exc))
errors += 1
if truthy(cfg["SYNC_RELEASES"]): if truthy(cfg["SYNC_RELEASES"]):
try: try:
stats = sync_releases(cfg, name, dry_run) stats = sync_releases(cfg, name, dry_run)
for key in release_totals: for key in release_totals:
release_totals[key] += stats[key] release_totals[key] += stats[key]
errors += stats["errors"]
except urllib.error.HTTPError as exc: except urllib.error.HTTPError as exc:
detail = exc.read().decode("utf-8", "replace")[:300] detail = exc.read().decode("utf-8", "replace")[:300]
fail("%s: release sync failed (HTTP %s): %s" log("ERROR", "%s: release sync failed (HTTP %s): %s"
% (name, exc.code, detail)) % (name, exc.code, detail))
errors += 1
except Exception as exc: except Exception as exc:
fail("%s: release sync failed: %s" % (name, exc)) log("ERROR", "%s: release sync failed: %s" % (name, exc))
errors += 1
github_names = {r["name"].lower() for r in repos} github_names = {r["name"].lower() for r in repos}
try: try:
page, orphans = 1, [] page, orphans = 1, []
while True: while True:
_, batch = api("%s/api/v1/user/repos?limit=50&page=%d" % (cfg["GITEA_URL"], page), batch = api("%s/api/v1/user/repos?limit=50&page=%d" % (cfg["GITEA_URL"], page),
token=cfg["GITEA_TOKEN"]) token=cfg["GITEA_TOKEN"]).body
if not batch: if not batch:
break break
orphans.extend( orphans.extend(
tests/test_sync_mirrors.py
+7 -7
View File
@@ -33,9 +33,6 @@ def github_release(**overrides):
class ReleaseSyncTests(unittest.TestCase): class ReleaseSyncTests(unittest.TestCase):
def setUp(self):
mirror.errors = 0
@mock.patch.object(mirror, "upload_release_asset") @mock.patch.object(mirror, "upload_release_asset")
@mock.patch.object(mirror, "download_github_asset") @mock.patch.object(mirror, "download_github_asset")
@mock.patch.object(mirror, "github_release_assets") @mock.patch.object(mirror, "github_release_assets")
@@ -59,13 +56,14 @@ class ReleaseSyncTests(unittest.TestCase):
gitea_tags.return_value = {"v1.0.0"} gitea_tags.return_value = {"v1.0.0"}
github_assets.return_value = [asset] github_assets.return_value = [asset]
download_asset.return_value = (io.BytesIO(b"zip"), 3) download_asset.return_value = (io.BytesIO(b"zip"), 3)
api.return_value = (201, {"id": 901, "tag_name": "v1.0.0", "assets": []}) api.return_value = mirror.Response(
201, {"id": 901, "tag_name": "v1.0.0", "assets": []}, {})
stats = mirror.sync_releases(config(), "project", False) stats = mirror.sync_releases(config(), "project", False)
self.assertEqual(1, stats["created"]) self.assertEqual(1, stats["created"])
self.assertEqual(1, stats["assets"]) self.assertEqual(1, stats["assets"])
self.assertEqual(0, mirror.errors) self.assertEqual(0, stats["errors"])
self.assertEqual("POST", api.call_args.kwargs["method"]) self.assertEqual("POST", api.call_args.kwargs["method"])
self.assertEqual("v1.0.0", api.call_args.kwargs["payload"]["tag_name"]) self.assertEqual("v1.0.0", api.call_args.kwargs["payload"]["tag_name"])
upload_asset.assert_called_once() upload_asset.assert_called_once()
@@ -118,9 +116,10 @@ class ReleaseSyncTests(unittest.TestCase):
@mock.patch.object(mirror, "api") @mock.patch.object(mirror, "api")
def test_creates_draft_without_requiring_mirrored_tag( def test_creates_draft_without_requiring_mirrored_tag(
self, api, github_releases, gitea_releases, gitea_tags): self, api, github_releases, gitea_releases, gitea_tags):
api.return_value = ( api.return_value = mirror.Response(
201, 201,
{"id": 901, "tag_name": "v1.0.0", "draft": True, "assets": []}, {"id": 901, "tag_name": "v1.0.0", "draft": True, "assets": []},
{},
) )
stats = mirror.sync_releases( stats = mirror.sync_releases(
@@ -148,7 +147,8 @@ class ReleaseSyncTests(unittest.TestCase):
} }
github_releases.return_value = [source] github_releases.return_value = [source]
gitea_releases.return_value = [target] gitea_releases.return_value = [target]
api.return_value = (200, dict(target, body="New notes", prerelease=True)) api.return_value = mirror.Response(
200, dict(target, body="New notes", prerelease=True), {})
stats = mirror.sync_releases( stats = mirror.sync_releases(
config(SYNC_RELEASE_ASSETS="false"), "project", False) config(SYNC_RELEASE_ASSETS="false"), "project", False)