"""TDD — musehub#224: user/identity account deletion (self-service + admin). Self-service deletion already existed (DELETE /api/me, GDPR Article 17) but had zero integration test coverage and no CLI/MCP wrapper. Admin-initiated deletion of a *different* account did not exist at all -- there was no way to remove a throwaway/spam/test account that wasn't the caller's own. Both paths now share one effect via musehub_account.delete_identity_account: hard-delete all registered auth keys, soft-delete the identity (deleted_at set). This file covers both paths through the real routes. Test IDs: AD_01 — self-delete (DELETE /api/me) soft-deletes identity + hard-deletes keys AD_02 — admin-delete (DELETE /api/admin/users/{handle}) does the same for another account AD_03 — non-admin caller gets 403 on the admin path AD_04 — unknown handle on the admin path gets 404 AD_05 — already-deleted handle on the admin path gets 404 (not re-deletable) AD_06 — admin cannot be blocked from deleting an account via the self path's auth shape (sanity: admin path works regardless of caller's own identity existing) """ from __future__ import annotations import hashlib as _hashlib from datetime import datetime, timezone import pytest from httpx import AsyncClient from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from muse.core.types import encode_pubkey, long_id from musehub.auth.request_signing import MSignContext, require_signed_request from musehub.core.genesis import compute_identity_id, compute_key_id from musehub.crypto.keys import key_fingerprint from musehub.db.musehub_auth_models import MusehubAuthKey from musehub.db.musehub_identity_models import MusehubIdentity from musehub.main import app _NOW = datetime.now(timezone.utc) _COUNTER: list[int] = [0] def _uid(tag: str = "") -> str: _COUNTER[0] += 1 return f"ad{tag}{_COUNTER[0]}" def _key_for(handle: str) -> str: raw = _hashlib.sha256(handle.encode()).digest() return encode_pubkey("ed25519", raw) def _make_ctx( is_admin: bool = False, handle: str = "actor", identity_id: str | None = None ) -> MSignContext: return MSignContext( identity_id=identity_id or long_id(handle.ljust(64, "0")[:64]), handle=handle, is_agent=False, is_admin=is_admin, ) async def _seed_identity_with_key(session: AsyncSession, handle: str) -> MusehubIdentity: import base64 pubkey_b64 = _key_for(handle) identity_id = compute_identity_id(handle.encode()) identity = MusehubIdentity( identity_id=identity_id, handle=handle, identity_type="human", is_admin=False, agent_capabilities=[], pinned_repo_ids=[], is_verified=False, created_at=_NOW, updated_at=_NOW, ) session.add(identity) await session.flush() _, raw_b64 = pubkey_b64.split(":", 1) raw_bytes = base64.urlsafe_b64decode(raw_b64 + "==") fp = key_fingerprint(raw_bytes) key = MusehubAuthKey( key_id=compute_key_id(identity_id, pubkey_b64), identity_id=identity_id, algorithm="ed25519", public_key_b64=pubkey_b64, fingerprint=fp, label="test-key", created_at=_NOW, ) session.add(key) await session.flush() return identity async def _key_count(session: AsyncSession, identity_id: str) -> int: rows = (await session.execute( select(MusehubAuthKey).where(MusehubAuthKey.identity_id == identity_id) )).scalars().all() return len(rows) # ── AD_01 — self-delete ────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_ad01_self_delete_soft_deletes_identity_and_hard_deletes_keys( client: AsyncClient, db_session: AsyncSession ) -> None: handle = f"selfdel-{_uid()}" identity = await _seed_identity_with_key(db_session, handle) await db_session.commit() identity_id = identity.identity_id app.dependency_overrides[require_signed_request] = lambda: _make_ctx( is_admin=False, handle=handle, identity_id=identity_id ) try: resp = await client.delete("/api/me") assert resp.status_code == 204, resp.text finally: app.dependency_overrides.pop(require_signed_request, None) await db_session.refresh(identity) assert identity.deleted_at is not None assert await _key_count(db_session, identity_id) == 0 # ── AD_02 — admin-delete another account ───────────────────────────────────── @pytest.mark.asyncio async def test_ad02_admin_can_delete_a_different_account( client: AsyncClient, db_session: AsyncSession ) -> None: handle = f"target-{_uid()}" target = await _seed_identity_with_key(db_session, handle) await db_session.commit() identity_id = target.identity_id app.dependency_overrides[require_signed_request] = lambda: _make_ctx( is_admin=True, handle="admin-actor" ) try: resp = await client.delete(f"/api/admin/users/{handle}") assert resp.status_code == 204, resp.text finally: app.dependency_overrides.pop(require_signed_request, None) await db_session.refresh(target) assert target.deleted_at is not None assert await _key_count(db_session, identity_id) == 0 # ── AD_03 — non-admin forbidden ─────────────────────────────────────────────── @pytest.mark.asyncio async def test_ad03_non_admin_gets_403(client: AsyncClient, db_session: AsyncSession) -> None: handle = f"target-{_uid()}" await _seed_identity_with_key(db_session, handle) await db_session.commit() app.dependency_overrides[require_signed_request] = lambda: _make_ctx( is_admin=False, handle="not-an-admin" ) try: resp = await client.delete(f"/api/admin/users/{handle}") assert resp.status_code == 403, resp.text finally: app.dependency_overrides.pop(require_signed_request, None) # ── AD_04 — unknown handle ──────────────────────────────────────────────────── @pytest.mark.asyncio async def test_ad04_unknown_handle_gets_404(client: AsyncClient) -> None: app.dependency_overrides[require_signed_request] = lambda: _make_ctx(is_admin=True) try: resp = await client.delete(f"/api/admin/users/ghost-{_uid()}") assert resp.status_code == 404, resp.text finally: app.dependency_overrides.pop(require_signed_request, None) # ── AD_05 — already-deleted handle is not re-deletable ─────────────────────── @pytest.mark.asyncio async def test_ad05_already_deleted_handle_gets_404( client: AsyncClient, db_session: AsyncSession ) -> None: handle = f"target-{_uid()}" target = await _seed_identity_with_key(db_session, handle) target.deleted_at = _NOW await db_session.commit() app.dependency_overrides[require_signed_request] = lambda: _make_ctx(is_admin=True) try: resp = await client.delete(f"/api/admin/users/{handle}") assert resp.status_code == 404, resp.text finally: app.dependency_overrides.pop(require_signed_request, None)