"""Phase 6 tests: MCP tool layer for proposal reimagination. Covers: - _proposal_data serialiser includes all new Phase 1-5 fields - _simulation_data serialiser shape - execute_create_proposal forwards new kwargs - execute_get_proposal returns enriched proposal - execute_run_simulation runs and persists result - execute_get_simulation reads cached result, returns not_found when missing - execute_list_simulations returns all cached simulations - Dispatcher routing: musehub_create_proposal new fields, simulation tools - Tool schema: new tools present in MUSEHUB_WRITE_TOOLS """ import asyncio import os from datetime import datetime, timezone from unittest.mock import AsyncMock, MagicMock, patch import pytest from muse.core.types import blob_id, fake_id, short_id from musehub.types.json_types import JSONObject, JSONValue # ── helpers ──────────────────────────────────────────────────────────────────── def _sid(label: str) -> str: """Deterministic sha256-prefixed ID from a label string.""" return fake_id(label) def _uid() -> str: """Unique per-run short ID — use for repo/proposal slugs.""" return short_id(blob_id(os.urandom(16)), strip=True) def _now() -> datetime: return datetime.now(timezone.utc) # ── _proposal_data serialiser ───────────────────────────────────────────────── class TestProposalDataSerializer: def _make_proposal_response(self, **overrides: JSONValue) -> "ProposalResponse": from musehub.models.musehub import ProposalResponse defaults = dict( proposal_id=_sid("p1"), title="Test proposal", body="", state="open", from_branch="feat/x", to_branch="dev", author="gabriel", proposal_number=1, proposal_type="state_merge", is_draft=False, merge_strategy="overlay", merge_conditions={}, selective_domains=[], merge_commit_id=None, created_at=_now(), merged_at=None, reviewer_count=0, comment_count=0, approval_count=0, changes_requested_count=0, merge_readiness=None, dimensional_risk={}, head_commit_id=_sid("commit1"), blocked_by=[], blocks=[], is_blocked=False, latest_simulations={}, ) defaults.update(overrides) return ProposalResponse(**defaults) def test_core_fields_present(self) -> None: from musehub.mcp.write_tools.proposals import _proposal_data p = self._make_proposal_response() data = _proposal_data(p) for key in ("proposal_id", "title", "body", "state", "from_branch", "to_branch", "author"): assert key in data def test_new_fields_present(self) -> None: from musehub.mcp.write_tools.proposals import _proposal_data p = self._make_proposal_response( proposal_type="midi_evolution", is_draft=True, merge_strategy="weave", selective_domains=["audio"], blocked_by=[1], blocks=[3], is_blocked=True, latest_simulations={"conflict_scan": {"conflict_count": 0}}, ) data = _proposal_data(p) assert data["proposal_type"] == "midi_evolution" assert data["is_draft"] is True assert data["merge_strategy"] == "weave" assert "merge_conditions" in data assert data["selective_domains"] == ["audio"] assert data["blocked_by"] == [1] assert data["blocks"] == [3] assert data["is_blocked"] is True assert "conflict_scan" in data["latest_simulations"] def test_dates_iso_format(self) -> None: from musehub.mcp.write_tools.proposals import _proposal_data now = _now() p = self._make_proposal_response(created_at=now, merged_at=now) data = _proposal_data(p) assert data["created_at"] == now.isoformat() assert data["merged_at"] == now.isoformat() def test_merged_at_none_handled(self) -> None: from musehub.mcp.write_tools.proposals import _proposal_data p = self._make_proposal_response(merged_at=None) data = _proposal_data(p) assert data["merged_at"] is None # ── _simulation_data serialiser ──────────────────────────────────────────────── class TestSimulationDataSerializer: def _make_sim(self, **overrides: JSONValue) -> "SimulationResponse": from musehub.models.musehub import SimulationResponse defaults = dict( simulation_id=_sid("sim1"), proposal_id=_sid("p1"), simulation_type="conflict_scan", result={"conflict_count": 3, "conflicting_files": ["a.py"]}, is_stale=False, from_branch_commit_id=_sid("c1"), duration_ms=42, created_at=_now(), expires_at=None, ) defaults.update(overrides) return SimulationResponse(**defaults) def test_all_fields_present(self) -> None: from musehub.mcp.write_tools.proposals import _simulation_data sim = self._make_sim() data = _simulation_data(sim) for key in ("simulation_id", "proposal_id", "simulation_type", "result", "is_stale", "from_branch_commit_id", "duration_ms", "created_at", "expires_at"): assert key in data def test_result_passthrough(self) -> None: from musehub.mcp.write_tools.proposals import _simulation_data result = {"conflict_count": 5, "risk_band": "high"} data = _simulation_data(self._make_sim(result=result)) assert data["result"] == result def test_is_stale_flag(self) -> None: from musehub.mcp.write_tools.proposals import _simulation_data data = _simulation_data(self._make_sim(is_stale=True)) assert data["is_stale"] is True def test_expires_at_iso_when_set(self) -> None: from musehub.mcp.write_tools.proposals import _simulation_data ts = _now() data = _simulation_data(self._make_sim(expires_at=ts)) assert data["expires_at"] == ts.isoformat() # ── execute_create_proposal — new fields forwarded ──────────────────────────── class TestExecuteCreateProposalNewFields: """Verify new Phase 1 kwargs are forwarded to musehub_proposals.create_proposal.""" @pytest.mark.asyncio async def test_new_kwargs_forwarded(self) -> None: from musehub.mcp.write_tools.proposals import execute_create_proposal mock_proposal = MagicMock() mock_proposal.proposal_id = _sid("p1") mock_proposal.title = "T" mock_proposal.body = "" mock_proposal.state = "open" mock_proposal.from_branch = "feat" mock_proposal.to_branch = "dev" mock_proposal.author = "gabriel" mock_proposal.merge_commit_id = None mock_proposal.created_at = None mock_proposal.merged_at = None mock_session = AsyncMock() mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=False) mock_repo = MagicMock(owner="gabriel", visibility="public") with ( patch("musehub.mcp.write_tools.proposals._check_db_available", return_value=None), patch("musehub.mcp.write_tools.proposals.AsyncSessionLocal", return_value=mock_session), patch("musehub.mcp.write_tools.proposals.musehub_repository.get_repo", new_callable=AsyncMock, return_value=mock_repo), patch("musehub.mcp.write_tools.proposals._require_public_or_write_access", new_callable=AsyncMock, return_value=None), patch("musehub.mcp.write_tools.proposals.musehub_repository.get_identity_id_for_handle", new_callable=AsyncMock, return_value=_sid("id1")), patch("musehub.mcp.write_tools.proposals.musehub_proposals.create_proposal", new_callable=AsyncMock, return_value=mock_proposal) as mock_create, ): result = await execute_create_proposal( repo_id=_sid("repo1"), title="T", from_branch="feat", to_branch="dev", proposal_type="midi_evolution", is_draft=True, merge_strategy="weave", merge_conditions={"require_approvals": 1}, selective_domains=["audio"], depends_on=[_sid("p0")], actor="gabriel", ) assert result.ok is True call_kwargs = mock_create.call_args.kwargs assert call_kwargs["proposal_type"] == "midi_evolution" assert call_kwargs["is_draft"] is True assert call_kwargs["merge_strategy"] == "weave" assert call_kwargs["merge_conditions"] == {"require_approvals": 1} assert call_kwargs["selective_domains"] == ["audio"] assert call_kwargs["depends_on"] == [_sid("p0")] @pytest.mark.asyncio async def test_no_actor_rejected(self) -> None: from musehub.mcp.write_tools.proposals import execute_create_proposal with patch("musehub.mcp.write_tools.proposals._check_db_available", return_value=None): result = await execute_create_proposal( repo_id=_sid("r"), title="T", from_branch="feat", to_branch="dev", actor="", ) # Will fail on branch == to_branch check first, but an empty actor proceeds to DB # where it would fail auth — we just test no crash here with mocked DB unavailable assert result is not None @pytest.mark.asyncio async def test_same_branch_rejected(self) -> None: from musehub.mcp.write_tools.proposals import execute_create_proposal with patch("musehub.mcp.write_tools.proposals._check_db_available", return_value=None): result = await execute_create_proposal( repo_id=_sid("r"), title="T", from_branch="dev", to_branch="dev", actor="gabriel", ) assert result.ok is False assert result.error_code == "invalid_args" # ── execute_get_proposal ─────────────────────────────────────────────────────── class TestExecuteGetProposal: def _mock_proposal(self) -> None: from musehub.models.musehub import ProposalResponse return ProposalResponse( proposal_id=_sid("p1"), title="My proposal", body="", state="open", from_branch="feat/x", to_branch="dev", author="gabriel", proposal_number=1, proposal_type="state_merge", is_draft=False, merge_strategy="overlay", merge_conditions={}, selective_domains=[], merge_commit_id=None, created_at=_now(), merged_at=None, reviewer_count=0, comment_count=0, approval_count=0, changes_requested_count=0, merge_readiness=None, dimensional_risk={}, head_commit_id=_sid("c1"), blocked_by=[], blocks=[], is_blocked=False, latest_simulations={"conflict_scan": {"conflict_count": 0}}, ) @pytest.mark.asyncio async def test_returns_enriched_proposal(self) -> None: from musehub.mcp.write_tools.proposals import execute_get_proposal mock_session = AsyncMock() mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=False) mock_repo = MagicMock(owner="gabriel", visibility="public") proposal = self._mock_proposal() with ( patch("musehub.mcp.write_tools.proposals._check_db_available", return_value=None), patch("musehub.mcp.write_tools.proposals.AsyncSessionLocal", return_value=mock_session), patch("musehub.mcp.write_tools.proposals.musehub_repository.get_repo", new_callable=AsyncMock, return_value=mock_repo), patch("musehub.mcp.write_tools.proposals.musehub_proposals.get_proposal", new_callable=AsyncMock, return_value=proposal), ): result = await execute_get_proposal( repo_id=_sid("repo"), proposal_id=_sid("p1"), actor="gabriel" ) assert result.ok is True assert result.data["proposal_id"] == _sid("p1") assert "latest_simulations" in result.data assert "conflict_scan" in result.data["latest_simulations"] @pytest.mark.asyncio async def test_not_found_when_proposal_missing(self) -> None: from musehub.mcp.write_tools.proposals import execute_get_proposal mock_session = AsyncMock() mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=False) mock_repo = MagicMock(owner="gabriel", visibility="public") with ( patch("musehub.mcp.write_tools.proposals._check_db_available", return_value=None), patch("musehub.mcp.write_tools.proposals.AsyncSessionLocal", return_value=mock_session), patch("musehub.mcp.write_tools.proposals.musehub_repository.get_repo", new_callable=AsyncMock, return_value=mock_repo), patch("musehub.mcp.write_tools.proposals.musehub_proposals.get_proposal", new_callable=AsyncMock, return_value=None), ): result = await execute_get_proposal( repo_id=_sid("repo"), proposal_id=_sid("p99"), actor="gabriel" ) assert result.ok is False assert result.error_code == "proposal_not_found" @pytest.mark.asyncio async def test_no_actor_rejected(self) -> None: from musehub.mcp.write_tools.proposals import execute_get_proposal with patch("musehub.mcp.write_tools.proposals._check_db_available", return_value=None): result = await execute_get_proposal( repo_id=_sid("repo"), proposal_id=_sid("p1"), actor="" ) assert result.ok is False assert result.error_code == "forbidden" # ── execute_run_simulation ───────────────────────────────────────────────────── class TestExecuteRunSimulation: def _mock_sim(self, simulation_type: str = "conflict_scan") -> "SimulationResponse": from musehub.models.musehub import SimulationResponse return SimulationResponse( simulation_id=_sid("sim1"), proposal_id=_sid("p1"), simulation_type=simulation_type, result={"conflict_count": 0}, is_stale=False, from_branch_commit_id=_sid("c1"), duration_ms=10, created_at=_now(), expires_at=None, ) @pytest.mark.asyncio async def test_runs_and_returns_result(self) -> None: from musehub.mcp.write_tools.proposals import execute_run_simulation mock_session = AsyncMock() mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=False) mock_repo = MagicMock(owner="gabriel", visibility="public") sim = self._mock_sim("risk_projection") with ( patch("musehub.mcp.write_tools.proposals._check_db_available", return_value=None), patch("musehub.mcp.write_tools.proposals.AsyncSessionLocal", return_value=mock_session), patch("musehub.mcp.write_tools.proposals.musehub_repository.get_repo", new_callable=AsyncMock, return_value=mock_repo), patch("musehub.mcp.write_tools.proposals.musehub_proposals.run_simulation", new_callable=AsyncMock, return_value=sim), ): result = await execute_run_simulation( repo_id=_sid("repo"), proposal_id=_sid("p1"), simulation_type="risk_projection", actor="gabriel", ) assert result.ok is True assert result.data["simulation_type"] == "risk_projection" assert "result" in result.data assert result.data["is_stale"] is False @pytest.mark.asyncio async def test_invalid_simulation_type_rejected(self) -> None: from musehub.mcp.write_tools.proposals import execute_run_simulation with patch("musehub.mcp.write_tools.proposals._check_db_available", return_value=None): result = await execute_run_simulation( repo_id=_sid("repo"), proposal_id=_sid("p1"), simulation_type="bad_type", actor="gabriel", ) assert result.ok is False assert result.error_code == "invalid_args" @pytest.mark.asyncio async def test_no_actor_rejected(self) -> None: from musehub.mcp.write_tools.proposals import execute_run_simulation with patch("musehub.mcp.write_tools.proposals._check_db_available", return_value=None): result = await execute_run_simulation( repo_id=_sid("repo"), proposal_id=_sid("p1"), simulation_type="conflict_scan", actor="", ) assert result.ok is False assert result.error_code == "forbidden" @pytest.mark.asyncio @pytest.mark.parametrize("sim_type", ["conflict_scan", "risk_projection", "dependency_order"]) async def test_all_valid_simulation_types_accepted(self, sim_type: str) -> None: from musehub.mcp.write_tools.proposals import execute_run_simulation mock_session = AsyncMock() mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=False) mock_repo = MagicMock(owner="gabriel", visibility="public") with ( patch("musehub.mcp.write_tools.proposals._check_db_available", return_value=None), patch("musehub.mcp.write_tools.proposals.AsyncSessionLocal", return_value=mock_session), patch("musehub.mcp.write_tools.proposals.musehub_repository.get_repo", new_callable=AsyncMock, return_value=mock_repo), patch("musehub.mcp.write_tools.proposals.musehub_proposals.run_simulation", new_callable=AsyncMock, return_value=self._mock_sim(sim_type)), ): result = await execute_run_simulation( repo_id=_sid("repo"), proposal_id=_sid("p1"), simulation_type=sim_type, actor="gabriel", ) assert result.ok is True # ── execute_get_simulation ───────────────────────────────────────────────────── class TestExecuteGetSimulation: def _mock_sim(self) -> "SimulationResponse": from musehub.models.musehub import SimulationResponse return SimulationResponse( simulation_id=_sid("sim1"), proposal_id=_sid("p1"), simulation_type="conflict_scan", result={"conflict_count": 2}, is_stale=True, from_branch_commit_id=_sid("c_old"), duration_ms=15, created_at=_now(), expires_at=None, ) @pytest.mark.asyncio async def test_returns_cached_result(self) -> None: from musehub.mcp.write_tools.proposals import execute_get_simulation mock_session = AsyncMock() mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=False) mock_repo = MagicMock(owner="gabriel", visibility="public") sim = self._mock_sim() with ( patch("musehub.mcp.write_tools.proposals._check_db_available", return_value=None), patch("musehub.mcp.write_tools.proposals.AsyncSessionLocal", return_value=mock_session), patch("musehub.mcp.write_tools.proposals.musehub_repository.get_repo", new_callable=AsyncMock, return_value=mock_repo), patch("musehub.mcp.write_tools.proposals.musehub_proposals.get_simulation", new_callable=AsyncMock, return_value=sim), ): result = await execute_get_simulation( repo_id=_sid("repo"), proposal_id=_sid("p1"), simulation_type="conflict_scan", actor="gabriel", ) assert result.ok is True assert result.data["is_stale"] is True assert result.data["result"]["conflict_count"] == 2 @pytest.mark.asyncio async def test_not_found_when_no_cached_simulation(self) -> None: from musehub.mcp.write_tools.proposals import execute_get_simulation mock_session = AsyncMock() mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=False) mock_repo = MagicMock(owner="gabriel", visibility="public") with ( patch("musehub.mcp.write_tools.proposals._check_db_available", return_value=None), patch("musehub.mcp.write_tools.proposals.AsyncSessionLocal", return_value=mock_session), patch("musehub.mcp.write_tools.proposals.musehub_repository.get_repo", new_callable=AsyncMock, return_value=mock_repo), patch("musehub.mcp.write_tools.proposals.musehub_proposals.get_simulation", new_callable=AsyncMock, return_value=None), ): result = await execute_get_simulation( repo_id=_sid("repo"), proposal_id=_sid("p1"), simulation_type="conflict_scan", actor="gabriel", ) assert result.ok is False assert result.error_code == "not_found" assert "musehub_run_proposal_simulation" in result.error_message @pytest.mark.asyncio async def test_invalid_type_rejected(self) -> None: from musehub.mcp.write_tools.proposals import execute_get_simulation with patch("musehub.mcp.write_tools.proposals._check_db_available", return_value=None): result = await execute_get_simulation( repo_id=_sid("repo"), proposal_id=_sid("p1"), simulation_type="nope", actor="gabriel", ) assert result.ok is False assert result.error_code == "invalid_args" # ── execute_list_simulations ─────────────────────────────────────────────────── class TestExecuteListSimulations: def _make_sim_list(self) -> None: from musehub.models.musehub import SimulationResponse, SimulationListResponse sims = [ SimulationResponse( simulation_id=_sid(f"sim{i}"), proposal_id=_sid("p1"), simulation_type=st, result={}, is_stale=False, from_branch_commit_id=_sid("c1"), duration_ms=5, created_at=_now(), expires_at=None, ) for i, st in enumerate(["conflict_scan", "risk_projection"]) ] return SimulationListResponse(simulations=sims, total=2) @pytest.mark.asyncio async def test_returns_all_simulations(self) -> None: from musehub.mcp.write_tools.proposals import execute_list_simulations mock_session = AsyncMock() mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=False) mock_repo = MagicMock(owner="gabriel", visibility="public") sim_list = self._make_sim_list() with ( patch("musehub.mcp.write_tools.proposals._check_db_available", return_value=None), patch("musehub.mcp.write_tools.proposals.AsyncSessionLocal", return_value=mock_session), patch("musehub.mcp.write_tools.proposals.musehub_repository.get_repo", new_callable=AsyncMock, return_value=mock_repo), patch("musehub.mcp.write_tools.proposals.musehub_proposals.list_simulations", new_callable=AsyncMock, return_value=sim_list), ): result = await execute_list_simulations( repo_id=_sid("repo"), proposal_id=_sid("p1"), actor="gabriel", ) assert result.ok is True assert result.data["total"] == 2 types = {s["simulation_type"] for s in result.data["simulations"]} assert types == {"conflict_scan", "risk_projection"} @pytest.mark.asyncio async def test_empty_list_when_none_run(self) -> None: from musehub.mcp.write_tools.proposals import execute_list_simulations from musehub.models.musehub import SimulationListResponse mock_session = AsyncMock() mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=False) mock_repo = MagicMock(owner="gabriel", visibility="public") with ( patch("musehub.mcp.write_tools.proposals._check_db_available", return_value=None), patch("musehub.mcp.write_tools.proposals.AsyncSessionLocal", return_value=mock_session), patch("musehub.mcp.write_tools.proposals.musehub_repository.get_repo", new_callable=AsyncMock, return_value=mock_repo), patch("musehub.mcp.write_tools.proposals.musehub_proposals.list_simulations", new_callable=AsyncMock, return_value=SimulationListResponse(simulations=[], total=0)), ): result = await execute_list_simulations( repo_id=_sid("repo"), proposal_id=_sid("p1"), actor="gabriel", ) assert result.ok is True assert result.data["total"] == 0 assert result.data["simulations"] == [] @pytest.mark.asyncio async def test_no_actor_rejected(self) -> None: from musehub.mcp.write_tools.proposals import execute_list_simulations with patch("musehub.mcp.write_tools.proposals._check_db_available", return_value=None): result = await execute_list_simulations( repo_id=_sid("repo"), proposal_id=_sid("p1"), actor="", ) assert result.ok is False assert result.error_code == "forbidden" # ── Tool schema presence ─────────────────────────────────────────────────────── class TestToolSchemas: def _get_tool_names(self) -> None: from musehub.mcp.tools.musehub import MUSEHUB_WRITE_TOOLS return {t["name"] for t in MUSEHUB_WRITE_TOOLS} def test_get_proposal_in_write_tools(self) -> None: assert "musehub_get_proposal" in self._get_tool_names() def test_run_simulation_in_write_tools(self) -> None: assert "musehub_run_proposal_simulation" in self._get_tool_names() def test_get_simulation_in_write_tools(self) -> None: assert "musehub_get_proposal_simulation" in self._get_tool_names() def test_list_simulations_in_write_tools(self) -> None: assert "musehub_list_proposal_simulations" in self._get_tool_names() def test_create_proposal_has_new_fields(self) -> None: from musehub.mcp.tools.musehub import MUSEHUB_WRITE_TOOLS schema = next(t for t in MUSEHUB_WRITE_TOOLS if t["name"] == "musehub_create_proposal") props = schema["inputSchema"]["properties"] for field in ("proposal_type", "is_draft", "merge_strategy", "merge_conditions", "selective_domains", "depends_on"): assert field in props, f"Missing field {field!r} in musehub_create_proposal schema" # Proposal type enum uses music-domain values assert "state_merge" in props["proposal_type"]["enum"] assert "midi_evolution" in props["proposal_type"]["enum"] def test_run_simulation_schema_enum(self) -> None: from musehub.mcp.tools.musehub import MUSEHUB_WRITE_TOOLS schema = next(t for t in MUSEHUB_WRITE_TOOLS if t["name"] == "musehub_run_proposal_simulation") sim_type_prop = schema["inputSchema"]["properties"]["simulation_type"] assert set(sim_type_prop["enum"]) == {"conflict_scan", "risk_projection", "dependency_order"} def test_simulation_tools_require_proposal_id(self) -> None: from musehub.mcp.tools.musehub import MUSEHUB_WRITE_TOOLS sim_tool_names = { "musehub_run_proposal_simulation", "musehub_get_proposal_simulation", "musehub_list_proposal_simulations", } for tool in MUSEHUB_WRITE_TOOLS: if tool["name"] in sim_tool_names: assert "proposal_id" in tool["inputSchema"].get("required", []), ( f"{tool['name']} must require proposal_id" ) def test_all_new_tools_in_tool_names_set(self) -> None: from musehub.mcp.tools.musehub import MUSEHUB_TOOL_NAMES for name in ("musehub_get_proposal", "musehub_run_proposal_simulation", "musehub_get_proposal_simulation", "musehub_list_proposal_simulations"): assert name in MUSEHUB_TOOL_NAMES # ── Dispatcher routing ───────────────────────────────────────────────────────── class TestDispatcherRouting: """Verify the dispatcher correctly routes new tool calls to the right executors.""" def _make_dispatcher_call(self, tool_name: str, arguments: JSONObject) -> None: """Simulate a dispatcher dispatch call synchronously.""" import asyncio from musehub.mcp import dispatcher as disp async def _run() -> None: return await disp.dispatch( name=tool_name, arguments=arguments, user_id="gabriel", session_id=None, request_context=None, ) return asyncio.get_event_loop().run_until_complete(_run()) @pytest.mark.asyncio async def test_dispatcher_routes_get_proposal(self) -> None: from musehub.mcp import dispatcher as disp mock_result = MagicMock() mock_result.ok = True mock_result.data = {"proposal_id": _sid("p1")} mock_result.error_code = None mock_result.error_message = None mock_result.hint = None with patch( "musehub.mcp.write_tools.proposals.execute_get_proposal", new_callable=AsyncMock, return_value=mock_result, ) as mock_exec: await disp.dispatch_tool( "musehub_get_proposal", {"repo_id": _sid("repo"), "proposal_id": _sid("p1")}, user_id="gabriel", ) mock_exec.assert_called_once() kwargs = mock_exec.call_args.kwargs assert kwargs["proposal_id"] == _sid("p1") assert kwargs["actor"] == "gabriel" @pytest.mark.asyncio async def test_dispatcher_routes_run_simulation(self) -> None: from musehub.mcp import dispatcher as disp mock_result = MagicMock() mock_result.ok = True mock_result.data = {} mock_result.error_code = None mock_result.error_message = None mock_result.hint = None with patch( "musehub.mcp.write_tools.proposals.execute_run_simulation", new_callable=AsyncMock, return_value=mock_result, ) as mock_exec: await disp.dispatch_tool( "musehub_run_proposal_simulation", { "repo_id": _sid("repo"), "proposal_id": _sid("p1"), "simulation_type": "conflict_scan", }, user_id="gabriel", ) mock_exec.assert_called_once() kwargs = mock_exec.call_args.kwargs assert kwargs["simulation_type"] == "conflict_scan" @pytest.mark.asyncio async def test_dispatcher_routes_list_simulations(self) -> None: from musehub.mcp import dispatcher as disp mock_result = MagicMock() mock_result.ok = True mock_result.data = {"simulations": [], "total": 0} mock_result.error_code = None mock_result.error_message = None mock_result.hint = None with patch( "musehub.mcp.write_tools.proposals.execute_list_simulations", new_callable=AsyncMock, return_value=mock_result, ) as mock_exec: await disp.dispatch_tool( "musehub_list_proposal_simulations", {"repo_id": _sid("repo"), "proposal_id": _sid("p1")}, user_id="gabriel", ) mock_exec.assert_called_once() @pytest.mark.asyncio async def test_dispatcher_create_proposal_forwards_new_fields(self) -> None: from musehub.mcp import dispatcher as disp mock_result = MagicMock() mock_result.ok = True mock_result.data = {} mock_result.error_code = None mock_result.error_message = None mock_result.hint = None with patch( "musehub.mcp.write_tools.proposals.execute_create_proposal", new_callable=AsyncMock, return_value=mock_result, ) as mock_exec: await disp.dispatch_tool( "musehub_create_proposal", { "repo_id": _sid("repo"), "title": "Test", "from_branch": "feat/x", "to_branch": "dev", "proposal_type": "midi_evolution", "is_draft": True, "merge_strategy": "weave", "selective_domains": ["audio", "midi"], "depends_on": [_sid("p0")], }, user_id="gabriel", ) kwargs = mock_exec.call_args.kwargs assert kwargs["proposal_type"] == "midi_evolution" assert kwargs["is_draft"] is True assert kwargs["merge_strategy"] == "weave" assert kwargs["selective_domains"] == ["audio", "midi"] assert kwargs["depends_on"] == [_sid("p0")]