"""Organization client for integration tests.""" import logging logger = logging.getLogger(__name__) class OrgsClient: """Wraps organization-related API calls.""" def __init__(self, client): self._client = client # ------------------------------------------------------------------ # Organization CRUD # ------------------------------------------------------------------ def create(self, name: str, slug: str | None = None, description: str | None = None) -> dict: """Create a new organization.""" payload: dict = {"name": name} if slug: payload["slug"] = slug if description: payload["description"] = description return self._client.post("/organizations", data=payload) def get(self, org_id: str) -> dict: """Get organization details.""" return self._client.get(f"/organizations/{org_id}") def update(self, org_id: str, **fields) -> dict: """Update organization fields (name, description, etc.).""" return self._client.patch(f"/organizations/{org_id}", data=fields) def delete(self, org_id: str, confirm: bool = False) -> dict: """Delete (soft-delete) an organization.""" return self._client.delete(f"/organizations/{org_id}", data={"confirm": confirm}) # ------------------------------------------------------------------ # Members # ------------------------------------------------------------------ def list_members(self, org_id: str) -> dict: """List members of an organization.""" return self._client.get(f"/organizations/{org_id}/members") def add_member(self, org_id: str, email: str, role: str = "member") -> dict: """Add an existing user as a member.""" return self._client.post( f"/organizations/{org_id}/members", data={"email": email, "role": role}, ) def remove_member(self, org_id: str, member_id: str) -> dict: """Remove a member from an organization.""" return self._client.delete(f"/organizations/{org_id}/members/{member_id}") def update_member_role(self, org_id: str, member_id: str, role: str) -> dict: """Update a member's role.""" return self._client.patch( f"/organizations/{org_id}/members/{member_id}/role", data={"role": role}, ) def transfer_ownership(self, org_id: str, new_owner_id: str) -> dict: """Transfer organization ownership.""" return self._client.post( f"/organizations/{org_id}/transfer-ownership", data={"new_owner_user_id": new_owner_id}, ) # ------------------------------------------------------------------ # Invites # ------------------------------------------------------------------ def list_invites(self, org_id: str) -> dict: """List pending invites.""" return self._client.get(f"/organizations/{org_id}/invites") def create_invite(self, org_id: str, email: str, role: str = "member") -> dict: """Create an invite for a new user.""" return self._client.post( f"/organizations/{org_id}/invites", data={"email": email, "role": role}, ) def cancel_invite(self, org_id: str, invite_id: str) -> dict: """Cancel a pending invite.""" return self._client.delete(f"/organizations/{org_id}/invites/{invite_id}") def get_invite_by_token(self, token: str) -> dict: """Get invite info by token (public endpoint).""" return self._client.get(f"/invites/{token}") def accept_invite(self, token: str, password: str | None = None, full_name: str | None = None, password_confirm: str | None = None) -> dict: """Accept an invite. For new users, password and full_name are required.""" payload: dict = {} if password: payload["password"] = password if password_confirm: payload["password_confirm"] = password_confirm if full_name: payload["full_name"] = full_name result = self._client.post(f"/invites/{token}/accept", data=payload) # Store token if returned (new user registration) token_val = result.get("data", {}).get("token") if token_val: self._client.set_token(token_val) return result # ------------------------------------------------------------------ # Principals & Departments # ------------------------------------------------------------------ def list_principals(self, org_id: str) -> dict: """List principals in an organization.""" return self._client.get(f"/organizations/{org_id}/principals") def create_principal(self, org_id: str, name: str, description: str | None = None) -> dict: """Create a principal.""" payload: dict = {"name": name} if description: payload["description"] = description return self._client.post(f"/organizations/{org_id}/principals", data=payload) def add_principal_member(self, org_id: str, principal_id: str, email: str) -> dict: """Add a user to a principal.""" return self._client.post( f"/organizations/{org_id}/principals/{principal_id}/members", data={"email": email}, ) def list_departments(self, org_id: str) -> dict: """List departments in an organization.""" return self._client.get(f"/organizations/{org_id}/departments") def create_department(self, org_id: str, name: str, description: str | None = None) -> dict: """Create a department.""" payload: dict = {"name": name} if description: payload["description"] = description return self._client.post(f"/organizations/{org_id}/departments", data=payload) def add_department_member(self, org_id: str, dept_id: str, email: str) -> dict: """Add a user to a department.""" return self._client.post( f"/organizations/{org_id}/departments/{dept_id}/members", data={"email": email}, ) def link_principal_department(self, org_id: str, principal_id: str, dept_id: str) -> dict: """Link a principal to a department.""" return self._client.post( f"/organizations/{org_id}/principals/{principal_id}/departments/{dept_id}", data={}, ) # ------------------------------------------------------------------ # CAs # ------------------------------------------------------------------ def list_cas(self, org_id: str) -> dict: """List CAs for an organization.""" return self._client.get(f"/organizations/{org_id}/cas") def create_ca(self, org_id: str, name: str, ca_type: str = "user", key_type: str = "ed25519") -> dict: """Create a Certificate Authority.""" return self._client.post( f"/organizations/{org_id}/cas", data={"name": name, "ca_type": ca_type, "key_type": key_type}, ) def get_ca(self, org_id: str, ca_id: str) -> dict: """Get a CA by ID.""" return self._client.get(f"/organizations/{org_id}/cas/{ca_id}") def rotate_ca(self, org_id: str, ca_id: str) -> dict: """Rotate a CA key.""" return self._client.post(f"/organizations/{org_id}/cas/{ca_id}/rotate") # ------------------------------------------------------------------ # API Keys # ------------------------------------------------------------------ def list_api_keys(self, org_id: str) -> dict: """List API keys.""" return self._client.get(f"/organizations/{org_id}/api-keys") def create_api_key(self, org_id: str, name: str, role: str = "member") -> dict: """Create an API key.""" return self._client.post( f"/organizations/{org_id}/api-keys", data={"name": name, "role": role}, ) def revoke_api_key(self, org_id: str, key_id: str) -> dict: """Revoke an API key.""" return self._client.delete(f"/organizations/{org_id}/api-keys/{key_id}")