Make distributable: config.toml support, blanked Matrix defaults, env creds fallback

- ~/.config/coe-mirror/config.toml (tomllib, Python 3.11+ required)
- Precedence uniform for every key: env var > config file > hardcoded default
- API_BASE, OWNERS, PCLOUD_TARGET, PENDING_DIR, matrix homeserver/room resolved
  via _resolve() helper
- Matrix homeserver/room now default to empty; _send_matrix short-circuits when
  either is blank so forks don't spam the original author's room
- _fetch_creds falls back to COE_MIRROR_MATRIX_USER/PASSWORD env vars when the
  Infisical creds binary is unavailable (required for external distribution)
- CLI subcommands: config init (writes example TOML, won't overwrite), config path
  Bare invocation and 'run' still work unchanged for the systemd timer
- README rewritten as a public quick-start with the precedence table and Matrix
  as opt-in
- Tests: 3 new for config precedence, 1 for Matrix silent no-op when blank,
  2 for creds env fallback (26 total, all pass)

DevTrack #1285
This commit is contained in:
n0mad1k
2026-07-07 11:04:07 -04:00
parent 5912ca92d2
commit 2093b737c3
4 changed files with 357 additions and 75 deletions
+2
View File
@@ -5,3 +5,5 @@ __pycache__/
.pytest_cache/ .pytest_cache/
*.egg-info/ *.egg-info/
.active-item.json .active-item.json
.workflow-state.json
agent-spawns/
+83 -38
View File
@@ -1,66 +1,111 @@
# coe-mirror # coe-mirror
Mirror Nightmare_Eclipse repos from Church of Malware to pCloud. 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.
## What it does ## What it does
Monitors `Nightmare_Eclipse` on `git.churchofmalware.org` for new or updated repos. Downloads per-commit zip snapshots to `~/pCloudDrive/Hacking/Nightmare_Eclipse/`. Falls back to local staging when pCloud is down. Runs at 00:00 and 12:00 daily with boot catch-up. For every owner in your config, it fetches the repo list, downloads a zip of the current HEAD of the default branch, and writes it to `<target>/<owner>/<repo>_<sha8>.zip`. Old snapshots stay as history — new commits produce new files. Runs unattended via a systemd user timer.
## Filename convention ## Requirements
`{repo}_{short_sha}.zip` where short_sha is the first 8 chars of the HEAD commit SHA on the default branch. New commits produce new files; old snapshots stay as history. - Python 3.11+ (uses stdlib `tomllib`)
- `pip install -r requirements.txt`
## Install ## Quick start
```
git clone <this repo>
cd coe-mirror
python3 -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
python coe_mirror.py config init # writes ~/.config/coe-mirror/config.toml
$EDITOR ~/.config/coe-mirror/config.toml # edit owners + target dir
python coe_mirror.py run # first mirror
```
Add or remove owners any time by editing the `owners = [...]` list in the config file.
## Configuration
**Precedence** (uniform for every key): env var > config file > hardcoded default.
**Config file** location: `~/.config/coe-mirror/config.toml` (override with `COE_MIRROR_CONFIG=/path/to/file.toml`).
Full schema:
```toml
api_base = "https://git.churchofmalware.org"
pcloud_target = "~/pCloudDrive/Hacking/ChurchOfMalware"
owners = ["Nightmare_Eclipse", "shai_hulud"]
# pending_dir = "~/.local/share/coe-mirror/pending" # optional
[matrix] # optional; leave blank to disable notifications
homeserver = ""
room = ""
```
**Env-var overrides** (all optional; each beats the config file):
| Var | Purpose |
|---|---|
| `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_PENDING_DIR` | Local staging dir |
| `COE_MIRROR_MATRIX_HOMESERVER` | Matrix homeserver URL |
| `COE_MIRROR_MATRIX_ROOM` | Matrix room ID |
| `COE_MIRROR_MATRIX_USER` | Matrix bot user (fallback if `creds` binary unavailable) |
| `COE_MIRROR_MATRIX_PASSWORD` | Matrix bot password (fallback) |
## CLI
```
python coe_mirror.py # run the mirror (default)
python coe_mirror.py run # same
python coe_mirror.py config init # write example config (won't overwrite)
python coe_mirror.py config path # print config file path
python coe_mirror.py --help
```
## Install as a systemd user timer
``` ```
cd ~/tools/coe-mirror
bash install.sh bash install.sh
``` ```
## Manual run Timer defaults to 00:00 and 12:00 daily with boot catch-up (`Persistent=true`). Missed windows run on next boot.
``` ## Matrix notifications (optional)
systemctl --user start coe-mirror.service
# or directly: - **Disabled by default.** Leave `[matrix]` blank to skip.
python3 ~/tools/coe-mirror/coe_mirror.py - If both `homeserver` and `room` are configured, the tool posts a summary after every run that lands at least one new zip (or reports any owner whose fetch failed).
``` - Credential lookup order: `creds get COE_MIRROR_BOT_USER matrix` / `COE_MIRROR_BOT_PASSWORD matrix` (Infisical CLI, if you use it), then `COE_MIRROR_MATRIX_USER` / `COE_MIRROR_MATRIX_PASSWORD` env vars. Tokens cached at `~/.local/share/coe-mirror/matrix-token.json` (mode 0600), auto-refreshed on expiry or 401.
- Matrix failures are non-fatal — a warning is logged and the run proceeds.
## Filename convention
- Final: `<target>/<owner>/<repo>_<sha8>.zip`
- Staging (while pending flush): `<pending_dir>/<owner>__<repo>_<sha8>.zip` — owner prefix prevents cross-owner collisions when two owners have identically-named repos.
## Behavior when the target is unavailable
Downloads land in `pending_dir`. Next run (or any subsequent invocation) flushes them to the target once it's healthy. Owner-scoped: one owner's fetch failure logs and continues; the next run retries.
## Logs ## Logs
- File: `~/.local/share/coe-mirror/coe-mirror.log` (rotated at 5 MB, 3 backups) - File: `~/.local/share/coe-mirror/coe-mirror.log` (rotated at 5 MB, 3 backups)
- Journal: `journalctl --user -u coe-mirror.service -n 100` - Journal: `journalctl --user -u coe-mirror.service -n 100`
## Storage estimate ## Storage
8 repos × 2 runs/day × ~1 commit/repo/day ≈ 16 zips/day ≈ ~5 MB/month at current zip sizes (~265 KB each). Old snapshots accumulate — no automatic pruning. Manually prune `~/pCloudDrive/Hacking/Nightmare_Eclipse/` if needed. Old snapshots accumulate — no automatic pruning. Prune the target directory manually if you need to reclaim space.
## Behavior when pCloud is down
Downloads land in `~/.local/share/coe-mirror/pending/`. Next run (or any subsequent invocation) flushes them to pCloud once the mount is healthy. No auto-start of the pCloud daemon — by design, because the daemon is a GUI AppImage with no systemd user unit. If pCloud is offline, ensure the user session is active and pCloud is running before the next scheduled run, or trigger manually.
## Catch-up after machine was off
The timer is `Persistent=true` and `OnBootSec=2min`. Missed windows run automatically on next boot. The tool also diffs against the pCloud directory, so even if a run is skipped entirely, the next run catches up everything that's missing.
## Matrix notification
After every run that lands at least one new zip, coe-mirror posts a summary
to the General room (`!MBpPtYlQYMcngtpvJt:m.mealeyfamily.com`) on
`m.mealeyfamily.com`.
- Authentication uses `MATRIX_USER` + `MATRIX_PASSWORD` from Infisical folder
`matrix`. Tokens are cached at `~/.local/share/coe-mirror/matrix-token.json`
(mode 0600) and auto-refreshed on expiry or 401.
- Failures are non-fatal: a Matrix outage logs a warning and the rest of the
run proceeds normally.
- Override the target room for testing: `COE_MIRROR_MATRIX_ROOM=! ...:host`.
- Override the homeserver: `COE_MIRROR_MATRIX_HOMESERVER=https://...`.
## Testing ## Testing
``` ```
cd ~/tools/coe-mirror python -m pytest tests/ -v
python3 -m pytest tests/ -v
``` ```
## Uninstall ## Uninstall
+167 -28
View File
@@ -1,8 +1,12 @@
"""Mirror repos from git.churchofmalware.org Forgejo instance to pCloud. """Mirror repos from a Forgejo instance to a local target dir (e.g. pCloud).
Fetches all repos owned by each user in OWNERS and stores per-commit zip snapshots Fetches all repos owned by each user in OWNERS and stores per-commit zip snapshots
under `PCLOUD_TARGET/<owner>/`. Runs unattended via systemd timer twice daily. under `PCLOUD_TARGET/<owner>/`. Runs unattended via systemd timer.
Local staging when pCloud unavailable. Local staging when the target is unavailable.
Configuration precedence (for each key): env var > config file > hardcoded default.
Config file: `~/.config/coe-mirror/config.toml` (see `python coe_mirror.py config init`).
Requires Python 3.11+ (for stdlib `tomllib`).
""" """
import json import json
@@ -14,6 +18,7 @@ import shutil
import subprocess import subprocess
import sys import sys
import time import time
import tomllib
import zipfile import zipfile
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
from pathlib import Path from pathlib import Path
@@ -23,11 +28,59 @@ from urllib.parse import quote
import requests import requests
# Module constants (overridable via env vars for testability) # --- Config file (optional) ------------------------------------------------
API_BASE = "https://git.churchofmalware.org" CONFIG_PATH = Path(
OWNERS = ["Nightmare_Eclipse", "shai_hulud", "bikini_mirror", "ek0mssavi0r"] os.getenv("COE_MIRROR_CONFIG")
PCLOUD_TARGET = Path.home() / "pCloudDrive" / "Hacking" / "ChurchOfMalware" or (Path(os.getenv("XDG_CONFIG_HOME") or Path.home() / ".config") / "coe-mirror" / "config.toml")
PENDING_DIR = Path.home() / ".local" / "share" / "coe-mirror" / "pending" )
def _load_config_file() -> dict:
"""Load config.toml if present. Never raises; malformed file → empty dict + stderr warning."""
try:
if CONFIG_PATH.is_file():
with open(CONFIG_PATH, "rb") as f:
return tomllib.load(f)
except Exception as exc:
print(f"coe-mirror: warning: could not parse {CONFIG_PATH}: {exc}", file=sys.stderr)
return {}
_CFG = _load_config_file()
_CFG_MATRIX = _CFG.get("matrix", {}) if isinstance(_CFG.get("matrix"), dict) else {}
def _resolve(env_key: str, cfg_value, default):
"""env var > config value > hardcoded default. Empty env/config values are ignored."""
env_val = os.getenv(env_key)
if env_val:
return env_val
if cfg_value not in (None, ""):
return cfg_value
return default
# --- Module constants (resolved via env > config > default) ---------------
API_BASE: str = _resolve("COE_MIRROR_API_BASE", _CFG.get("api_base"), "https://git.churchofmalware.org")
OWNERS: List[str] = (
[o.strip() for o in os.getenv("COE_MIRROR_OWNERS", "").split(",") if o.strip()]
or (_CFG.get("owners") if isinstance(_CFG.get("owners"), list) else None)
or ["Nightmare_Eclipse", "shai_hulud", "bikini_mirror", "ek0mssavi0r"]
)
PCLOUD_TARGET: Path = Path(
_resolve(
"COE_MIRROR_PCLOUD_TARGET",
_CFG.get("pcloud_target"),
str(Path.home() / "pCloudDrive" / "Hacking" / "ChurchOfMalware"),
)
).expanduser()
PENDING_DIR: Path = Path(
_resolve(
"COE_MIRROR_PENDING_DIR",
_CFG.get("pending_dir"),
str(Path.home() / ".local" / "share" / "coe-mirror" / "pending"),
)
).expanduser()
LOG_DIR = Path.home() / ".local" / "share" / "coe-mirror" LOG_DIR = Path.home() / ".local" / "share" / "coe-mirror"
LOG_FILE = LOG_DIR / "coe-mirror.log" LOG_FILE = LOG_DIR / "coe-mirror.log"
LOCK_NAME = "coe-mirror.lock" LOCK_NAME = "coe-mirror.lock"
@@ -38,25 +91,15 @@ RETRY_BACKOFF = [0, 5, 30] # seconds between attempts; len = max attempts
USER_AGENT = "coe-mirror/1.0" USER_AGENT = "coe-mirror/1.0"
REPO_NAME_RE = re.compile(r"^[a-zA-Z0-9._-]+$") REPO_NAME_RE = re.compile(r"^[a-zA-Z0-9._-]+$")
# Matrix configuration # Matrix configuration (fully opt-in; blank defaults => notifications silently skipped)
MATRIX_HOMESERVER = "https://m.mealeyfamily.com" MATRIX_HOMESERVER: str = _resolve("COE_MIRROR_MATRIX_HOMESERVER", _CFG_MATRIX.get("homeserver"), "")
MATRIX_ROOM = "!MBpPtYlQYMcngtpvJt:m.mealeyfamily.com" # General MATRIX_ROOM: str = _resolve("COE_MIRROR_MATRIX_ROOM", _CFG_MATRIX.get("room"), "")
MATRIX_TOKEN_CACHE = LOG_DIR / "matrix-token.json" MATRIX_TOKEN_CACHE = LOG_DIR / "matrix-token.json"
MATRIX_TIMEOUT = 8 MATRIX_TIMEOUT = 8
MATRIX_REFRESH_THRESHOLD_SECONDS = 1800 # refresh if <30 min left MATRIX_REFRESH_THRESHOLD_SECONDS = 1800 # refresh if <30 min left
COE_MIRROR_BOT_USER_KEY = "COE_MIRROR_BOT_USER" COE_MIRROR_BOT_USER_KEY = "COE_MIRROR_BOT_USER"
COE_MIRROR_BOT_PASSWORD_KEY = "COE_MIRROR_BOT_PASSWORD" COE_MIRROR_BOT_PASSWORD_KEY = "COE_MIRROR_BOT_PASSWORD"
# Allow env var overrides for testing and deployment flexibility
if env_pcloud := os.getenv("COE_MIRROR_PCLOUD_TARGET"):
PCLOUD_TARGET = Path(env_pcloud)
if env_pending := os.getenv("COE_MIRROR_PENDING_DIR"):
PENDING_DIR = Path(env_pending)
if env_hs := os.getenv("COE_MIRROR_MATRIX_HOMESERVER"):
MATRIX_HOMESERVER = env_hs
if env_room := os.getenv("COE_MIRROR_MATRIX_ROOM"):
MATRIX_ROOM = env_room
logger: logging.Logger = None # Initialized in setup_logging() logger: logging.Logger = None # Initialized in setup_logging()
@@ -362,7 +405,12 @@ def _write_token_cache(access_token: str, expires_at_iso: str) -> None:
def _fetch_creds() -> Optional[Tuple[str, str]]: def _fetch_creds() -> Optional[Tuple[str, str]]:
"""Fetch (user, password) from Infisical via creds. Return None on failure.""" """Fetch (user, password) for Matrix.
Order: `creds` binary (Infisical) → `COE_MIRROR_MATRIX_USER` / `COE_MIRROR_MATRIX_PASSWORD`
env vars → None. Distributed users typically set the env vars in the systemd
EnvironmentFile; the Infisical path is convenience for hosts that already have it.
"""
try: try:
user = subprocess.run( user = subprocess.run(
["creds", "get", COE_MIRROR_BOT_USER_KEY, "matrix"], ["creds", "get", COE_MIRROR_BOT_USER_KEY, "matrix"],
@@ -376,14 +424,20 @@ def _fetch_creds() -> Optional[Tuple[str, str]]:
u, p = user.stdout.strip(), pw.stdout.strip() u, p = user.stdout.strip(), pw.stdout.strip()
if u and p: if u and p:
return u, p return u, p
logger.warning("creds returned non-zero for %s/%s", logger.debug("creds returned non-zero for %s/%s; trying env fallback",
COE_MIRROR_BOT_USER_KEY, COE_MIRROR_BOT_PASSWORD_KEY) COE_MIRROR_BOT_USER_KEY, COE_MIRROR_BOT_PASSWORD_KEY)
except FileNotFoundError: except FileNotFoundError:
logger.warning("creds binary not found on PATH") logger.debug("creds binary not on PATH; trying env fallback")
except subprocess.TimeoutExpired: except subprocess.TimeoutExpired:
logger.warning("creds timed out fetching Matrix credentials") logger.warning("creds timed out; trying env fallback")
except Exception as exc: except Exception as exc:
logger.warning("creds fetch failed: %s", exc) logger.warning("creds fetch failed: %s; trying env fallback", exc)
env_u = os.getenv("COE_MIRROR_MATRIX_USER", "").strip()
env_p = os.getenv("COE_MIRROR_MATRIX_PASSWORD", "").strip()
if env_u and env_p:
return env_u, env_p
logger.warning("no Matrix credentials available (creds + env both empty); notifications disabled")
return None return None
@@ -436,7 +490,11 @@ def _send_matrix(message: str) -> bool:
Non-fatal: any failure logs WARN and returns False. Never raises. Non-fatal: any failure logs WARN and returns False. Never raises.
Implements one 401 retry with token refresh. Implements one 401 retry with token refresh.
Silently skips (returns False) if MATRIX_HOMESERVER or MATRIX_ROOM is unconfigured.
""" """
if not MATRIX_HOMESERVER or not MATRIX_ROOM:
logger.debug("matrix notifications disabled (homeserver or room unset)")
return False
with requests.Session() as session: with requests.Session() as session:
session.headers.update({"User-Agent": USER_AGENT}) session.headers.update({"User-Agent": USER_AGENT})
@@ -605,5 +663,86 @@ def main() -> int:
release_lock(lock_fd) release_lock(lock_fd)
EXAMPLE_CONFIG = """\
# coe-mirror configuration
# Precedence for every key: env var > this file > hardcoded default.
# Env var names are COE_MIRROR_<UPPER_KEY>; matrix keys are prefixed COE_MIRROR_MATRIX_.
api_base = "https://git.churchofmalware.org"
# Where mirrored zips are stored. One subdirectory per owner is created here.
pcloud_target = "~/pCloudDrive/Hacking/ChurchOfMalware"
# Optional: local staging when pcloud_target is unavailable.
# pending_dir = "~/.local/share/coe-mirror/pending"
# Forgejo owners (users or orgs) to mirror. Edit this list to add/remove.
owners = [
"Nightmare_Eclipse",
"shai_hulud",
"bikini_mirror",
"ek0mssavi0r",
]
# Optional Matrix notifications. Leave both blank to disable.
# User/password come from COE_MIRROR_MATRIX_USER / COE_MIRROR_MATRIX_PASSWORD env vars
# (or the `creds` binary if you use Infisical).
[matrix]
homeserver = ""
room = ""
"""
def cmd_config_init() -> int:
"""Write an example config to CONFIG_PATH. Won't overwrite existing files."""
if CONFIG_PATH.exists():
print(f"coe-mirror: config already exists at {CONFIG_PATH}; not overwriting", file=sys.stderr)
return 1
CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True)
CONFIG_PATH.write_text(EXAMPLE_CONFIG)
print(f"coe-mirror: wrote example config to {CONFIG_PATH}")
return 0
def cmd_config_path() -> int:
"""Print the config file path (resolved), regardless of whether it exists."""
print(str(CONFIG_PATH))
return 0
USAGE = """\
Usage: coe_mirror.py [command]
Commands:
(no args) Run the mirror (default; used by systemd timer)
run Same as no args
config init Write an example config.toml to the standard path
config path Print the config file path
-h, --help Show this message
Config file: {path}
Precedence: env var > config file > hardcoded default
"""
def dispatch(argv: List[str]) -> int:
if not argv or argv[0] == "run":
return main()
if argv[0] in ("-h", "--help", "help"):
print(USAGE.format(path=CONFIG_PATH))
return 0
if argv[0] == "config":
sub = argv[1] if len(argv) > 1 else ""
if sub == "init":
return cmd_config_init()
if sub == "path":
return cmd_config_path()
print(USAGE.format(path=CONFIG_PATH), file=sys.stderr)
return 2
print(f"coe-mirror: unknown command {argv[0]!r}", file=sys.stderr)
print(USAGE.format(path=CONFIG_PATH), file=sys.stderr)
return 2
if __name__ == "__main__": if __name__ == "__main__":
sys.exit(main()) sys.exit(dispatch(sys.argv[1:]))
+105 -9
View File
@@ -42,6 +42,8 @@ def mock_paths(tmp_path, monkeypatch):
monkeypatch.setattr(coe_mirror, "LOG_DIR", log_dir) monkeypatch.setattr(coe_mirror, "LOG_DIR", log_dir)
monkeypatch.setattr(coe_mirror, "LOG_FILE", log_dir / "test.log") monkeypatch.setattr(coe_mirror, "LOG_FILE", log_dir / "test.log")
monkeypatch.setattr(coe_mirror, "OWNERS", ["TestOwner"]) 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 { return {
"pcloud_target": pcloud_target, "pcloud_target": pcloud_target,
@@ -625,8 +627,8 @@ class TestMatrixSendSuccess:
# Mock send endpoint to return 200 # Mock send endpoint to return 200
with requests_mock.Mocker() as m: with requests_mock.Mocker() as m:
m.post( m.post(
"https://m.mealeyfamily.com/_matrix/client/r0/rooms/" f"{coe_mirror.MATRIX_HOMESERVER}/_matrix/client/r0/rooms/"
"%21MBpPtYlQYMcngtpvJt%3Am.mealeyfamily.com/send/m.room.message", "%21testroom%3Atest.local/send/m.room.message",
json={"event_id": "$event1"}, json={"event_id": "$event1"},
status_code=200 status_code=200
) )
@@ -662,8 +664,8 @@ class TestMatrixSendFailureNonfatal:
with requests_mock.Mocker() as m: with requests_mock.Mocker() as m:
m.post( m.post(
"https://m.mealeyfamily.com/_matrix/client/r0/rooms/" f"{coe_mirror.MATRIX_HOMESERVER}/_matrix/client/r0/rooms/"
"%21MBpPtYlQYMcngtpvJt%3Am.mealeyfamily.com/send/m.room.message", "%21testroom%3Atest.local/send/m.room.message",
status_code=500, status_code=500,
text="Internal Server Error" text="Internal Server Error"
) )
@@ -714,8 +716,8 @@ class TestMatrixTokenRefreshOn401:
with requests_mock.Mocker() as m: with requests_mock.Mocker() as m:
# First send returns 401 # First send returns 401
m.post( m.post(
"https://m.mealeyfamily.com/_matrix/client/r0/rooms/" f"{coe_mirror.MATRIX_HOMESERVER}/_matrix/client/r0/rooms/"
"%21MBpPtYlQYMcngtpvJt%3Am.mealeyfamily.com/send/m.room.message", "%21testroom%3Atest.local/send/m.room.message",
[ [
{"status_code": 401, "text": "Unauthorized"}, {"status_code": 401, "text": "Unauthorized"},
{"status_code": 200, "json": {"event_id": "$event1"}} {"status_code": 200, "json": {"event_id": "$event1"}}
@@ -724,7 +726,7 @@ class TestMatrixTokenRefreshOn401:
# Login endpoint # Login endpoint
m.post( m.post(
"https://m.mealeyfamily.com/_matrix/client/r0/login", f"{coe_mirror.MATRIX_HOMESERVER}/_matrix/client/r0/login",
json={"access_token": "new-token", "expires_in_ms": 604800000} json={"access_token": "new-token", "expires_in_ms": 604800000}
) )
@@ -765,7 +767,7 @@ class TestMatrixRoomEnvOverride:
with requests_mock.Mocker() as m: with requests_mock.Mocker() as m:
m.post( m.post(
"https://m.mealeyfamily.com/_matrix/client/r0/rooms/" f"{coe_mirror.MATRIX_HOMESERVER}/_matrix/client/r0/rooms/"
"%21override%3Aexample.org/send/m.room.message", "%21override%3Aexample.org/send/m.room.message",
json={"event_id": "$event1"}, json={"event_id": "$event1"},
status_code=200 status_code=200
@@ -806,7 +808,7 @@ class TestMatrixLoginWritesCacheChmod600:
with requests_mock.Mocker() as m: with requests_mock.Mocker() as m:
m.post( m.post(
"https://m.mealeyfamily.com/_matrix/client/r0/login", f"{coe_mirror.MATRIX_HOMESERVER}/_matrix/client/r0/login",
json={"access_token": "new-token", "expires_in_ms": 604800000} json={"access_token": "new-token", "expires_in_ms": 604800000}
) )
@@ -913,3 +915,97 @@ class TestMultiOwner:
assert captured["failed_owners"] == ["OwnerA"] assert captured["failed_owners"] == ["OwnerA"]
# summary contains only the OwnerB entry # summary contains only the OwnerB entry
assert any(owner == "OwnerB" and repo == "onlyone" for owner, repo, *_ in captured["summary"]) 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'
'pcloud_target = "/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"):
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.PCLOUD_TARGET == 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_PCLOUD_TARGET",
"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