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

Genericize: drop pCloud terminology, use TARGET_DIR / target_dir

Browse Source 
pCloud was never a hard dependency - the tool writes to any local directory.
Renaming makes the tool clearly generic before publishing.

- PCLOUD_TARGET → TARGET_DIR
- probe_pcloud → probe_target
- place_into_pcloud → place_into_target
- pcloud_target config key → target_dir
- COE_MIRROR_PCLOUD_TARGET env var → COE_MIRROR_TARGET_DIR
- Hardcoded default: ~/pCloudDrive/Hacking/ChurchOfMalware → ~/coe-mirror
- Log messages and docstrings updated
- Tests updated to match

DevTrack #1286
This commit is contained in:
n0mad1k
2026-07-07 11:12:46 -04:00
parent 376115a9de
commit 759d4a6e23
3 changed files with 111 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+3 -3
View File
@@ -1,6 +1,6 @@
# coe-mirror
Mirror repos from a Forgejo/Gitea instance to a local directory (typically pCloud). Built for `git.churchofmalware.org` but usable against any Forgejo/Gitea host.
Mirror repos from a Forgejo/Gitea instance to a local directory. Built for `git.churchofmalware.org` but usable against any Forgejo/Gitea host.
## What it does
@@ -36,7 +36,7 @@ Full schema:
```toml
api_base = "https://git.churchofmalware.org"
pcloud_target = "~/pCloudDrive/Hacking/ChurchOfMalware"
target_dir = "~/coe-mirror"
owners = ["Nightmare_Eclipse", "shai_hulud"]
# pending_dir = "~/.local/share/coe-mirror/pending" # optional
@@ -52,7 +52,7 @@ room = ""
| `COE_MIRROR_CONFIG` | Alternate path to config.toml |
| `COE_MIRROR_API_BASE` | Forgejo/Gitea base URL |
| `COE_MIRROR_OWNERS` | Comma-separated owners (`"a,b,c"`) |
| `COE_MIRROR_PCLOUD_TARGET` | Target directory |
| `COE_MIRROR_TARGET_DIR` | Target directory |
| `COE_MIRROR_PENDING_DIR` | Local staging dir |
| `COE_MIRROR_MATRIX_HOMESERVER` | Matrix homeserver URL |
| `COE_MIRROR_MATRIX_ROOM` | Matrix room ID |
coe_mirror.py
+42 -42
View File
@@ -1,7 +1,7 @@
"""Mirror repos from a Forgejo instance to a local target dir (e.g. pCloud).
"""Mirror repos from a Forgejo/Gitea instance to a local target directory.
Fetches all repos owned by each user in OWNERS and stores per-commit zip snapshots
under `PCLOUD_TARGET/<owner>/`. Runs unattended via systemd timer.
under `TARGET_DIR/<owner>/`. Runs unattended via systemd timer.
Local staging when the target is unavailable.
Configuration precedence (for each key): env var > config file > hardcoded default.
@@ -67,11 +67,11 @@ OWNERS: List[str] = (
or (_CFG.get("owners") if isinstance(_CFG.get("owners"), list) else None)
or ["Nightmare_Eclipse", "shai_hulud", "bikini_mirror", "ek0mssavi0r"]
)
PCLOUD_TARGET: Path = Path(
TARGET_DIR: Path = Path(
_resolve(
"COE_MIRROR_PCLOUD_TARGET",
_CFG.get("pcloud_target"),
str(Path.home() / "pCloudDrive" / "Hacking" / "ChurchOfMalware"),
"COE_MIRROR_TARGET_DIR",
_CFG.get("target_dir"),
str(Path.home() / "coe-mirror"),
)
).expanduser()
PENDING_DIR: Path = Path(
@@ -165,11 +165,11 @@ def release_lock(lock_fd: int) -> None:
pass
def probe_pcloud() -> bool:
"""Probe pCloud health. Write tiny file, read back, delete. Return True if healthy."""
def probe_target() -> bool:
"""Probe target directory health. Write tiny file, read back, delete. Return True if healthy."""
try:
probe_file = PCLOUD_TARGET / ".write_probe"
PCLOUD_TARGET.mkdir(parents=True, exist_ok=True)
probe_file = TARGET_DIR / ".write_probe"
TARGET_DIR.mkdir(parents=True, exist_ok=True)
probe_file.write_text("probe")
# fsync on file descriptor, not stat
@@ -179,15 +179,15 @@ def probe_pcloud() -> bool:
probe_file.unlink()
if content != "probe":
logger.warning("pCloud write verification failed (content mismatch); pcloud unavailable; queuing locally")
logger.warning("target directory write verification failed (content mismatch); target unavailable; queuing locally")
return False
return True
except Exception as exc:
logger.warning("pCloud probe failed: %s; pcloud unavailable; queuing locally", exc)
logger.warning("target directory probe failed: %s; target unavailable; queuing locally", exc)
return False
def place_into_pcloud(src: Path, dst: Path) -> bool:
def place_into_target(src: Path, dst: Path) -> bool:
"""Copy src → dst with size verification; unlink src only on success.
Returns True on success, False on any failure. Never raises.
@@ -277,7 +277,7 @@ def download_repo_zip(session: requests.Session, owner: str, repo_name: str, ful
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.
when routing to the per-owner target subdir.
"""
# Defense in depth: validate repo_name even though main() should have done so
if not _safe_repo_name(repo_name) or not _safe_repo_name(owner):
@@ -339,16 +339,16 @@ def download_repo_zip(session: requests.Session, owner: str, repo_name: str, ful
return None
def flush_pending(pcloud_healthy: bool) -> list:
"""Move all *.zip files from PENDING_DIR to PCLOUD_TARGET/<owner>/ if healthy.
def flush_pending(target_healthy: bool) -> list:
"""Move all *.zip files from PENDING_DIR to TARGET_DIR/<owner>/ if healthy.
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).
when placing into target (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():
if not target_healthy or not PENDING_DIR.exists():
return moved
for pending_file in PENDING_DIR.glob("*.zip"):
@@ -360,16 +360,16 @@ def flush_pending(pcloud_healthy: bool) -> list:
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 = TARGET_DIR / owner
owner_dir.mkdir(parents=True, exist_ok=True)
dst = owner_dir / rest
if place_into_pcloud(pending_file, dst):
if place_into_target(pending_file, dst):
# 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((owner, repo, short_sha, str(dst)))
logger.info("Flushed %s to pCloud", pending_file.name)
logger.info("Flushed %s to target directory", pending_file.name)
return moved
@@ -567,12 +567,12 @@ def main() -> int:
try:
logger.info("coe-mirror starting")
# Probe pCloud health
pcloud_healthy = probe_pcloud()
logger.info("pCloud health check: %s", "healthy" if pcloud_healthy else "unavailable")
# Probe target directory health
target_healthy = probe_target()
logger.info("target directory health check: %s", "healthy" if target_healthy else "unavailable")
# Flush pending at start
initial_flush = flush_pending(pcloud_healthy)
initial_flush = flush_pending(target_healthy)
# Create session and iterate owners
session = requests.Session()
@@ -595,13 +595,13 @@ def main() -> int:
logger.info("Fetched %d repos for %s", len(repos), owner)
# Ensure per-owner pcloud subdir exists (best-effort; probe already covered parent)
owner_pcloud = PCLOUD_TARGET / owner
if pcloud_healthy:
# Ensure per-owner target subdir exists (best-effort; probe already covered parent)
owner_target = TARGET_DIR / owner
if target_healthy:
try:
owner_pcloud.mkdir(parents=True, exist_ok=True)
owner_target.mkdir(parents=True, exist_ok=True)
except OSError as exc:
logger.warning("could not create owner subdir %s: %s", owner_pcloud, exc)
logger.warning("could not create owner subdir %s: %s", owner_target, exc)
# Process each repo for this owner
for repo in repos:
@@ -623,13 +623,13 @@ def main() -> int:
continue
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
# final name in target: {repo}_{sha8}.zip inside owner subdir
target_final = f"{repo_name}_{short_sha}.zip"
target_path = owner_target / target_final
# pending name carries owner prefix to avoid cross-owner collisions
pending_name = f"{owner}__{pcloud_final}"
pending_name = f"{owner}__{target_final}"
pending_path = PENDING_DIR / pending_name
already_have = (pcloud_path.exists() and pcloud_path.stat().st_size > 0) \
already_have = (target_path.exists() and target_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)", pending_name)
@@ -638,19 +638,19 @@ def main() -> int:
downloaded = download_repo_zip(session, owner, repo_name, sha)
if downloaded:
# Try to place it
if pcloud_healthy:
dst = owner_pcloud / pcloud_final
if place_into_pcloud(downloaded, dst):
if target_healthy:
dst = owner_target / target_final
if place_into_target(downloaded, 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((owner, repo_name, short_sha, str(downloaded)))
else:
logger.info("pCloud unhealthy; %s queued in pending", downloaded.name)
logger.info("target directory unhealthy; %s queued in pending", downloaded.name)
new_files.append((owner, repo_name, short_sha, str(downloaded)))
# Flush pending at end
final_flush = flush_pending(pcloud_healthy)
final_flush = flush_pending(target_healthy)
new_files.extend(final_flush)
# Notify
@@ -671,9 +671,9 @@ EXAMPLE_CONFIG = """\
api_base = "https://git.churchofmalware.org"
# Where mirrored zips are stored. One subdirectory per owner is created here.
pcloud_target = "~/pCloudDrive/Hacking/ChurchOfMalware"
target_dir = "~/coe-mirror"
# Optional: local staging when pcloud_target is unavailable.
# Optional: local staging when target_dir is unavailable.
# pending_dir = "~/.local/share/coe-mirror/pending"
# Forgejo owners (users or orgs) to mirror. Edit this list to add/remove.
tests/test_coe_mirror.py
+66 -66
View File
@@ -28,16 +28,16 @@ import coe_mirror
@pytest.fixture
def mock_paths(tmp_path, monkeypatch):
"""Redirect all paths to tmp_path for test isolation."""
pcloud_target = tmp_path / "pcloud"
target_dir = tmp_path / "target"
pending_dir = tmp_path / "pending"
log_dir = tmp_path / "logs"
pcloud_target.mkdir()
(pcloud_target / "TestOwner").mkdir()
target_dir.mkdir()
(target_dir / "TestOwner").mkdir()
pending_dir.mkdir()
log_dir.mkdir()
monkeypatch.setattr(coe_mirror, "PCLOUD_TARGET", pcloud_target)
monkeypatch.setattr(coe_mirror, "TARGET_DIR", target_dir)
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")
@@ -46,7 +46,7 @@ def mock_paths(tmp_path, monkeypatch):
monkeypatch.setattr(coe_mirror, "MATRIX_ROOM", "!testroom:test.local")
return {
"pcloud_target": pcloud_target,
"target_dir": target_dir,
"pending_dir": pending_dir,
"log_dir": log_dir,
}
@@ -84,16 +84,16 @@ def mock_lock(monkeypatch):
class TestNewRepoDownloadsAndPlacesInPcloud:
"""Test new repo downloads and places in pCloud."""
"""Test new repo downloads and places in target directory."""
def test_new_repo_downloads_and_places_in_pcloud(self, mock_paths, setup_logging, monkeypatch):
"""Mock list returning one repo, assert file lands in pCloud with expected name."""
"""Mock list returning one repo, assert file lands in target directory with expected name."""
monkeypatch.setattr(coe_mirror, "acquire_lock", lambda: 999)
monkeypatch.setattr(coe_mirror, "release_lock", lambda fd: None)
monkeypatch.setattr(time, "sleep", lambda x: None) # Skip actual sleep
# Mock pCloud probe to return healthy
monkeypatch.setattr(coe_mirror, "probe_pcloud", lambda: True)
# Mock target directory probe to return healthy
monkeypatch.setattr(coe_mirror, "probe_target", lambda: True)
with requests_mock.Mocker() as m:
# Mock repo list
@@ -121,8 +121,8 @@ class TestNewRepoDownloadsAndPlacesInPcloud:
result = coe_mirror.main()
assert result == 0
# Verify file landed in pCloud with expected name
expected_file = mock_paths["pcloud_target"] / "TestOwner" / "test-repo_aaaabbbb.zip"
# Verify file landed in target directory with expected name
expected_file = mock_paths["target_dir"] / "TestOwner" / "test-repo_aaaabbbb.zip"
assert expected_file.exists()
assert expected_file.stat().st_size > 0
@@ -136,11 +136,11 @@ class TestNewCommitCreatesSnapshot:
monkeypatch.setattr(coe_mirror, "release_lock", lambda fd: None)
monkeypatch.setattr(time, "sleep", lambda x: None)
# Mock pCloud probe to return healthy
monkeypatch.setattr(coe_mirror, "probe_pcloud", lambda: True)
# Mock target directory probe to return healthy
monkeypatch.setattr(coe_mirror, "probe_target", lambda: True)
# Pre-create old file
old_file = mock_paths["pcloud_target"] / "TestOwner" / "test-repo_aaaabbbb.zip"
old_file = mock_paths["target_dir"] / "TestOwner" / "test-repo_aaaabbbb.zip"
create_valid_zip(old_file, "old")
with requests_mock.Mocker() as m:
@@ -171,29 +171,29 @@ class TestNewCommitCreatesSnapshot:
# Verify old file still exists
assert old_file.exists()
# Verify new file landed
new_file = mock_paths["pcloud_target"] / "TestOwner" / "test-repo_bbbbcccc.zip"
new_file = mock_paths["target_dir"] / "TestOwner" / "test-repo_bbbbcccc.zip"
assert new_file.exists()
class TestPcloudFailureKeepsPending:
"""Test pCloud write failure keeps file in pending with WARN log."""
"""Test target directory write failure keeps file in pending with WARN log."""
def test_pcloud_write_fails_keeps_in_pending(self, mock_paths, setup_logging, monkeypatch):
"""Make pCloud target non-writable, assert download stays in pending with WARN log."""
"""Make target directory target non-writable, assert download stays in pending with WARN log."""
monkeypatch.setattr(coe_mirror, "acquire_lock", lambda: 999)
monkeypatch.setattr(coe_mirror, "release_lock", lambda fd: None)
monkeypatch.setattr(time, "sleep", lambda x: None)
# Make pCloud target a file (not a dir) to cause write failure
pcloud_as_file = mock_paths["pcloud_target"]
shutil.rmtree(pcloud_as_file)
pcloud_as_file.write_text("block")
# Make target directory target a file (not a dir) to cause write failure
target_as_file = mock_paths["target_dir"]
shutil.rmtree(target_as_file)
target_as_file.write_text("block")
# Mock place_into_pcloud to fail
# Mock place_into_target to fail
def mock_place(src, dst):
return False
monkeypatch.setattr(coe_mirror, "place_into_pcloud", mock_place)
monkeypatch.setattr(coe_mirror, "place_into_target", mock_place)
with requests_mock.Mocker() as m:
m.get(
@@ -224,16 +224,16 @@ class TestPcloudFailureKeepsPending:
class TestPendingFlushesWhenPcloudRecoveres:
"""Test pending files flush when pCloud recovers."""
"""Test pending files flush when target directory recovers."""
def test_pending_flushes_when_pcloud_recovers(self, mock_paths, setup_logging, monkeypatch):
"""Drop valid zip in pending, make pCloud healthy, assert file lands in pCloud."""
"""Drop valid zip in pending, make target directory healthy, assert file lands in target directory."""
monkeypatch.setattr(coe_mirror, "acquire_lock", lambda: 999)
monkeypatch.setattr(coe_mirror, "release_lock", lambda fd: None)
monkeypatch.setattr(time, "sleep", lambda x: None)
# Mock pCloud probe to return healthy
monkeypatch.setattr(coe_mirror, "probe_pcloud", lambda: True)
# Mock target directory probe to return healthy
monkeypatch.setattr(coe_mirror, "probe_target", lambda: True)
# Create a pending file
pending_file = mock_paths["pending_dir"] / "TestOwner__test-repo_aaaabbbb.zip"
@@ -249,9 +249,9 @@ class TestPendingFlushesWhenPcloudRecoveres:
result = coe_mirror.main()
assert result == 0
# Verify file moved to pCloud
pcloud_file = mock_paths["pcloud_target"] / "TestOwner" / "test-repo_aaaabbbb.zip"
assert pcloud_file.exists()
# Verify file moved to target directory
target_file = mock_paths["target_dir"] / "TestOwner" / "test-repo_aaaabbbb.zip"
assert target_file.exists()
assert not pending_file.exists()
@@ -259,7 +259,7 @@ class TestMalformedZipRejected:
"""Test that malformed zip is rejected and not placed."""
def test_malformed_zip_rejected_and_not_placed(self, mock_paths, setup_logging, monkeypatch):
"""Mock zip download returning garbage, assert nothing lands in pCloud."""
"""Mock zip download returning garbage, assert nothing lands in target directory."""
monkeypatch.setattr(coe_mirror, "acquire_lock", lambda: 999)
monkeypatch.setattr(coe_mirror, "release_lock", lambda fd: None)
monkeypatch.setattr(time, "sleep", lambda x: None)
@@ -282,9 +282,9 @@ class TestMalformedZipRejected:
result = coe_mirror.main()
assert result == 0
# Verify nothing in pCloud or pending
pcloud_file = mock_paths["pcloud_target"] / "TestOwner" / "test-repo_aaaabbbb.zip"
assert not pcloud_file.exists()
# Verify nothing in target directory or pending
target_file = mock_paths["target_dir"] / "TestOwner" / "test-repo_aaaabbbb.zip"
assert not target_file.exists()
pending_file = mock_paths["pending_dir"] / "TestOwner__test-repo_aaaabbbb.zip"
assert not pending_file.exists()
@@ -405,7 +405,7 @@ class TestSizeMismatchKeepsPending:
assert result == 0
# Verify bad dst was removed
bad_dst = mock_paths["pcloud_target"] / "TestOwner" / "test-repo_aaaabbbb.zip"
bad_dst = mock_paths["target_dir"] / "TestOwner" / "test-repo_aaaabbbb.zip"
assert not bad_dst.exists()
# Verify src still in pending
@@ -414,7 +414,7 @@ class TestSizeMismatchKeepsPending:
class TestExistenceCheckSkipsZeroByte:
"""Test that zero-byte file in pCloud target causes re-download."""
"""Test that zero-byte file in target directory target causes re-download."""
def test_existence_check_skips_when_pcloud_has_zero_byte_file(self, mock_paths, setup_logging, monkeypatch):
"""Pre-create zero-byte file, assert download still happens because st_size check rejects it."""
@@ -422,18 +422,18 @@ class TestExistenceCheckSkipsZeroByte:
monkeypatch.setattr(coe_mirror, "release_lock", lambda fd: None)
monkeypatch.setattr(time, "sleep", lambda x: None)
# Pre-create zero-byte file in pCloud target
zero_file = mock_paths["pcloud_target"] / "TestOwner" / "test-repo_aaaabbbb.zip"
# Pre-create zero-byte file in target directory target
zero_file = mock_paths["target_dir"] / "TestOwner" / "test-repo_aaaabbbb.zip"
zero_file.write_bytes(b"")
assert zero_file.stat().st_size == 0
# Mock place_into_pcloud to succeed (make pCloud healthy)
# Mock place_into_target to succeed (make target directory healthy)
def mock_place(src, dst):
shutil.copy2(src, dst)
src.unlink()
return True
monkeypatch.setattr(coe_mirror, "place_into_pcloud", mock_place)
monkeypatch.setattr(coe_mirror, "place_into_target", mock_place)
with requests_mock.Mocker() as m:
m.get(
@@ -500,8 +500,8 @@ class TestUnsafeRepoNameRejected:
monkeypatch.setattr(coe_mirror, "release_lock", lambda fd: None)
monkeypatch.setattr(time, "sleep", lambda x: None)
# Mock pCloud probe to return healthy
monkeypatch.setattr(coe_mirror, "probe_pcloud", lambda: True)
# Mock target directory probe to return healthy
monkeypatch.setattr(coe_mirror, "probe_target", lambda: True)
with requests_mock.Mocker() as m:
# Mock repo list returning unsafe repo name
@@ -514,27 +514,27 @@ class TestUnsafeRepoNameRejected:
assert result == 0
# Verify no file was written anywhere with traversal path
assert not (mock_paths["pcloud_target"] / "etc").exists()
assert not (mock_paths["target_dir"] / "etc").exists()
assert not (mock_paths["pending_dir"] / "etc").exists()
# Also check that no file matching the unsafe name exists
unsafe_paths = list(mock_paths["pcloud_target"].glob("*..*.zip")) + list(mock_paths["pending_dir"].glob("*..*.zip"))
unsafe_paths = list(mock_paths["target_dir"].glob("*..*.zip")) + list(mock_paths["pending_dir"].glob("*..*.zip"))
assert len(unsafe_paths) == 0
class TestSkipWhenFileAlreadyInPcloud:
"""Test that file already in pCloud is skipped (Fix 3 - SameFileError)."""
"""Test that file already in target directory is skipped (Fix 3 - SameFileError)."""
def test_skip_when_file_already_in_pcloud(self, mock_paths, setup_logging, monkeypatch):
"""Pre-create non-empty zip in pCloud, assert download_repo_zip not called and no duplicate placed."""
"""Pre-create non-empty zip in target directory, assert download_repo_zip not called and no duplicate placed."""
monkeypatch.setattr(coe_mirror, "acquire_lock", lambda: 999)
monkeypatch.setattr(coe_mirror, "release_lock", lambda fd: None)
monkeypatch.setattr(time, "sleep", lambda x: None)
# Mock pCloud probe to return healthy
monkeypatch.setattr(coe_mirror, "probe_pcloud", lambda: True)
# Mock target directory probe to return healthy
monkeypatch.setattr(coe_mirror, "probe_target", lambda: True)
# Pre-create a non-empty file in pCloud
existing_file = mock_paths["pcloud_target"] / "TestOwner" / "RepoX_abcdef12.zip"
# Pre-create a non-empty file in target directory
existing_file = mock_paths["target_dir"] / "TestOwner" / "RepoX_abcdef12.zip"
create_valid_zip(existing_file, "existing")
# Spy on download_repo_zip to ensure it's not called
@@ -575,17 +575,17 @@ class TestSkipWhenFileAlreadyInPcloud:
class TestPlaceIntoPloudSameFileIsNoop:
"""Test that place_into_pcloud with same src/dst is a no-op (Fix 3 - belt and suspenders)."""
"""Test that place_into_target with same src/dst is a no-op (Fix 3 - belt and suspenders)."""
def test_place_into_pcloud_same_file_is_noop(self, mock_paths, setup_logging):
"""Call place_into_pcloud(p, p) for existing path, assert returns True and file unchanged."""
def test_place_into_target_same_file_is_noop(self, mock_paths, setup_logging):
"""Call place_into_target(p, p) for existing path, assert returns True and file unchanged."""
# Create a test file
test_file = mock_paths["pcloud_target"] / "test-file.zip"
test_file = mock_paths["target_dir"] / "test-file.zip"
create_valid_zip(test_file, "content")
original_mtime = test_file.stat().st_mtime
# Call with same src and dst
result = coe_mirror.place_into_pcloud(test_file, test_file)
result = coe_mirror.place_into_target(test_file, test_file)
# Assert returns True (success)
assert result is True
@@ -836,8 +836,8 @@ class TestMultiOwner:
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)
(mock_paths["target_dir"] / "OwnerA").mkdir(exist_ok=True)
(mock_paths["target_dir"] / "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)
@@ -866,16 +866,16 @@ class TestMultiOwner:
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"
a_file = mock_paths["target_dir"] / "OwnerA" / f"shared_{sha_a[:8]}.zip"
b_file = mock_paths["target_dir"] / "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)
(mock_paths["target_dir"] / "OwnerA").mkdir(exist_ok=True)
(mock_paths["target_dir"] / "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)
@@ -910,7 +910,7 @@ class TestMultiOwner:
rc = coe_mirror.main()
assert rc == 0
b_file = mock_paths["pcloud_target"] / "OwnerB" / f"onlyone_{sha_b[:8]}.zip"
b_file = mock_paths["target_dir"] / "OwnerB" / f"onlyone_{sha_b[:8]}.zip"
assert b_file.exists()
assert captured["failed_owners"] == ["OwnerA"]
# summary contains only the OwnerB entry
@@ -925,11 +925,11 @@ class TestConfigLoading:
cfg = tmp_path / "config.toml"
cfg.write_text(
'api_base = "https://forgejo.example.org"\n'
'pcloud_target = "/tmp/coe-mirror-cfg-test"\n'
'target_dir = "/tmp/coe-mirror-cfg-test"\n'
'owners = ["Alice", "Bob"]\n'
)
# Clear any env overrides
for k in ("COE_MIRROR_API_BASE", "COE_MIRROR_OWNERS", "COE_MIRROR_PCLOUD_TARGET"):
for k in ("COE_MIRROR_API_BASE", "COE_MIRROR_OWNERS", "COE_MIRROR_TARGET_DIR"):
monkeypatch.delenv(k, raising=False)
monkeypatch.setenv("COE_MIRROR_CONFIG", str(cfg))
# Re-import to re-run module-level config resolution
@@ -938,7 +938,7 @@ class TestConfigLoading:
try:
assert cm.API_BASE == "https://forgejo.example.org"
assert cm.OWNERS == ["Alice", "Bob"]
assert cm.PCLOUD_TARGET == Path("/tmp/coe-mirror-cfg-test")
assert cm.TARGET_DIR == Path("/tmp/coe-mirror-cfg-test")
finally:
monkeypatch.delenv("COE_MIRROR_CONFIG", raising=False)
importlib.reload(cm)
@@ -946,7 +946,7 @@ class TestConfigLoading:
def test_missing_config_falls_back_to_defaults(self, tmp_path, monkeypatch):
"""No config file + no env overrides → hardcoded defaults."""
missing = tmp_path / "does-not-exist.toml"
for k in ("COE_MIRROR_API_BASE", "COE_MIRROR_OWNERS", "COE_MIRROR_PCLOUD_TARGET",
for k in ("COE_MIRROR_API_BASE", "COE_MIRROR_OWNERS", "COE_MIRROR_TARGET_DIR",
"COE_MIRROR_MATRIX_HOMESERVER", "COE_MIRROR_MATRIX_ROOM"):
monkeypatch.delenv(k, raising=False)
monkeypatch.setenv("COE_MIRROR_CONFIG", str(missing))