gabriel / muse public
test_domains_publish.py python
379 lines 14.2 KB
Raw
86000da9 fix: replace typer CliRunner with argparse-compatible test helper Gabriel Cardona <gabriel@tellurstori.com> 1d ago
1 """Tests for ``muse domains publish`` — the MuseHub marketplace publisher.
2
3 Covers:
4 Unit tests for ``_post_json`` helper:
5 - Sends correct HTTP method, URL, headers, and JSON body.
6 - Returns a typed _PublishResponse on 200.
7 - Raises HTTPError on non-2xx.
8 - Raises ValueError on non-object JSON.
9
10 CLI integration tests (via Typer CliRunner, no real HTTP):
11 - Successful publish with --capabilities JSON emits domain_id / scoped_id.
12 - Successful publish with --json emits machine-readable JSON.
13 - Missing required args → UsageError (non-zero exit).
14 - No auth token → exit 1 with clear message.
15 - HTTP 409 conflict → exit 1 with "already registered" message.
16 - HTTP 401 → exit 1 with "Authentication failed" message.
17 - Network error (URLError) → exit 1 with "Could not reach" message.
18 - Non-JSON response body → exit 1 with "Unexpected response" message.
19 - Capabilities derived from plugin schema when --capabilities is omitted.
20 """
21 from __future__ import annotations
22
23 import http.client
24 import io
25 import json
26 import pathlib
27 import urllib.error
28 import urllib.request
29 import urllib.response
30 import unittest.mock
31 from typing import TYPE_CHECKING
32
33 import pytest
34 from tests.cli_test_helper import CliRunner
35
36 from muse._version import __version__
37 cli = None # argparse migration — CliRunner ignores this arg
38 from muse.cli.commands.domains import _post_json, _PublishPayload, _Capabilities, _DimensionDef
39
40 if TYPE_CHECKING:
41 pass
42
43 runner = CliRunner()
44
45 # ---------------------------------------------------------------------------
46 # Fixture: minimal Muse repo with auth token
47 # ---------------------------------------------------------------------------
48
49 _BASE_CAPS_JSON = json.dumps({
50 "dimensions": [{"name": "notes", "description": "Note events"}],
51 "artifact_types": ["mid"],
52 "merge_semantics": "three_way",
53 "supported_commands": ["commit", "diff"],
54 })
55
56 _REQUIRED_ARGS = [
57 "domains", "publish",
58 "--author", "testuser",
59 "--slug", "genomics",
60 "--name", "Genomics",
61 "--description", "Version DNA sequences",
62 "--viewer-type", "genome",
63 "--capabilities", _BASE_CAPS_JSON,
64 "--hub", "https://hub.test",
65 ]
66
67
68 @pytest.fixture()
69 def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
70 """Minimal .muse/ repo; auth token is mocked via get_auth_token."""
71 muse_dir = tmp_path / ".muse"
72 (muse_dir / "refs" / "heads").mkdir(parents=True)
73 (muse_dir / "objects").mkdir()
74 (muse_dir / "commits").mkdir()
75 (muse_dir / "snapshots").mkdir()
76 (muse_dir / "repo.json").write_text(
77 json.dumps({"repo_id": "test-repo", "schema_version": __version__, "domain": "midi"})
78 )
79 (muse_dir / "HEAD").write_text("ref: refs/heads/main\n")
80 monkeypatch.chdir(tmp_path)
81 monkeypatch.setattr("muse.cli.commands.domains.get_auth_token", lambda *a, **kw: "test-token-abc")
82 return tmp_path
83
84
85 @pytest.fixture()
86 def repo_no_token(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
87 """Minimal .muse/ repo where get_auth_token returns None (no token)."""
88 muse_dir = tmp_path / ".muse"
89 (muse_dir / "refs" / "heads").mkdir(parents=True)
90 (muse_dir / "objects").mkdir()
91 (muse_dir / "commits").mkdir()
92 (muse_dir / "snapshots").mkdir()
93 (muse_dir / "repo.json").write_text(
94 json.dumps({"repo_id": "test-repo", "schema_version": __version__, "domain": "midi"})
95 )
96 (muse_dir / "HEAD").write_text("ref: refs/heads/main\n")
97 monkeypatch.chdir(tmp_path)
98 monkeypatch.setattr("muse.cli.commands.domains.get_auth_token", lambda *a, **kw: None)
99 return tmp_path
100
101
102 # ---------------------------------------------------------------------------
103 # Helper: build a mock urllib response
104 # ---------------------------------------------------------------------------
105
106
107 def _mock_urlopen(response_body: str | bytes, status: int = 200) -> unittest.mock.MagicMock:
108 """Return a context-manager mock that yields a fake HTTP response."""
109 if isinstance(response_body, str):
110 response_body = response_body.encode()
111 mock_resp = unittest.mock.MagicMock()
112 mock_resp.read.return_value = response_body
113 mock_resp.__enter__ = unittest.mock.MagicMock(return_value=mock_resp)
114 mock_resp.__exit__ = unittest.mock.MagicMock(return_value=False)
115 return mock_resp
116
117
118 # ---------------------------------------------------------------------------
119 # Unit tests for _post_json
120 # ---------------------------------------------------------------------------
121
122
123 def test_post_json_sends_correct_request() -> None:
124 """_post_json should POST JSON to the given URL with Auth header."""
125 payload = _PublishPayload(
126 author_slug="alice",
127 slug="spatial",
128 display_name="Spatial 3D",
129 description="Version 3D scenes",
130 capabilities=_Capabilities(
131 dimensions=[_DimensionDef(name="geometry", description="Mesh data")],
132 artifact_types=["glb"],
133 merge_semantics="three_way",
134 supported_commands=["commit"],
135 ),
136 viewer_type="spatial",
137 version="0.1.0",
138 )
139
140 captured: list[urllib.request.Request] = []
141
142 def _fake_urlopen(
143 req: urllib.request.Request, timeout: float | None = None
144 ) -> unittest.mock.MagicMock:
145 captured.append(req)
146 mock_resp = _mock_urlopen(json.dumps({"domain_id": "d1", "scoped_id": "@alice/spatial", "manifest_hash": "abc"}))
147 return mock_resp
148
149 with unittest.mock.patch("urllib.request.urlopen", side_effect=_fake_urlopen):
150 result = _post_json("https://hub.test/api/v1/domains", payload, "tok-123")
151
152 assert len(captured) == 1
153 req = captured[0]
154 assert req.get_method() == "POST"
155 assert req.full_url == "https://hub.test/api/v1/domains"
156 assert req.get_header("Authorization") == "Bearer tok-123"
157 assert req.get_header("Content-type") == "application/json"
158 assert req.data is not None
159 body = json.loads(req.data.decode())
160 assert body["author_slug"] == "alice"
161 assert body["slug"] == "spatial"
162
163 assert result["scoped_id"] == "@alice/spatial"
164
165
166 def test_post_json_raises_on_non_object_response() -> None:
167 """_post_json should raise ValueError when server returns a JSON array."""
168 payload = _PublishPayload(
169 author_slug="bob", slug="s", display_name="S", description="d",
170 capabilities=_Capabilities(), viewer_type="v", version="0.1.0",
171 )
172 mock_resp = _mock_urlopen(json.dumps(["not", "an", "object"]))
173 with unittest.mock.patch("urllib.request.urlopen", return_value=mock_resp):
174 with pytest.raises(ValueError, match="Expected JSON object"):
175 _post_json("https://hub.test/api/v1/domains", payload, "tok")
176
177
178 def test_post_json_raises_http_error() -> None:
179 """_post_json should propagate HTTPError from urlopen."""
180 payload = _PublishPayload(
181 author_slug="bob", slug="s", display_name="S", description="d",
182 capabilities=_Capabilities(), viewer_type="v", version="0.1.0",
183 )
184 err = urllib.error.HTTPError(
185 url="https://hub.test/api/v1/domains",
186 code=409,
187 msg="Conflict",
188 hdrs=http.client.HTTPMessage(),
189 fp=io.BytesIO(b'{"error": "already_exists"}'),
190 )
191 with unittest.mock.patch("urllib.request.urlopen", side_effect=err):
192 with pytest.raises(urllib.error.HTTPError):
193 _post_json("https://hub.test/api/v1/domains", payload, "tok")
194
195
196 # ---------------------------------------------------------------------------
197 # CLI integration tests
198 # ---------------------------------------------------------------------------
199
200
201 def test_publish_success(repo: pathlib.Path) -> None:
202 """Successful publish prints domain scoped_id and manifest_hash."""
203 server_resp = json.dumps({
204 "domain_id": "dom-001",
205 "scoped_id": "@testuser/genomics",
206 "manifest_hash": "sha256:abc123",
207 })
208 mock_resp = _mock_urlopen(server_resp)
209
210 with unittest.mock.patch("urllib.request.urlopen", return_value=mock_resp):
211 result = runner.invoke(cli, _REQUIRED_ARGS)
212
213 assert result.exit_code == 0, result.output
214 assert "@testuser/genomics" in result.output
215 assert "sha256:abc123" in result.output
216
217
218 def test_publish_json_flag(repo: pathlib.Path) -> None:
219 """--json flag emits machine-readable JSON."""
220 server_resp = json.dumps({
221 "domain_id": "dom-002",
222 "scoped_id": "@testuser/genomics",
223 "manifest_hash": "sha256:def456",
224 })
225 mock_resp = _mock_urlopen(server_resp)
226
227 with unittest.mock.patch("urllib.request.urlopen", return_value=mock_resp):
228 result = runner.invoke(cli, _REQUIRED_ARGS + ["--json"])
229
230 assert result.exit_code == 0, result.output
231 data = json.loads(result.output.strip())
232 assert data["scoped_id"] == "@testuser/genomics"
233
234
235 def test_publish_no_token(repo_no_token: pathlib.Path) -> None:
236 """Missing auth token should exit 1 with clear message."""
237 result = runner.invoke(cli, _REQUIRED_ARGS)
238 assert result.exit_code != 0
239 assert "token" in result.output.lower() or "auth" in result.output.lower()
240
241
242 def test_publish_http_409_conflict(repo: pathlib.Path) -> None:
243 """HTTP 409 should exit 1 with 'already registered' message."""
244 err = urllib.error.HTTPError(
245 url="https://hub.test/api/v1/domains",
246 code=409,
247 msg="Conflict",
248 hdrs=http.client.HTTPMessage(),
249 fp=io.BytesIO(b'{"error": "already_exists"}'),
250 )
251 with unittest.mock.patch("urllib.request.urlopen", side_effect=err):
252 result = runner.invoke(cli, _REQUIRED_ARGS)
253
254 assert result.exit_code != 0
255 assert "already registered" in result.output.lower() or "409" in result.output
256
257
258 def test_publish_http_401_unauthorized(repo: pathlib.Path) -> None:
259 """HTTP 401 should exit 1 with authentication message."""
260 err = urllib.error.HTTPError(
261 url="https://hub.test/api/v1/domains",
262 code=401,
263 msg="Unauthorized",
264 hdrs=http.client.HTTPMessage(),
265 fp=io.BytesIO(b'{"error": "unauthorized"}'),
266 )
267 with unittest.mock.patch("urllib.request.urlopen", side_effect=err):
268 result = runner.invoke(cli, _REQUIRED_ARGS)
269
270 assert result.exit_code != 0
271 assert "authentication" in result.output.lower() or "token" in result.output.lower()
272
273
274 def test_publish_network_error(repo: pathlib.Path) -> None:
275 """URLError (network failure) should exit 1 with 'Could not reach' message."""
276 err = urllib.error.URLError(reason="Connection refused")
277 with unittest.mock.patch("urllib.request.urlopen", side_effect=err):
278 result = runner.invoke(cli, _REQUIRED_ARGS)
279
280 assert result.exit_code != 0
281 assert "could not reach" in result.output.lower()
282
283
284 def test_publish_bad_json_response(repo: pathlib.Path) -> None:
285 """Non-JSON server response should exit 1 with 'Unexpected response' message."""
286 mock_resp = _mock_urlopen(b"not json at all")
287 with unittest.mock.patch("urllib.request.urlopen", return_value=mock_resp):
288 result = runner.invoke(cli, _REQUIRED_ARGS)
289
290 assert result.exit_code != 0
291 assert "unexpected" in result.output.lower()
292
293
294 def test_publish_missing_required_author(repo: pathlib.Path) -> None:
295 """Omitting --author should produce a non-zero exit and usage message."""
296 args = [a for a in _REQUIRED_ARGS if a != "--author" and a != "testuser"]
297 result = runner.invoke(cli, args)
298 assert result.exit_code != 0
299
300
301 def test_publish_missing_required_slug(repo: pathlib.Path) -> None:
302 """Omitting --slug should produce a non-zero exit."""
303 args = [a for a in _REQUIRED_ARGS if a != "--slug" and a != "genomics"]
304 result = runner.invoke(cli, args)
305 assert result.exit_code != 0
306
307
308 def test_publish_capabilities_from_plugin_schema(repo: pathlib.Path) -> None:
309 """When --capabilities is omitted, schema is derived from active domain plugin."""
310 # Remove --capabilities from args
311 args_no_caps = [
312 a for i, a in enumerate(_REQUIRED_ARGS)
313 if a not in ("--capabilities", _BASE_CAPS_JSON)
314 and not (i > 0 and _REQUIRED_ARGS[i - 1] == "--capabilities")
315 ]
316
317 server_resp = json.dumps({
318 "domain_id": "dom-plugin",
319 "scoped_id": "@testuser/genomics",
320 "manifest_hash": "sha256:plugin",
321 })
322 mock_resp = _mock_urlopen(server_resp)
323
324 with unittest.mock.patch("urllib.request.urlopen", return_value=mock_resp):
325 result = runner.invoke(cli, args_no_caps)
326
327 # Should succeed (midi plugin schema is available)
328 assert result.exit_code == 0, result.output
329 assert "@testuser/genomics" in result.output
330
331
332 def test_publish_invalid_capabilities_json(repo: pathlib.Path) -> None:
333 """--capabilities with invalid JSON should exit 1."""
334 bad_caps_args = [
335 "domains", "publish",
336 "--author", "testuser",
337 "--slug", "genomics",
338 "--name", "Genomics",
339 "--description", "Version DNA sequences",
340 "--viewer-type", "genome",
341 "--capabilities", "{not valid json",
342 "--hub", "https://hub.test",
343 ]
344 result = runner.invoke(cli, bad_caps_args)
345 assert result.exit_code != 0
346 assert "json" in result.output.lower()
347
348
349 def test_publish_http_500_server_error(repo: pathlib.Path) -> None:
350 """HTTP 5xx should exit 1 with HTTP error code shown."""
351 err = urllib.error.HTTPError(
352 url="https://hub.test/api/v1/domains",
353 code=500,
354 msg="Internal Server Error",
355 hdrs=http.client.HTTPMessage(),
356 fp=io.BytesIO(b'{"error": "server_error"}'),
357 )
358 with unittest.mock.patch("urllib.request.urlopen", side_effect=err):
359 result = runner.invoke(cli, _REQUIRED_ARGS)
360
361 assert result.exit_code != 0
362 assert "500" in result.output
363
364
365 def test_publish_custom_version(repo: pathlib.Path) -> None:
366 """--version flag is passed to the server payload."""
367 captured_bodies: list[bytes] = []
368
369 def _capture(req: urllib.request.Request, timeout: float | None = None) -> unittest.mock.MagicMock:
370 raw = req.data
371 captured_bodies.append(raw if raw is not None else b"")
372 return _mock_urlopen(json.dumps({"domain_id": "d", "scoped_id": "@testuser/genomics", "manifest_hash": "h"}))
373
374 with unittest.mock.patch("urllib.request.urlopen", side_effect=_capture):
375 result = runner.invoke(cli, _REQUIRED_ARGS + ["--version", "1.2.3"])
376
377 assert result.exit_code == 0, result.output
378 body = json.loads(captured_bodies[0])
379 assert body["version"] == "1.2.3"