gabriel / muse public
test_transport_hub_json.py python
225 lines 11.2 KB
Raw
sha256:79ffe87f5fe2ec146e35f05521218bbf54dffdb0440c07f970bad05f16efb89f chore: merge main — carry all urllib/typing/test fixes from dev Sonnet 4.6 minor ⚠ breaking 19 days ago
1 """TDD tests for HttpTransport.hub_json and HttpTransport.hub_bytes."""
2
3 from __future__ import annotations
4
5 import json
6 import unittest.mock
7
8 import pytest
9
10 from muse.core.transport import HttpTransport, SigningIdentity, TransportError
11
12
13 # ---------------------------------------------------------------------------
14 # Helpers
15 # ---------------------------------------------------------------------------
16
17
18 def _make_signing() -> SigningIdentity:
19 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
20 return SigningIdentity(handle="testuser", private_key=Ed25519PrivateKey.generate())
21
22
23 def _mock_urllib_do(body: bytes, status: int = 200) -> "Callable[..., bytes]":
24 """Patch _urllib_do to return body or raise TransportError for non-2xx."""
25 calls: list[tuple[str, str, dict, bytes | None]] = []
26
27 def _side_effect(method: str, url: str, headers: "dict[str, str]", data: "bytes | None" = None, **kwargs: "str | int | bool") -> bytes:
28 calls.append((method, url, headers, data))
29 if status >= 400:
30 raise TransportError(f"HTTP {status}", status)
31 return body
32
33 _side_effect.calls = calls
34 return _side_effect
35
36
37 # ---------------------------------------------------------------------------
38 # hub_json
39 # ---------------------------------------------------------------------------
40
41
42 class TestHubJson:
43 def test_returns_parsed_dict(self) -> None:
44 payload = {"muse_version": "0.1.0", "mist_id": "abc123", "ok": True}
45 do = _mock_urllib_do(json.dumps(payload).encode())
46 with unittest.mock.patch("muse.core.transport._urllib_do", side_effect=do):
47 result = HttpTransport().hub_json("GET", "https://hub.example.com/mist/abc123", None)
48 assert result == payload
49
50 def test_post_with_body_sends_json_bytes(self) -> None:
51 body = {"title": "hello", "content": "world"}
52 do = _mock_urllib_do(b'{"ok": true}')
53 with unittest.mock.patch("muse.core.transport._urllib_do", side_effect=do):
54 HttpTransport().hub_json("POST", "https://hub.example.com/mists", None, body=body)
55 _, _, _, sent_data = do.calls[0]
56 assert sent_data == json.dumps(body).encode()
57
58 def test_post_with_body_sets_content_type_json(self) -> None:
59 do = _mock_urllib_do(b'{"ok": true}')
60 with unittest.mock.patch("muse.core.transport._urllib_do", side_effect=do):
61 HttpTransport().hub_json("POST", "https://hub.example.com/mists", None, body={"x": 1})
62 _, _, headers, _ = do.calls[0]
63 ct = headers.get("Content-Type") or headers.get("content-type", "")
64 assert "application/json" in ct
65
66 def test_accept_json_header_sent(self) -> None:
67 do = _mock_urllib_do(b'{"ok": true}')
68 with unittest.mock.patch("muse.core.transport._urllib_do", side_effect=do):
69 HttpTransport().hub_json("GET", "https://hub.example.com/mist/x", None)
70 _, _, headers, _ = do.calls[0]
71 accept = headers.get("Accept") or headers.get("accept", "")
72 assert "application/json" in accept
73
74 def test_msign_auth_header_sent_when_signing_provided(self) -> None:
75 signing = _make_signing()
76 do = _mock_urllib_do(b'{"ok": true}')
77 with unittest.mock.patch("muse.core.transport._urllib_do", side_effect=do):
78 with unittest.mock.patch("muse.core.hub_trust.check_and_pin"):
79 HttpTransport().hub_json("GET", "https://hub.example.com/mist/x", signing)
80 _, _, headers, _ = do.calls[0]
81 auth = headers.get("Authorization") or headers.get("authorization", "")
82 assert auth.startswith('MSign handle="testuser"')
83
84 def test_no_auth_header_when_signing_is_none(self) -> None:
85 do = _mock_urllib_do(b'{"ok": true}')
86 with unittest.mock.patch("muse.core.transport._urllib_do", side_effect=do):
87 HttpTransport().hub_json("GET", "https://hub.example.com/mist/x", None)
88 _, _, headers, _ = do.calls[0]
89 auth = headers.get("Authorization") or headers.get("authorization")
90 assert auth is None
91
92 def test_non_dict_json_response_returns_empty_dict(self) -> None:
93 do = _mock_urllib_do(b'["a", "b", "c"]')
94 with unittest.mock.patch("muse.core.transport._urllib_do", side_effect=do):
95 result = HttpTransport().hub_json("GET", "https://hub.example.com/mist/x", None)
96 assert result == {}
97
98 def test_invalid_json_raises_transport_error(self) -> None:
99 do = _mock_urllib_do(b"not json at all")
100 with unittest.mock.patch("muse.core.transport._urllib_do", side_effect=do):
101 with pytest.raises(TransportError) as exc_info:
102 HttpTransport().hub_json("GET", "https://hub.example.com/mist/x", None)
103 assert "invalid json" in str(exc_info.value).lower()
104
105 def test_http_404_raises_transport_error(self) -> None:
106 do = _mock_urllib_do(b"Not Found", status=404)
107 with unittest.mock.patch("muse.core.transport._urllib_do", side_effect=do):
108 with pytest.raises(TransportError) as exc_info:
109 HttpTransport().hub_json("GET", "https://hub.example.com/mist/missing", None)
110 assert exc_info.value.status_code == 404
111
112 def test_http_401_raises_transport_error(self) -> None:
113 do = _mock_urllib_do(b"Unauthorized", status=401)
114 with unittest.mock.patch("muse.core.transport._urllib_do", side_effect=do):
115 with pytest.raises(TransportError) as exc_info:
116 HttpTransport().hub_json("POST", "https://hub.example.com/mists", None, body={})
117 assert exc_info.value.status_code == 401
118
119 def test_http_500_raises_transport_error(self) -> None:
120 do = _mock_urllib_do(b"Server Error", status=500)
121 with unittest.mock.patch("muse.core.transport._urllib_do", side_effect=do):
122 with pytest.raises(TransportError) as exc_info:
123 HttpTransport().hub_json("GET", "https://hub.example.com/mist/x", None)
124 assert exc_info.value.status_code == 500
125
126 def test_network_error_raises_transport_error_with_code_0(self) -> None:
127 def _fail(method: str, url: str, headers: "dict[str, str]", data: "bytes | None" = None, **kwargs: "str | int | bool") -> bytes:
128 raise TransportError("connection refused", 0)
129 with unittest.mock.patch("muse.core.transport._urllib_do", side_effect=_fail):
130 with pytest.raises(TransportError) as exc_info:
131 HttpTransport().hub_json("GET", "https://hub.example.com/mist/x", None)
132 assert exc_info.value.status_code == 0
133
134 def test_get_with_no_body_sends_no_content(self) -> None:
135 do = _mock_urllib_do(b'{"ok": true}')
136 with unittest.mock.patch("muse.core.transport._urllib_do", side_effect=do):
137 HttpTransport().hub_json("GET", "https://hub.example.com/mist/x", None)
138 _, _, _, sent_data = do.calls[0]
139 assert sent_data is None
140
141 def test_uses_urllib_not_open_url(self) -> None:
142 """hub_json must go through _urllib_do, not _open_url (the fetch seam)."""
143 do = _mock_urllib_do(b'{"ok": true}')
144 with unittest.mock.patch("muse.core.transport._urllib_do", side_effect=do) as mock_do:
145 with unittest.mock.patch("muse.core.transport._open_url") as mock_open:
146 HttpTransport().hub_json("GET", "https://hub.example.com/mist/x", None)
147 assert mock_do.called
148 mock_open.assert_not_called()
149
150
151 # ---------------------------------------------------------------------------
152 # hub_bytes
153 # ---------------------------------------------------------------------------
154
155
156 class TestHubBytes:
157 def test_returns_raw_bytes(self) -> None:
158 expected = b"# Muse Basics\nHello world"
159 do = _mock_urllib_do(expected)
160 with unittest.mock.patch("muse.core.transport._urllib_do", side_effect=do):
161 result = HttpTransport().hub_bytes("https://hub.example.com/mist/abc/raw", None)
162 assert result == expected
163
164 def test_sends_get_method(self) -> None:
165 do = _mock_urllib_do(b"data")
166 with unittest.mock.patch("muse.core.transport._urllib_do", side_effect=do):
167 HttpTransport().hub_bytes("https://hub.example.com/mist/abc/raw", None)
168 method, _, _, _ = do.calls[0]
169 assert method == "GET"
170
171 def test_accept_star_header_sent(self) -> None:
172 do = _mock_urllib_do(b"data")
173 with unittest.mock.patch("muse.core.transport._urllib_do", side_effect=do):
174 HttpTransport().hub_bytes("https://hub.example.com/mist/abc/raw", None)
175 _, _, headers, _ = do.calls[0]
176 accept = headers.get("Accept") or headers.get("accept", "")
177 assert accept == "*/*"
178
179 def test_msign_header_sent_when_signing_provided(self) -> None:
180 signing = _make_signing()
181 do = _mock_urllib_do(b"data")
182 with unittest.mock.patch("muse.core.transport._urllib_do", side_effect=do):
183 with unittest.mock.patch("muse.core.hub_trust.check_and_pin"):
184 HttpTransport().hub_bytes("https://hub.example.com/mist/abc/raw", signing)
185 _, _, headers, _ = do.calls[0]
186 auth = headers.get("Authorization") or headers.get("authorization", "")
187 assert auth.startswith('MSign handle="testuser"')
188
189 def test_no_auth_header_when_signing_is_none(self) -> None:
190 do = _mock_urllib_do(b"data")
191 with unittest.mock.patch("muse.core.transport._urllib_do", side_effect=do):
192 HttpTransport().hub_bytes("https://hub.example.com/mist/abc/raw", None)
193 _, _, headers, _ = do.calls[0]
194 auth = headers.get("Authorization") or headers.get("authorization")
195 assert auth is None
196
197 def test_http_404_raises_transport_error(self) -> None:
198 do = _mock_urllib_do(b"Not Found", status=404)
199 with unittest.mock.patch("muse.core.transport._urllib_do", side_effect=do):
200 with pytest.raises(TransportError) as exc_info:
201 HttpTransport().hub_bytes("https://hub.example.com/mist/missing/raw", None)
202 assert exc_info.value.status_code == 404
203
204 def test_network_error_raises_transport_error_with_code_0(self) -> None:
205 def _fail(method: str, url: str, headers: "dict[str, str]", data: "bytes | None" = None, **kwargs: "str | int | bool") -> bytes:
206 raise TransportError("timed out", 0)
207 with unittest.mock.patch("muse.core.transport._urllib_do", side_effect=_fail):
208 with pytest.raises(TransportError) as exc_info:
209 HttpTransport().hub_bytes("https://hub.example.com/mist/abc/raw", None)
210 assert exc_info.value.status_code == 0
211
212 def test_uses_urllib_not_open_url(self) -> None:
213 """hub_bytes must go through _urllib_do, not _open_url (the fetch seam)."""
214 do = _mock_urllib_do(b"data")
215 with unittest.mock.patch("muse.core.transport._urllib_do", side_effect=do) as mock_do:
216 with unittest.mock.patch("muse.core.transport._open_url") as mock_open:
217 HttpTransport().hub_bytes("https://hub.example.com/mist/abc/raw", None)
218 assert mock_do.called
219 mock_open.assert_not_called()
220
221 def test_returns_empty_bytes_on_empty_200_response(self) -> None:
222 do = _mock_urllib_do(b"")
223 with unittest.mock.patch("muse.core.transport._urllib_do", side_effect=do):
224 result = HttpTransport().hub_bytes("https://hub.example.com/mist/empty/raw", None)
225 assert result == b""
File History 2 commits
sha256:79ffe87f5fe2ec146e35f05521218bbf54dffdb0440c07f970bad05f16efb89f chore: merge main — carry all urllib/typing/test fixes from dev Sonnet 4.6 minor ⚠ 19 days ago
sha256:0bea7600d1eee83e87950be49933b1006fa9dc2c71e7c4ee748d324f61138156 chore: bump version to 0.2.0rc11; fix typing audit violatio… Sonnet 4.6 minor ⚠ 19 days ago