| 1 | """Tests for the pass(1) credential source in lib/env.py. |
| 2 | |
| 3 | Covers: |
| 4 | - missing `pass` binary returns {} |
| 5 | - successful lookups return parsed key/value pairs at the prefix convention |
| 6 | - first-line extraction + whitespace stripping |
| 7 | - subprocess timeout / OSError are swallowed |
| 8 | - the path prefix is honored (default + LAST30DAYS_PASS_PREFIX override) |
| 9 | - get_config merges pass below keychain and below explicit env, and labels |
| 10 | _CONFIG_SOURCE = 'pass' when pass is the effective source |
| 11 | - lib/env.py KEYCHAIN_KEYS and setup-pass.sh ALL_KEYS stay in lockstep |
| 12 | """ |
| 13 | |
| 14 | from __future__ import annotations |
| 15 | |
| 16 | import re |
| 17 | import subprocess |
| 18 | from pathlib import Path |
| 19 | from unittest import mock |
| 20 | |
| 21 | import pytest |
| 22 | |
| 23 | from lib import env |
| 24 | |
| 25 | SETUP_PASS_SH = Path(__file__).resolve().parents[1] / "skills" / "last30days" / "scripts" / "setup-pass.sh" |
| 26 | |
| 27 | # --------------------------------------------------------------------------- |
| 28 | # _load_pass unit tests |
| 29 | # --------------------------------------------------------------------------- |
| 30 | |
| 31 | |
| 32 | def _run_result(returncode: int, stdout: str = "") -> subprocess.CompletedProcess: |
| 33 | return subprocess.CompletedProcess(args=[], returncode=returncode, stdout=stdout, stderr="") |
| 34 | |
| 35 | |
| 36 | def test_load_pass_returns_empty_when_pass_missing(): |
| 37 | with mock.patch("shutil.which", return_value=None): |
| 38 | assert env._load_pass(["XAI_API_KEY"], "last30days/") == {} |
| 39 | |
| 40 | |
| 41 | def test_load_pass_loads_present_keys_skips_missing(): |
| 42 | def fake_run(cmd, **kwargs): |
| 43 | path = cmd[-1] # [pass_bin, "show", "<prefix><key>"] |
| 44 | if path == "last30days/XAI_API_KEY": |
| 45 | return _run_result(0, "xai-abc\n") |
| 46 | if path == "last30days/BRAVE_API_KEY": |
| 47 | return _run_result(0, "brv-xyz\n") |
| 48 | return _run_result(1) # pass exits non-zero for a missing entry |
| 49 | |
| 50 | with mock.patch("shutil.which", return_value="/usr/bin/pass"), \ |
| 51 | mock.patch("subprocess.run", side_effect=fake_run): |
| 52 | result = env._load_pass(["XAI_API_KEY", "BRAVE_API_KEY", "OPENAI_API_KEY"], "last30days/") |
| 53 | |
| 54 | assert result == {"XAI_API_KEY": "xai-abc", "BRAVE_API_KEY": "brv-xyz"} |
| 55 | |
| 56 | |
| 57 | def test_load_pass_takes_first_line_only(): |
| 58 | # pass entries keep the secret on line 1; metadata may follow. |
| 59 | with mock.patch("shutil.which", return_value="/usr/bin/pass"), \ |
| 60 | mock.patch("subprocess.run", return_value=_run_result(0, "sk-secret\nurl: https://x\nuser: bob\n")): |
| 61 | assert env._load_pass(["OPENAI_API_KEY"], "last30days/") == {"OPENAI_API_KEY": "sk-secret"} |
| 62 | |
| 63 | |
| 64 | def test_load_pass_strips_whitespace(): |
| 65 | with mock.patch("shutil.which", return_value="/usr/bin/pass"), \ |
| 66 | mock.patch("subprocess.run", return_value=_run_result(0, " hello-key \n")): |
| 67 | assert env._load_pass(["FOO"], "last30days/") == {"FOO": "hello-key"} |
| 68 | |
| 69 | |
| 70 | def test_load_pass_skips_empty_and_whitespace_only_stdout(): |
| 71 | with mock.patch("shutil.which", return_value="/usr/bin/pass"), \ |
| 72 | mock.patch("subprocess.run", return_value=_run_result(0, " \n")): |
| 73 | assert env._load_pass(["XAI_API_KEY"], "last30days/") == {} |
| 74 | |
| 75 | |
| 76 | def test_load_pass_swallows_timeout(): |
| 77 | def fake_run(cmd, **kwargs): |
| 78 | raise subprocess.TimeoutExpired(cmd=cmd, timeout=5) |
| 79 | |
| 80 | with mock.patch("shutil.which", return_value="/usr/bin/pass"), \ |
| 81 | mock.patch("subprocess.run", side_effect=fake_run): |
| 82 | assert env._load_pass(["XAI_API_KEY"], "last30days/") == {} |
| 83 | |
| 84 | |
| 85 | def test_load_pass_swallows_oserror(): |
| 86 | with mock.patch("shutil.which", return_value="/usr/bin/pass"), \ |
| 87 | mock.patch("subprocess.run", side_effect=OSError("boom")): |
| 88 | assert env._load_pass(["XAI_API_KEY"], "last30days/") == {} |
| 89 | |
| 90 | |
| 91 | def test_load_pass_stops_probing_after_timeout(): |
| 92 | # A hanging store (GPG/pinentry) must not be probed once per key — otherwise |
| 93 | # a locked store stalls every config load by 5s x len(keys). |
| 94 | calls = {"n": 0} |
| 95 | |
| 96 | def fake_run(cmd, **kwargs): |
| 97 | calls["n"] += 1 |
| 98 | raise subprocess.TimeoutExpired(cmd=cmd, timeout=5) |
| 99 | |
| 100 | with mock.patch("shutil.which", return_value="/usr/bin/pass"), \ |
| 101 | mock.patch("subprocess.run", side_effect=fake_run): |
| 102 | result = env._load_pass(["XAI_API_KEY", "BRAVE_API_KEY", "OPENAI_API_KEY"], "last30days/") |
| 103 | |
| 104 | assert result == {} |
| 105 | assert calls["n"] == 1 # stopped after the first timeout, didn't probe the rest |
| 106 | |
| 107 | |
| 108 | def test_load_pass_honors_prefix(): |
| 109 | seen = {} |
| 110 | |
| 111 | def fake_run(cmd, **kwargs): |
| 112 | seen["path"] = cmd[-1] |
| 113 | return _run_result(0, "v\n") |
| 114 | |
| 115 | with mock.patch("shutil.which", return_value="/usr/bin/pass"), \ |
| 116 | mock.patch("subprocess.run", side_effect=fake_run): |
| 117 | env._load_pass(["XAI_API_KEY"], "secrets/l30/") |
| 118 | |
| 119 | assert seen["path"] == "secrets/l30/XAI_API_KEY" |
| 120 | |
| 121 | |
| 122 | # --------------------------------------------------------------------------- |
| 123 | # get_config integration tests (pass merged below keychain and explicit env) |
| 124 | # --------------------------------------------------------------------------- |
| 125 | |
| 126 | |
| 127 | @pytest.fixture |
| 128 | def clean_env(monkeypatch, tmp_path): |
| 129 | for var in [ |
| 130 | "OPENAI_API_KEY", "XAI_API_KEY", "BRAVE_API_KEY", "AUTH_TOKEN", "CT0", |
| 131 | "SCRAPECREATORS_API_KEY", "APIFY_API_TOKEN", "BSKY_HANDLE", |
| 132 | "BSKY_APP_PASSWORD", "TRUTHSOCIAL_TOKEN", "EXA_API_KEY", |
| 133 | "SERPER_API_KEY", "OPENROUTER_API_KEY", "PERPLEXITY_API_KEY", "PARALLEL_API_KEY", |
| 134 | "XQUIK_API_KEY", "GOOGLE_API_KEY", "GEMINI_API_KEY", |
| 135 | "GOOGLE_GENAI_API_KEY", "INCLUDE_SOURCES", "FROM_BROWSER", |
| 136 | ]: |
| 137 | monkeypatch.delenv(var, raising=False) |
| 138 | monkeypatch.setattr(env, "CONFIG_FILE", tmp_path / "does-not-exist.env") |
| 139 | monkeypatch.chdir(tmp_path) |
| 140 | |
| 141 | |
| 142 | def test_get_config_reports_pass_source(clean_env): |
| 143 | with mock.patch.object(env, "_load_keychain", return_value={}), \ |
| 144 | mock.patch.object(env, "_load_pass", return_value={"XAI_API_KEY": "xai-from-pass"}): |
| 145 | cfg = env.get_config() |
| 146 | assert cfg["_CONFIG_SOURCE"] == "pass" |
| 147 | assert cfg["XAI_API_KEY"] == "xai-from-pass" |
| 148 | |
| 149 | |
| 150 | def test_get_config_keychain_outranks_pass(clean_env): |
| 151 | with mock.patch.object(env, "_load_keychain", return_value={"XAI_API_KEY": "xai-from-kc"}), \ |
| 152 | mock.patch.object(env, "_load_pass", return_value={"XAI_API_KEY": "xai-from-pass"}): |
| 153 | cfg = env.get_config() |
| 154 | assert cfg["XAI_API_KEY"] == "xai-from-kc" |
| 155 | assert cfg["_CONFIG_SOURCE"] == "keychain" |
| 156 | |
| 157 | |
| 158 | def test_get_config_env_var_overrides_pass(clean_env, monkeypatch): |
| 159 | monkeypatch.setenv("XAI_API_KEY", "xai-from-env") |
| 160 | with mock.patch.object(env, "_load_keychain", return_value={}), \ |
| 161 | mock.patch.object(env, "_load_pass", return_value={"XAI_API_KEY": "xai-from-pass"}): |
| 162 | cfg = env.get_config() |
| 163 | assert cfg["XAI_API_KEY"] == "xai-from-env" |
| 164 | |
| 165 | |
| 166 | def test_get_config_global_file_outranks_pass(clean_env, tmp_path, monkeypatch): |
| 167 | cfg_file = tmp_path / "global.env" |
| 168 | cfg_file.write_text("XAI_API_KEY=xai-from-file\n") |
| 169 | monkeypatch.setattr(env, "CONFIG_FILE", cfg_file) |
| 170 | with mock.patch.object(env, "_load_keychain", return_value={}), \ |
| 171 | mock.patch.object(env, "_load_pass", return_value={"XAI_API_KEY": "xai-from-pass"}): |
| 172 | cfg = env.get_config() |
| 173 | assert cfg["XAI_API_KEY"] == "xai-from-file" |
| 174 | assert cfg["_CONFIG_SOURCE"].startswith("global:") |
| 175 | |
| 176 | |
| 177 | def test_get_config_probes_pass_only_for_missing_keys(clean_env, monkeypatch): |
| 178 | # A key already supplied by a higher-priority source must not be probed in |
| 179 | # pass — that's what keeps a `pass`-installed but `.env`-using box off gpg. |
| 180 | monkeypatch.setenv("XAI_API_KEY", "xai-from-env") |
| 181 | seen = {} |
| 182 | |
| 183 | def fake_load_pass(keys, prefix): |
| 184 | seen["keys"] = list(keys) |
| 185 | return {} |
| 186 | |
| 187 | with mock.patch.object(env, "_load_keychain", return_value={}), \ |
| 188 | mock.patch.object(env, "_load_pass", side_effect=fake_load_pass): |
| 189 | env.get_config() |
| 190 | |
| 191 | assert "XAI_API_KEY" not in seen["keys"] # already supplied by env |
| 192 | assert "BRAVE_API_KEY" in seen["keys"] # still missing, so probed |
| 193 | |
| 194 | |
| 195 | def test_get_config_pass_prefix_resolved_from_config_file(clean_env, tmp_path, monkeypatch): |
| 196 | # LAST30DAYS_PASS_PREFIX set in the .env config layer (not shell-exported) |
| 197 | # must reach _load_pass — i.e. the prefix is resolved at call time. |
| 198 | cfg_file = tmp_path / "global.env" |
| 199 | cfg_file.write_text("LAST30DAYS_PASS_PREFIX=secrets/l30/\n") |
| 200 | monkeypatch.setattr(env, "CONFIG_FILE", cfg_file) |
| 201 | seen = {} |
| 202 | |
| 203 | def fake_load_pass(keys, prefix): |
| 204 | seen["prefix"] = prefix |
| 205 | return {} |
| 206 | |
| 207 | with mock.patch.object(env, "_load_keychain", return_value={}), \ |
| 208 | mock.patch.object(env, "_load_pass", side_effect=fake_load_pass): |
| 209 | env.get_config() |
| 210 | |
| 211 | assert seen["prefix"] == "secrets/l30/" |
| 212 | |
| 213 | |
| 214 | def test_get_config_openai_key_can_come_from_pass(clean_env): |
| 215 | with mock.patch.object(env, "_load_keychain", return_value={}), \ |
| 216 | mock.patch.object(env, "_load_pass", return_value={"OPENAI_API_KEY": "sk-from-pass"}): |
| 217 | cfg = env.get_config() |
| 218 | assert cfg["OPENAI_API_KEY"] == "sk-from-pass" |
| 219 | assert cfg["OPENAI_AUTH_SOURCE"] == "api_key" |
| 220 | |
| 221 | |
| 222 | # --------------------------------------------------------------------------- |
| 223 | # Drift guard: lib/env.py KEYCHAIN_KEYS and setup-pass.sh ALL_KEYS must stay in |
| 224 | # lockstep, same as the Keychain helper. A mismatch means a key stored via the |
| 225 | # helper wouldn't be picked up by the loader, or vice versa. |
| 226 | # --------------------------------------------------------------------------- |
| 227 | |
| 228 | |
| 229 | def _parse_all_keys_from_shell(script: Path) -> list[str]: |
| 230 | text = script.read_text(encoding="utf-8") |
| 231 | match = re.search(r"ALL_KEYS=\(\s*(.*?)\s*\)", text, re.DOTALL) |
| 232 | if not match: |
| 233 | raise AssertionError(f"ALL_KEYS=( ... ) array not found in {script}") |
| 234 | body = re.sub(r"#[^\n]*", "", match.group(1)) |
| 235 | return [tok for tok in body.split() if tok] |
| 236 | |
| 237 | |
| 238 | def test_pass_keys_match_setup_script(): |
| 239 | shell_keys = _parse_all_keys_from_shell(SETUP_PASS_SH) |
| 240 | python_keys = list(env.KEYCHAIN_KEYS) |
| 241 | assert shell_keys == python_keys, ( |
| 242 | "lib/env.py::KEYCHAIN_KEYS and scripts/setup-pass.sh::ALL_KEYS have " |
| 243 | f"drifted.\n python: {python_keys}\n shell: {shell_keys}" |
| 244 | ) |
| 245 |