cgcardona / muse / 368bcde6 / tests / test_cli_hub.py
cgcardona / muse public
test_cli_hub.py python
329 lines 13.7 KB
Raw
368bcde6 Add security test coverage and reference documentation Gabriel Cardona <gabriel@tellurstori.com> 9h ago
1 """Tests for `muse hub` CLI commands — connect, status, disconnect, ping.
2
3 All network calls are mocked — no real HTTP traffic occurs. The identity
4 store is isolated per test using a tmp_path override.
5 """
6
7 from __future__ import annotations
8
9 import io
10 import json
11 import pathlib
12 import unittest.mock
13 import urllib.error
14 import urllib.request
15 import urllib.response
16
17 import pytest
18 from typer.testing import CliRunner
19
20 from muse.cli.app import cli
21 from muse.cli.commands.hub import _hub_hostname, _normalise_url, _ping_hub
22 from muse.cli.config import get_hub_url, set_hub_url
23 from muse.core.identity import IdentityEntry, save_identity
24
25 runner = CliRunner()
26
27
28 # ---------------------------------------------------------------------------
29 # Fixture: minimal Muse repo
30 # ---------------------------------------------------------------------------
31
32
33 @pytest.fixture()
34 def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
35 """Minimal .muse/ repo; hub tests don't need commits."""
36 muse_dir = tmp_path / ".muse"
37 (muse_dir / "refs" / "heads").mkdir(parents=True)
38 (muse_dir / "objects").mkdir()
39 (muse_dir / "commits").mkdir()
40 (muse_dir / "snapshots").mkdir()
41 (muse_dir / "repo.json").write_text(
42 json.dumps({"repo_id": "test-repo", "schema_version": "2", "domain": "midi"})
43 )
44 (muse_dir / "HEAD").write_text("refs/heads/main\n")
45 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
46 monkeypatch.chdir(tmp_path)
47 # Redirect the identity store to tmp_path so tests never touch ~/.muse/
48 fake_identity_dir = tmp_path / "fake_home" / ".muse"
49 fake_identity_dir.mkdir(parents=True)
50 fake_identity_file = fake_identity_dir / "identity.toml"
51 monkeypatch.setattr("muse.core.identity._IDENTITY_DIR", fake_identity_dir)
52 monkeypatch.setattr("muse.core.identity._IDENTITY_FILE", fake_identity_file)
53 return tmp_path
54
55
56 # ---------------------------------------------------------------------------
57 # Unit tests for pure helper functions
58 # ---------------------------------------------------------------------------
59
60
61 class TestNormaliseUrl:
62 def test_bare_hostname_gets_https(self) -> None:
63 assert _normalise_url("musehub.ai") == "https://musehub.ai"
64
65 def test_https_url_unchanged(self) -> None:
66 assert _normalise_url("https://musehub.ai") == "https://musehub.ai"
67
68 def test_trailing_slash_stripped(self) -> None:
69 assert _normalise_url("https://musehub.ai/") == "https://musehub.ai"
70
71 def test_http_url_raises(self) -> None:
72 with pytest.raises(ValueError, match="Insecure"):
73 _normalise_url("http://musehub.ai")
74
75 def test_http_suggests_https(self) -> None:
76 with pytest.raises(ValueError, match="https://"):
77 _normalise_url("http://musehub.ai")
78
79 def test_whitespace_stripped(self) -> None:
80 assert _normalise_url(" https://musehub.ai ") == "https://musehub.ai"
81
82
83 class TestHubHostname:
84 def test_extracts_hostname_from_https_url(self) -> None:
85 assert _hub_hostname("https://musehub.ai/repos/r1") == "musehub.ai"
86
87 def test_bare_hostname(self) -> None:
88 assert _hub_hostname("musehub.ai") == "musehub.ai"
89
90 def test_strips_path(self) -> None:
91 assert _hub_hostname("https://musehub.ai/deep/path") == "musehub.ai"
92
93 def test_preserves_port(self) -> None:
94 assert _hub_hostname("https://musehub.ai:8443") == "musehub.ai:8443"
95
96
97 class TestPingHub:
98 def test_2xx_returns_true(self) -> None:
99 mock_resp = unittest.mock.MagicMock()
100 mock_resp.status = 200
101 mock_resp.__enter__ = lambda s: s
102 mock_resp.__exit__ = unittest.mock.MagicMock(return_value=False)
103 with unittest.mock.patch("muse.cli.commands.hub._PING_OPENER.open", return_value=mock_resp):
104 ok, msg = _ping_hub("https://musehub.ai")
105 assert ok is True
106 assert "200" in msg
107
108 def test_5xx_returns_false(self) -> None:
109 mock_resp = unittest.mock.MagicMock()
110 mock_resp.status = 503
111 mock_resp.__enter__ = lambda s: s
112 mock_resp.__exit__ = unittest.mock.MagicMock(return_value=False)
113 with unittest.mock.patch("muse.cli.commands.hub._PING_OPENER.open", return_value=mock_resp):
114 ok, msg = _ping_hub("https://musehub.ai")
115 assert ok is False
116
117 def test_http_error_returns_false(self) -> None:
118 err = urllib.error.HTTPError("https://musehub.ai/health", 401, "Unauthorized", {}, io.BytesIO(b"Unauthorized"))
119 with unittest.mock.patch("muse.cli.commands.hub._PING_OPENER.open", side_effect=err):
120 ok, msg = _ping_hub("https://musehub.ai")
121 assert ok is False
122 assert "401" in msg
123
124 def test_url_error_returns_false(self) -> None:
125 err = urllib.error.URLError("name resolution failure")
126 with unittest.mock.patch("muse.cli.commands.hub._PING_OPENER.open", side_effect=err):
127 ok, msg = _ping_hub("https://musehub.ai")
128 assert ok is False
129
130 def test_timeout_error_returns_false(self) -> None:
131 with unittest.mock.patch("muse.cli.commands.hub._PING_OPENER.open", side_effect=TimeoutError()):
132 ok, msg = _ping_hub("https://musehub.ai")
133 assert ok is False
134 assert "timed out" in msg
135
136 def test_os_error_returns_false(self) -> None:
137 with unittest.mock.patch("muse.cli.commands.hub._PING_OPENER.open", side_effect=OSError("network down")):
138 ok, msg = _ping_hub("https://musehub.ai")
139 assert ok is False
140
141 def test_health_endpoint_used(self) -> None:
142 calls: list[str] = []
143
144 def _fake_open(req: urllib.request.Request, timeout: int = 0) -> urllib.response.addinfourl:
145 calls.append(req.full_url)
146 raise urllib.error.URLError("stop")
147
148 with unittest.mock.patch("muse.cli.commands.hub._PING_OPENER.open", side_effect=_fake_open):
149 _ping_hub("https://musehub.ai")
150 assert calls and calls[0] == "https://musehub.ai/health"
151
152
153 # ---------------------------------------------------------------------------
154 # hub connect
155 # ---------------------------------------------------------------------------
156
157
158 class TestHubConnect:
159 def test_connect_bare_hostname(self, repo: pathlib.Path) -> None:
160 result = runner.invoke(cli, ["hub", "connect", "musehub.ai"])
161 assert result.exit_code == 0
162 assert "Connected" in result.output
163
164 def test_connect_stores_https_url(self, repo: pathlib.Path) -> None:
165 runner.invoke(cli, ["hub", "connect", "musehub.ai"])
166 stored = get_hub_url(repo)
167 assert stored == "https://musehub.ai"
168
169 def test_connect_https_url_directly(self, repo: pathlib.Path) -> None:
170 result = runner.invoke(cli, ["hub", "connect", "https://musehub.ai"])
171 assert result.exit_code == 0
172 assert get_hub_url(repo) == "https://musehub.ai"
173
174 def test_connect_http_rejected(self, repo: pathlib.Path) -> None:
175 result = runner.invoke(cli, ["hub", "connect", "http://musehub.ai"])
176 assert result.exit_code != 0
177 assert "Insecure" in result.output or "rejected" in result.output
178
179 def test_connect_warns_on_hub_switch(self, repo: pathlib.Path) -> None:
180 runner.invoke(cli, ["hub", "connect", "https://hub1.example.com"])
181 result = runner.invoke(cli, ["hub", "connect", "https://hub2.example.com"])
182 assert result.exit_code == 0
183 assert "hub1.example.com" in result.output or "Switching" in result.output
184
185 def test_connect_shows_identity_if_already_logged_in(self, repo: pathlib.Path) -> None:
186 entry: IdentityEntry = {"type": "human", "token": "tok123", "name": "Alice"}
187 save_identity("https://musehub.ai", entry)
188 result = runner.invoke(cli, ["hub", "connect", "https://musehub.ai"])
189 assert result.exit_code == 0
190 assert "Alice" in result.output or "human" in result.output
191
192 def test_connect_prompts_login_when_no_identity(self, repo: pathlib.Path) -> None:
193 result = runner.invoke(cli, ["hub", "connect", "https://musehub.ai"])
194 assert result.exit_code == 0
195 assert "muse auth login" in result.output
196
197 def test_connect_fails_outside_repo(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
198 monkeypatch.chdir(tmp_path)
199 monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
200 result = runner.invoke(cli, ["hub", "connect", "https://musehub.ai"])
201 assert result.exit_code != 0
202
203
204 # ---------------------------------------------------------------------------
205 # hub status
206 # ---------------------------------------------------------------------------
207
208
209 class TestHubStatus:
210 def _setup_hub(self, repo: pathlib.Path) -> None:
211 set_hub_url("https://musehub.ai", repo)
212
213 def test_no_hub_exits_nonzero(self, repo: pathlib.Path) -> None:
214 result = runner.invoke(cli, ["hub", "status"])
215 assert result.exit_code != 0
216
217 def test_hub_url_shown(self, repo: pathlib.Path) -> None:
218 self._setup_hub(repo)
219 result = runner.invoke(cli, ["hub", "status"])
220 assert result.exit_code == 0
221 assert "musehub.ai" in result.output
222
223 def test_not_authenticated_shown(self, repo: pathlib.Path) -> None:
224 self._setup_hub(repo)
225 result = runner.invoke(cli, ["hub", "status"])
226 assert "not authenticated" in result.output or "auth login" in result.output
227
228 def test_identity_fields_shown_when_logged_in(self, repo: pathlib.Path) -> None:
229 self._setup_hub(repo)
230 entry: IdentityEntry = {"type": "agent", "token": "tok", "name": "bot", "id": "agt_001"}
231 save_identity("https://musehub.ai", entry)
232 result = runner.invoke(cli, ["hub", "status"])
233 assert "agent" in result.output
234 assert "bot" in result.output
235
236 def test_json_output_structure(self, repo: pathlib.Path) -> None:
237 self._setup_hub(repo)
238 result = runner.invoke(cli, ["hub", "status", "--json"])
239 assert result.exit_code == 0
240 data = json.loads(result.output)
241 assert "hub_url" in data
242 assert "hostname" in data
243 assert "authenticated" in data
244
245 def test_json_output_with_identity(self, repo: pathlib.Path) -> None:
246 self._setup_hub(repo)
247 entry: IdentityEntry = {"type": "human", "token": "t", "name": "Alice", "id": "usr_1"}
248 save_identity("https://musehub.ai", entry)
249 result = runner.invoke(cli, ["hub", "status", "--json"])
250 data = json.loads(result.output)
251 assert data["authenticated"] is True
252 assert data["identity_type"] == "human"
253 assert data["identity_name"] == "Alice"
254
255
256 # ---------------------------------------------------------------------------
257 # hub disconnect
258 # ---------------------------------------------------------------------------
259
260
261 class TestHubDisconnect:
262 def test_disconnect_clears_hub_url(self, repo: pathlib.Path) -> None:
263 set_hub_url("https://musehub.ai", repo)
264 result = runner.invoke(cli, ["hub", "disconnect"])
265 assert result.exit_code == 0
266 assert get_hub_url(repo) is None
267
268 def test_disconnect_shows_hostname(self, repo: pathlib.Path) -> None:
269 set_hub_url("https://musehub.ai", repo)
270 result = runner.invoke(cli, ["hub", "disconnect"])
271 assert "musehub.ai" in result.output
272
273 def test_disconnect_nothing_to_do(self, repo: pathlib.Path) -> None:
274 result = runner.invoke(cli, ["hub", "disconnect"])
275 assert result.exit_code == 0
276 assert "nothing" in result.output.lower() or "No hub" in result.output
277
278 def test_disconnect_preserves_identity(self, repo: pathlib.Path) -> None:
279 """Credentials in identity.toml must survive hub disconnect."""
280 set_hub_url("https://musehub.ai", repo)
281 entry: IdentityEntry = {"type": "human", "token": "secret"}
282 save_identity("https://musehub.ai", entry)
283 runner.invoke(cli, ["hub", "disconnect"])
284 from muse.core.identity import load_identity
285 assert load_identity("https://musehub.ai") is not None
286
287 def test_disconnect_fails_outside_repo(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
288 monkeypatch.chdir(tmp_path)
289 monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
290 result = runner.invoke(cli, ["hub", "disconnect"])
291 assert result.exit_code != 0
292
293
294 # ---------------------------------------------------------------------------
295 # hub ping
296 # ---------------------------------------------------------------------------
297
298
299 class TestHubPing:
300 def _setup_hub(self, repo: pathlib.Path) -> None:
301 set_hub_url("https://musehub.ai", repo)
302
303 def test_ping_success(self, repo: pathlib.Path) -> None:
304 self._setup_hub(repo)
305 mock_resp = unittest.mock.MagicMock()
306 mock_resp.status = 200
307 mock_resp.__enter__ = lambda s: s
308 mock_resp.__exit__ = unittest.mock.MagicMock(return_value=False)
309 with unittest.mock.patch("muse.cli.commands.hub._PING_OPENER.open", return_value=mock_resp):
310 result = runner.invoke(cli, ["hub", "ping"])
311 assert result.exit_code == 0
312 assert "200" in result.output or "OK" in result.output.upper()
313
314 def test_ping_failure_exits_nonzero(self, repo: pathlib.Path) -> None:
315 self._setup_hub(repo)
316 err = urllib.error.URLError("no route to host")
317 with unittest.mock.patch("muse.cli.commands.hub._PING_OPENER.open", side_effect=err):
318 result = runner.invoke(cli, ["hub", "ping"])
319 assert result.exit_code != 0
320
321 def test_ping_no_hub_exits_nonzero(self, repo: pathlib.Path) -> None:
322 result = runner.invoke(cli, ["hub", "ping"])
323 assert result.exit_code != 0
324
325 def test_ping_fails_outside_repo(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
326 monkeypatch.chdir(tmp_path)
327 monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
328 result = runner.invoke(cli, ["hub", "ping"])
329 assert result.exit_code != 0