gabriel / musehub public
test_profile_reimagination.py python
736 lines 27.0 KB
Raw
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65 refactor: enforce gRPC framing on all MWP wire traffic Sonnet 4.6 minor ⚠ breaking 156 days ago
1 """Tests for Issue #1 — Profile Page Reimagination.
2
3 Covers:
4 1. Genesis ID functions for attestation and MPay
5 2. Canonical form / determinism / collision resistance
6 3. Attestation signature verification (pure, no DB)
7 4. MPay signature verification (pure, no DB)
8 5. Pydantic schema validation (ProfileManifest, AttestationResponse, MPayClaimResponse)
9 6. Service layer — attestation CRUD (async DB mocks)
10 7. Service layer — MPay claim record + ledger (async DB mocks)
11 8. Service layer — profile manifest builder (async DB mocks)
12 9. API route smoke tests (FastAPI TestClient)
13 """
14 from __future__ import annotations
15
16 import base64
17 import json
18 from datetime import datetime, timezone
19 from unittest.mock import AsyncMock, MagicMock, patch
20
21 import pytest
22 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
23
24 from musehub.core.genesis import compute_attestation_id, compute_mpay_claim_id
25 from musehub.models.musehub import (
26 ActivityDomain,
27 AttestationBadge,
28 AttestationRequest,
29 AttestationResponse,
30 MPayClaimRequest,
31 MPayClaimResponse,
32 MPayLedgerResponse,
33 OrgManifest,
34 ProfileManifest,
35 ProfileRepoSummary,
36 TrustChainEntry,
37 )
38 from musehub.services.musehub_attestations import (
39 attestation_to_badge,
40 verify_attestation_signature,
41 )
42 from musehub.services.musehub_mpay import verify_mpay_signature
43
44 # ---------------------------------------------------------------------------
45 # Shared fixtures
46 # ---------------------------------------------------------------------------
47
48 _TS = "2026-04-21T12:00:00+00:00"
49 _ATTESTER = "gabriel"
50 _SUBJECT = "aria"
51 _CLAIM = '{"type":"human","confidence":0.99}'
52 _ATTESTER2 = "maestro"
53
54 _SENDER = "gabriel"
55 _RECIPIENT = "aria"
56 _AMOUNT_NANO = 1_000_000
57 _NONCE_HEX = "a" * 64
58
59
60 def _make_ed25519_pair() -> tuple[Ed25519PrivateKey, str, str]:
61 """Return (privkey, pubkey_prefixed, sig_for_attest_message)."""
62 privkey = Ed25519PrivateKey.generate()
63 pubkey_bytes = privkey.public_key().public_bytes_raw()
64 pubkey_b64 = base64.urlsafe_b64encode(pubkey_bytes).rstrip(b"=").decode()
65 return privkey, f"ed25519:{pubkey_b64}"
66
67
68 def _sign_attest(
69 privkey: Ed25519PrivateKey,
70 attester: str,
71 subject: str,
72 claim: str,
73 ts: str,
74 ) -> str:
75 msg = f"ATTEST\n{attester}\n{subject}\n{claim}\n{ts}".encode()
76 sig_bytes = privkey.sign(msg)
77 sig_b64 = base64.urlsafe_b64encode(sig_bytes).rstrip(b"=").decode()
78 return f"ed25519:{sig_b64}"
79
80
81 def _sign_mpay(
82 privkey: Ed25519PrivateKey,
83 sender: str,
84 recipient: str,
85 amount_nano: int,
86 nonce_hex: str,
87 ) -> str:
88 msg = f"MPAY\n{sender}\n{recipient}\n{amount_nano}\n{nonce_hex}".encode()
89 sig_bytes = privkey.sign(msg)
90 sig_b64 = base64.urlsafe_b64encode(sig_bytes).rstrip(b"=").decode()
91 return f"ed25519:{sig_b64}"
92
93
94 # ---------------------------------------------------------------------------
95 # 1 & 2. Genesis ID canonical form, determinism, collision resistance
96 # ---------------------------------------------------------------------------
97
98
99 class TestAttestationGenesisId:
100 def test_canonical_form(self):
101 aid = compute_attestation_id(_ATTESTER, _SUBJECT, _CLAIM, _TS)
102 assert aid.startswith("sha256:")
103 assert len(aid) == 71
104
105 def test_deterministic(self):
106 a1 = compute_attestation_id(_ATTESTER, _SUBJECT, _CLAIM, _TS)
107 a2 = compute_attestation_id(_ATTESTER, _SUBJECT, _CLAIM, _TS)
108 assert a1 == a2
109
110 def test_distinct_attester(self):
111 a1 = compute_attestation_id(_ATTESTER, _SUBJECT, _CLAIM, _TS)
112 a2 = compute_attestation_id(_ATTESTER2, _SUBJECT, _CLAIM, _TS)
113 assert a1 != a2
114
115 def test_distinct_subject(self):
116 a1 = compute_attestation_id(_ATTESTER, _SUBJECT, _CLAIM, _TS)
117 a2 = compute_attestation_id(_ATTESTER, "other_subject", _CLAIM, _TS)
118 assert a1 != a2
119
120 def test_distinct_claim(self):
121 a1 = compute_attestation_id(_ATTESTER, _SUBJECT, _CLAIM, _TS)
122 a2 = compute_attestation_id(_ATTESTER, _SUBJECT, '{"type":"org"}', _TS)
123 assert a1 != a2
124
125 def test_distinct_timestamp(self):
126 a1 = compute_attestation_id(_ATTESTER, _SUBJECT, _CLAIM, _TS)
127 a2 = compute_attestation_id(_ATTESTER, _SUBJECT, _CLAIM, "2026-05-01T00:00:00+00:00")
128 assert a1 != a2
129
130
131 class TestMPayGenesisId:
132 def test_canonical_form(self):
133 cid = compute_mpay_claim_id(_SENDER, _RECIPIENT, _AMOUNT_NANO, _NONCE_HEX)
134 assert cid.startswith("sha256:")
135 assert len(cid) == 71
136
137 def test_deterministic(self):
138 c1 = compute_mpay_claim_id(_SENDER, _RECIPIENT, _AMOUNT_NANO, _NONCE_HEX)
139 c2 = compute_mpay_claim_id(_SENDER, _RECIPIENT, _AMOUNT_NANO, _NONCE_HEX)
140 assert c1 == c2
141
142 def test_distinct_amount(self):
143 c1 = compute_mpay_claim_id(_SENDER, _RECIPIENT, 1_000_000, _NONCE_HEX)
144 c2 = compute_mpay_claim_id(_SENDER, _RECIPIENT, 2_000_000, _NONCE_HEX)
145 assert c1 != c2
146
147 def test_distinct_nonce(self):
148 c1 = compute_mpay_claim_id(_SENDER, _RECIPIENT, _AMOUNT_NANO, "a" * 64)
149 c2 = compute_mpay_claim_id(_SENDER, _RECIPIENT, _AMOUNT_NANO, "b" * 64)
150 assert c1 != c2
151
152
153 # ---------------------------------------------------------------------------
154 # 3. Attestation signature verification
155 # ---------------------------------------------------------------------------
156
157
158 class TestAttestationSignatureVerification:
159 def test_valid_signature(self):
160 privkey, pubkey = _make_ed25519_pair()
161 sig = _sign_attest(privkey, _ATTESTER, _SUBJECT, _CLAIM, _TS)
162 ok, reason = verify_attestation_signature(
163 _ATTESTER, _SUBJECT, _CLAIM, _TS, sig, pubkey
164 )
165 assert ok is True
166 assert reason == ""
167
168 def test_wrong_message_fails(self):
169 privkey, pubkey = _make_ed25519_pair()
170 # Sign with different claim
171 sig = _sign_attest(privkey, _ATTESTER, _SUBJECT, '{"type":"org"}', _TS)
172 ok, reason = verify_attestation_signature(
173 _ATTESTER, _SUBJECT, _CLAIM, _TS, sig, pubkey
174 )
175 assert ok is False
176
177 def test_missing_prefix_fails(self):
178 privkey, pubkey = _make_ed25519_pair()
179 sig = _sign_attest(privkey, _ATTESTER, _SUBJECT, _CLAIM, _TS)
180 bare_sig = sig[len("ed25519:"):]
181 ok, reason = verify_attestation_signature(
182 _ATTESTER, _SUBJECT, _CLAIM, _TS, bare_sig, pubkey
183 )
184 assert ok is False
185 assert "ed25519:" in reason
186
187 def test_wrong_public_key_fails(self):
188 privkey, _ = _make_ed25519_pair()
189 _, other_pubkey = _make_ed25519_pair()
190 sig = _sign_attest(privkey, _ATTESTER, _SUBJECT, _CLAIM, _TS)
191 ok, _ = verify_attestation_signature(
192 _ATTESTER, _SUBJECT, _CLAIM, _TS, sig, other_pubkey
193 )
194 assert ok is False
195
196
197 # ---------------------------------------------------------------------------
198 # 4. MPay signature verification
199 # ---------------------------------------------------------------------------
200
201
202 class TestMPaySignatureVerification:
203 def test_valid_signature(self):
204 privkey, pubkey = _make_ed25519_pair()
205 sig = _sign_mpay(privkey, _SENDER, _RECIPIENT, _AMOUNT_NANO, _NONCE_HEX)
206 ok, reason = verify_mpay_signature(
207 _SENDER, _RECIPIENT, _AMOUNT_NANO, _NONCE_HEX, sig, pubkey
208 )
209 assert ok is True
210 assert reason == ""
211
212 def test_tampered_amount_fails(self):
213 privkey, pubkey = _make_ed25519_pair()
214 sig = _sign_mpay(privkey, _SENDER, _RECIPIENT, _AMOUNT_NANO, _NONCE_HEX)
215 # Verify with different amount
216 ok, _ = verify_mpay_signature(
217 _SENDER, _RECIPIENT, _AMOUNT_NANO + 1, _NONCE_HEX, sig, pubkey
218 )
219 assert ok is False
220
221 def test_missing_prefix_fails(self):
222 privkey, pubkey = _make_ed25519_pair()
223 sig = _sign_mpay(privkey, _SENDER, _RECIPIENT, _AMOUNT_NANO, _NONCE_HEX)
224 bare_sig = sig[len("ed25519:"):]
225 ok, reason = verify_mpay_signature(
226 _SENDER, _RECIPIENT, _AMOUNT_NANO, _NONCE_HEX, bare_sig, pubkey
227 )
228 assert ok is False
229
230
231 # ---------------------------------------------------------------------------
232 # 5. Pydantic schema validation
233 # ---------------------------------------------------------------------------
234
235
236 class TestPydanticSchemas:
237 def _make_repo_summary(self) -> ProfileRepoSummary:
238 return ProfileRepoSummary(
239 repo_id="sha256:" + "a" * 64,
240 name="test-repo",
241 owner="gabriel",
242 slug="test-repo",
243 visibility="public",
244 last_activity_at=None,
245 created_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
246 )
247
248 def test_profile_manifest_human(self):
249 manifest = ProfileManifest(
250 identity_id="sha256:" + "a" * 64,
251 handle="gabriel",
252 identity_type="human",
253 avax_address="0xABCDEF",
254 created_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
255 updated_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
256 )
257 assert manifest.identity_type == "human"
258 assert manifest.avax_address == "0xABCDEF"
259 assert manifest.org is None
260 assert manifest.trust_chain == []
261
262 def test_profile_manifest_agent(self):
263 manifest = ProfileManifest(
264 identity_id="sha256:" + "b" * 64,
265 handle="aria",
266 identity_type="agent",
267 agent_model="claude-sonnet-4-6",
268 agent_capabilities=["push", "pull"],
269 trust_chain=[
270 TrustChainEntry(handle="aria", identity_type="agent", spawned_by="gabriel"),
271 TrustChainEntry(handle="gabriel", identity_type="human", spawned_by=None),
272 ],
273 created_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
274 updated_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
275 )
276 assert manifest.identity_type == "agent"
277 assert len(manifest.trust_chain) == 2
278 assert manifest.trust_chain[-1].identity_type == "human"
279
280 def test_profile_manifest_org(self):
281 manifest = ProfileManifest(
282 identity_id="sha256:" + "c" * 64,
283 handle="acme",
284 identity_type="org",
285 org=OrgManifest(members=["gabriel", "aria"], quorum=2),
286 created_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
287 updated_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
288 )
289 assert manifest.org is not None
290 assert manifest.org.quorum == 2
291
292 def test_activity_domain_grid_length(self):
293 grid = [0] * 364
294 domain = ActivityDomain(domain="code", grid=grid, peak=0, total=0)
295 assert len(domain.grid) == 364
296
297 def test_attestation_response(self):
298 issued = datetime(2026, 4, 21, 12, 0, 0, tzinfo=timezone.utc)
299 resp = AttestationResponse(
300 attestation_id="sha256:" + "d" * 64,
301 attester="gabriel",
302 subject="aria",
303 claim='{"type":"human"}',
304 signature="ed25519:fakesig",
305 attester_public_key="ed25519:fakepubkey",
306 issued_at=issued,
307 )
308 assert resp.revoked_at is None
309
310 def test_attestation_badge_claim_type(self):
311 issued = datetime(2026, 4, 21, 12, 0, 0, tzinfo=timezone.utc)
312 a = AttestationResponse(
313 attestation_id="sha256:" + "e" * 64,
314 attester="gabriel",
315 subject="aria",
316 claim='{"type":"human","confidence":0.99}',
317 signature="ed25519:sig",
318 attester_public_key="ed25519:pk",
319 issued_at=issued,
320 )
321 badge = attestation_to_badge(a)
322 assert badge.claim_type == "type"
323
324 def test_mpay_claim_response(self):
325 now = datetime(2026, 4, 21, tzinfo=timezone.utc)
326 resp = MPayClaimResponse(
327 claim_id="sha256:" + "f" * 64,
328 sender="gabriel",
329 recipient="aria",
330 amount_nano=1_000_000,
331 nonce_hex="a" * 64,
332 signature="ed25519:sig",
333 sender_public_key="ed25519:pk",
334 created_at=now,
335 )
336 assert resp.confirmed_at is None
337 assert resp.voided_at is None
338
339 def test_mpay_ledger_totals(self):
340 now = datetime(2026, 4, 21, tzinfo=timezone.utc)
341
342 def _claim(amount: int) -> MPayClaimResponse:
343 return MPayClaimResponse(
344 claim_id="sha256:" + "a" * 64,
345 sender="gabriel",
346 recipient="aria",
347 amount_nano=amount,
348 nonce_hex="b" * 64,
349 signature="ed25519:sig",
350 sender_public_key="ed25519:pk",
351 created_at=now,
352 )
353
354 ledger = MPayLedgerResponse(
355 handle="gabriel",
356 sent=[_claim(500_000), _claim(250_000)],
357 received=[_claim(1_000_000)],
358 total_sent_nano=750_000,
359 total_received_nano=1_000_000,
360 )
361 assert ledger.total_sent_nano == 750_000
362 assert ledger.total_received_nano == 1_000_000
363
364
365 # ---------------------------------------------------------------------------
366 # 6. Service layer — attestation (async DB mocks)
367 # ---------------------------------------------------------------------------
368
369
370 class TestAttestationService:
371 @pytest.mark.asyncio
372 async def test_issue_attestation_valid_sig(self):
373 privkey, pubkey = _make_ed25519_pair()
374 issued_at = datetime(2026, 4, 21, 12, 0, 0, tzinfo=timezone.utc)
375 sig = _sign_attest(privkey, _ATTESTER, _SUBJECT, _CLAIM, issued_at.isoformat())
376
377 req = AttestationRequest(
378 attester=_ATTESTER,
379 subject=_SUBJECT,
380 claim=_CLAIM,
381 signature=sig,
382 attester_public_key=pubkey,
383 issued_at=issued_at,
384 )
385
386 # Mock DB: no existing record → insert
387 mock_db = AsyncMock()
388 mock_result = MagicMock()
389 mock_result.mappings.return_value.one_or_none.return_value = None
390 mock_db.execute.return_value = mock_result
391
392 from musehub.services.musehub_attestations import issue_attestation
393 result = await issue_attestation(mock_db, req)
394
395 assert result.attester == _ATTESTER
396 assert result.subject == _SUBJECT
397 assert result.revoked_at is None
398 mock_db.commit.assert_called_once()
399
400 @pytest.mark.asyncio
401 async def test_issue_attestation_invalid_sig_raises(self):
402 _, pubkey = _make_ed25519_pair()
403 issued_at = datetime(2026, 4, 21, 12, 0, 0, tzinfo=timezone.utc)
404
405 req = AttestationRequest(
406 attester=_ATTESTER,
407 subject=_SUBJECT,
408 claim=_CLAIM,
409 signature="ed25519:invalidsignaturedata",
410 attester_public_key=pubkey,
411 issued_at=issued_at,
412 )
413 mock_db = AsyncMock()
414 mock_result = MagicMock()
415 mock_result.mappings.return_value.one_or_none.return_value = None
416 mock_db.execute.return_value = mock_result
417
418 from musehub.services.musehub_attestations import issue_attestation
419 with pytest.raises(ValueError, match="Invalid attestation signature"):
420 await issue_attestation(mock_db, req)
421
422 @pytest.mark.asyncio
423 async def test_issue_attestation_idempotent(self):
424 privkey, pubkey = _make_ed25519_pair()
425 issued_at = datetime(2026, 4, 21, 12, 0, 0, tzinfo=timezone.utc)
426 sig = _sign_attest(privkey, _ATTESTER, _SUBJECT, _CLAIM, issued_at.isoformat())
427 aid = compute_attestation_id(_ATTESTER, _SUBJECT, _CLAIM, issued_at.isoformat())
428
429 req = AttestationRequest(
430 attester=_ATTESTER,
431 subject=_SUBJECT,
432 claim=_CLAIM,
433 signature=sig,
434 attester_public_key=pubkey,
435 issued_at=issued_at,
436 )
437
438 existing_row = {
439 "attestation_id": aid,
440 "attester": _ATTESTER,
441 "subject": _SUBJECT,
442 "claim": _CLAIM,
443 "signature": sig,
444 "attester_public_key": pubkey,
445 "issued_at": issued_at,
446 "revoked_at": None,
447 }
448 mock_db = AsyncMock()
449 mock_result = MagicMock()
450 mock_result.mappings.return_value.one_or_none.return_value = existing_row
451 mock_db.execute.return_value = mock_result
452
453 from musehub.services.musehub_attestations import issue_attestation
454 result = await issue_attestation(mock_db, req)
455
456 # Should return existing without insert
457 assert result.attestation_id == aid
458 mock_db.commit.assert_not_called()
459
460 @pytest.mark.asyncio
461 async def test_revoke_attestation_wrong_revoker_raises(self):
462 aid = compute_attestation_id(_ATTESTER, _SUBJECT, _CLAIM, _TS)
463 existing_row = {
464 "attestation_id": aid,
465 "attester": _ATTESTER,
466 "subject": _SUBJECT,
467 "claim": _CLAIM,
468 "signature": "ed25519:sig",
469 "attester_public_key": "ed25519:pk",
470 "issued_at": datetime(2026, 4, 21, tzinfo=timezone.utc),
471 "revoked_at": None,
472 }
473 mock_db = AsyncMock()
474 mock_result = MagicMock()
475 mock_result.mappings.return_value.one_or_none.return_value = existing_row
476 mock_db.execute.return_value = mock_result
477
478 from musehub.services.musehub_attestations import revoke_attestation
479 with pytest.raises(PermissionError):
480 await revoke_attestation(mock_db, aid, revoker="not_gabriel")
481
482 @pytest.mark.asyncio
483 async def test_revoke_attestation_not_found_raises(self):
484 mock_db = AsyncMock()
485 mock_result = MagicMock()
486 mock_result.mappings.return_value.one_or_none.return_value = None
487 mock_db.execute.return_value = mock_result
488
489 from musehub.services.musehub_attestations import revoke_attestation
490 with pytest.raises(KeyError):
491 await revoke_attestation(mock_db, "sha256:" + "x" * 64, revoker=_ATTESTER)
492
493
494 # ---------------------------------------------------------------------------
495 # 7. Service layer — MPay claim (async DB mocks)
496 # ---------------------------------------------------------------------------
497
498
499 class TestMPayService:
500 @pytest.mark.asyncio
501 async def test_record_claim_valid_sig(self):
502 privkey, pubkey = _make_ed25519_pair()
503 sig = _sign_mpay(privkey, _SENDER, _RECIPIENT, _AMOUNT_NANO, _NONCE_HEX)
504
505 req = MPayClaimRequest(
506 sender=_SENDER,
507 recipient=_RECIPIENT,
508 amount_nano=_AMOUNT_NANO,
509 nonce_hex=_NONCE_HEX,
510 signature=sig,
511 sender_public_key=pubkey,
512 )
513
514 mock_db = AsyncMock()
515 mock_result = MagicMock()
516 mock_result.mappings.return_value.one_or_none.return_value = None
517 mock_db.execute.return_value = mock_result
518
519 from musehub.services.musehub_mpay import record_mpay_claim
520 result = await record_mpay_claim(mock_db, req)
521
522 assert result.sender == _SENDER
523 assert result.recipient == _RECIPIENT
524 assert result.amount_nano == _AMOUNT_NANO
525 mock_db.commit.assert_called_once()
526
527 @pytest.mark.asyncio
528 async def test_record_claim_invalid_sig_raises(self):
529 _, pubkey = _make_ed25519_pair()
530
531 req = MPayClaimRequest(
532 sender=_SENDER,
533 recipient=_RECIPIENT,
534 amount_nano=_AMOUNT_NANO,
535 nonce_hex=_NONCE_HEX,
536 signature="ed25519:badsig",
537 sender_public_key=pubkey,
538 )
539 mock_db = AsyncMock()
540 mock_result = MagicMock()
541 mock_result.mappings.return_value.one_or_none.return_value = None
542 mock_db.execute.return_value = mock_result
543
544 from musehub.services.musehub_mpay import record_mpay_claim
545 with pytest.raises(ValueError, match="Invalid MPay signature"):
546 await record_mpay_claim(mock_db, req)
547
548 @pytest.mark.asyncio
549 async def test_record_claim_idempotent(self):
550 privkey, pubkey = _make_ed25519_pair()
551 sig = _sign_mpay(privkey, _SENDER, _RECIPIENT, _AMOUNT_NANO, _NONCE_HEX)
552 claim_id = compute_mpay_claim_id(_SENDER, _RECIPIENT, _AMOUNT_NANO, _NONCE_HEX)
553 now = datetime(2026, 4, 21, tzinfo=timezone.utc)
554
555 existing_row = {
556 "claim_id": claim_id,
557 "sender": _SENDER,
558 "recipient": _RECIPIENT,
559 "amount_nano": _AMOUNT_NANO,
560 "nonce_hex": _NONCE_HEX,
561 "signature": sig,
562 "sender_public_key": pubkey,
563 "memo": None,
564 "created_at": now,
565 "confirmed_at": None,
566 "voided_at": None,
567 }
568 mock_db = AsyncMock()
569 mock_result = MagicMock()
570 mock_result.mappings.return_value.one_or_none.return_value = existing_row
571 mock_db.execute.return_value = mock_result
572
573 req = MPayClaimRequest(
574 sender=_SENDER,
575 recipient=_RECIPIENT,
576 amount_nano=_AMOUNT_NANO,
577 nonce_hex=_NONCE_HEX,
578 signature=sig,
579 sender_public_key=pubkey,
580 )
581
582 from musehub.services.musehub_mpay import record_mpay_claim
583 result = await record_mpay_claim(mock_db, req)
584
585 assert result.claim_id == claim_id
586 mock_db.commit.assert_not_called()
587
588
589 # ---------------------------------------------------------------------------
590 # 8. Service layer — profile manifest builder
591 # ---------------------------------------------------------------------------
592
593
594 class TestProfileManifestBuilder:
595 def _make_identity(
596 self,
597 handle: str = "gabriel",
598 identity_type: str = "human",
599 spawned_by: str | None = None,
600 ):
601 identity = MagicMock()
602 identity.identity_id = "sha256:" + "a" * 64
603 identity.handle = handle
604 identity.identity_type = identity_type
605 identity.display_name = f"Display {handle}"
606 identity.bio = None
607 identity.avatar_url = None
608 identity.location = None
609 identity.website_url = None
610 identity.social_url = None
611 identity.is_verified = False
612 identity.cc_license = None
613 identity.pinned_repo_ids = []
614 identity.avax_address = "0xABC" if identity_type == "human" else None
615 identity.agent_model = "claude-sonnet-4-6" if identity_type == "agent" else None
616 identity.agent_capabilities = ["push"] if identity_type == "agent" else []
617 identity.org_members = ["gabriel", "aria"] if identity_type == "org" else None
618 identity.org_quorum = 2 if identity_type == "org" else None
619 identity.org_treasury_address = None
620 identity.spawned_by = spawned_by
621 identity.deleted_at = None
622 identity.created_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
623 identity.updated_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
624 return identity
625
626 @pytest.mark.asyncio
627 async def test_manifest_returns_none_for_unknown_handle(self):
628 from musehub.services.musehub_profile import build_profile_manifest
629
630 with patch(
631 "musehub.services.musehub_profile.get_profile_by_username",
632 new=AsyncMock(return_value=None),
633 ):
634 result = await build_profile_manifest(AsyncMock(), "unknown")
635 assert result is None
636
637 @pytest.mark.asyncio
638 async def test_manifest_human_fields(self):
639 from musehub.services.musehub_profile import build_profile_manifest
640
641 identity = self._make_identity("gabriel", "human")
642
643 with (
644 patch("musehub.services.musehub_profile.get_profile_by_username", new=AsyncMock(return_value=identity)),
645 patch("musehub.services.musehub_profile.get_public_repos", new=AsyncMock(return_value=[])),
646 patch("musehub.services.musehub_profile.build_activity_canvas", new=AsyncMock(return_value=[])),
647 ):
648 result = await build_profile_manifest(AsyncMock(), "gabriel")
649
650 assert result is not None
651 assert result.identity_type == "human"
652 assert result.avax_address == "0xABC"
653 assert result.org is None
654 assert result.trust_chain == []
655
656 @pytest.mark.asyncio
657 async def test_manifest_org_fields(self):
658 from musehub.services.musehub_profile import build_profile_manifest
659
660 identity = self._make_identity("acme", "org")
661
662 with (
663 patch("musehub.services.musehub_profile.get_profile_by_username", new=AsyncMock(return_value=identity)),
664 patch("musehub.services.musehub_profile.get_public_repos", new=AsyncMock(return_value=[])),
665 patch("musehub.services.musehub_profile.build_activity_canvas", new=AsyncMock(return_value=[])),
666 ):
667 result = await build_profile_manifest(AsyncMock(), "acme")
668
669 assert result is not None
670 assert result.org is not None
671 assert result.org.members == ["gabriel", "aria"]
672 assert result.org.quorum == 2
673 assert result.avax_address is None
674
675 @pytest.mark.asyncio
676 async def test_manifest_mpay_totals_passed_through(self):
677 from musehub.services.musehub_profile import build_profile_manifest
678
679 identity = self._make_identity("gabriel", "human")
680
681 with (
682 patch("musehub.services.musehub_profile.get_profile_by_username", new=AsyncMock(return_value=identity)),
683 patch("musehub.services.musehub_profile.get_public_repos", new=AsyncMock(return_value=[])),
684 patch("musehub.services.musehub_profile.build_activity_canvas", new=AsyncMock(return_value=[])),
685 ):
686 result = await build_profile_manifest(
687 AsyncMock(), "gabriel", mpay_sent_nano=500_000, mpay_received_nano=1_000_000
688 )
689
690 assert result is not None
691 assert result.mpay_total_sent_nano == 500_000
692 assert result.mpay_total_received_nano == 1_000_000
693
694
695 # ---------------------------------------------------------------------------
696 # 9. Activity canvas unit tests
697 # ---------------------------------------------------------------------------
698
699
700 class TestActivityCanvas:
701 def test_grid_index_today_is_last(self):
702 from musehub.services.musehub_profile import _date_to_grid_index, _GRID_DAYS
703
704 today = datetime(2026, 4, 21, tzinfo=timezone.utc)
705 idx = _date_to_grid_index(today, today)
706 assert idx == _GRID_DAYS - 1
707
708 def test_grid_index_out_of_range_returns_none(self):
709 from musehub.services.musehub_profile import _date_to_grid_index
710
711 today = datetime(2026, 4, 21, tzinfo=timezone.utc)
712 past = datetime(2025, 1, 1, tzinfo=timezone.utc) # >364 days ago
713 idx = _date_to_grid_index(today, past)
714 assert idx is None
715
716 def test_grid_index_future_returns_none(self):
717 from musehub.services.musehub_profile import _date_to_grid_index
718
719 today = datetime(2026, 4, 21, tzinfo=timezone.utc)
720 future = datetime(2026, 5, 1, tzinfo=timezone.utc)
721 idx = _date_to_grid_index(today, future)
722 assert idx is None
723
724 def test_empty_grid_length(self):
725 from musehub.services.musehub_profile import _empty_grid, _GRID_DAYS
726
727 assert len(_empty_grid()) == _GRID_DAYS
728
729 def test_grid_to_domain(self):
730 from musehub.services.musehub_profile import _grid_to_domain
731
732 grid = [0] * 363 + [5]
733 domain = _grid_to_domain("code", grid)
734 assert domain.peak == 5
735 assert domain.total == 5
736 assert domain.domain == "code"
File History 1 commit
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65 refactor: enforce gRPC framing on all MWP wire traffic Sonnet 4.6 minor ⚠ 156 days ago