"""Admin client for integration tests.""" import logging logger = logging.getLogger(__name__) class AdminClient: """Wraps admin-only API calls.""" def __init__(self, client): self._client = client def list_users(self) -> dict: """List all users (paginated).""" return self._client.get("/admin/users") def get_user(self, user_id: str) -> dict: """Get a single user by ID.""" return self._client.get(f"/admin/users/{user_id}") def suspend_user(self, user_id: str) -> dict: """Suspend a user account.""" return self._client.post(f"/admin/users/{user_id}/suspend") def unsuspend_user(self, user_id: str) -> dict: """Unsuspend a user account.""" return self._client.post(f"/admin/users/{user_id}/unsuspend") def verify_user_email(self, user_id: str) -> dict: """Admin-verify a user's email.""" return self._client.post(f"/admin/users/{user_id}/verify-email") def set_user_password(self, user_id: str, new_password: str) -> dict: """Set a user's password (admin override).""" return self._client.post( f"/admin/users/{user_id}/password", data={"password": new_password}, ) def remove_user_mfa(self, user_id: str, mfa_type: str = "totp") -> dict: """Remove a user's MFA method.""" return self._client.delete(f"/admin/users/{user_id}/mfa/{mfa_type}") def hard_delete_user(self, user_id: str, confirm: bool = False) -> dict: """Hard-delete a user.""" return self._client.post( f"/admin/users/{user_id}/delete", data={"confirm": confirm}, ) def get_user_ssh_certificates(self, user_id: str, **params) -> dict: """List all SSH certificates for a user (admin view). Args: user_id: Target user ID **params: Optional query parameters — status, active, cert_type, page, per_page """ path = f"/admin/users/{user_id}/ssh-certificates" if params: from urllib.parse import urlencode query = urlencode({k: v for k, v in params.items() if v is not None}) if query: path = f"{path}?{query}" return self._client.get(path) def list_audit_logs(self) -> dict: """List system-wide audit logs.""" return self._client.get("/audit-logs")