From 376a6d3a25bf50e12b1bcd502a87e45a53cd1605 Mon Sep 17 00:00:00 2001 From: n0mad1k Date: Tue, 7 Jul 2026 11:12:46 -0400 Subject: [PATCH] Genericize: drop pCloud terminology, use TARGET_DIR / target_dir MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- README.md | 6 +- coe_mirror.py | 84 ++++++++++++------------- tests/test_coe_mirror.py | 132 +++++++++++++++++++-------------------- 3 files changed, 111 insertions(+), 111 deletions(-) diff --git a/README.md b/README.md index 076095b..a9e218c 100644 --- a/README.md +++ b/README.md @@ -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 | diff --git a/coe_mirror.py b/coe_mirror.py index 8c45116..83540a2 100644 --- a/coe_mirror.py +++ b/coe_mirror.py @@ -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//`. Runs unattended via systemd timer. +under `TARGET_DIR//`. 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// if healthy. +def flush_pending(target_healthy: bool) -> list: + """Move all *.zip files from PENDING_DIR to TARGET_DIR// 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. diff --git a/tests/test_coe_mirror.py b/tests/test_coe_mirror.py index 6cd416b..9f397f6 100644 --- a/tests/test_coe_mirror.py +++ b/tests/test_coe_mirror.py @@ -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))