"""Unit tests for musehub/services/musehub_profile.py. Tests the service-layer profile functions directly (no HTTP), covering: - Profile CRUD (create, get_by_username, get_by_user_id, update) - Contribution graph shape and zero-commit baseline - get_public_repos filters private repos - get_session_credits baseline (no sessions → 0) """ from __future__ import annotations import pytest from sqlalchemy.ext.asyncio import AsyncSession from musehub.models.musehub import ProfileUpdateRequest from musehub.services import musehub_profile from tests.factories import create_profile, create_repo from muse.core.types import long_id # --------------------------------------------------------------------------- # create_profile / get_profile_by_username # --------------------------------------------------------------------------- async def test_create_profile_and_get_by_username(db_session: AsyncSession) -> None: profile = await create_profile(db_session, username="artistone", display_name="Artist One") found = await musehub_profile.get_profile_by_username(db_session, "artistone") assert found is not None assert found.handle == "artistone" assert found.display_name == "Artist One" async def test_get_profile_by_username_missing_returns_none(db_session: AsyncSession) -> None: result = await musehub_profile.get_profile_by_username(db_session, "ghost-user") assert result is None async def test_get_profile_by_user_id(db_session: AsyncSession) -> None: profile = await create_profile(db_session, username="byid-user") found = await musehub_profile.get_profile_by_user_id(db_session, profile.identity_id) assert found is not None assert found.handle == "byid-user" async def test_get_profile_by_user_id_missing_returns_none(db_session: AsyncSession) -> None: result = await musehub_profile.get_profile_by_user_id(db_session, "00000000-dead-beef-0000-000000000000") assert result is None # --------------------------------------------------------------------------- # update_profile # --------------------------------------------------------------------------- async def test_update_profile_bio(db_session: AsyncSession) -> None: orm_profile = await create_profile(db_session, username="bio-user", bio="old bio") await musehub_profile.update_profile( db_session, orm_profile, ProfileUpdateRequest(bio="new bio"), ) updated = await musehub_profile.get_profile_by_username(db_session, "bio-user") assert updated is not None assert updated.bio == "new bio" async def test_update_profile_display_name(db_session: AsyncSession) -> None: orm_profile = await create_profile(db_session, username="name-user", display_name="Old Name") await musehub_profile.update_profile( db_session, orm_profile, ProfileUpdateRequest(display_name="New Name"), ) updated = await musehub_profile.get_profile_by_username(db_session, "name-user") assert updated is not None assert updated.display_name == "New Name" # --------------------------------------------------------------------------- # get_public_repos # --------------------------------------------------------------------------- async def test_get_public_repos_returns_public_only(db_session: AsyncSession) -> None: profile = await create_profile(db_session, username="pub-repo-user") await create_repo( db_session, owner="pub-repo-user", owner_user_id=profile.identity_id, slug="public-one", visibility="public", ) await create_repo( db_session, owner="pub-repo-user", owner_user_id=profile.identity_id, slug="private-one", visibility="private", ) repos = await musehub_profile.get_public_repos(db_session, profile.handle) slugs = [r.slug for r in repos] assert "public-one" in slugs assert "private-one" not in slugs async def test_get_public_repos_empty_for_no_repos(db_session: AsyncSession) -> None: profile = await create_profile(db_session, username="no-repos-user") repos = await musehub_profile.get_public_repos(db_session, profile.handle) assert repos == [] # --------------------------------------------------------------------------- # get_session_credits # --------------------------------------------------------------------------- async def test_session_credits_zero_baseline(db_session: AsyncSession) -> None: profile = await create_profile(db_session, username="credit-user") credits = await musehub_profile.get_session_credits(db_session, profile.handle) assert credits == 0 # --------------------------------------------------------------------------- # get_full_profile # --------------------------------------------------------------------------- async def test_get_full_profile_returns_structured_response(db_session: AsyncSession) -> None: profile = await create_profile( db_session, username="full-profile-user", bio="Full profile bio", display_name="Full User", ) result = await musehub_profile.get_full_profile(db_session, "full-profile-user") assert result is not None assert result.username == "full-profile-user" assert result.bio == "Full profile bio" assert result.display_name == "Full User" assert isinstance(result.repos, list) assert result.session_credits == 0 async def test_get_full_profile_missing_returns_none(db_session: AsyncSession) -> None: result = await musehub_profile.get_full_profile(db_session, "nobody-at-all") assert result is None # --------------------------------------------------------------------------- # get_public_repos — domain field # --------------------------------------------------------------------------- async def test_get_public_repos_domain_defaults_to_code(db_session: AsyncSession) -> None: profile = await create_profile(db_session, username="domain-default-user") await create_repo( db_session, owner="domain-default-user", owner_user_id=profile.identity_id, slug="no-domain-repo", visibility="public", ) repos = await musehub_profile.get_public_repos(db_session, profile.handle) assert len(repos) == 1 assert repos[0].domain == "code" async def test_get_public_repos_preserves_explicit_domain(db_session: AsyncSession) -> None: """domain_id is a plain string label — no musehub_domains join required.""" profile = await create_profile(db_session, username="domain-explicit-user") await create_repo( db_session, owner="domain-explicit-user", owner_user_id=profile.identity_id, slug="midi-repo", visibility="public", domain_id="midi", ) repos = await musehub_profile.get_public_repos(db_session, profile.handle) assert len(repos) == 1 assert repos[0].domain == "midi" # --------------------------------------------------------------------------- # build_activity_canvas — domain-driven, no phantom rows # --------------------------------------------------------------------------- async def test_canvas_empty_for_user_with_no_repos(db_session: AsyncSession) -> None: """Canvas returns [] for a user with no repos — no phantom domain rows.""" profile = await create_profile(db_session, username="no-repos-canvas-user") canvas = await musehub_profile.build_activity_canvas(db_session, profile.handle) assert canvas == [] async def test_canvas_shows_only_domains_with_commits(db_session: AsyncSession) -> None: """Canvas only includes domains where the user has commits. domain_id is a plain string.""" import secrets from datetime import datetime, timezone, timedelta from musehub.db.musehub_repo_models import MusehubCommit, MusehubCommitRef, MusehubRepo from musehub.core.genesis import compute_repo_id, compute_identity_id handle = f"canvas-real-{secrets.token_hex(4)}" await create_profile(db_session, username=handle) owner_id = compute_identity_id(handle.encode()) ts = datetime.now(tz=timezone.utc) - timedelta(days=5) # Code repo with one commit — must appear code_repo = MusehubRepo( repo_id=compute_repo_id(owner_id, "code-proj", "code", ts.isoformat()), name="code-proj", owner=handle, slug="code-proj", visibility="public", owner_user_id=owner_id, domain_id="code", ) db_session.add(code_repo) _cid1 = long_id(secrets.token_hex(32)) db_session.add(MusehubCommit( commit_id=_cid1, branch="main", parent_ids=[], author=handle, message="init", timestamp=ts, )) db_session.add(MusehubCommitRef(repo_id=code_repo.repo_id, commit_id=_cid1)) # Midi repo with no commits — must NOT appear midi_repo = MusehubRepo( repo_id=compute_repo_id(owner_id, "midi-proj", "midi", ts.isoformat()), name="midi-proj", owner=handle, slug="midi-proj", visibility="public", owner_user_id=owner_id, domain_id="midi", ) db_session.add(midi_repo) await db_session.commit() canvas = await musehub_profile.build_activity_canvas(db_session, handle) domain_names = [d.domain for d in canvas] assert "code" in domain_names, "code domain must appear — it has commits" assert "midi" not in domain_names, "midi domain must not appear — no commits" async def test_canvas_null_domain_id_counts_as_code(db_session: AsyncSession) -> None: """Repos with domain_id=null must appear in the 'code' bucket on the canvas.""" import secrets from datetime import datetime, timezone, timedelta from musehub.db.musehub_repo_models import MusehubCommit, MusehubCommitRef, MusehubRepo from musehub.core.genesis import compute_repo_id, compute_identity_id handle = f"canvas-null-{secrets.token_hex(4)}" await create_profile(db_session, username=handle) owner_id = compute_identity_id(handle.encode()) ts = datetime.now(tz=timezone.utc) - timedelta(days=3) repo = MusehubRepo( repo_id=compute_repo_id(owner_id, "legacy-repo", "", ts.isoformat()), name="legacy-repo", owner=handle, slug="legacy-repo", visibility="public", owner_user_id=owner_id, domain_id=None, ) db_session.add(repo) _cid2 = long_id(secrets.token_hex(32)) db_session.add(MusehubCommit( commit_id=_cid2, branch="main", parent_ids=[], author=handle, message="init", timestamp=ts, )) db_session.add(MusehubCommitRef(repo_id=repo.repo_id, commit_id=_cid2)) await db_session.commit() canvas = await musehub_profile.build_activity_canvas(db_session, handle) domain_names = [d.domain for d in canvas] assert "code" in domain_names, "null domain_id repos must count toward 'code'" async def test_canvas_excludes_internal_domains(db_session: AsyncSession) -> None: """identity and social repos never appear in the activity canvas.""" import secrets from datetime import datetime, timezone, timedelta from musehub.db.musehub_repo_models import MusehubCommit, MusehubCommitRef, MusehubRepo from musehub.core.genesis import compute_repo_id, compute_identity_id handle = f"canvas-internal-{secrets.token_hex(4)}" await create_profile(db_session, username=handle) owner_id = compute_identity_id(handle.encode()) ts = datetime.now(tz=timezone.utc) - timedelta(days=2) for domain_id in ("identity", "social"): repo = MusehubRepo( repo_id=compute_repo_id(owner_id, f"{domain_id}-repo", domain_id, ts.isoformat()), name=f"{domain_id}-repo", owner=handle, slug=f"{domain_id}-repo", visibility="private", owner_user_id=owner_id, domain_id=domain_id, ) db_session.add(repo) _cid = long_id(secrets.token_hex(32)) db_session.add(MusehubCommit( commit_id=_cid, branch="main", parent_ids=[], author=handle, message="sys", timestamp=ts, )) db_session.add(MusehubCommitRef(repo_id=repo.repo_id, commit_id=_cid)) await db_session.commit() canvas = await musehub_profile.build_activity_canvas(db_session, handle) domain_names = [d.domain for d in canvas] assert "identity" not in domain_names assert "social" not in domain_names