gabriel / muse public
test_cli_hub.py python
330 lines 13.8 KB
Raw
8aa515d5 refactor: consolidate schema_version to single source of truth Gabriel Cardona <gabriel@tellurstori.com> 3d 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._version import __version__
21 from muse.cli.app import cli
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