test_domains_publish.py
python
| 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 typer.testing import CliRunner |
| 35 | |
| 36 | from muse._version import __version__ |
| 37 | from muse.cli.app import cli |
| 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" |