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