"""TDD tests for HttpTransport.hub_json and HttpTransport.hub_bytes.""" from __future__ import annotations import json import unittest.mock import pytest from muse.core.transport import HttpTransport, SigningIdentity, TransportError # --------------------------------------------------------------------------- # Helpers (mirror test_core_transport.py patterns) # --------------------------------------------------------------------------- def _make_signing() -> SigningIdentity: from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey return SigningIdentity(handle="testuser", private_key=Ed25519PrivateKey.generate()) def _mock_httpx_client_resp(body: bytes, status: int = 200) -> unittest.mock.MagicMock: resp = unittest.mock.MagicMock() resp.status_code = status resp.content = body resp.text = body.decode("utf-8", errors="replace") client = unittest.mock.MagicMock() client.is_closed = False client.request = unittest.mock.MagicMock(return_value=resp) client.__enter__ = unittest.mock.MagicMock(return_value=client) client.__exit__ = unittest.mock.MagicMock(return_value=False) return client # --------------------------------------------------------------------------- # hub_json # --------------------------------------------------------------------------- class TestHubJson: def test_returns_parsed_dict(self) -> None: payload = {"muse_version": "0.1.0", "mist_id": "abc123", "ok": True} client = _mock_httpx_client_resp(json.dumps(payload).encode()) with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod: mock_mod.Client = unittest.mock.MagicMock(return_value=client) result = HttpTransport().hub_json("GET", "https://hub.example.com/mist/abc123", None) assert result == payload def test_post_with_body_sends_json_bytes(self) -> None: client = _mock_httpx_client_resp(b'{"ok": true}') body = {"title": "hello", "content": "world"} with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod: mock_mod.Client = unittest.mock.MagicMock(return_value=client) HttpTransport().hub_json("POST", "https://hub.example.com/mists", None, body=body) call_kwargs = client.request.call_args sent_body = call_kwargs.kwargs.get("content") or call_kwargs[1].get("content") assert sent_body == json.dumps(body).encode() def test_post_with_body_sets_content_type_json(self) -> None: client = _mock_httpx_client_resp(b'{"ok": true}') with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod: mock_mod.Client = unittest.mock.MagicMock(return_value=client) HttpTransport().hub_json( "POST", "https://hub.example.com/mists", None, body={"x": 1} ) headers = client.request.call_args.kwargs.get("headers", {}) ct = headers.get("Content-Type") or headers.get("content-type", "") assert "application/json" in ct def test_accept_json_header_sent(self) -> None: client = _mock_httpx_client_resp(b'{"ok": true}') with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod: mock_mod.Client = unittest.mock.MagicMock(return_value=client) HttpTransport().hub_json("GET", "https://hub.example.com/mist/x", None) headers = client.request.call_args.kwargs.get("headers", {}) accept = headers.get("Accept") or headers.get("accept", "") assert "application/json" in accept def test_msign_auth_header_sent_when_signing_provided(self) -> None: signing = _make_signing() client = _mock_httpx_client_resp(b'{"ok": true}') with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod: mock_mod.Client = unittest.mock.MagicMock(return_value=client) HttpTransport().hub_json("GET", "https://hub.example.com/mist/x", signing) headers = client.request.call_args.kwargs.get("headers", {}) auth = headers.get("Authorization") or headers.get("authorization", "") assert auth.startswith('MSign handle="testuser"') def test_no_auth_header_when_signing_is_none(self) -> None: client = _mock_httpx_client_resp(b'{"ok": true}') with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod: mock_mod.Client = unittest.mock.MagicMock(return_value=client) HttpTransport().hub_json("GET", "https://hub.example.com/mist/x", None) headers = client.request.call_args.kwargs.get("headers", {}) auth = headers.get("Authorization") or headers.get("authorization") assert auth is None def test_non_dict_json_response_returns_empty_dict(self) -> None: client = _mock_httpx_client_resp(b'["a", "b", "c"]') with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod: mock_mod.Client = unittest.mock.MagicMock(return_value=client) result = HttpTransport().hub_json("GET", "https://hub.example.com/mist/x", None) assert result == {} def test_invalid_json_raises_transport_error(self) -> None: client = _mock_httpx_client_resp(b"not json at all") with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod: mock_mod.Client = unittest.mock.MagicMock(return_value=client) with pytest.raises(TransportError) as exc_info: HttpTransport().hub_json("GET", "https://hub.example.com/mist/x", None) assert "invalid json" in str(exc_info.value).lower() def test_http_404_raises_transport_error(self) -> None: client = _mock_httpx_client_resp(b"Not Found", status=404) with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod: mock_mod.Client = unittest.mock.MagicMock(return_value=client) with pytest.raises(TransportError) as exc_info: HttpTransport().hub_json("GET", "https://hub.example.com/mist/missing", None) assert exc_info.value.status_code == 404 def test_http_401_raises_transport_error(self) -> None: client = _mock_httpx_client_resp(b"Unauthorized", status=401) with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod: mock_mod.Client = unittest.mock.MagicMock(return_value=client) with pytest.raises(TransportError) as exc_info: HttpTransport().hub_json("POST", "https://hub.example.com/mists", None, body={}) assert exc_info.value.status_code == 401 def test_http_500_raises_transport_error(self) -> None: client = _mock_httpx_client_resp(b"Server Error", status=500) with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod: mock_mod.Client = unittest.mock.MagicMock(return_value=client) with pytest.raises(TransportError) as exc_info: HttpTransport().hub_json("GET", "https://hub.example.com/mist/x", None) assert exc_info.value.status_code == 500 def test_network_error_raises_transport_error_with_code_0(self) -> None: client = unittest.mock.MagicMock() client.is_closed = False client.request = unittest.mock.MagicMock(side_effect=Exception("connection refused")) client.__enter__ = unittest.mock.MagicMock(return_value=client) client.__exit__ = unittest.mock.MagicMock(return_value=False) with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod: mock_mod.Client = unittest.mock.MagicMock(return_value=client) with pytest.raises(TransportError) as exc_info: HttpTransport().hub_json("GET", "https://hub.example.com/mist/x", None) assert exc_info.value.status_code == 0 def test_get_with_no_body_sends_no_content(self) -> None: client = _mock_httpx_client_resp(b'{"ok": true}') with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod: mock_mod.Client = unittest.mock.MagicMock(return_value=client) HttpTransport().hub_json("GET", "https://hub.example.com/mist/x", None) sent_body = client.request.call_args.kwargs.get("content") assert sent_body is None def test_uses_httpx_client_not_urllib(self) -> None: """hub_json must go through _httpx_mod.Client, not urllib.""" client = _mock_httpx_client_resp(b'{"ok": true}') with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod: mock_mod.Client = unittest.mock.MagicMock(return_value=client) with unittest.mock.patch("muse.core.transport._open_url") as mock_urllib: HttpTransport().hub_json("GET", "https://hub.example.com/mist/x", None) mock_mod.Client.assert_called_once() mock_urllib.assert_not_called() # --------------------------------------------------------------------------- # hub_bytes # --------------------------------------------------------------------------- class TestHubBytes: def test_returns_raw_bytes(self) -> None: expected = b"# Muse Basics\nHello world" client = _mock_httpx_client_resp(expected) with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod: mock_mod.Client = unittest.mock.MagicMock(return_value=client) result = HttpTransport().hub_bytes("https://hub.example.com/mist/abc/raw", None) assert result == expected def test_sends_get_method(self) -> None: client = _mock_httpx_client_resp(b"data") with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod: mock_mod.Client = unittest.mock.MagicMock(return_value=client) HttpTransport().hub_bytes("https://hub.example.com/mist/abc/raw", None) method_called = client.request.call_args[0][0] assert method_called == "GET" def test_accept_star_header_sent(self) -> None: client = _mock_httpx_client_resp(b"data") with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod: mock_mod.Client = unittest.mock.MagicMock(return_value=client) HttpTransport().hub_bytes("https://hub.example.com/mist/abc/raw", None) headers = client.request.call_args.kwargs.get("headers", {}) accept = headers.get("Accept") or headers.get("accept", "") assert accept == "*/*" def test_msign_header_sent_when_signing_provided(self) -> None: signing = _make_signing() client = _mock_httpx_client_resp(b"data") with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod: mock_mod.Client = unittest.mock.MagicMock(return_value=client) HttpTransport().hub_bytes("https://hub.example.com/mist/abc/raw", signing) headers = client.request.call_args.kwargs.get("headers", {}) auth = headers.get("Authorization") or headers.get("authorization", "") assert auth.startswith('MSign handle="testuser"') def test_no_auth_header_when_signing_is_none(self) -> None: client = _mock_httpx_client_resp(b"data") with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod: mock_mod.Client = unittest.mock.MagicMock(return_value=client) HttpTransport().hub_bytes("https://hub.example.com/mist/abc/raw", None) headers = client.request.call_args.kwargs.get("headers", {}) auth = headers.get("Authorization") or headers.get("authorization") assert auth is None def test_http_404_raises_transport_error(self) -> None: client = _mock_httpx_client_resp(b"Not Found", status=404) with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod: mock_mod.Client = unittest.mock.MagicMock(return_value=client) with pytest.raises(TransportError) as exc_info: HttpTransport().hub_bytes("https://hub.example.com/mist/missing/raw", None) assert exc_info.value.status_code == 404 def test_network_error_raises_transport_error_with_code_0(self) -> None: client = unittest.mock.MagicMock() client.is_closed = False client.request = unittest.mock.MagicMock(side_effect=Exception("timed out")) client.__enter__ = unittest.mock.MagicMock(return_value=client) client.__exit__ = unittest.mock.MagicMock(return_value=False) with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod: mock_mod.Client = unittest.mock.MagicMock(return_value=client) with pytest.raises(TransportError) as exc_info: HttpTransport().hub_bytes("https://hub.example.com/mist/abc/raw", None) assert exc_info.value.status_code == 0 def test_uses_httpx_client_not_urllib(self) -> None: """hub_bytes must go through _httpx_mod.Client, not urllib.""" client = _mock_httpx_client_resp(b"data") with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod: mock_mod.Client = unittest.mock.MagicMock(return_value=client) with unittest.mock.patch("muse.core.transport._open_url") as mock_urllib: HttpTransport().hub_bytes("https://hub.example.com/mist/abc/raw", None) mock_mod.Client.assert_called_once() mock_urllib.assert_not_called() def test_returns_empty_bytes_on_empty_200_response(self) -> None: client = _mock_httpx_client_resp(b"") with unittest.mock.patch("muse.core.transport._httpx_mod") as mock_mod: mock_mod.Client = unittest.mock.MagicMock(return_value=client) result = HttpTransport().hub_bytes("https://hub.example.com/mist/empty/raw", None) assert result == b""