n0mad1k
376a6d3a25
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
1012 lines
40 KiB
Python
1012 lines
40 KiB
Python
"""Tests for coe_mirror.py
|
|
|
|
Lock test (test_lock_contention_exits_zero) may have ordering dependencies
|
|
and should run in isolation if flaky. Other tests are independent.
|
|
"""
|
|
|
|
import io
|
|
import logging
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import threading
|
|
import time
|
|
import zipfile
|
|
from fcntl import flock, LOCK_EX, LOCK_NB
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
import pytest
|
|
import requests
|
|
import requests_mock
|
|
|
|
|
|
# Import the module under test
|
|
import coe_mirror
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_paths(tmp_path, monkeypatch):
|
|
"""Redirect all paths to tmp_path for test isolation."""
|
|
target_dir = tmp_path / "target"
|
|
pending_dir = tmp_path / "pending"
|
|
log_dir = tmp_path / "logs"
|
|
|
|
target_dir.mkdir()
|
|
(target_dir / "TestOwner").mkdir()
|
|
pending_dir.mkdir()
|
|
log_dir.mkdir()
|
|
|
|
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")
|
|
monkeypatch.setattr(coe_mirror, "OWNERS", ["TestOwner"])
|
|
monkeypatch.setattr(coe_mirror, "MATRIX_HOMESERVER", "https://matrix.test.local")
|
|
monkeypatch.setattr(coe_mirror, "MATRIX_ROOM", "!testroom:test.local")
|
|
|
|
return {
|
|
"target_dir": target_dir,
|
|
"pending_dir": pending_dir,
|
|
"log_dir": log_dir,
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def setup_logging(mock_paths, monkeypatch):
|
|
"""Set up logging for tests."""
|
|
coe_mirror.setup_logging()
|
|
yield
|
|
# Clean up logger
|
|
logger = logging.getLogger("coe-mirror")
|
|
for handler in logger.handlers[:]:
|
|
handler.close()
|
|
logger.removeHandler(handler)
|
|
|
|
|
|
def create_valid_zip(path: Path, content: str = "test") -> None:
|
|
"""Helper to create a valid zip file."""
|
|
with zipfile.ZipFile(path, "w") as zf:
|
|
zf.writestr("README.md", content)
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_lock(monkeypatch):
|
|
"""Mock locking to avoid actual file locks during tests."""
|
|
# For most tests, just return a valid fd
|
|
original_open = os.open
|
|
|
|
def mock_open(path, flags, mode=0o777):
|
|
# Return a dummy fd that won't conflict
|
|
return original_open(path, flags, mode)
|
|
|
|
monkeypatch.setattr(os, "open", mock_open)
|
|
|
|
|
|
class TestNewRepoDownloadsAndPlacesInPcloud:
|
|
"""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 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 target directory probe to return healthy
|
|
monkeypatch.setattr(coe_mirror, "probe_target", lambda: True)
|
|
|
|
with requests_mock.Mocker() as m:
|
|
# Mock repo list
|
|
m.get(
|
|
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/TestOwner/test-repo/branches/main",
|
|
json={"commit": {"id": "aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj"}},
|
|
)
|
|
|
|
# Create valid zip
|
|
zip_buffer = io.BytesIO()
|
|
with zipfile.ZipFile(zip_buffer, "w") as zf:
|
|
zf.writestr("README.md", "test")
|
|
zip_buffer.seek(0)
|
|
|
|
m.get(
|
|
f"{coe_mirror.API_BASE}/TestOwner/test-repo/archive/aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj.zip",
|
|
content=zip_buffer.getvalue(),
|
|
)
|
|
|
|
result = coe_mirror.main()
|
|
assert result == 0
|
|
|
|
# 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
|
|
|
|
|
|
class TestNewCommitCreatesSnapshot:
|
|
"""Test that new commit on existing repo creates new snapshot."""
|
|
|
|
def test_new_commit_on_existing_repo_creates_new_snapshot(self, mock_paths, setup_logging, monkeypatch):
|
|
"""Pre-create old snapshot, fetch new SHA, assert new file lands and old untouched."""
|
|
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 target directory probe to return healthy
|
|
monkeypatch.setattr(coe_mirror, "probe_target", lambda: True)
|
|
|
|
# Pre-create old file
|
|
old_file = mock_paths["target_dir"] / "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/TestOwner/repos",
|
|
json=[{"name": "test-repo", "default_branch": "main"}],
|
|
)
|
|
# Return new SHA
|
|
m.get(
|
|
f"{coe_mirror.API_BASE}/api/v1/repos/TestOwner/test-repo/branches/main",
|
|
json={"commit": {"id": "bbbbccccddddeeeeffffgggghhhhiiiijjjjkkkk"}},
|
|
)
|
|
|
|
# Create valid zip
|
|
zip_buffer = io.BytesIO()
|
|
with zipfile.ZipFile(zip_buffer, "w") as zf:
|
|
zf.writestr("README.md", "new")
|
|
zip_buffer.seek(0)
|
|
|
|
m.get(
|
|
f"{coe_mirror.API_BASE}/TestOwner/test-repo/archive/bbbbccccddddeeeeffffgggghhhhiiiijjjjkkkk.zip",
|
|
content=zip_buffer.getvalue(),
|
|
)
|
|
|
|
result = coe_mirror.main()
|
|
assert result == 0
|
|
|
|
# Verify old file still exists
|
|
assert old_file.exists()
|
|
# Verify new file landed
|
|
new_file = mock_paths["target_dir"] / "TestOwner" / "test-repo_bbbbcccc.zip"
|
|
assert new_file.exists()
|
|
|
|
|
|
class TestPcloudFailureKeepsPending:
|
|
"""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 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 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_target to fail
|
|
def mock_place(src, dst):
|
|
return False
|
|
|
|
monkeypatch.setattr(coe_mirror, "place_into_target", mock_place)
|
|
|
|
with requests_mock.Mocker() as m:
|
|
m.get(
|
|
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/TestOwner/test-repo/branches/main",
|
|
json={"commit": {"id": "aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj"}},
|
|
)
|
|
|
|
zip_buffer = io.BytesIO()
|
|
with zipfile.ZipFile(zip_buffer, "w") as zf:
|
|
zf.writestr("README.md", "test")
|
|
zip_buffer.seek(0)
|
|
|
|
m.get(
|
|
f"{coe_mirror.API_BASE}/TestOwner/test-repo/archive/aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj.zip",
|
|
content=zip_buffer.getvalue(),
|
|
)
|
|
|
|
result = coe_mirror.main()
|
|
assert result == 0
|
|
|
|
# Verify file is in pending
|
|
pending_file = mock_paths["pending_dir"] / "TestOwner__test-repo_aaaabbbb.zip"
|
|
assert pending_file.exists()
|
|
|
|
|
|
class TestPendingFlushesWhenPcloudRecoveres:
|
|
"""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 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 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"
|
|
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/TestOwner/repos",
|
|
json=[],
|
|
)
|
|
|
|
result = coe_mirror.main()
|
|
assert result == 0
|
|
|
|
# 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()
|
|
|
|
|
|
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 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)
|
|
|
|
with requests_mock.Mocker() as m:
|
|
m.get(
|
|
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/TestOwner/test-repo/branches/main",
|
|
json={"commit": {"id": "aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj"}},
|
|
)
|
|
# Return binary garbage
|
|
m.get(
|
|
f"{coe_mirror.API_BASE}/TestOwner/test-repo/archive/aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj.zip",
|
|
content=b"GARBAGE_DATA_NOT_A_ZIP",
|
|
)
|
|
|
|
result = coe_mirror.main()
|
|
assert result == 0
|
|
|
|
# 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()
|
|
|
|
|
|
class TestNotifyStub:
|
|
"""Test notify stub is no-op."""
|
|
|
|
def test_notify_stub_no_op(self, setup_logging, monkeypatch):
|
|
"""Assert notify function does not call network code."""
|
|
# Spy on notify
|
|
original_notify = coe_mirror.notify
|
|
call_log = []
|
|
|
|
def spy_notify(summary, failed_owners=None):
|
|
call_log.append(summary)
|
|
return original_notify(summary, failed_owners=failed_owners)
|
|
|
|
monkeypatch.setattr(coe_mirror, "notify", spy_notify)
|
|
|
|
# Test with no files
|
|
coe_mirror.notify([])
|
|
assert len(call_log) == 1
|
|
assert call_log[0] == []
|
|
|
|
# Test with files
|
|
call_log.clear()
|
|
coe_mirror.notify([("TestOwner", "repo", "aaaabbbb", "/path")])
|
|
assert len(call_log) == 1
|
|
|
|
|
|
class TestApiRetriesOn429:
|
|
"""Test API retries on 429 with backoff."""
|
|
|
|
def test_api_retries_on_429(self, mock_paths, setup_logging, monkeypatch):
|
|
"""Mock 429 twice then 200, assert success after retries."""
|
|
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
|
|
|
|
attempt_log = []
|
|
|
|
def track_attempts(status):
|
|
def callback(req, ctx):
|
|
attempt_log.append(status)
|
|
ctx.status_code = status
|
|
if status == 200:
|
|
return [{"name": "test-repo", "default_branch": "main"}]
|
|
return None
|
|
|
|
return callback
|
|
|
|
with requests_mock.Mocker() as m:
|
|
m.get(
|
|
f"{coe_mirror.API_BASE}/api/v1/users/TestOwner/repos",
|
|
[
|
|
{"status_code": 429, "json": None},
|
|
{"status_code": 429, "json": None},
|
|
{"json": [{"name": "test-repo", "default_branch": "main"}]},
|
|
],
|
|
)
|
|
m.get(
|
|
f"{coe_mirror.API_BASE}/api/v1/repos/TestOwner/test-repo/branches/main",
|
|
json={"commit": {"id": "aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj"}},
|
|
)
|
|
|
|
zip_buffer = io.BytesIO()
|
|
with zipfile.ZipFile(zip_buffer, "w") as zf:
|
|
zf.writestr("README.md", "test")
|
|
zip_buffer.seek(0)
|
|
|
|
m.get(
|
|
f"{coe_mirror.API_BASE}/TestOwner/test-repo/archive/aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj.zip",
|
|
content=zip_buffer.getvalue(),
|
|
)
|
|
|
|
result = coe_mirror.main()
|
|
assert result == 0
|
|
|
|
|
|
class TestSizeMismatchKeepsPending:
|
|
"""Test size mismatch during placement keeps file in pending."""
|
|
|
|
def test_size_mismatch_keeps_pending_unlinks_bad_dst(self, mock_paths, setup_logging, monkeypatch):
|
|
"""Mock copy to write short file, assert bad dst removed and src in pending."""
|
|
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 shutil.copy2 to write a short file
|
|
def mock_copy2(src, dst):
|
|
# Write dst but make it smaller than src
|
|
with open(dst, "wb") as f:
|
|
f.write(b"SHORT")
|
|
|
|
monkeypatch.setattr(shutil, "copy2", mock_copy2)
|
|
|
|
with requests_mock.Mocker() as m:
|
|
m.get(
|
|
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/TestOwner/test-repo/branches/main",
|
|
json={"commit": {"id": "aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj"}},
|
|
)
|
|
|
|
zip_buffer = io.BytesIO()
|
|
with zipfile.ZipFile(zip_buffer, "w") as zf:
|
|
zf.writestr("README.md", "test content here is longer")
|
|
zip_buffer.seek(0)
|
|
|
|
m.get(
|
|
f"{coe_mirror.API_BASE}/TestOwner/test-repo/archive/aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj.zip",
|
|
content=zip_buffer.getvalue(),
|
|
)
|
|
|
|
result = coe_mirror.main()
|
|
assert result == 0
|
|
|
|
# Verify bad dst was removed
|
|
bad_dst = mock_paths["target_dir"] / "TestOwner" / "test-repo_aaaabbbb.zip"
|
|
assert not bad_dst.exists()
|
|
|
|
# Verify src still in pending
|
|
pending_src = mock_paths["pending_dir"] / "TestOwner__test-repo_aaaabbbb.zip"
|
|
assert pending_src.exists()
|
|
|
|
|
|
class TestExistenceCheckSkipsZeroByte:
|
|
"""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."""
|
|
monkeypatch.setattr(coe_mirror, "acquire_lock", lambda: 999)
|
|
monkeypatch.setattr(coe_mirror, "release_lock", lambda fd: None)
|
|
monkeypatch.setattr(time, "sleep", lambda x: None)
|
|
|
|
# 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_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_target", mock_place)
|
|
|
|
with requests_mock.Mocker() as m:
|
|
m.get(
|
|
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/TestOwner/test-repo/branches/main",
|
|
json={"commit": {"id": "aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj"}},
|
|
)
|
|
|
|
zip_buffer = io.BytesIO()
|
|
with zipfile.ZipFile(zip_buffer, "w") as zf:
|
|
zf.writestr("README.md", "real content")
|
|
zip_buffer.seek(0)
|
|
|
|
m.get(
|
|
f"{coe_mirror.API_BASE}/TestOwner/test-repo/archive/aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj.zip",
|
|
content=zip_buffer.getvalue(),
|
|
)
|
|
|
|
result = coe_mirror.main()
|
|
assert result == 0
|
|
|
|
# Verify zero-byte file was overwritten with real content
|
|
assert zero_file.exists()
|
|
assert zero_file.stat().st_size > 0
|
|
|
|
|
|
class TestLockContention:
|
|
"""Test lock contention exits gracefully."""
|
|
|
|
def test_lock_contention_exits_zero(self, mock_paths, setup_logging, monkeypatch):
|
|
"""Acquire lock externally, assert main() returns 0 without work."""
|
|
# Create lock file and acquire it
|
|
xdg_runtime = os.getenv("XDG_RUNTIME_DIR")
|
|
lock_dir = Path(xdg_runtime) if xdg_runtime else Path("/tmp")
|
|
lock_file = lock_dir / coe_mirror.LOCK_NAME
|
|
|
|
with open(lock_file, "w") as f:
|
|
try:
|
|
flock(f, LOCK_EX | LOCK_NB)
|
|
# Lock acquired; now test that main() respects it
|
|
result = coe_mirror.main()
|
|
assert result == 0
|
|
finally:
|
|
try:
|
|
f.close()
|
|
except:
|
|
pass
|
|
# Clean up
|
|
try:
|
|
lock_file.unlink()
|
|
except:
|
|
pass
|
|
|
|
|
|
class TestUnsafeRepoNameRejected:
|
|
"""Test that unsafe repo names are rejected (Fix 1 - path traversal)."""
|
|
|
|
def test_unsafe_repo_name_rejected(self, mock_paths, setup_logging, monkeypatch):
|
|
"""Mock API returning repo with unsafe name ../etc/passwd, assert rejected with WARN/ERROR 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)
|
|
|
|
# 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
|
|
m.get(
|
|
f"{coe_mirror.API_BASE}/api/v1/users/TestOwner/repos",
|
|
json=[{"name": "../../../etc/passwd", "default_branch": "main"}],
|
|
)
|
|
|
|
result = coe_mirror.main()
|
|
assert result == 0
|
|
|
|
# Verify no file was written anywhere with traversal path
|
|
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["target_dir"].glob("*..*.zip")) + list(mock_paths["pending_dir"].glob("*..*.zip"))
|
|
assert len(unsafe_paths) == 0
|
|
|
|
|
|
class TestSkipWhenFileAlreadyInPcloud:
|
|
"""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 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 target directory probe to return healthy
|
|
monkeypatch.setattr(coe_mirror, "probe_target", lambda: True)
|
|
|
|
# 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
|
|
original_download = coe_mirror.download_repo_zip
|
|
download_calls = []
|
|
|
|
def spy_download(session, repo_name, full_sha):
|
|
download_calls.append((repo_name, full_sha))
|
|
return original_download(session, repo_name, full_sha)
|
|
|
|
monkeypatch.setattr(coe_mirror, "download_repo_zip", spy_download)
|
|
|
|
with requests_mock.Mocker() as m:
|
|
# Mock repo list
|
|
m.get(
|
|
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/TestOwner/RepoX/branches/main",
|
|
json={"commit": {"id": "abcdef1234567890"}},
|
|
)
|
|
|
|
result = coe_mirror.main()
|
|
assert result == 0
|
|
|
|
# Assert download_repo_zip was not called (file already exists, skipped at main level)
|
|
assert len(download_calls) == 0
|
|
|
|
# Verify existing file is still there and unchanged
|
|
assert existing_file.exists()
|
|
assert existing_file.stat().st_size > 0
|
|
|
|
# Verify new_files does not contain this repo (no notification generated)
|
|
# This is verified indirectly: notify should be called with empty list or list not containing RepoX
|
|
# Hard to verify without capturing notify, but the skipped download is the key test
|
|
|
|
|
|
class TestPlaceIntoPloudSameFileIsNoop:
|
|
"""Test that place_into_target with same src/dst is a no-op (Fix 3 - belt and suspenders)."""
|
|
|
|
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["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_target(test_file, test_file)
|
|
|
|
# Assert returns True (success)
|
|
assert result is True
|
|
|
|
# Assert file still exists and has same content
|
|
assert test_file.exists()
|
|
assert test_file.stat().st_size > 0
|
|
|
|
|
|
# Tests for Matrix notification implementation.
|
|
# Note: Tests requiring monkeypatched subprocess.run must use raising=False because
|
|
# subprocess.run is not a class attribute on the coe_mirror module by default
|
|
# — patch via monkeypatch.setattr("coe_mirror.subprocess.run", fake_run).
|
|
|
|
|
|
class TestMatrixSendSuccess:
|
|
"""Test successful Matrix message send with cached token."""
|
|
|
|
def test_matrix_send_success(self, mock_paths, setup_logging, monkeypatch):
|
|
"""Pre-write valid token, mock send endpoint 200, assert returns True."""
|
|
# Pre-write a valid (un-expired) token to cache
|
|
token_cache_file = mock_paths["log_dir"] / "matrix-token.json"
|
|
import json
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
future_time = datetime.now(timezone.utc) + timedelta(days=1)
|
|
cache_data = {
|
|
"access_token": "cached-token-12345",
|
|
"expires_at": future_time.isoformat()
|
|
}
|
|
token_cache_file.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(token_cache_file, "w") as f:
|
|
json.dump(cache_data, f)
|
|
os.chmod(token_cache_file, 0o600)
|
|
|
|
# Patch MATRIX_TOKEN_CACHE to point to our temp file
|
|
monkeypatch.setattr(coe_mirror, "MATRIX_TOKEN_CACHE", token_cache_file)
|
|
|
|
# Mock send endpoint to return 200
|
|
with requests_mock.Mocker() as m:
|
|
m.post(
|
|
f"{coe_mirror.MATRIX_HOMESERVER}/_matrix/client/r0/rooms/"
|
|
"%21testroom%3Atest.local/send/m.room.message",
|
|
json={"event_id": "$event1"},
|
|
status_code=200
|
|
)
|
|
|
|
# Call _send_matrix
|
|
result = coe_mirror._send_matrix("hello")
|
|
|
|
# Assert success
|
|
assert result is True
|
|
# Verify the request was made
|
|
assert m.called
|
|
|
|
|
|
class TestMatrixSendFailureNonfatal:
|
|
"""Test that Matrix send failure is non-fatal."""
|
|
|
|
def test_matrix_send_failure_nonfatal(self, mock_paths, setup_logging, monkeypatch):
|
|
"""Pre-write token, mock send 500, assert returns False, no exception."""
|
|
token_cache_file = mock_paths["log_dir"] / "matrix-token.json"
|
|
import json
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
future_time = datetime.now(timezone.utc) + timedelta(days=1)
|
|
cache_data = {
|
|
"access_token": "cached-token-12345",
|
|
"expires_at": future_time.isoformat()
|
|
}
|
|
token_cache_file.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(token_cache_file, "w") as f:
|
|
json.dump(cache_data, f)
|
|
|
|
monkeypatch.setattr(coe_mirror, "MATRIX_TOKEN_CACHE", token_cache_file)
|
|
|
|
with requests_mock.Mocker() as m:
|
|
m.post(
|
|
f"{coe_mirror.MATRIX_HOMESERVER}/_matrix/client/r0/rooms/"
|
|
"%21testroom%3Atest.local/send/m.room.message",
|
|
status_code=500,
|
|
text="Internal Server Error"
|
|
)
|
|
|
|
result = coe_mirror._send_matrix("hello")
|
|
|
|
assert result is False
|
|
|
|
|
|
class TestMatrixTokenRefreshOn401:
|
|
"""Test that 401 on send triggers token refresh and retry."""
|
|
|
|
def test_matrix_token_refresh_on_401(self, mock_paths, setup_logging, monkeypatch):
|
|
"""Pre-write token, mock send 401 then 200, mock login, assert refreshes and succeeds."""
|
|
token_cache_file = mock_paths["log_dir"] / "matrix-token.json"
|
|
import json
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
future_time = datetime.now(timezone.utc) + timedelta(days=1)
|
|
cache_data = {
|
|
"access_token": "old-token",
|
|
"expires_at": future_time.isoformat()
|
|
}
|
|
token_cache_file.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(token_cache_file, "w") as f:
|
|
json.dump(cache_data, f)
|
|
|
|
monkeypatch.setattr(coe_mirror, "MATRIX_TOKEN_CACHE", token_cache_file)
|
|
|
|
# Mock creds to return user/pass
|
|
fake_result_user = MagicMock()
|
|
fake_result_user.returncode = 0
|
|
fake_result_user.stdout = "test_user"
|
|
|
|
fake_result_pass = MagicMock()
|
|
fake_result_pass.returncode = 0
|
|
fake_result_pass.stdout = "test_pass"
|
|
|
|
def mock_subprocess_run(cmd, **kwargs):
|
|
if cmd[2] == coe_mirror.COE_MIRROR_BOT_USER_KEY:
|
|
return fake_result_user
|
|
elif cmd[2] == coe_mirror.COE_MIRROR_BOT_PASSWORD_KEY:
|
|
return fake_result_pass
|
|
return MagicMock(returncode=1, stdout="")
|
|
|
|
monkeypatch.setattr(coe_mirror.subprocess, "run", mock_subprocess_run)
|
|
|
|
with requests_mock.Mocker() as m:
|
|
# First send returns 401
|
|
m.post(
|
|
f"{coe_mirror.MATRIX_HOMESERVER}/_matrix/client/r0/rooms/"
|
|
"%21testroom%3Atest.local/send/m.room.message",
|
|
[
|
|
{"status_code": 401, "text": "Unauthorized"},
|
|
{"status_code": 200, "json": {"event_id": "$event1"}}
|
|
]
|
|
)
|
|
|
|
# Login endpoint
|
|
m.post(
|
|
f"{coe_mirror.MATRIX_HOMESERVER}/_matrix/client/r0/login",
|
|
json={"access_token": "new-token", "expires_in_ms": 604800000}
|
|
)
|
|
|
|
result = coe_mirror._send_matrix("hello")
|
|
|
|
# Assert succeeded after refresh
|
|
assert result is True
|
|
|
|
# Verify new token is in cache
|
|
with open(token_cache_file) as f:
|
|
cached = json.load(f)
|
|
assert cached["access_token"] == "new-token"
|
|
|
|
|
|
class TestMatrixRoomEnvOverride:
|
|
"""Test that COE_MIRROR_MATRIX_ROOM env var overrides the default room."""
|
|
|
|
def test_matrix_room_env_override(self, mock_paths, setup_logging, monkeypatch):
|
|
"""Set env override, assert send POSTs to override room URL."""
|
|
token_cache_file = mock_paths["log_dir"] / "matrix-token.json"
|
|
import json
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
future_time = datetime.now(timezone.utc) + timedelta(days=1)
|
|
cache_data = {
|
|
"access_token": "test-token",
|
|
"expires_at": future_time.isoformat()
|
|
}
|
|
token_cache_file.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(token_cache_file, "w") as f:
|
|
json.dump(cache_data, f)
|
|
|
|
monkeypatch.setattr(coe_mirror, "MATRIX_TOKEN_CACHE", token_cache_file)
|
|
|
|
# Override the room
|
|
override_room = "!override:example.org"
|
|
monkeypatch.setattr(coe_mirror, "MATRIX_ROOM", override_room)
|
|
|
|
with requests_mock.Mocker() as m:
|
|
m.post(
|
|
f"{coe_mirror.MATRIX_HOMESERVER}/_matrix/client/r0/rooms/"
|
|
"%21override%3Aexample.org/send/m.room.message",
|
|
json={"event_id": "$event1"},
|
|
status_code=200
|
|
)
|
|
|
|
result = coe_mirror._send_matrix("hello")
|
|
|
|
assert result is True
|
|
assert m.called
|
|
|
|
|
|
class TestMatrixLoginWritesCacheChmod600:
|
|
"""Test that _matrix_login writes cache file with 0600 permissions."""
|
|
|
|
def test_matrix_login_writes_cache_chmod_600(self, mock_paths, setup_logging, monkeypatch):
|
|
"""Start with no cache, mock creds and login, assert cache exists with 0o600."""
|
|
token_cache_file = mock_paths["log_dir"] / "matrix-token.json"
|
|
|
|
monkeypatch.setattr(coe_mirror, "MATRIX_TOKEN_CACHE", token_cache_file)
|
|
|
|
# Mock creds to return user/pass
|
|
fake_result_user = MagicMock()
|
|
fake_result_user.returncode = 0
|
|
fake_result_user.stdout = "testuser"
|
|
|
|
fake_result_pass = MagicMock()
|
|
fake_result_pass.returncode = 0
|
|
fake_result_pass.stdout = "testpass"
|
|
|
|
def mock_subprocess_run(cmd, **kwargs):
|
|
if cmd[2] == coe_mirror.COE_MIRROR_BOT_USER_KEY:
|
|
return fake_result_user
|
|
elif cmd[2] == coe_mirror.COE_MIRROR_BOT_PASSWORD_KEY:
|
|
return fake_result_pass
|
|
return MagicMock(returncode=1, stdout="")
|
|
|
|
monkeypatch.setattr(coe_mirror.subprocess, "run", mock_subprocess_run)
|
|
|
|
with requests_mock.Mocker() as m:
|
|
m.post(
|
|
f"{coe_mirror.MATRIX_HOMESERVER}/_matrix/client/r0/login",
|
|
json={"access_token": "new-token", "expires_in_ms": 604800000}
|
|
)
|
|
|
|
session = requests.Session()
|
|
token = coe_mirror._get_matrix_token(session, force_refresh=True)
|
|
|
|
assert token == "new-token"
|
|
assert token_cache_file.exists()
|
|
|
|
# Check permissions
|
|
import stat
|
|
mode = stat.S_IMODE(token_cache_file.stat().st_mode)
|
|
assert mode == 0o600
|
|
|
|
# Check content
|
|
import json
|
|
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["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)
|
|
|
|
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["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["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)
|
|
|
|
# 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["target_dir"] / "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"])
|
|
|
|
|
|
class TestConfigLoading:
|
|
"""Config precedence: env var > config.toml > hardcoded default."""
|
|
|
|
def test_config_file_populates_owners_and_pcloud(self, tmp_path, monkeypatch):
|
|
"""When config.toml exists and no env override, its values load."""
|
|
cfg = tmp_path / "config.toml"
|
|
cfg.write_text(
|
|
'api_base = "https://forgejo.example.org"\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_TARGET_DIR"):
|
|
monkeypatch.delenv(k, raising=False)
|
|
monkeypatch.setenv("COE_MIRROR_CONFIG", str(cfg))
|
|
# Re-import to re-run module-level config resolution
|
|
import importlib, coe_mirror as cm
|
|
importlib.reload(cm)
|
|
try:
|
|
assert cm.API_BASE == "https://forgejo.example.org"
|
|
assert cm.OWNERS == ["Alice", "Bob"]
|
|
assert cm.TARGET_DIR == Path("/tmp/coe-mirror-cfg-test")
|
|
finally:
|
|
monkeypatch.delenv("COE_MIRROR_CONFIG", raising=False)
|
|
importlib.reload(cm)
|
|
|
|
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_TARGET_DIR",
|
|
"COE_MIRROR_MATRIX_HOMESERVER", "COE_MIRROR_MATRIX_ROOM"):
|
|
monkeypatch.delenv(k, raising=False)
|
|
monkeypatch.setenv("COE_MIRROR_CONFIG", str(missing))
|
|
import importlib, coe_mirror as cm
|
|
importlib.reload(cm)
|
|
try:
|
|
assert cm.API_BASE == "https://git.churchofmalware.org"
|
|
assert "Nightmare_Eclipse" in cm.OWNERS
|
|
# Matrix defaults must be blank so notify no-ops on fresh checkouts
|
|
assert cm.MATRIX_HOMESERVER == ""
|
|
assert cm.MATRIX_ROOM == ""
|
|
finally:
|
|
monkeypatch.delenv("COE_MIRROR_CONFIG", raising=False)
|
|
importlib.reload(cm)
|
|
|
|
def test_env_var_beats_config_file(self, tmp_path, monkeypatch):
|
|
"""When both env var and config file are set, env wins."""
|
|
cfg = tmp_path / "config.toml"
|
|
cfg.write_text('owners = ["FromConfig"]\napi_base = "https://from-config.example"\n')
|
|
monkeypatch.setenv("COE_MIRROR_CONFIG", str(cfg))
|
|
monkeypatch.setenv("COE_MIRROR_OWNERS", "FromEnvA,FromEnvB")
|
|
monkeypatch.setenv("COE_MIRROR_API_BASE", "https://from-env.example")
|
|
import importlib, coe_mirror as cm
|
|
importlib.reload(cm)
|
|
try:
|
|
assert cm.OWNERS == ["FromEnvA", "FromEnvB"]
|
|
assert cm.API_BASE == "https://from-env.example"
|
|
finally:
|
|
for k in ("COE_MIRROR_CONFIG", "COE_MIRROR_OWNERS", "COE_MIRROR_API_BASE"):
|
|
monkeypatch.delenv(k, raising=False)
|
|
importlib.reload(cm)
|
|
|
|
|
|
class TestMatrixSilentWhenUnconfigured:
|
|
"""_send_matrix must no-op silently when homeserver/room are blank."""
|
|
|
|
def test_send_matrix_returns_false_when_blank(self, mock_paths, setup_logging, monkeypatch):
|
|
monkeypatch.setattr(coe_mirror, "MATRIX_HOMESERVER", "")
|
|
monkeypatch.setattr(coe_mirror, "MATRIX_ROOM", "")
|
|
# If this hits the network it will fail loudly; the point is it must NOT hit the network
|
|
assert coe_mirror._send_matrix("hello") is False
|
|
|
|
|
|
class TestCredsEnvFallback:
|
|
"""`_fetch_creds` falls through to env vars when `creds` binary is unavailable."""
|
|
|
|
def test_env_fallback_when_creds_missing(self, setup_logging, monkeypatch):
|
|
# Force `creds` subprocess to raise FileNotFoundError
|
|
def raise_fnf(*args, **kwargs):
|
|
raise FileNotFoundError("no creds binary")
|
|
monkeypatch.setattr(coe_mirror.subprocess, "run", raise_fnf)
|
|
monkeypatch.setenv("COE_MIRROR_MATRIX_USER", "@mirror:example.org")
|
|
monkeypatch.setenv("COE_MIRROR_MATRIX_PASSWORD", "s3cret")
|
|
assert coe_mirror._fetch_creds() == ("@mirror:example.org", "s3cret")
|
|
|
|
def test_returns_none_when_both_unavailable(self, setup_logging, monkeypatch):
|
|
def raise_fnf(*args, **kwargs):
|
|
raise FileNotFoundError("no creds binary")
|
|
monkeypatch.setattr(coe_mirror.subprocess, "run", raise_fnf)
|
|
monkeypatch.delenv("COE_MIRROR_MATRIX_USER", raising=False)
|
|
monkeypatch.delenv("COE_MIRROR_MATRIX_PASSWORD", raising=False)
|
|
assert coe_mirror._fetch_creds() is None
|