gabriel / muse public
test_core_transport.py python
462 lines 18.4 KB
Raw
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 150 days ago
1 """Tests for muse.core.transport — HttpTransport and response parsers."""
2
3 from __future__ import annotations
4
5 import json
6 import unittest.mock
7 import urllib.error
8 import urllib.request
9 from io import BytesIO
10
11 import msgpack
12 import pytest
13
14 from muse.core._types import MsgpackDict
15 from muse.core.pack import PackBundle, RemoteInfo
16 from muse.core.transport import (
17 HttpTransport,
18 TransportError,
19 _parse_bundle,
20 _parse_push_result,
21 _parse_remote_info,
22 )
23
24
25 # ---------------------------------------------------------------------------
26 # Helpers
27 # ---------------------------------------------------------------------------
28
29
30 def _make_signing() -> "SigningIdentity":
31 """Generate a fresh Ed25519 SigningIdentity for tests."""
32 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
33 from muse.core.transport import SigningIdentity
34 return SigningIdentity(handle="testuser", private_key=Ed25519PrivateKey.generate())
35
36
37 def _mock_response(
38 body: bytes,
39 status: int = 200,
40 content_type: str = "application/x-msgpack",
41 streaming: bool = False,
42 ) -> unittest.mock.MagicMock:
43 """Return a mock urllib response context manager.
44
45 When *streaming* is True, ``read()`` returns *body* on the first call and
46 ``b""`` on all subsequent calls — matching the chunked-read loop used by
47 :meth:`~muse.core.transport.HttpTransport._execute_fetch`.
48 """
49 resp = unittest.mock.MagicMock()
50 if streaming:
51 resp.read.side_effect = [body, b""]
52 else:
53 resp.read.return_value = body
54 resp.headers = {"Content-Type": content_type}
55 resp.__enter__ = lambda s: s
56 resp.__exit__ = unittest.mock.MagicMock(return_value=False)
57 return resp
58
59
60 def _mp(data: MsgpackDict) -> bytes:
61 """Encode data as msgpack."""
62 return msgpack.packb(data, use_bin_type=True)
63
64
65 def _http_error(code: int, body: bytes = b"") -> urllib.error.HTTPError:
66 return urllib.error.HTTPError(
67 url="https://example.com",
68 code=code,
69 msg=str(code),
70 hdrs=None,
71 fp=BytesIO(body),
72 )
73
74
75 # ---------------------------------------------------------------------------
76 # _parse_remote_info
77 # ---------------------------------------------------------------------------
78
79
80 class TestParseRemoteInfo:
81 def test_valid_response(self) -> None:
82 raw = _mp(
83 {
84 "repo_id": "r123",
85 "domain": "midi",
86 "default_branch": "main",
87 "branch_heads": {"main": "abc123", "dev": "def456"},
88 }
89 )
90 info = _parse_remote_info(raw)
91 assert info["repo_id"] == "r123"
92 assert info["domain"] == "midi"
93 assert info["default_branch"] == "main"
94 assert info["branch_heads"] == {"main": "abc123", "dev": "def456"}
95
96 def test_invalid_msgpack_raises_transport_error(self) -> None:
97 with pytest.raises(TransportError):
98 _parse_remote_info(b"\xff\xff\xff\xff\xff invalid")
99
100 def test_non_dict_response_returns_defaults(self) -> None:
101 raw = _mp([1, 2, 3])
102 info = _parse_remote_info(raw)
103 assert info["repo_id"] == ""
104 assert info["branch_heads"] == {}
105
106 def test_missing_fields_get_defaults(self) -> None:
107 raw = _mp({"repo_id": "x"})
108 info = _parse_remote_info(raw)
109 assert info["repo_id"] == "x"
110 assert info["domain"] == "midi"
111 assert info["default_branch"] == "main"
112 assert info["branch_heads"] == {}
113
114 def test_non_string_branch_heads_excluded(self) -> None:
115 raw = _mp({"branch_heads": {"main": "abc", "bad": 123}})
116 info = _parse_remote_info(raw)
117 assert "main" in info["branch_heads"]
118 assert "bad" not in info["branch_heads"]
119
120
121 # ---------------------------------------------------------------------------
122 # _parse_bundle
123 # ---------------------------------------------------------------------------
124
125
126 class TestParseBundle:
127 def test_empty_msgpack_object_returns_empty_bundle(self) -> None:
128 bundle = _parse_bundle(_mp({}))
129 assert bundle == {}
130
131 def test_non_dict_returns_empty_bundle(self) -> None:
132 bundle = _parse_bundle(_mp([]))
133 assert bundle == {}
134
135 def test_commits_extracted(self) -> None:
136 raw = _mp(
137 {
138 "commits": [
139 {
140 "commit_id": "c1",
141 "repo_id": "r1",
142 "branch": "main",
143 "snapshot_id": "1" * 64,
144 "message": "test",
145 "committed_at": "2026-01-01T00:00:00+00:00",
146 "parent_commit_id": None,
147 "parent2_commit_id": None,
148 "author": "bob",
149 "metadata": {},
150 }
151 ]
152 }
153 )
154 bundle = _parse_bundle(raw)
155 commits = bundle.get("commits") or []
156 assert len(commits) == 1
157 assert commits[0]["commit_id"] == "c1"
158
159 def test_objects_extracted(self) -> None:
160 raw = _mp(
161 {
162 "objects": [
163 {
164 "object_id": "abc123",
165 "content": b"hello",
166 }
167 ]
168 }
169 )
170 bundle = _parse_bundle(raw)
171 objs = bundle.get("objects") or []
172 assert len(objs) == 1
173 assert objs[0]["object_id"] == "abc123"
174 assert objs[0]["content"] == b"hello"
175
176 def test_object_missing_content_excluded(self) -> None:
177 raw = _mp({"objects": [{"object_id": "abc"}]})
178 bundle = _parse_bundle(raw)
179 assert (bundle.get("objects") or []) == []
180
181 def test_branch_heads_extracted(self) -> None:
182 raw = _mp({"branch_heads": {"main": "abc123"}})
183 bundle = _parse_bundle(raw)
184 assert bundle.get("branch_heads") == {"main": "abc123"}
185
186
187 # ---------------------------------------------------------------------------
188 # _parse_push_result
189 # ---------------------------------------------------------------------------
190
191
192 class TestParsePushResult:
193 def test_success_response(self) -> None:
194 raw = _mp({"ok": True, "message": "pushed", "branch_heads": {"main": "abc"}})
195 result = _parse_push_result(raw)
196 assert result["ok"] is True
197 assert result["message"] == "pushed"
198 assert result["branch_heads"] == {"main": "abc"}
199
200 def test_failure_response(self) -> None:
201 raw = _mp({"ok": False, "message": "rejected", "branch_heads": {}})
202 result = _parse_push_result(raw)
203 assert result["ok"] is False
204 assert result["message"] == "rejected"
205
206 def test_non_msgpack_raises_transport_error(self) -> None:
207 with pytest.raises(TransportError):
208 _parse_push_result(b"\xff\xff invalid msgpack")
209
210 def test_missing_ok_defaults_false(self) -> None:
211 raw = _mp({"message": "hm", "branch_heads": {}})
212 result = _parse_push_result(raw)
213 assert result["ok"] is False
214
215
216 # ---------------------------------------------------------------------------
217 # HttpTransport — mocked urlopen
218 # ---------------------------------------------------------------------------
219
220
221 class TestHttpTransportFetchRemoteInfo:
222 def test_calls_correct_endpoint(self) -> None:
223 body = _mp(
224 {
225 "repo_id": "r1",
226 "domain": "midi",
227 "default_branch": "main",
228 "branch_heads": {"main": "abc"},
229 }
230 )
231 mock_resp = _mock_response(body)
232 with unittest.mock.patch("muse.core.transport._open_url", return_value=mock_resp) as m:
233 transport = HttpTransport()
234 info = transport.fetch_remote_info("https://hub.example.com/repos/r1", None)
235 req = m.call_args[0][0]
236 assert req.full_url == "https://hub.example.com/repos/r1/refs"
237 assert info["repo_id"] == "r1"
238
239 def test_msign_header_sent(self) -> None:
240 body = _mp(
241 {"repo_id": "r1", "domain": "midi", "default_branch": "main", "branch_heads": {}}
242 )
243 mock_resp = _mock_response(body)
244 signing = _make_signing()
245 with unittest.mock.patch("muse.core.transport._open_url", return_value=mock_resp) as m:
246 HttpTransport().fetch_remote_info("https://hub.example.com/repos/r1", signing)
247 req = m.call_args[0][0]
248 assert req.get_header("Authorization").startswith("MSign handle=\"testuser\"")
249
250 def test_no_token_no_auth_header(self) -> None:
251 body = _mp(
252 {"repo_id": "r1", "domain": "midi", "default_branch": "main", "branch_heads": {}}
253 )
254 mock_resp = _mock_response(body)
255 with unittest.mock.patch("muse.core.transport._open_url", return_value=mock_resp) as m:
256 HttpTransport().fetch_remote_info("https://hub.example.com/repos/r1", None)
257 req = m.call_args[0][0]
258 assert req.get_header("Authorization") is None
259
260 def test_http_401_raises_transport_error(self) -> None:
261 with unittest.mock.patch(
262 "muse.core.transport._open_url", side_effect=_http_error(401, b"Unauthorized")
263 ):
264 with pytest.raises(TransportError) as exc_info:
265 HttpTransport().fetch_remote_info("https://hub.example.com/repos/r1", None)
266 assert exc_info.value.status_code == 401
267
268 def test_http_404_raises_transport_error(self) -> None:
269 with unittest.mock.patch(
270 "muse.core.transport._open_url", side_effect=_http_error(404)
271 ):
272 with pytest.raises(TransportError) as exc_info:
273 HttpTransport().fetch_remote_info("https://hub.example.com/repos/r1", None)
274 assert exc_info.value.status_code == 404
275
276 def test_http_500_raises_transport_error(self) -> None:
277 with unittest.mock.patch(
278 "muse.core.transport._open_url", side_effect=_http_error(500, b"Internal Error")
279 ):
280 with pytest.raises(TransportError) as exc_info:
281 HttpTransport().fetch_remote_info("https://hub.example.com/repos/r1", None)
282 assert exc_info.value.status_code == 500
283
284 def test_url_error_raises_transport_error_with_code_0(self) -> None:
285 with unittest.mock.patch(
286 "muse.core.transport._open_url",
287 side_effect=urllib.error.URLError("Name or service not known"),
288 ):
289 with pytest.raises(TransportError) as exc_info:
290 HttpTransport().fetch_remote_info("https://bad.host/r", None)
291 assert exc_info.value.status_code == 0
292
293 def test_trailing_slash_stripped_from_url(self) -> None:
294 body = _mp(
295 {"repo_id": "r", "domain": "midi", "default_branch": "main", "branch_heads": {}}
296 )
297 mock_resp = _mock_response(body)
298 with unittest.mock.patch("muse.core.transport._open_url", return_value=mock_resp) as m:
299 HttpTransport().fetch_remote_info("https://hub.example.com/repos/r1/", None)
300 req = m.call_args[0][0]
301 assert req.full_url == "https://hub.example.com/repos/r1/refs"
302
303
304 class TestHttpTransportFetchPack:
305 def test_posts_to_fetch_endpoint(self) -> None:
306 bundle_body = _mp(
307 {
308 "commits": [],
309 "snapshots": [],
310 "objects": [],
311 "branch_heads": {"main": "abc"},
312 }
313 )
314 # fetch_pack uses _execute_fetch which reads in a chunked loop until
315 # read() returns b"" — use streaming=True so the mock terminates.
316 mock_resp = _mock_response(bundle_body, streaming=True)
317 with unittest.mock.patch("muse.core.transport._open_url", return_value=mock_resp) as m:
318 transport = HttpTransport()
319 bundle = transport.fetch_pack(
320 "https://hub.example.com/repos/r1",
321 _make_signing(),
322 want=["abc"],
323 have=["def"],
324 )
325 req = m.call_args[0][0]
326 assert req.full_url == "https://hub.example.com/repos/r1/fetch"
327 sent = msgpack.unpackb(req.data, raw=False)
328 assert sent["want"] == ["abc"]
329 assert sent["have"] == ["def"]
330 assert bundle.get("branch_heads") == {"main": "abc"}
331
332 def test_http_409_raises_transport_error(self) -> None:
333 with unittest.mock.patch(
334 "muse.core.transport._open_url", side_effect=_http_error(409)
335 ):
336 with pytest.raises(TransportError) as exc_info:
337 HttpTransport().fetch_pack("https://hub.example.com/r", None, [], [])
338 assert exc_info.value.status_code == 409
339
340
341 class TestHttpTransportPushPack:
342 def test_posts_to_push_endpoint(self) -> None:
343 push_body = _mp({"ok": True, "message": "ok", "branch_heads": {"main": "new"}})
344 mock_resp = _mock_response(push_body)
345 bundle: PackBundle = {"commits": [], "snapshots": [], "objects": []}
346 with unittest.mock.patch("muse.core.transport._open_url", return_value=mock_resp) as m:
347 result = HttpTransport().push_pack(
348 "https://hub.example.com/repos/r1", _make_signing(), bundle, "main", False
349 )
350 req = m.call_args[0][0]
351 assert req.full_url == "https://hub.example.com/repos/r1/push"
352 sent = msgpack.unpackb(req.data, raw=False)
353 assert sent["branch"] == "main"
354 assert sent["force"] is False
355 assert result["ok"] is True
356
357 def test_force_flag_sent(self) -> None:
358 push_body = _mp({"ok": True, "message": "", "branch_heads": {}})
359 mock_resp = _mock_response(push_body)
360 bundle: PackBundle = {}
361 with unittest.mock.patch("muse.core.transport._open_url", return_value=mock_resp) as m:
362 HttpTransport().push_pack("https://hub.example.com/r", None, bundle, "main", True)
363 req = m.call_args[0][0]
364 sent = msgpack.unpackb(req.data, raw=False)
365 assert sent["force"] is True
366
367 def test_push_rejected_raises_transport_error(self) -> None:
368 with unittest.mock.patch(
369 "muse.core.transport._open_url", side_effect=_http_error(409, b"non-fast-forward")
370 ):
371 with pytest.raises(TransportError) as exc_info:
372 HttpTransport().push_pack(
373 "https://hub.example.com/r", None, {}, "main", False
374 )
375 assert exc_info.value.status_code == 409
376
377
378 # ---------------------------------------------------------------------------
379 # HttpTransport._build_request — credential security and loopback allowlist
380 # ---------------------------------------------------------------------------
381
382
383 class TestBuildRequest:
384 """_build_request enforces HTTPS for non-loopback URLs with signing identity."""
385
386 def _build(self, url: str, with_signing: bool = True) -> urllib.request.Request:
387 signing = _make_signing() if with_signing else None
388 return HttpTransport()._build_request("GET", url, signing)
389
390 # ── Loopback hosts allowed over plain HTTP ────────────────────────────
391
392 def test_localhost_http_with_token_allowed(self) -> None:
393 req = self._build("http://localhost:10003/repo/refs")
394 assert req.get_header("Authorization").startswith("MSign ")
395
396 def test_127_0_0_1_http_with_token_allowed(self) -> None:
397 req = self._build("http://127.0.0.1:10003/repo/refs")
398 assert req.get_header("Authorization").startswith("MSign ")
399
400 def test_ipv6_loopback_http_with_token_allowed(self) -> None:
401 req = self._build("http://[::1]:10003/repo/refs")
402 assert req.get_header("Authorization").startswith("MSign ")
403
404 def test_host_docker_internal_http_with_token_allowed(self) -> None:
405 """host.docker.internal is Docker Desktop's alias for the host loopback.
406
407 Agent swarms run inside Docker and use http://host.docker.internal:10003
408 to reach a local MuseHub instance. Credentials must be sent over this
409 plain-HTTP connection — the traffic never leaves the machine.
410 """
411 req = self._build("http://host.docker.internal:10003/gabriel/repo/refs")
412 assert req.get_header("Authorization").startswith("MSign ")
413
414 def test_https_any_host_with_token_allowed(self) -> None:
415 req = self._build("https://musehub.ai/gabriel/repo/refs")
416 assert req.get_header("Authorization").startswith("MSign ")
417
418 # ── Non-loopback HTTP with token must be rejected ─────────────────────
419
420 def test_non_loopback_http_token_raises_transport_error(self) -> None:
421 with pytest.raises(TransportError, match="non-HTTPS"):
422 self._build("http://musehub.ai/gabriel/repo/refs")
423
424 def test_arbitrary_hostname_http_token_raises(self) -> None:
425 with pytest.raises(TransportError, match="non-HTTPS"):
426 self._build("http://evil.example.com/steal")
427
428 def test_localhost_lookalike_http_token_raises(self) -> None:
429 """'localhost.evil.com' must NOT be mistaken for the loopback interface."""
430 with pytest.raises(TransportError, match="non-HTTPS"):
431 self._build("http://localhost.evil.com/repo/refs")
432
433 def test_host_docker_internal_lookalike_http_token_raises(self) -> None:
434 """'host.docker.internal.evil.com' must not bypass the check."""
435 with pytest.raises(TransportError, match="non-HTTPS"):
436 self._build("http://host.docker.internal.evil.com/repo")
437
438 # ── No token — scheme restriction does not apply ──────────────────────
439
440 def test_non_loopback_http_without_token_allowed(self) -> None:
441 req = self._build("http://musehub.ai/public/repo/refs", with_signing=False)
442 assert req.get_header("Authorization") is None
443
444 # ── Request structure ─────────────────────────────────────────────────
445
446 def test_accept_header_always_set(self) -> None:
447 req = self._build("https://musehub.ai/repo/refs")
448 assert "msgpack" in req.get_header("Accept")
449
450 def test_method_preserved(self) -> None:
451 req = HttpTransport()._build_request("POST", "https://musehub.ai/x", None)
452 assert req.get_method() == "POST"
453
454 def test_body_sets_content_type(self) -> None:
455 req = HttpTransport()._build_request(
456 "POST", "https://musehub.ai/x", None, body_bytes=b"data"
457 )
458 assert req.get_header("Content-type") == "application/x-msgpack"
459
460 def test_no_body_omits_content_type(self) -> None:
461 req = self._build("https://musehub.ai/x", with_signing=False)
462 assert req.get_header("Content-type") is None
File History 1 commit
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 150 days ago