diff --git a/.gitignore b/.gitignore index 53ee67e..35cc7cf 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,5 @@ __pycache__/ .pytest_cache/ *.egg-info/ .active-item.json +.workflow-state.json +agent-spawns/ diff --git a/README.md b/README.md index cb2d29e..076095b 100644 --- a/README.md +++ b/README.md @@ -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 `//_.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 +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: `//_.zip` +- Staging (while pending flush): `/___.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 (`!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://...`. +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 diff --git a/coe_mirror.py b/coe_mirror.py index b6a5063..8c45116 100644 --- a/coe_mirror.py +++ b/coe_mirror.py @@ -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//`. Runs unattended via systemd timer twice daily. -Local staging when pCloud unavailable. +under `PCLOUD_TARGET//`. 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.mealeyfamily.com" -MATRIX_ROOM = "!MBpPtYlQYMcngtpvJt:m.mealeyfamily.com" # 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_; 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:])) diff --git a/tests/test_coe_mirror.py b/tests/test_coe_mirror.py index 3c1cf0a..6cd416b 100644 --- a/tests/test_coe_mirror.py +++ b/tests/test_coe_mirror.py @@ -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.mealeyfamily.com/_matrix/client/r0/rooms/" - "%21MBpPtYlQYMcngtpvJt%3Am.mealeyfamily.com/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.mealeyfamily.com/_matrix/client/r0/rooms/" - "%21MBpPtYlQYMcngtpvJt%3Am.mealeyfamily.com/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.mealeyfamily.com/_matrix/client/r0/rooms/" - "%21MBpPtYlQYMcngtpvJt%3Am.mealeyfamily.com/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.mealeyfamily.com/_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.mealeyfamily.com/_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.mealeyfamily.com/_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