Cobra
4d0fbff039
Replaces shared @intel-bot credentials with a Matrix user created specifically for this tool. Reads Infisical keys via constants COE_MIRROR_BOT_USER_KEY and COE_MIRROR_BOT_PASSWORD_KEY so test assertions track the code rather than hardcoded strings.
827 lines
31 KiB
Python
827 lines
31 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."""
|
|
pcloud_target = tmp_path / "pcloud"
|
|
pending_dir = tmp_path / "pending"
|
|
log_dir = tmp_path / "logs"
|
|
|
|
pcloud_target.mkdir()
|
|
pending_dir.mkdir()
|
|
log_dir.mkdir()
|
|
|
|
monkeypatch.setattr(coe_mirror, "PCLOUD_TARGET", pcloud_target)
|
|
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")
|
|
|
|
return {
|
|
"pcloud_target": pcloud_target,
|
|
"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 pCloud."""
|
|
|
|
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."""
|
|
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)
|
|
|
|
with requests_mock.Mocker() as m:
|
|
# Mock repo list
|
|
m.get(
|
|
f"{coe_mirror.API_BASE}/api/v1/users/{coe_mirror.OWNER}/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",
|
|
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}/{coe_mirror.OWNER}/test-repo/archive/aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj.zip",
|
|
content=zip_buffer.getvalue(),
|
|
)
|
|
|
|
result = coe_mirror.main()
|
|
assert result == 0
|
|
|
|
# Verify file landed in pCloud with expected name
|
|
expected_file = mock_paths["pcloud_target"] / "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 pCloud probe to return healthy
|
|
monkeypatch.setattr(coe_mirror, "probe_pcloud", lambda: True)
|
|
|
|
# Pre-create old file
|
|
old_file = mock_paths["pcloud_target"] / "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",
|
|
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",
|
|
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}/{coe_mirror.OWNER}/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["pcloud_target"] / "test-repo_bbbbcccc.zip"
|
|
assert new_file.exists()
|
|
|
|
|
|
class TestPcloudFailureKeepsPending:
|
|
"""Test pCloud 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."""
|
|
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"]
|
|
pcloud_as_file.rmdir()
|
|
pcloud_as_file.write_text("block")
|
|
|
|
# Mock place_into_pcloud to fail
|
|
def mock_place(src, dst):
|
|
return False
|
|
|
|
monkeypatch.setattr(coe_mirror, "place_into_pcloud", mock_place)
|
|
|
|
with requests_mock.Mocker() as m:
|
|
m.get(
|
|
f"{coe_mirror.API_BASE}/api/v1/users/{coe_mirror.OWNER}/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",
|
|
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}/{coe_mirror.OWNER}/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"] / "test-repo_aaaabbbb.zip"
|
|
assert pending_file.exists()
|
|
|
|
|
|
class TestPendingFlushesWhenPcloudRecoveres:
|
|
"""Test pending files flush when pCloud 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."""
|
|
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)
|
|
|
|
# Create a pending file
|
|
pending_file = mock_paths["pending_dir"] / "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",
|
|
json=[],
|
|
)
|
|
|
|
result = coe_mirror.main()
|
|
assert result == 0
|
|
|
|
# Verify file moved to pCloud
|
|
pcloud_file = mock_paths["pcloud_target"] / "test-repo_aaaabbbb.zip"
|
|
assert pcloud_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 pCloud."""
|
|
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/{coe_mirror.OWNER}/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",
|
|
json={"commit": {"id": "aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj"}},
|
|
)
|
|
# Return binary garbage
|
|
m.get(
|
|
f"{coe_mirror.API_BASE}/{coe_mirror.OWNER}/test-repo/archive/aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj.zip",
|
|
content=b"GARBAGE_DATA_NOT_A_ZIP",
|
|
)
|
|
|
|
result = coe_mirror.main()
|
|
assert result == 0
|
|
|
|
# Verify nothing in pCloud or pending
|
|
pcloud_file = mock_paths["pcloud_target"] / "test-repo_aaaabbbb.zip"
|
|
assert not pcloud_file.exists()
|
|
pending_file = mock_paths["pending_dir"] / "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):
|
|
call_log.append(summary)
|
|
return original_notify(summary)
|
|
|
|
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([("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/{coe_mirror.OWNER}/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/{coe_mirror.OWNER}/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}/{coe_mirror.OWNER}/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/{coe_mirror.OWNER}/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",
|
|
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}/{coe_mirror.OWNER}/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["pcloud_target"] / "test-repo_aaaabbbb.zip"
|
|
assert not bad_dst.exists()
|
|
|
|
# Verify src still in pending
|
|
pending_src = mock_paths["pending_dir"] / "test-repo_aaaabbbb.zip"
|
|
assert pending_src.exists()
|
|
|
|
|
|
class TestExistenceCheckSkipsZeroByte:
|
|
"""Test that zero-byte file in pCloud 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 pCloud target
|
|
zero_file = mock_paths["pcloud_target"] / "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)
|
|
def mock_place(src, dst):
|
|
shutil.copy2(src, dst)
|
|
src.unlink()
|
|
return True
|
|
|
|
monkeypatch.setattr(coe_mirror, "place_into_pcloud", mock_place)
|
|
|
|
with requests_mock.Mocker() as m:
|
|
m.get(
|
|
f"{coe_mirror.API_BASE}/api/v1/users/{coe_mirror.OWNER}/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",
|
|
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}/{coe_mirror.OWNER}/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 pCloud probe to return healthy
|
|
monkeypatch.setattr(coe_mirror, "probe_pcloud", 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/{coe_mirror.OWNER}/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["pcloud_target"] / "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"))
|
|
assert len(unsafe_paths) == 0
|
|
|
|
|
|
class TestSkipWhenFileAlreadyInPcloud:
|
|
"""Test that file already in pCloud 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."""
|
|
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)
|
|
|
|
# Pre-create a non-empty file in pCloud
|
|
existing_file = mock_paths["pcloud_target"] / "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/{coe_mirror.OWNER}/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",
|
|
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_pcloud 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."""
|
|
# Create a test file
|
|
test_file = mock_paths["pcloud_target"] / "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)
|
|
|
|
# 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(
|
|
"https://m.example.org/_matrix/client/r0/rooms/"
|
|
"%21REDACTEDROOM%3Am.example.org/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(
|
|
"https://m.example.org/_matrix/client/r0/rooms/"
|
|
"%21REDACTEDROOM%3Am.example.org/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(
|
|
"https://m.example.org/_matrix/client/r0/rooms/"
|
|
"%21REDACTEDROOM%3Am.example.org/send/m.room.message",
|
|
[
|
|
{"status_code": 401, "text": "Unauthorized"},
|
|
{"status_code": 200, "json": {"event_id": "$event1"}}
|
|
]
|
|
)
|
|
|
|
# Login endpoint
|
|
m.post(
|
|
"https://m.example.org/_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(
|
|
"https://m.example.org/_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(
|
|
"https://m.example.org/_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"
|