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

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

Browse Source 
- ~/.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 9ffe38cbbc
commit 376115a9de
4 changed files with 357 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+2
View File
@@ -5,3 +5,5 @@ __pycache__/
.pytest_cache/
*.egg-info/
.active-item.json
.workflow-state.json
agent-spawns/
README.md
+83 -38
View File
@@ -1,66 +1,111 @@
# 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
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
```
## Manual run
Timer defaults to 00:00 and 12:00 daily with boot catch-up (`Persistent=true`). Missed windows run on next boot.
```
systemctl --user start coe-mirror.service
# or directly:
python3 ~/tools/coe-mirror/coe_mirror.py
```
## Matrix notifications (optional)
- **Disabled by default.** Leave `[matrix]` blank to skip.
- 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
- File: `~/.local/share/coe-mirror/coe-mirror.log` (rotated at 5 MB, 3 backups)
- 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.
## 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 (`!REDACTEDROOM:m.example.org`) on
`m.example.org`.
- 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://...`.
Old snapshots accumulate — no automatic pruning. Prune the target directory manually if you need to reclaim space.
## Testing
```
cd ~/tools/coe-mirror
python3 -m pytest tests/ -v
python -m pytest tests/ -v
```
## Uninstall
coe_mirror.py
+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
under `PCLOUD_TARGET/<owner>/`. Runs unattended via systemd timer twice daily.
Local staging when pCloud unavailable.
under `PCLOUD_TARGET/<owner>/`. Runs unattended via systemd timer.
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
@@ -14,6 +18,7 @@ import shutil
import subprocess
import sys
import time
import tomllib
import zipfile
from datetime import datetime, timedelta, timezone
from pathlib import Path
@@ -23,11 +28,59 @@ from urllib.parse import quote
import requests
# Module constants (overridable via env vars for testability)
API_BASE = "https://git.churchofmalware.org"
OWNERS = ["Nightmare_Eclipse", "shai_hulud", "bikini_mirror", "ek0mssavi0r"]
PCLOUD_TARGET = Path.home() / "pCloudDrive" / "Hacking" / "ChurchOfMalware"
PENDING_DIR = Path.home() / ".local" / "share" / "coe-mirror" / "pending"
# --- Config file (optional) ------------------------------------------------
CONFIG_PATH = Path(
os.getenv("COE_MIRROR_CONFIG")
or (Path(os.getenv("XDG_CONFIG_HOME") or Path.home() / ".config") / "coe-mirror" / "config.toml")
)
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_FILE = LOG_DIR / "coe-mirror.log"
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"
REPO_NAME_RE = re.compile(r"^[a-zA-Z0-9._-]+$")
# Matrix configuration
MATRIX_HOMESERVER = "https://m.example.org"
MATRIX_ROOM = "!REDACTEDROOM:m.example.org" # General
# Matrix configuration (fully opt-in; blank defaults => notifications silently skipped)
MATRIX_HOMESERVER: str = _resolve("COE_MIRROR_MATRIX_HOMESERVER", _CFG_MATRIX.get("homeserver"), "")
MATRIX_ROOM: str = _resolve("COE_MIRROR_MATRIX_ROOM", _CFG_MATRIX.get("room"), "")
MATRIX_TOKEN_CACHE = LOG_DIR / "matrix-token.json"
MATRIX_TIMEOUT = 8
MATRIX_REFRESH_THRESHOLD_SECONDS = 1800 # refresh if <30 min left
COE_MIRROR_BOT_USER_KEY = "COE_MIRROR_BOT_USER"
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()
@@ -362,7 +405,12 @@ def _write_token_cache(access_token: str, expires_at_iso: str) -> None:
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:
user = subprocess.run(
["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()
if u and p:
return u, p
logger.warning("creds returned non-zero for %s/%s",
COE_MIRROR_BOT_USER_KEY, COE_MIRROR_BOT_PASSWORD_KEY)
logger.debug("creds returned non-zero for %s/%s; trying env fallback",
COE_MIRROR_BOT_USER_KEY, COE_MIRROR_BOT_PASSWORD_KEY)
except FileNotFoundError:
logger.warning("creds binary not found on PATH")
logger.debug("creds binary not on PATH; trying env fallback")
except subprocess.TimeoutExpired:
logger.warning("creds timed out fetching Matrix credentials")
logger.warning("creds timed out; trying env fallback")
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
@@ -436,7 +490,11 @@ def _send_matrix(message: str) -> bool:
Non-fatal: any failure logs WARN and returns False. Never raises.
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:
session.headers.update({"User-Agent": USER_AGENT})
@@ -605,5 +663,86 @@ def main() -> int:
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__":
sys.exit(main())
sys.exit(dispatch(sys.argv[1:]))
tests/test_coe_mirror.py
+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_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 {
"pcloud_target": pcloud_target,
@@ -625,8 +627,8 @@ class TestMatrixSendSuccess:
# 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",
f"{coe_mirror.MATRIX_HOMESERVER}/_matrix/client/r0/rooms/"
"%21testroom%3Atest.local/send/m.room.message",
json={"event_id": "$event1"},
status_code=200
)
@@ -662,8 +664,8 @@ class TestMatrixSendFailureNonfatal:
with requests_mock.Mocker() as m:
m.post(
"https://m.example.org/_matrix/client/r0/rooms/"
"%21REDACTEDROOM%3Am.example.org/send/m.room.message",
f"{coe_mirror.MATRIX_HOMESERVER}/_matrix/client/r0/rooms/"
"%21testroom%3Atest.local/send/m.room.message",
status_code=500,
text="Internal Server Error"
)
@@ -714,8 +716,8 @@ class TestMatrixTokenRefreshOn401:
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",
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"}}
@@ -724,7 +726,7 @@ class TestMatrixTokenRefreshOn401:
# Login endpoint
m.post(
"https://m.example.org/_matrix/client/r0/login",
f"{coe_mirror.MATRIX_HOMESERVER}/_matrix/client/r0/login",
json={"access_token": "new-token", "expires_in_ms": 604800000}
)
@@ -765,7 +767,7 @@ class TestMatrixRoomEnvOverride:
with requests_mock.Mocker() as m:
m.post(
"https://m.example.org/_matrix/client/r0/rooms/"
f"{coe_mirror.MATRIX_HOMESERVER}/_matrix/client/r0/rooms/"
"%21override%3Aexample.org/send/m.room.message",
json={"event_id": "$event1"},
status_code=200
@@ -806,7 +808,7 @@ class TestMatrixLoginWritesCacheChmod600:
with requests_mock.Mocker() as m:
m.post(
"https://m.example.org/_matrix/client/r0/login",
f"{coe_mirror.MATRIX_HOMESERVER}/_matrix/client/r0/login",
json={"access_token": "new-token", "expires_in_ms": 604800000}
)
@@ -913,3 +915,97 @@ class TestMultiOwner:
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'
'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