gabriel / musehub public
test_mcp_profile_tools.py python
793 lines 32.8 KB
Raw
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ breaking 20 days ago
1 """TDD tests for Phase 5 MCP profile + attestation + MPay executor functions.
2
3 Tests are written RED-first against the public contracts described in issue #2.
4 They drive the implementation of six new executor functions and their dispatcher
5 routing.
6
7 Covered executors:
8 1. execute_read_profile_manifest — full archetype-aware manifest
9 2. execute_issue_attestation — verify sig + persist
10 3. execute_revoke_attestation — revoke by id, attester-only
11 4. execute_list_attestations — query by subject
12 5. execute_record_mpay_claim — verify sig + persist, idempotent
13 6. execute_get_mpay_ledger — sent + received totals
14 """
15 from __future__ import annotations
16
17 from datetime import datetime, timezone
18 from unittest.mock import AsyncMock, MagicMock, patch
19
20 import pytest
21 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
22 from muse.core.types import encode_pubkey, encode_sig, long_id
23
24 # ---------------------------------------------------------------------------
25 # Helpers — shared across all test groups
26 # ---------------------------------------------------------------------------
27
28 def _make_ed25519_pair() -> tuple[Ed25519PrivateKey, str]:
29 privkey = Ed25519PrivateKey.generate()
30 pub_bytes = privkey.public_key().public_bytes_raw()
31 return privkey, encode_pubkey("ed25519", pub_bytes)
32
33
34 def _sign_attest(privkey: Ed25519PrivateKey, attester: str, subject: str, claim: str, ts: str) -> str:
35 msg = f"ATTEST\n{attester}\n{subject}\n{claim}\n{ts}".encode()
36 return encode_sig("ed25519", privkey.sign(msg))
37
38
39 def _sign_mpay(privkey: Ed25519PrivateKey, sender: str, recipient: str, amount: int, nonce: str) -> str:
40 msg = f"MPAY\n{sender}\n{recipient}\n{amount}\n{nonce}".encode()
41 return encode_sig("ed25519", privkey.sign(msg))
42
43
44 _ATTESTER = "gabriel"
45 _SUBJECT = "aria"
46 _CLAIM = '{"type": "human", "confidence": 0.99}'
47 _TS = "2026-04-21T12:00:00+00:00"
48 _SENDER = "gabriel"
49 _RECIPIENT = "aria"
50 _AMOUNT_NANO = 500_000
51 _NONCE_HEX = "b" * 64
52
53
54 # ---------------------------------------------------------------------------
55 # 1. execute_read_profile_manifest
56 # ---------------------------------------------------------------------------
57
58 class TestExecuteReadProfileManifest:
59 """execute_read_profile_manifest must return the full archetype-aware manifest,
60 not just basic bio/avatar fields."""
61
62 def _make_manifest(self, identity_type: str = "human") -> MagicMock:
63 from musehub.models.musehub import (
64 ActivityDomain, AttestationBadge, OrgManifest,
65 ProfileManifest, ProfileRepoSummary, TrustChainEntry,
66 )
67 now = datetime.now(timezone.utc)
68 return ProfileManifest(
69 identity_id=long_id("a" * 64),
70 handle="gabriel",
71 identity_type=identity_type,
72 display_name="Carlos Gabriel Cardona",
73 bio="Building the sound of the future",
74 avatar_url="https://staging.musehub.ai/avatars/gabriel.png",
75 location="San Francisco",
76 website_url="https://gabriel.dev",
77 social_url="https://x.com/gabriel",
78 is_verified=True,
79 pinned_repo_ids=[],
80 repos=[],
81 created_at=now,
82 updated_at=now,
83 activity=[
84 ActivityDomain(domain="code", grid=[0] * 364, peak=0, total=0),
85 ActivityDomain(domain="music", grid=[3] * 364, peak=3, total=3 * 364),
86 ],
87 attestations=[
88 AttestationBadge(
89 attestation_id=long_id("b" * 64),
90 attester="aaronrene",
91 subject="gabriel",
92 claim_type="collab",
93 claim='{"type":"collab"}',
94 issued_at=now,
95 )
96 ],
97 avax_address="0x1a2b3c4d5e6f" if identity_type == "human" else None,
98 agent_model="claude-sonnet-4-6" if identity_type == "agent" else None,
99 agent_capabilities=["read:repos", "write:repos"] if identity_type == "agent" else [],
100 trust_chain=[TrustChainEntry(handle="gabriel", identity_type="human")] if identity_type == "agent" else [],
101 org=OrgManifest(members=["gabriel", "aria"], quorum=2) if identity_type == "org" else None,
102 mpay_total_sent_nano=1_000_000,
103 mpay_total_received_nano=500_000,
104 )
105
106 @pytest.mark.asyncio
107 async def test_returns_full_manifest_fields(self) -> None:
108 """Result data must include identity_type, activity, attestations, avax_address, mpay totals."""
109 from musehub.services.musehub_mcp_executor import execute_read_profile_manifest
110
111 manifest = self._make_manifest("human")
112
113 with (
114 patch("musehub.services.musehub_mcp_executor._check_db_available", return_value=None),
115 patch("musehub.services.musehub_mcp_executor.AsyncSessionLocal") as mock_session_cls,
116 patch("musehub.services.musehub_profile.build_profile_manifest", new=AsyncMock(return_value=manifest)),
117 patch("musehub.services.musehub_attestations.get_attestations_for_subject", new=AsyncMock()),
118 patch("musehub.services.musehub_attestations.attestation_to_badge", return_value=manifest.attestations[0]),
119 patch("musehub.services.musehub_mpay.get_mpay_ledger", new=AsyncMock()),
120 ):
121 mock_session = AsyncMock()
122 mock_session.__aenter__ = AsyncMock(return_value=mock_session)
123 mock_session.__aexit__ = AsyncMock(return_value=False)
124 mock_session_cls.return_value = mock_session
125
126 result = await execute_read_profile_manifest(handle="gabriel")
127
128 assert result.ok is True
129 d = result.data
130 assert d is not None
131 assert d["handle"] == "gabriel"
132 assert d["identity_type"] == "human"
133 assert "activity" in d
134 assert "attestations" in d
135 assert d["avax_address"] == "0x1a2b3c4d5e6f"
136 assert d["mpay_total_sent_nano"] == 1_000_000
137 assert d["mpay_total_received_nano"] == 500_000
138
139 @pytest.mark.asyncio
140 async def test_returns_agent_specific_fields(self) -> None:
141 """Agent manifest includes agent_model, agent_capabilities, trust_chain."""
142 from musehub.services.musehub_mcp_executor import execute_read_profile_manifest
143
144 manifest = self._make_manifest("agent")
145
146 with (
147 patch("musehub.services.musehub_mcp_executor._check_db_available", return_value=None),
148 patch("musehub.services.musehub_mcp_executor.AsyncSessionLocal") as mock_session_cls,
149 patch("musehub.services.musehub_profile.build_profile_manifest", new=AsyncMock(return_value=manifest)),
150 patch("musehub.services.musehub_attestations.get_attestations_for_subject", new=AsyncMock()),
151 patch("musehub.services.musehub_attestations.attestation_to_badge", return_value=manifest.attestations[0]),
152 patch("musehub.services.musehub_mpay.get_mpay_ledger", new=AsyncMock()),
153 ):
154 mock_session = AsyncMock()
155 mock_session.__aenter__ = AsyncMock(return_value=mock_session)
156 mock_session.__aexit__ = AsyncMock(return_value=False)
157 mock_session_cls.return_value = mock_session
158
159 result = await execute_read_profile_manifest(handle="mix-engine-7")
160
161 assert result.ok is True
162 d = result.data
163 assert d["identity_type"] == "agent"
164 assert d["agent_model"] == "claude-sonnet-4-6"
165 assert "read:repos" in d["agent_capabilities"]
166 assert len(d["trust_chain"]) == 1
167
168 @pytest.mark.asyncio
169 async def test_returns_org_specific_fields(self) -> None:
170 """Org manifest includes org manifest with members and quorum."""
171 from musehub.services.musehub_mcp_executor import execute_read_profile_manifest
172
173 manifest = self._make_manifest("org")
174
175 with (
176 patch("musehub.services.musehub_mcp_executor._check_db_available", return_value=None),
177 patch("musehub.services.musehub_mcp_executor.AsyncSessionLocal") as mock_session_cls,
178 patch("musehub.services.musehub_profile.build_profile_manifest", new=AsyncMock(return_value=manifest)),
179 patch("musehub.services.musehub_attestations.get_attestations_for_subject", new=AsyncMock()),
180 patch("musehub.services.musehub_attestations.attestation_to_badge", return_value=manifest.attestations[0]),
181 patch("musehub.services.musehub_mpay.get_mpay_ledger", new=AsyncMock()),
182 ):
183 mock_session = AsyncMock()
184 mock_session.__aenter__ = AsyncMock(return_value=mock_session)
185 mock_session.__aexit__ = AsyncMock(return_value=False)
186 mock_session_cls.return_value = mock_session
187
188 result = await execute_read_profile_manifest(handle="darkroom-collective")
189
190 assert result.ok is True
191 d = result.data
192 assert d["identity_type"] == "org"
193 assert d["org"] is not None
194 assert "gabriel" in d["org"]["members"]
195 assert d["org"]["quorum"] == 2
196
197 @pytest.mark.asyncio
198 async def test_not_found_returns_error(self) -> None:
199 from musehub.services.musehub_mcp_executor import execute_read_profile_manifest
200
201 with (
202 patch("musehub.services.musehub_mcp_executor._check_db_available", return_value=None),
203 patch("musehub.services.musehub_mcp_executor.AsyncSessionLocal") as mock_session_cls,
204 patch("musehub.services.musehub_profile.build_profile_manifest", new=AsyncMock(return_value=None)),
205 patch("musehub.services.musehub_attestations.get_attestations_for_subject", new=AsyncMock()),
206 patch("musehub.services.musehub_mpay.get_mpay_ledger", new=AsyncMock()),
207 ):
208 mock_session = AsyncMock()
209 mock_session.__aenter__ = AsyncMock(return_value=mock_session)
210 mock_session.__aexit__ = AsyncMock(return_value=False)
211 mock_session_cls.return_value = mock_session
212
213 result = await execute_read_profile_manifest(handle="nobody")
214
215 assert result.ok is False
216 assert result.error_code == "profile_not_found"
217
218
219 # ---------------------------------------------------------------------------
220 # 2. execute_issue_attestation
221 # ---------------------------------------------------------------------------
222
223 class TestExecuteIssueAttestation:
224 """execute_issue_attestation must verify the Ed25519 signature and persist."""
225
226 @pytest.mark.asyncio
227 async def test_valid_attestation_returns_ok(self) -> None:
228 from musehub.services.musehub_mcp_executor import execute_issue_attestation
229
230 privkey, pubkey = _make_ed25519_pair()
231 sig = _sign_attest(privkey, _ATTESTER, _SUBJECT, _CLAIM, _TS)
232
233 issued_at = datetime.fromisoformat(_TS)
234 mock_resp = MagicMock()
235 mock_resp.attestation_id = long_id("c" * 64)
236 mock_resp.attester = _ATTESTER
237 mock_resp.subject = _SUBJECT
238 mock_resp.claim = _CLAIM
239 mock_resp.issued_at = issued_at
240 mock_resp.revoked_at = None
241
242 with (
243 patch("musehub.services.musehub_mcp_executor._check_db_available", return_value=None),
244 patch("musehub.services.musehub_mcp_executor.AsyncSessionLocal") as mock_session_cls,
245 patch("musehub.services.musehub_attestations.issue_attestation", new=AsyncMock(return_value=mock_resp)),
246 ):
247 mock_session = AsyncMock()
248 mock_session.__aenter__ = AsyncMock(return_value=mock_session)
249 mock_session.__aexit__ = AsyncMock(return_value=False)
250 mock_session_cls.return_value = mock_session
251
252 result = await execute_issue_attestation(
253 attester=_ATTESTER,
254 subject=_SUBJECT,
255 claim=_CLAIM,
256 issued_at_iso=_TS,
257 signature=sig,
258 attester_public_key=pubkey,
259 )
260
261 assert result.ok is True
262 d = result.data
263 assert d["attester"] == _ATTESTER
264 assert d["subject"] == _SUBJECT
265 assert d["attestation_id"].startswith("sha256:")
266 assert d["revoked_at"] is None
267
268 @pytest.mark.asyncio
269 async def test_invalid_signature_returns_error(self) -> None:
270 from musehub.services.musehub_mcp_executor import execute_issue_attestation
271
272 _, pubkey = _make_ed25519_pair()
273 bad_sig = f"ed25519:{"Z" * 88}" # garbage
274
275 with patch("musehub.services.musehub_mcp_executor._check_db_available", return_value=None):
276 result = await execute_issue_attestation(
277 attester=_ATTESTER,
278 subject=_SUBJECT,
279 claim=_CLAIM,
280 issued_at_iso=_TS,
281 signature=bad_sig,
282 attester_public_key=pubkey,
283 )
284
285 assert result.ok is False
286 assert result.error_code == "invalid_attestation_signature"
287
288 @pytest.mark.asyncio
289 async def test_wrong_key_returns_error(self) -> None:
290 """Signing key doesn't match supplied public key → verification fails."""
291 from musehub.services.musehub_mcp_executor import execute_issue_attestation
292
293 privkey, _ = _make_ed25519_pair()
294 _, other_pubkey = _make_ed25519_pair()
295 sig = _sign_attest(privkey, _ATTESTER, _SUBJECT, _CLAIM, _TS)
296
297 with patch("musehub.services.musehub_mcp_executor._check_db_available", return_value=None):
298 result = await execute_issue_attestation(
299 attester=_ATTESTER,
300 subject=_SUBJECT,
301 claim=_CLAIM,
302 issued_at_iso=_TS,
303 signature=sig,
304 attester_public_key=other_pubkey,
305 )
306
307 assert result.ok is False
308 assert result.error_code == "invalid_attestation_signature"
309
310
311 # ---------------------------------------------------------------------------
312 # 3. execute_revoke_attestation
313 # ---------------------------------------------------------------------------
314
315 class TestExecuteRevokeAttestation:
316
317 @pytest.mark.asyncio
318 async def test_revoke_returns_revoked_at(self) -> None:
319 from musehub.services.musehub_mcp_executor import execute_revoke_attestation
320
321 now = datetime.now(timezone.utc)
322 mock_resp = MagicMock()
323 mock_resp.attestation_id = long_id("d" * 64)
324 mock_resp.attester = _ATTESTER
325 mock_resp.subject = _SUBJECT
326 mock_resp.claim = _CLAIM
327 mock_resp.issued_at = datetime.fromisoformat(_TS)
328 mock_resp.revoked_at = now
329
330 with (
331 patch("musehub.services.musehub_mcp_executor._check_db_available", return_value=None),
332 patch("musehub.services.musehub_mcp_executor.AsyncSessionLocal") as mock_session_cls,
333 patch("musehub.services.musehub_attestations.revoke_attestation", new=AsyncMock(return_value=mock_resp)),
334 ):
335 mock_session = AsyncMock()
336 mock_session.__aenter__ = AsyncMock(return_value=mock_session)
337 mock_session.__aexit__ = AsyncMock(return_value=False)
338 mock_session_cls.return_value = mock_session
339
340 result = await execute_revoke_attestation(
341 attestation_id=long_id("d" * 64),
342 revoker=_ATTESTER,
343 )
344
345 assert result.ok is True
346 assert result.data["revoked_at"] is not None
347
348 @pytest.mark.asyncio
349 async def test_not_found_returns_error(self) -> None:
350 from musehub.services.musehub_mcp_executor import execute_revoke_attestation
351
352 with (
353 patch("musehub.services.musehub_mcp_executor._check_db_available", return_value=None),
354 patch("musehub.services.musehub_mcp_executor.AsyncSessionLocal") as mock_session_cls,
355 patch(
356 "musehub.services.musehub_attestations.revoke_attestation",
357 new=AsyncMock(side_effect=KeyError("not found")),
358 ),
359 ):
360 mock_session = AsyncMock()
361 mock_session.__aenter__ = AsyncMock(return_value=mock_session)
362 mock_session.__aexit__ = AsyncMock(return_value=False)
363 mock_session_cls.return_value = mock_session
364
365 result = await execute_revoke_attestation(
366 attestation_id=long_id("e" * 64),
367 revoker=_ATTESTER,
368 )
369
370 assert result.ok is False
371 assert result.error_code == "attestation_not_found"
372
373 @pytest.mark.asyncio
374 async def test_wrong_revoker_returns_forbidden(self) -> None:
375 from musehub.services.musehub_mcp_executor import execute_revoke_attestation
376
377 with (
378 patch("musehub.services.musehub_mcp_executor._check_db_available", return_value=None),
379 patch("musehub.services.musehub_mcp_executor.AsyncSessionLocal") as mock_session_cls,
380 patch(
381 "musehub.services.musehub_attestations.revoke_attestation",
382 new=AsyncMock(side_effect=PermissionError("not your attestation")),
383 ),
384 ):
385 mock_session = AsyncMock()
386 mock_session.__aenter__ = AsyncMock(return_value=mock_session)
387 mock_session.__aexit__ = AsyncMock(return_value=False)
388 mock_session_cls.return_value = mock_session
389
390 result = await execute_revoke_attestation(
391 attestation_id=long_id("d" * 64),
392 revoker="impostor",
393 )
394
395 assert result.ok is False
396 assert result.error_code == "forbidden"
397
398
399 # ---------------------------------------------------------------------------
400 # 4. execute_list_attestations
401 # ---------------------------------------------------------------------------
402
403 class TestExecuteListAttestations:
404
405 @pytest.mark.asyncio
406 async def test_returns_list_with_count(self) -> None:
407 from musehub.services.musehub_mcp_executor import execute_list_attestations
408 from musehub.models.musehub import AttestationListResponse, AttestationResponse
409
410 now = datetime.now(timezone.utc)
411 badge = AttestationResponse(
412 attestation_id=long_id("f" * 64),
413 attester=_ATTESTER,
414 subject=_SUBJECT,
415 claim=_CLAIM,
416 signature="ed25519:abc",
417 attester_public_key="ed25519:def",
418 issued_at=now,
419 revoked_at=None,
420 )
421 mock_resp = AttestationListResponse(subject=_SUBJECT, attestations=[badge], total=1)
422
423 with (
424 patch("musehub.services.musehub_mcp_executor._check_db_available", return_value=None),
425 patch("musehub.services.musehub_mcp_executor.AsyncSessionLocal") as mock_session_cls,
426 patch(
427 "musehub.services.musehub_attestations.get_attestations_for_subject",
428 new=AsyncMock(return_value=mock_resp),
429 ),
430 ):
431 mock_session = AsyncMock()
432 mock_session.__aenter__ = AsyncMock(return_value=mock_session)
433 mock_session.__aexit__ = AsyncMock(return_value=False)
434 mock_session_cls.return_value = mock_session
435
436 result = await execute_list_attestations(subject=_SUBJECT, include_revoked=False)
437
438 assert result.ok is True
439 assert result.data["subject"] == _SUBJECT
440 assert result.data["total"] == 1
441 assert len(result.data["attestations"]) == 1
442 assert result.data["attestations"][0]["attester"] == _ATTESTER
443
444 @pytest.mark.asyncio
445 async def test_empty_subject_returns_empty_list(self) -> None:
446 from musehub.services.musehub_mcp_executor import execute_list_attestations
447 from musehub.models.musehub import AttestationListResponse
448
449 mock_resp = AttestationListResponse(subject="nobody", attestations=[], total=0)
450
451 with (
452 patch("musehub.services.musehub_mcp_executor._check_db_available", return_value=None),
453 patch("musehub.services.musehub_mcp_executor.AsyncSessionLocal") as mock_session_cls,
454 patch(
455 "musehub.services.musehub_attestations.get_attestations_for_subject",
456 new=AsyncMock(return_value=mock_resp),
457 ),
458 ):
459 mock_session = AsyncMock()
460 mock_session.__aenter__ = AsyncMock(return_value=mock_session)
461 mock_session.__aexit__ = AsyncMock(return_value=False)
462 mock_session_cls.return_value = mock_session
463
464 result = await execute_list_attestations(subject="nobody")
465
466 assert result.ok is True
467 assert result.data["total"] == 0
468 assert result.data["attestations"] == []
469
470
471 # ---------------------------------------------------------------------------
472 # 5. execute_record_mpay_claim
473 # ---------------------------------------------------------------------------
474
475 class TestExecuteRecordMpayClaim:
476
477 @pytest.mark.asyncio
478 async def test_valid_claim_returns_claim_id(self) -> None:
479 from musehub.services.musehub_mcp_executor import execute_record_mpay_claim
480
481 privkey, pubkey = _make_ed25519_pair()
482 sig = _sign_mpay(privkey, _SENDER, _RECIPIENT, _AMOUNT_NANO, _NONCE_HEX)
483 now = datetime.now(timezone.utc)
484
485 mock_resp = MagicMock()
486 mock_resp.claim_id = long_id("a" * 64)
487 mock_resp.sender = _SENDER
488 mock_resp.recipient = _RECIPIENT
489 mock_resp.amount_nano = _AMOUNT_NANO
490 mock_resp.nonce_hex = _NONCE_HEX
491 mock_resp.created_at = now
492 mock_resp.confirmed_at = None
493 mock_resp.voided_at = None
494 mock_resp.memo = None
495
496 with (
497 patch("musehub.services.musehub_mcp_executor._check_db_available", return_value=None),
498 patch("musehub.services.musehub_mcp_executor.AsyncSessionLocal") as mock_session_cls,
499 patch("musehub.services.musehub_mpay.record_mpay_claim", new=AsyncMock(return_value=mock_resp)),
500 ):
501 mock_session = AsyncMock()
502 mock_session.__aenter__ = AsyncMock(return_value=mock_session)
503 mock_session.__aexit__ = AsyncMock(return_value=False)
504 mock_session_cls.return_value = mock_session
505
506 result = await execute_record_mpay_claim(
507 sender=_SENDER,
508 recipient=_RECIPIENT,
509 amount_nano=_AMOUNT_NANO,
510 nonce_hex=_NONCE_HEX,
511 signature=sig,
512 sender_public_key=pubkey,
513 )
514
515 assert result.ok is True
516 d = result.data
517 assert d["sender"] == _SENDER
518 assert d["recipient"] == _RECIPIENT
519 assert d["amount_nano"] == _AMOUNT_NANO
520 assert d["claim_id"].startswith("sha256:")
521
522 @pytest.mark.asyncio
523 async def test_invalid_mpay_signature_returns_error(self) -> None:
524 from musehub.services.musehub_mcp_executor import execute_record_mpay_claim
525
526 _, pubkey = _make_ed25519_pair()
527 bad_sig = f"ed25519:{"Z" * 88}"
528
529 with patch("musehub.services.musehub_mcp_executor._check_db_available", return_value=None):
530 result = await execute_record_mpay_claim(
531 sender=_SENDER,
532 recipient=_RECIPIENT,
533 amount_nano=_AMOUNT_NANO,
534 nonce_hex=_NONCE_HEX,
535 signature=bad_sig,
536 sender_public_key=pubkey,
537 )
538
539 assert result.ok is False
540 assert result.error_code == "invalid_mpay_signature"
541
542 @pytest.mark.asyncio
543 async def test_zero_amount_returns_error(self) -> None:
544 """Amount must be > 0."""
545 from musehub.services.musehub_mcp_executor import execute_record_mpay_claim
546
547 privkey, pubkey = _make_ed25519_pair()
548 sig = _sign_mpay(privkey, _SENDER, _RECIPIENT, 0, _NONCE_HEX)
549
550 with patch("musehub.services.musehub_mcp_executor._check_db_available", return_value=None):
551 result = await execute_record_mpay_claim(
552 sender=_SENDER,
553 recipient=_RECIPIENT,
554 amount_nano=0,
555 nonce_hex=_NONCE_HEX,
556 signature=sig,
557 sender_public_key=pubkey,
558 )
559
560 assert result.ok is False
561 assert result.error_code == "invalid_amount"
562
563 @pytest.mark.asyncio
564 async def test_self_payment_returns_error(self) -> None:
565 """Sender and recipient must be different handles."""
566 from musehub.services.musehub_mcp_executor import execute_record_mpay_claim
567
568 privkey, pubkey = _make_ed25519_pair()
569 sig = _sign_mpay(privkey, _SENDER, _SENDER, _AMOUNT_NANO, _NONCE_HEX)
570
571 with patch("musehub.services.musehub_mcp_executor._check_db_available", return_value=None):
572 result = await execute_record_mpay_claim(
573 sender=_SENDER,
574 recipient=_SENDER,
575 amount_nano=_AMOUNT_NANO,
576 nonce_hex=_NONCE_HEX,
577 signature=sig,
578 sender_public_key=pubkey,
579 )
580
581 assert result.ok is False
582 assert result.error_code == "self_payment"
583
584
585 # ---------------------------------------------------------------------------
586 # 6. execute_get_mpay_ledger
587 # ---------------------------------------------------------------------------
588
589 class TestExecuteGetMpayLedger:
590
591 @pytest.mark.asyncio
592 async def test_returns_sent_received_totals(self) -> None:
593 from musehub.services.musehub_mcp_executor import execute_get_mpay_ledger
594 from musehub.models.musehub import MPayLedgerResponse, MPayClaimResponse
595
596 now = datetime.now(timezone.utc)
597 claim = MPayClaimResponse(
598 claim_id=long_id("a" * 64),
599 sender=_SENDER,
600 recipient=_RECIPIENT,
601 amount_nano=_AMOUNT_NANO,
602 nonce_hex=_NONCE_HEX,
603 signature="ed25519:abc",
604 sender_public_key="ed25519:def",
605 memo=None,
606 created_at=now,
607 confirmed_at=None,
608 voided_at=None,
609 )
610 mock_ledger = MPayLedgerResponse(
611 handle=_SENDER,
612 sent=[claim],
613 received=[],
614 total_sent_nano=_AMOUNT_NANO,
615 total_received_nano=0,
616 )
617
618 with (
619 patch("musehub.services.musehub_mcp_executor._check_db_available", return_value=None),
620 patch("musehub.services.musehub_mcp_executor.AsyncSessionLocal") as mock_session_cls,
621 patch("musehub.services.musehub_mpay.get_mpay_ledger", new=AsyncMock(return_value=mock_ledger)),
622 ):
623 mock_session = AsyncMock()
624 mock_session.__aenter__ = AsyncMock(return_value=mock_session)
625 mock_session.__aexit__ = AsyncMock(return_value=False)
626 mock_session_cls.return_value = mock_session
627
628 result = await execute_get_mpay_ledger(handle=_SENDER, limit=50)
629
630 assert result.ok is True
631 d = result.data
632 assert d["handle"] == _SENDER
633 assert d["total_sent_nano"] == _AMOUNT_NANO
634 assert d["total_received_nano"] == 0
635 assert len(d["sent"]) == 1
636 assert d["sent"][0]["amount_nano"] == _AMOUNT_NANO
637
638 @pytest.mark.asyncio
639 async def test_limit_clamped_to_500(self) -> None:
640 """Limit above 500 is clamped silently."""
641 from musehub.services.musehub_mcp_executor import execute_get_mpay_ledger
642 from musehub.models.musehub import MPayLedgerResponse
643
644 mock_ledger = MPayLedgerResponse(
645 handle=_SENDER, sent=[], received=[],
646 total_sent_nano=0, total_received_nano=0,
647 )
648
649 captured: dict[str, int] = {}
650
651 async def _mock_ledger(db: MagicMock, handle: str, limit: int = 100) -> MPayLedgerResponse:
652 captured["limit"] = limit
653 return mock_ledger
654
655 with (
656 patch("musehub.services.musehub_mcp_executor._check_db_available", return_value=None),
657 patch("musehub.services.musehub_mcp_executor.AsyncSessionLocal") as mock_session_cls,
658 patch("musehub.services.musehub_mpay.get_mpay_ledger", new=_mock_ledger),
659 ):
660 mock_session = AsyncMock()
661 mock_session.__aenter__ = AsyncMock(return_value=mock_session)
662 mock_session.__aexit__ = AsyncMock(return_value=False)
663 mock_session_cls.return_value = mock_session
664
665 await execute_get_mpay_ledger(handle=_SENDER, limit=9999)
666
667 assert captured["limit"] <= 500
668
669
670 # ---------------------------------------------------------------------------
671 # 7. Dispatcher routing
672 # ---------------------------------------------------------------------------
673
674 class TestDispatcherRouting:
675 """Dispatcher must route the 6 new tool names to the correct executors."""
676
677 @pytest.mark.asyncio
678 async def test_dispatch_read_profile_manifest(self) -> None:
679 from musehub.mcp.dispatcher import dispatch_tool
680
681 mock_result = MagicMock(ok=True, data={"handle": "gabriel"}, error_code=None, error_message=None, hint=None)
682 with patch(
683 "musehub.services.musehub_mcp_executor.execute_read_profile_manifest",
684 new=AsyncMock(return_value=mock_result),
685 ) as mock_exe:
686 await dispatch_tool(
687 name="musehub_read_profile_manifest",
688 arguments={"handle": "gabriel"},
689 user_id="gabriel",
690 session_context=None,
691 )
692 mock_exe.assert_called_once_with(handle="gabriel")
693
694 @pytest.mark.asyncio
695 async def test_dispatch_issue_attestation(self) -> None:
696 from musehub.mcp.dispatcher import dispatch_tool
697
698 mock_result = MagicMock(ok=True, data={}, error_code=None, error_message=None, hint=None)
699 with patch(
700 "musehub.services.musehub_mcp_executor.execute_issue_attestation",
701 new=AsyncMock(return_value=mock_result),
702 ) as mock_exe:
703 args = {
704 "attester": "gabriel",
705 "subject": "aria",
706 "claim": '{"type":"human"}',
707 "issued_at_iso": _TS,
708 "signature": "ed25519:abc",
709 "attester_public_key": "ed25519:def",
710 }
711 await dispatch_tool(
712 name="musehub_issue_attestation",
713 arguments=args,
714 user_id="gabriel",
715 session_context=None,
716 )
717 mock_exe.assert_called_once()
718
719 @pytest.mark.asyncio
720 async def test_dispatch_revoke_attestation(self) -> None:
721 from musehub.mcp.dispatcher import dispatch_tool
722
723 mock_result = MagicMock(ok=True, data={}, error_code=None, error_message=None, hint=None)
724 with patch(
725 "musehub.services.musehub_mcp_executor.execute_revoke_attestation",
726 new=AsyncMock(return_value=mock_result),
727 ) as mock_exe:
728 await dispatch_tool(
729 name="musehub_revoke_attestation",
730 arguments={"attestation_id": long_id("a" * 64), "revoker": "gabriel"},
731 user_id="gabriel",
732 session_context=None,
733 )
734 mock_exe.assert_called_once()
735
736 @pytest.mark.asyncio
737 async def test_dispatch_list_attestations(self) -> None:
738 from musehub.mcp.dispatcher import dispatch_tool
739
740 mock_result = MagicMock(ok=True, data={}, error_code=None, error_message=None, hint=None)
741 with patch(
742 "musehub.services.musehub_mcp_executor.execute_list_attestations",
743 new=AsyncMock(return_value=mock_result),
744 ) as mock_exe:
745 await dispatch_tool(
746 name="musehub_list_attestations",
747 arguments={"subject": "aria"},
748 user_id="gabriel",
749 session_context=None,
750 )
751 mock_exe.assert_called_once_with(subject="aria", include_revoked=False)
752
753 @pytest.mark.asyncio
754 async def test_dispatch_record_mpay_claim(self) -> None:
755 from musehub.mcp.dispatcher import dispatch_tool
756
757 mock_result = MagicMock(ok=True, data={}, error_code=None, error_message=None, hint=None)
758 with patch(
759 "musehub.services.musehub_mcp_executor.execute_record_mpay_claim",
760 new=AsyncMock(return_value=mock_result),
761 ) as mock_exe:
762 args = {
763 "sender": _SENDER,
764 "recipient": _RECIPIENT,
765 "amount_nano": _AMOUNT_NANO,
766 "nonce_hex": _NONCE_HEX,
767 "signature": "ed25519:abc",
768 "sender_public_key": "ed25519:def",
769 }
770 await dispatch_tool(
771 name="musehub_record_mpay_claim",
772 arguments=args,
773 user_id="gabriel",
774 session_context=None,
775 )
776 mock_exe.assert_called_once()
777
778 @pytest.mark.asyncio
779 async def test_dispatch_get_mpay_ledger(self) -> None:
780 from musehub.mcp.dispatcher import dispatch_tool
781
782 mock_result = MagicMock(ok=True, data={}, error_code=None, error_message=None, hint=None)
783 with patch(
784 "musehub.services.musehub_mcp_executor.execute_get_mpay_ledger",
785 new=AsyncMock(return_value=mock_result),
786 ) as mock_exe:
787 await dispatch_tool(
788 name="musehub_get_mpay_ledger",
789 arguments={"handle": "gabriel", "limit": 50},
790 user_id="gabriel",
791 session_context=None,
792 )
793 mock_exe.assert_called_once_with(handle="gabriel", limit=50)
File History 1 commit
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ 20 days ago