n0mad1k/coe-mirror
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Mirror all four Church of Malware owners, per-owner pcloud subdirs

Browse Source 
- OWNERS list replaces single OWNER: Nightmare_Eclipse, shai_hulud, bikini_mirror, ek0mssavi0r
- PCLOUD_TARGET moves to ChurchOfMalware/ parent, per-owner subdirs at runtime
- Owner threaded through fetch_repos, fetch_branch_sha, download_repo_zip
- Pending filenames get owner__ prefix to prevent cross-owner collisions
- flush_pending strips owner prefix when routing to owner subdir
- Per-owner fetch failure logs and continues; failed_owners reported in notify
- notify groups new files by owner; also picks up flushes done at run start
- Tests: parametrize on OWNERS via monkeypatch, add multi-owner collision and
  per-owner failure isolation tests

DevTrack #1284
This commit is contained in:
n0mad1k
2026-07-06 12:35:56 -04:00
parent 4d0fbff039
commit 9ffe38cbbc
3 changed files with 252 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -4,3 +4,4 @@ __pycache__/
*.pyo
.pytest_cache/
*.egg-info/
.active-item.json
coe_mirror.py
+100 -53
View File
@@ -1,7 +1,8 @@
"""Mirror repos from git.churchofmalware.org Forgejo instance to pCloud.
Fetches all repos owned by Nightmare_Eclipse and stores per-commit zip snapshots.
Runs unattended via systemd timer twice daily. Local staging when pCloud unavailable.
Fetches all repos owned by each user in OWNERS and stores per-commit zip snapshots
under `PCLOUD_TARGET/<owner>/`. Runs unattended via systemd timer twice daily.
Local staging when pCloud unavailable.
"""
import json
@@ -24,8 +25,8 @@ import requests
# Module constants (overridable via env vars for testability)
API_BASE = "https://git.churchofmalware.org"
OWNER = "Nightmare_Eclipse"
PCLOUD_TARGET = Path.home() / "pCloudDrive" / "Hacking" / "Nightmare_Eclipse"
OWNERS = ["Nightmare_Eclipse", "shai_hulud", "bikini_mirror", "ek0mssavi0r"]
PCLOUD_TARGET = Path.home() / "pCloudDrive" / "Hacking" / "ChurchOfMalware"
PENDING_DIR = Path.home() / ".local" / "share" / "coe-mirror" / "pending"
LOG_DIR = Path.home() / ".local" / "share" / "coe-mirror"
LOG_FILE = LOG_DIR / "coe-mirror.log"
@@ -174,9 +175,9 @@ def place_into_pcloud(src: Path, dst: Path) -> bool:
return False
def fetch_repos(session: requests.Session) -> Optional[list]:
"""Fetch list of repos owned by OWNER. Return list of repo dicts or None on fatal error."""
url = f"{API_BASE}/api/v1/users/{OWNER}/repos"
def fetch_repos(session: requests.Session, owner: str) -> Optional[list]:
"""Fetch list of repos owned by `owner`. Return list of repo dicts or None on fatal error."""
url = f"{API_BASE}/api/v1/users/{owner}/repos"
for attempt, backoff in enumerate(RETRY_BACKOFF):
if attempt > 0:
@@ -200,9 +201,9 @@ def fetch_repos(session: requests.Session) -> Optional[list]:
return None
def fetch_branch_sha(session: requests.Session, repo_name: str, default_branch: str) -> Optional[str]:
def fetch_branch_sha(session: requests.Session, owner: str, repo_name: str, default_branch: str) -> Optional[str]:
"""Fetch commit SHA for repo's default branch. Return full hex SHA or None on error."""
url = f"{API_BASE}/api/v1/repos/{OWNER}/{quote(repo_name, safe='')}/branches/{quote(default_branch, safe='')}"
url = f"{API_BASE}/api/v1/repos/{owner}/{quote(repo_name, safe='')}/branches/{quote(default_branch, safe='')}"
for attempt, backoff in enumerate(RETRY_BACKOFF):
if attempt > 0:
@@ -228,19 +229,24 @@ def fetch_branch_sha(session: requests.Session, repo_name: str, default_branch:
return None
def download_repo_zip(session: requests.Session, repo_name: str, full_sha: str) -> Optional[Path]:
"""Download repo zip for given SHA. Return Path to downloaded file (in PENDING_DIR) or None."""
def download_repo_zip(session: requests.Session, owner: str, repo_name: str, full_sha: str) -> Optional[Path]:
"""Download repo zip for given SHA. Return Path to downloaded file (in PENDING_DIR) or None.
Pending filename is `{owner}__{repo}_{sha8}.zip` so owners with identical repo names
don't collide in the shared PENDING_DIR. flush_pending() strips the owner prefix
when routing to the per-owner pcloud subdir.
"""
# Defense in depth: validate repo_name even though main() should have done so
if not _safe_repo_name(repo_name):
logger.error("refusing unsafe repo name in download_repo_zip: %r", repo_name)
if not _safe_repo_name(repo_name) or not _safe_repo_name(owner):
logger.error("refusing unsafe owner/repo name in download_repo_zip: %r/%r", owner, repo_name)
return None
filename = f"{repo_name}_{full_sha[:SHORT_SHA_LEN]}.zip"
filename = f"{owner}__{repo_name}_{full_sha[:SHORT_SHA_LEN]}.zip"
partial_path = PENDING_DIR / f"{filename}.partial"
final_path = PENDING_DIR / filename
# URL-encode repo_name and full_sha for safety (defense in depth after main-level validation)
url = f"{API_BASE}/{OWNER}/{quote(repo_name, safe='')}/archive/{quote(full_sha, safe='')}.zip"
url = f"{API_BASE}/{owner}/{quote(repo_name, safe='')}/archive/{quote(full_sha, safe='')}.zip"
for attempt, backoff in enumerate(RETRY_BACKOFF):
if attempt > 0:
@@ -291,22 +297,35 @@ def download_repo_zip(session: requests.Session, repo_name: str, full_sha: str)
def flush_pending(pcloud_healthy: bool) -> list:
"""Move all *.zip files from PENDING_DIR to PCLOUD_TARGET if healthy.
"""Move all *.zip files from PENDING_DIR to PCLOUD_TARGET/<owner>/ if healthy.
Return list of (repo, short_sha, final_location) tuples for files moved.
Pending filenames are `{owner}__{repo}_{sha8}.zip`. The owner prefix is stripped
when placing into pcloud (final name is `{repo}_{sha8}.zip` inside owner subdir).
Return list of (owner, repo, short_sha, final_location) tuples for files moved.
"""
moved = []
if not pcloud_healthy or not PENDING_DIR.exists():
return moved
for pending_file in PENDING_DIR.glob("*.zip"):
dst = PCLOUD_TARGET / pending_file.name
# Expect `{owner}__{repo}_{sha8}.zip`
if "__" not in pending_file.name:
logger.warning("skipping pending file without owner prefix: %s", pending_file.name)
continue
owner, rest = pending_file.name.split("__", 1)
if not _safe_repo_name(owner):
logger.warning("refusing unsafe owner prefix in pending file: %r", owner)
continue
owner_dir = PCLOUD_TARGET / owner
owner_dir.mkdir(parents=True, exist_ok=True)
dst = owner_dir / rest
if place_into_pcloud(pending_file, dst):
# Extract repo name and short sha from filename
name_parts = pending_file.stem.rsplit("_", 1)
# Extract repo name and short sha from remainder (`{repo}_{sha8}.zip`)
name_parts = Path(rest).stem.rsplit("_", 1)
if len(name_parts) == 2:
repo, short_sha = name_parts
moved.append((repo, short_sha, str(dst)))
moved.append((owner, repo, short_sha, str(dst)))
logger.info("Flushed %s to pCloud", pending_file.name)
return moved
@@ -454,15 +473,25 @@ def _send_matrix(message: str) -> bool:
return False
def notify(summary: List[Tuple[str, str, str]]) -> None:
"""Build a Matrix message and POST it. Non-fatal."""
if not summary:
def notify(summary: List[Tuple[str, str, str, str]], failed_owners: Optional[List[str]] = None) -> None:
"""Build a Matrix message and POST it. Non-fatal.
`summary` is a list of (owner, repo, short_sha, final_location).
`failed_owners` is a list of owners whose repo-list fetch failed this run.
"""
failed_owners = failed_owners or []
if not summary and not failed_owners:
return
repos = [f"{repo}@{short_sha}" for repo, short_sha, _ in summary]
repos_str = ", ".join(repos)
text = (
f"coe-mirror: {len(repos)} new from Nightmare_Eclipse: {repos_str}"
)
parts = []
if summary:
by_owner: dict = {}
for owner, repo, short_sha, _ in summary:
by_owner.setdefault(owner, []).append(f"{repo}@{short_sha}")
chunks = [f"{owner}: {', '.join(items)}" for owner, items in by_owner.items()]
parts.append(f"coe-mirror: {len(summary)} new — " + " | ".join(chunks))
if failed_owners:
parts.append(f"{len(failed_owners)} owner(s) failed: {', '.join(failed_owners)}")
text = " ; ".join(parts)
# Always log the intended message so behavior is visible even when Matrix is down
logger.info("notify: %s", text)
_send_matrix(text)
@@ -487,69 +516,87 @@ def main() -> int:
# Flush pending at start
initial_flush = flush_pending(pcloud_healthy)
# Create session and fetch repos
# Create session and iterate owners
session = requests.Session()
session.headers.update({"User-Agent": USER_AGENT})
repos = fetch_repos(session)
new_files = list(initial_flush)
failed_owners: List[str] = []
for owner in OWNERS:
if not _safe_repo_name(owner):
logger.warning("refusing unsafe owner name: %r", owner)
failed_owners.append(owner)
continue
repos = fetch_repos(session, owner)
if repos is None:
logger.error("Fatal: could not fetch repos")
return 1
logger.error("Could not fetch repo list for %s; continuing to next owner", owner)
failed_owners.append(owner)
continue
logger.info("Fetched %d repos", len(repos))
logger.info("Fetched %d repos for %s", len(repos), owner)
new_files = []
# Ensure per-owner pcloud subdir exists (best-effort; probe already covered parent)
owner_pcloud = PCLOUD_TARGET / owner
if pcloud_healthy:
try:
owner_pcloud.mkdir(parents=True, exist_ok=True)
except OSError as exc:
logger.warning("could not create owner subdir %s: %s", owner_pcloud, exc)
# Process each repo
# Process each repo for this owner
for repo in repos:
repo_name = repo.get("name")
default_branch = repo.get("default_branch", "main")
if not repo_name:
logger.warning("Repo object missing 'name'")
logger.warning("Repo object missing 'name' (owner=%s)", owner)
continue
# Fix 1: Reject unsafe repo names to prevent path traversal
# Reject unsafe repo names to prevent path traversal
if not _safe_repo_name(repo_name):
logger.warning("refusing unsafe repo name: %r", repo_name)
logger.warning("refusing unsafe repo name: %r (owner=%s)", repo_name, owner)
continue
sha = fetch_branch_sha(session, repo_name, default_branch)
sha = fetch_branch_sha(session, owner, repo_name, default_branch)
if sha is None:
logger.error("Skipping %s (could not fetch branch SHA)", repo_name)
logger.error("Skipping %s/%s (could not fetch branch SHA)", owner, repo_name)
continue
# Fix 3: Check existence before download to avoid SameFileError
filename = f"{repo_name}_{sha[:SHORT_SHA_LEN]}.zip"
pcloud_path = PCLOUD_TARGET / filename
pending_path = PENDING_DIR / filename
short_sha = sha[:SHORT_SHA_LEN]
# pcloud final name: {repo}_{sha8}.zip inside owner subdir
pcloud_final = f"{repo_name}_{short_sha}.zip"
pcloud_path = owner_pcloud / pcloud_final
# pending name carries owner prefix to avoid cross-owner collisions
pending_name = f"{owner}__{pcloud_final}"
pending_path = PENDING_DIR / pending_name
already_have = (pcloud_path.exists() and pcloud_path.stat().st_size > 0) \
or (pending_path.exists() and pending_path.stat().st_size > 0)
if already_have:
logger.info("Skipping %s (already present)", filename)
logger.info("Skipping %s (already present)", pending_name)
continue
downloaded = download_repo_zip(session, repo_name, sha)
downloaded = download_repo_zip(session, owner, repo_name, sha)
if downloaded:
short_sha = sha[:SHORT_SHA_LEN]
# Try to place it
if pcloud_healthy:
dst = PCLOUD_TARGET / downloaded.name
dst = owner_pcloud / pcloud_final
if place_into_pcloud(downloaded, dst):
new_files.append((repo_name, short_sha, str(dst)))
new_files.append((owner, repo_name, short_sha, str(dst)))
else:
logger.info("File %s left in pending (placement failed)", downloaded.name)
new_files.append((repo_name, short_sha, str(downloaded)))
new_files.append((owner, repo_name, short_sha, str(downloaded)))
else:
logger.info("pCloud unhealthy; %s queued in pending", downloaded.name)
new_files.append((repo_name, short_sha, str(downloaded)))
new_files.append((owner, repo_name, short_sha, str(downloaded)))
# Flush pending at end
final_flush = flush_pending(pcloud_healthy)
new_files.extend(final_flush)
# Notify
notify(new_files)
notify(new_files, failed_owners=failed_owners)
logger.info("coe-mirror complete")
return 0
tests/test_coe_mirror.py
+130 -41
View File
@@ -33,6 +33,7 @@ def mock_paths(tmp_path, monkeypatch):
log_dir = tmp_path / "logs"
pcloud_target.mkdir()
(pcloud_target / "TestOwner").mkdir()
pending_dir.mkdir()
log_dir.mkdir()
@@ -40,6 +41,7 @@ def mock_paths(tmp_path, monkeypatch):
monkeypatch.setattr(coe_mirror, "PENDING_DIR", pending_dir)
monkeypatch.setattr(coe_mirror, "LOG_DIR", log_dir)
monkeypatch.setattr(coe_mirror, "LOG_FILE", log_dir / "test.log")
monkeypatch.setattr(coe_mirror, "OWNERS", ["TestOwner"])
return {
"pcloud_target": pcloud_target,
@@ -94,12 +96,12 @@ class TestNewRepoDownloadsAndPlacesInPcloud:
with requests_mock.Mocker() as m:
# Mock repo list
m.get(
f"{coe_mirror.API_BASE}/api/v1/users/{coe_mirror.OWNER}/repos",
f"{coe_mirror.API_BASE}/api/v1/users/TestOwner/repos",
json=[{"name": "test-repo", "default_branch": "main"}],
)
# Mock branch info
m.get(
f"{coe_mirror.API_BASE}/api/v1/repos/{coe_mirror.OWNER}/test-repo/branches/main",
f"{coe_mirror.API_BASE}/api/v1/repos/TestOwner/test-repo/branches/main",
json={"commit": {"id": "aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj"}},
)
@@ -110,7 +112,7 @@ class TestNewRepoDownloadsAndPlacesInPcloud:
zip_buffer.seek(0)
m.get(
f"{coe_mirror.API_BASE}/{coe_mirror.OWNER}/test-repo/archive/aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj.zip",
f"{coe_mirror.API_BASE}/TestOwner/test-repo/archive/aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj.zip",
content=zip_buffer.getvalue(),
)
@@ -118,7 +120,7 @@ class TestNewRepoDownloadsAndPlacesInPcloud:
assert result == 0
# Verify file landed in pCloud with expected name
expected_file = mock_paths["pcloud_target"] / "test-repo_aaaabbbb.zip"
expected_file = mock_paths["pcloud_target"] / "TestOwner" / "test-repo_aaaabbbb.zip"
assert expected_file.exists()
assert expected_file.stat().st_size > 0
@@ -136,17 +138,17 @@ class TestNewCommitCreatesSnapshot:
monkeypatch.setattr(coe_mirror, "probe_pcloud", lambda: True)
# Pre-create old file
old_file = mock_paths["pcloud_target"] / "test-repo_aaaabbbb.zip"
old_file = mock_paths["pcloud_target"] / "TestOwner" / "test-repo_aaaabbbb.zip"
create_valid_zip(old_file, "old")
with requests_mock.Mocker() as m:
m.get(
f"{coe_mirror.API_BASE}/api/v1/users/{coe_mirror.OWNER}/repos",
f"{coe_mirror.API_BASE}/api/v1/users/TestOwner/repos",
json=[{"name": "test-repo", "default_branch": "main"}],
)
# Return new SHA
m.get(
f"{coe_mirror.API_BASE}/api/v1/repos/{coe_mirror.OWNER}/test-repo/branches/main",
f"{coe_mirror.API_BASE}/api/v1/repos/TestOwner/test-repo/branches/main",
json={"commit": {"id": "bbbbccccddddeeeeffffgggghhhhiiiijjjjkkkk"}},
)
@@ -157,7 +159,7 @@ class TestNewCommitCreatesSnapshot:
zip_buffer.seek(0)
m.get(
f"{coe_mirror.API_BASE}/{coe_mirror.OWNER}/test-repo/archive/bbbbccccddddeeeeffffgggghhhhiiiijjjjkkkk.zip",
f"{coe_mirror.API_BASE}/TestOwner/test-repo/archive/bbbbccccddddeeeeffffgggghhhhiiiijjjjkkkk.zip",
content=zip_buffer.getvalue(),
)
@@ -167,7 +169,7 @@ class TestNewCommitCreatesSnapshot:
# Verify old file still exists
assert old_file.exists()
# Verify new file landed
new_file = mock_paths["pcloud_target"] / "test-repo_bbbbcccc.zip"
new_file = mock_paths["pcloud_target"] / "TestOwner" / "test-repo_bbbbcccc.zip"
assert new_file.exists()
@@ -182,7 +184,7 @@ class TestPcloudFailureKeepsPending:
# Make pCloud target a file (not a dir) to cause write failure
pcloud_as_file = mock_paths["pcloud_target"]
pcloud_as_file.rmdir()
shutil.rmtree(pcloud_as_file)
pcloud_as_file.write_text("block")
# Mock place_into_pcloud to fail
@@ -193,11 +195,11 @@ class TestPcloudFailureKeepsPending:
with requests_mock.Mocker() as m:
m.get(
f"{coe_mirror.API_BASE}/api/v1/users/{coe_mirror.OWNER}/repos",
f"{coe_mirror.API_BASE}/api/v1/users/TestOwner/repos",
json=[{"name": "test-repo", "default_branch": "main"}],
)
m.get(
f"{coe_mirror.API_BASE}/api/v1/repos/{coe_mirror.OWNER}/test-repo/branches/main",
f"{coe_mirror.API_BASE}/api/v1/repos/TestOwner/test-repo/branches/main",
json={"commit": {"id": "aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj"}},
)
@@ -207,7 +209,7 @@ class TestPcloudFailureKeepsPending:
zip_buffer.seek(0)
m.get(
f"{coe_mirror.API_BASE}/{coe_mirror.OWNER}/test-repo/archive/aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj.zip",
f"{coe_mirror.API_BASE}/TestOwner/test-repo/archive/aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj.zip",
content=zip_buffer.getvalue(),
)
@@ -215,7 +217,7 @@ class TestPcloudFailureKeepsPending:
assert result == 0
# Verify file is in pending
pending_file = mock_paths["pending_dir"] / "test-repo_aaaabbbb.zip"
pending_file = mock_paths["pending_dir"] / "TestOwner__test-repo_aaaabbbb.zip"
assert pending_file.exists()
@@ -232,13 +234,13 @@ class TestPendingFlushesWhenPcloudRecoveres:
monkeypatch.setattr(coe_mirror, "probe_pcloud", lambda: True)
# Create a pending file
pending_file = mock_paths["pending_dir"] / "test-repo_aaaabbbb.zip"
pending_file = mock_paths["pending_dir"] / "TestOwner__test-repo_aaaabbbb.zip"
create_valid_zip(pending_file, "pending")
with requests_mock.Mocker() as m:
# Empty repo list (no new downloads)
m.get(
f"{coe_mirror.API_BASE}/api/v1/users/{coe_mirror.OWNER}/repos",
f"{coe_mirror.API_BASE}/api/v1/users/TestOwner/repos",
json=[],
)
@@ -246,7 +248,7 @@ class TestPendingFlushesWhenPcloudRecoveres:
assert result == 0
# Verify file moved to pCloud
pcloud_file = mock_paths["pcloud_target"] / "test-repo_aaaabbbb.zip"
pcloud_file = mock_paths["pcloud_target"] / "TestOwner" / "test-repo_aaaabbbb.zip"
assert pcloud_file.exists()
assert not pending_file.exists()
@@ -262,16 +264,16 @@ class TestMalformedZipRejected:
with requests_mock.Mocker() as m:
m.get(
f"{coe_mirror.API_BASE}/api/v1/users/{coe_mirror.OWNER}/repos",
f"{coe_mirror.API_BASE}/api/v1/users/TestOwner/repos",
json=[{"name": "test-repo", "default_branch": "main"}],
)
m.get(
f"{coe_mirror.API_BASE}/api/v1/repos/{coe_mirror.OWNER}/test-repo/branches/main",
f"{coe_mirror.API_BASE}/api/v1/repos/TestOwner/test-repo/branches/main",
json={"commit": {"id": "aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj"}},
)
# Return binary garbage
m.get(
f"{coe_mirror.API_BASE}/{coe_mirror.OWNER}/test-repo/archive/aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj.zip",
f"{coe_mirror.API_BASE}/TestOwner/test-repo/archive/aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj.zip",
content=b"GARBAGE_DATA_NOT_A_ZIP",
)
@@ -279,9 +281,9 @@ class TestMalformedZipRejected:
assert result == 0
# Verify nothing in pCloud or pending
pcloud_file = mock_paths["pcloud_target"] / "test-repo_aaaabbbb.zip"
pcloud_file = mock_paths["pcloud_target"] / "TestOwner" / "test-repo_aaaabbbb.zip"
assert not pcloud_file.exists()
pending_file = mock_paths["pending_dir"] / "test-repo_aaaabbbb.zip"
pending_file = mock_paths["pending_dir"] / "TestOwner__test-repo_aaaabbbb.zip"
assert not pending_file.exists()
@@ -294,9 +296,9 @@ class TestNotifyStub:
original_notify = coe_mirror.notify
call_log = []
def spy_notify(summary):
def spy_notify(summary, failed_owners=None):
call_log.append(summary)
return original_notify(summary)
return original_notify(summary, failed_owners=failed_owners)
monkeypatch.setattr(coe_mirror, "notify", spy_notify)
@@ -307,7 +309,7 @@ class TestNotifyStub:
# Test with files
call_log.clear()
coe_mirror.notify([("repo", "aaaabbbb", "/path")])
coe_mirror.notify([("TestOwner", "repo", "aaaabbbb", "/path")])
assert len(call_log) == 1
@@ -334,7 +336,7 @@ class TestApiRetriesOn429:
with requests_mock.Mocker() as m:
m.get(
f"{coe_mirror.API_BASE}/api/v1/users/{coe_mirror.OWNER}/repos",
f"{coe_mirror.API_BASE}/api/v1/users/TestOwner/repos",
[
{"status_code": 429, "json": None},
{"status_code": 429, "json": None},
@@ -342,7 +344,7 @@ class TestApiRetriesOn429:
],
)
m.get(
f"{coe_mirror.API_BASE}/api/v1/repos/{coe_mirror.OWNER}/test-repo/branches/main",
f"{coe_mirror.API_BASE}/api/v1/repos/TestOwner/test-repo/branches/main",
json={"commit": {"id": "aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj"}},
)
@@ -352,7 +354,7 @@ class TestApiRetriesOn429:
zip_buffer.seek(0)
m.get(
f"{coe_mirror.API_BASE}/{coe_mirror.OWNER}/test-repo/archive/aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj.zip",
f"{coe_mirror.API_BASE}/TestOwner/test-repo/archive/aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj.zip",
content=zip_buffer.getvalue(),
)
@@ -379,11 +381,11 @@ class TestSizeMismatchKeepsPending:
with requests_mock.Mocker() as m:
m.get(
f"{coe_mirror.API_BASE}/api/v1/users/{coe_mirror.OWNER}/repos",
f"{coe_mirror.API_BASE}/api/v1/users/TestOwner/repos",
json=[{"name": "test-repo", "default_branch": "main"}],
)
m.get(
f"{coe_mirror.API_BASE}/api/v1/repos/{coe_mirror.OWNER}/test-repo/branches/main",
f"{coe_mirror.API_BASE}/api/v1/repos/TestOwner/test-repo/branches/main",
json={"commit": {"id": "aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj"}},
)
@@ -393,7 +395,7 @@ class TestSizeMismatchKeepsPending:
zip_buffer.seek(0)
m.get(
f"{coe_mirror.API_BASE}/{coe_mirror.OWNER}/test-repo/archive/aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj.zip",
f"{coe_mirror.API_BASE}/TestOwner/test-repo/archive/aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj.zip",
content=zip_buffer.getvalue(),
)
@@ -401,11 +403,11 @@ class TestSizeMismatchKeepsPending:
assert result == 0
# Verify bad dst was removed
bad_dst = mock_paths["pcloud_target"] / "test-repo_aaaabbbb.zip"
bad_dst = mock_paths["pcloud_target"] / "TestOwner" / "test-repo_aaaabbbb.zip"
assert not bad_dst.exists()
# Verify src still in pending
pending_src = mock_paths["pending_dir"] / "test-repo_aaaabbbb.zip"
pending_src = mock_paths["pending_dir"] / "TestOwner__test-repo_aaaabbbb.zip"
assert pending_src.exists()
@@ -419,7 +421,7 @@ class TestExistenceCheckSkipsZeroByte:
monkeypatch.setattr(time, "sleep", lambda x: None)
# Pre-create zero-byte file in pCloud target
zero_file = mock_paths["pcloud_target"] / "test-repo_aaaabbbb.zip"
zero_file = mock_paths["pcloud_target"] / "TestOwner" / "test-repo_aaaabbbb.zip"
zero_file.write_bytes(b"")
assert zero_file.stat().st_size == 0
@@ -433,11 +435,11 @@ class TestExistenceCheckSkipsZeroByte:
with requests_mock.Mocker() as m:
m.get(
f"{coe_mirror.API_BASE}/api/v1/users/{coe_mirror.OWNER}/repos",
f"{coe_mirror.API_BASE}/api/v1/users/TestOwner/repos",
json=[{"name": "test-repo", "default_branch": "main"}],
)
m.get(
f"{coe_mirror.API_BASE}/api/v1/repos/{coe_mirror.OWNER}/test-repo/branches/main",
f"{coe_mirror.API_BASE}/api/v1/repos/TestOwner/test-repo/branches/main",
json={"commit": {"id": "aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj"}},
)
@@ -447,7 +449,7 @@ class TestExistenceCheckSkipsZeroByte:
zip_buffer.seek(0)
m.get(
f"{coe_mirror.API_BASE}/{coe_mirror.OWNER}/test-repo/archive/aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj.zip",
f"{coe_mirror.API_BASE}/TestOwner/test-repo/archive/aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj.zip",
content=zip_buffer.getvalue(),
)
@@ -502,7 +504,7 @@ class TestUnsafeRepoNameRejected:
with requests_mock.Mocker() as m:
# Mock repo list returning unsafe repo name
m.get(
f"{coe_mirror.API_BASE}/api/v1/users/{coe_mirror.OWNER}/repos",
f"{coe_mirror.API_BASE}/api/v1/users/TestOwner/repos",
json=[{"name": "../../../etc/passwd", "default_branch": "main"}],
)
@@ -530,7 +532,7 @@ class TestSkipWhenFileAlreadyInPcloud:
monkeypatch.setattr(coe_mirror, "probe_pcloud", lambda: True)
# Pre-create a non-empty file in pCloud
existing_file = mock_paths["pcloud_target"] / "RepoX_abcdef12.zip"
existing_file = mock_paths["pcloud_target"] / "TestOwner" / "RepoX_abcdef12.zip"
create_valid_zip(existing_file, "existing")
# Spy on download_repo_zip to ensure it's not called
@@ -546,12 +548,12 @@ class TestSkipWhenFileAlreadyInPcloud:
with requests_mock.Mocker() as m:
# Mock repo list
m.get(
f"{coe_mirror.API_BASE}/api/v1/users/{coe_mirror.OWNER}/repos",
f"{coe_mirror.API_BASE}/api/v1/users/TestOwner/repos",
json=[{"name": "RepoX", "default_branch": "main"}],
)
# Return matching SHA (first 8 chars = abcdef12)
m.get(
f"{coe_mirror.API_BASE}/api/v1/repos/{coe_mirror.OWNER}/RepoX/branches/main",
f"{coe_mirror.API_BASE}/api/v1/repos/TestOwner/RepoX/branches/main",
json={"commit": {"id": "abcdef1234567890"}},
)
@@ -824,3 +826,90 @@ class TestMatrixLoginWritesCacheChmod600:
with open(token_cache_file) as f:
cached = json.load(f)
assert cached["access_token"] == "new-token"
class TestMultiOwner:
"""Multi-owner behavior: collisions, per-owner failure isolation."""
def test_same_repo_name_across_owners_does_not_collide(self, mock_paths, setup_logging, monkeypatch):
"""Two owners each have a repo named 'shared' — both land in their own subdirs."""
monkeypatch.setattr(coe_mirror, "OWNERS", ["OwnerA", "OwnerB"])
(mock_paths["pcloud_target"] / "OwnerA").mkdir(exist_ok=True)
(mock_paths["pcloud_target"] / "OwnerB").mkdir(exist_ok=True)
monkeypatch.setattr(coe_mirror, "acquire_lock", lambda: 999)
monkeypatch.setattr(coe_mirror, "release_lock", lambda fd: None)
monkeypatch.setattr(time, "sleep", lambda x: None)
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w") as zf:
zf.writestr("README.md", "hi")
zip_bytes = buf.getvalue()
sha_a = "a" * 40
sha_b = "b" * 40
with requests_mock.Mocker() as m:
for owner, sha in (("OwnerA", sha_a), ("OwnerB", sha_b)):
m.get(
f"{coe_mirror.API_BASE}/api/v1/users/{owner}/repos",
json=[{"name": "shared", "default_branch": "main"}],
)
m.get(
f"{coe_mirror.API_BASE}/api/v1/repos/{owner}/shared/branches/main",
json={"commit": {"id": sha}},
)
m.get(
f"{coe_mirror.API_BASE}/{owner}/shared/archive/{sha}.zip",
content=zip_bytes,
)
rc = coe_mirror.main()
assert rc == 0
a_file = mock_paths["pcloud_target"] / "OwnerA" / f"shared_{sha_a[:8]}.zip"
b_file = mock_paths["pcloud_target"] / "OwnerB" / f"shared_{sha_b[:8]}.zip"
assert a_file.exists() and a_file.stat().st_size > 0
assert b_file.exists() and b_file.stat().st_size > 0
def test_one_owner_failure_does_not_block_others(self, mock_paths, setup_logging, monkeypatch):
"""OwnerA repo-list fetch 500s; OwnerB still processed and reported healthy."""
monkeypatch.setattr(coe_mirror, "OWNERS", ["OwnerA", "OwnerB"])
(mock_paths["pcloud_target"] / "OwnerA").mkdir(exist_ok=True)
(mock_paths["pcloud_target"] / "OwnerB").mkdir(exist_ok=True)
monkeypatch.setattr(coe_mirror, "acquire_lock", lambda: 999)
monkeypatch.setattr(coe_mirror, "release_lock", lambda fd: None)
monkeypatch.setattr(time, "sleep", lambda x: None)
# Capture notify args
captured = {}
def spy_notify(summary, failed_owners=None):
captured["summary"] = summary
captured["failed_owners"] = failed_owners
monkeypatch.setattr(coe_mirror, "notify", spy_notify)
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w") as zf:
zf.writestr("README.md", "hi")
zip_bytes = buf.getvalue()
sha_b = "b" * 40
with requests_mock.Mocker() as m:
m.get(f"{coe_mirror.API_BASE}/api/v1/users/OwnerA/repos", status_code=500)
m.get(
f"{coe_mirror.API_BASE}/api/v1/users/OwnerB/repos",
json=[{"name": "onlyone", "default_branch": "main"}],
)
m.get(
f"{coe_mirror.API_BASE}/api/v1/repos/OwnerB/onlyone/branches/main",
json={"commit": {"id": sha_b}},
)
m.get(
f"{coe_mirror.API_BASE}/OwnerB/onlyone/archive/{sha_b}.zip",
content=zip_bytes,
)
rc = coe_mirror.main()
assert rc == 0
b_file = mock_paths["pcloud_target"] / "OwnerB" / f"onlyone_{sha_b[:8]}.zip"
assert b_file.exists()
assert captured["failed_owners"] == ["OwnerA"]
# summary contains only the OwnerB entry
assert any(owner == "OwnerB" and repo == "onlyone" for owner, repo, *_ in captured["summary"])