feat: adds support for keycloak organizations

src/keycloak/keycloak_admin.py

@@ -428,6 +428,572 @@ class KeycloakAdmin:
             expected_codes=[HTTP_NO_CONTENT],
         )
 
+    def get_organizations(self, query: dict | None = None) -> list:
+        """
+        Fetch all organizations.
+
+        Returns a list of organizations, filtered according to query parameters
+
+        OrganizationRepresentation
+        https://www.keycloak.org/docs-api/26.1.4/rest-api/index.html#OrganizationRepresentation
+
+        :return: List of organizations
+        :rtype: list
+        """
+        query = query or {}
+        params_path = {"realm-name": self.connection.realm_name}
+        url = urls_patterns.URL_ADMIN_ORGANIZATIONS.format(**params_path)
+
+        if "first" in query or "max" in query:
+            return self.__fetch_paginated(url, query)
+
+        return self.__fetch_all(url, query)
+
+    async def a_get_organizations(self, query: dict | None = None) -> list:
+        """
+        Fetch all organizations asynchronously.
+
+        Returns a list of organizations, filtered according to query parameters
+
+        OrganizationRepresentation
+        https://www.keycloak.org/docs-api/26.1.4/rest-api/index.html#OrganizationRepresentation
+
+        :return: List of organizations
+        :rtype: list
+        """
+        query = query or {}
+        params_path = {"realm-name": self.connection.realm_name}
+        url = urls_patterns.URL_ADMIN_ORGANIZATIONS.format(**params_path)
+
+        if "first" in query or "max" in query:
+            return self.a___fetch_paginated(url, query)
+
+        return self.a___fetch_all(url, query)
+
+    def get_organization(self, organization_id: str) -> dict:
+        """
+        Get representation of the organization.
+
+        OrganizationRepresentation:
+        https://www.keycloak.org/docs-api/26.1.4/rest-api/index.html#OrganizationRepresentation
+
+        :param organization_id: ID of the organization
+        :type organization_id: str
+
+        :return: Organization details
+        :rtype: dict
+        """
+        params_path = {
+            "realm-name": self.connection.realm_name,
+            "organization_id": organization_id,
+        }
+        data_raw = self.connection.raw_get(
+            urls_patterns.URL_ADMIN_ORGANIZATION_BY_ID.format(**params_path)
+        )
+
+        return raise_error_from_response(data_raw, KeycloakGetError)
+
+    async def a_get_organization(self, organization_id: str) -> dict:
+        """
+        Get representation of the organization asynchronously.
+
+        OrganizationRepresentation:
+        https://www.keycloak.org/docs-api/26.1.4/rest-api/index.html#OrganizationRepresentation
+
+        :param organization_id: ID of the organization
+        :type organization_id: str
+
+        :return: Organization details
+        :rtype: dict
+        """
+        params_path = {
+            "realm-name": self.connection.realm_name,
+            "organization_id": organization_id,
+        }
+        data_raw = self.connection.a_raw_get(
+            urls_patterns.URL_ADMIN_ORGANIZATION_BY_ID.format(**params_path)
+        )
+
+        return raise_error_from_response(data_raw, KeycloakGetError)
+
+    def create_organization(self, payload: dict) -> str | None:
+        """
+        Create a new organization.
+
+        Organization name and alias must be unique.
+
+        OrganizationRepresentation:
+        https://www.keycloak.org/docs-api/26.1.4/rest-api/index.html#OrganizationRepresentation
+
+        :param payload: Dictionary containing organization details
+        :type payload: dict
+        :return: org_id
+        :rtype: str
+        """
+        params_path = {"realm-name": self.connection.realm_name}
+
+        data_raw = self.connection.raw_post(
+            urls_patterns.URL_ADMIN_ORGANIZATIONS.format(**params_path),
+            data=json.dumps(payload),
+        )
+        raise_error_from_response(data_raw, KeycloakPostError, expected_codes=[HTTP_CREATED])
+        try:
+            _last_slash_idx = data_raw.headers["Location"].rindex("/")
+            return data_raw.headers["Location"][_last_slash_idx + 1 :]
+        except KeyError:
+            return None
+
+    async def a_create_organization(self, payload: dict) -> str | None:
+        """
+        Create a new organization asynchronously.
+
+        Organization name and alias must be unique.
+
+        OrganizationRepresentation:
+        https://www.keycloak.org/docs-api/26.1.4/rest-api/index.html#OrganizationRepresentation
+
+        :param payload: Dictionary containing organization details
+        :type payload: dict
+        :return: org_id
+        :rtype: str
+        """
+        params_path = {"realm-name": self.connection.realm_name}
+
+        data_raw = self.connection.a_raw_post(
+            urls_patterns.URL_ADMIN_ORGANIZATIONS.format(**params_path),
+            data=json.dumps(payload),
+        )
+        raise_error_from_response(data_raw, KeycloakPostError, expected_codes=[HTTP_CREATED])
+        try:
+            _last_slash_idx = data_raw.headers["Location"].rindex("/")
+            return data_raw.headers["Location"][_last_slash_idx + 1 :]
+        except KeyError:
+            return None
+
+    def update_organization(self, organization_id: str, payload: dict) -> dict | bytes:
+        """
+        Update an existing organization.
+
+        OrganizationRepresentation:
+        https://www.keycloak.org/docs-api/26.1.4/rest-api/index.html#OrganizationRepresentation
+
+        :param organization_id: ID of the organization
+        :type organization_id: str
+        :param payload: Dictionary with updated organization details
+        :type payload: dict
+        :return: Response from Keycloak
+        :rtype: dict | bytes
+        """
+        params_path = {
+            "realm-name": self.connection.realm_name,
+            "organization_id": organization_id,
+        }
+
+        data_raw = self.connection.raw_put(
+            urls_patterns.URL_ADMIN_ORGANIZATION_BY_ID.format(**params_path),
+            data=json.dumps(payload),
+        )
+        return raise_error_from_response(
+            data_raw, KeycloakPutError, expected_codes=[HTTP_NO_CONTENT]
+        )
+
+    async def a_update_organization(self, organization_id: str, payload: dict) -> dict | bytes:
+        """
+        Update an existing organization asynchronously.
+
+        OrganizationRepresentation:
+        https://www.keycloak.org/docs-api/26.1.4/rest-api/index.html#OrganizationRepresentation
+
+        :param organization_id: ID of the organization
+        :type organization_id: str
+        :param payload: Dictionary with updated organization details
+        :type payload: dict
+        :return: Response from Keycloak
+        :rtype: dict | bytes
+        """
+        params_path = {
+            "realm-name": self.connection.realm_name,
+            "organization_id": organization_id,
+        }
+
+        data_raw = self.connection.a_raw_put(
+            urls_patterns.URL_ADMIN_ORGANIZATION_BY_ID.format(**params_path),
+            data=json.dumps(payload),
+        )
+        return raise_error_from_response(
+            data_raw, KeycloakPutError, expected_codes=[HTTP_NO_CONTENT]
+        )
+
+    def delete_organization(self, organization_id: str) -> dict | bytes:
+        """
+        Delete an organization.
+
+        :param organization_id: ID of the organization
+        :type organization_id: str
+        :return: Response from Keycloak
+        :rtype: dict | bytes
+        """
+        params_path = {
+            "realm-name": self.connection.realm_name,
+            "organization_id": organization_id,
+        }
+
+        data_raw = self.connection.raw_delete(
+            urls_patterns.URL_ADMIN_ORGANIZATION_BY_ID.format(**params_path)
+        )
+
+        return raise_error_from_response(
+            data_raw, KeycloakDeleteError, expected_codes=[HTTP_NO_CONTENT]
+        )
+
+    async def a_delete_organization(self, organization_id: str) -> dict | bytes:
+        """
+        Delete an organization asynchronously.
+
+        :param organization_id: ID of the organization
+        :type organization_id: str
+        :return: Response from Keycloak
+        :rtype: dict | bytes
+        """
+        params_path = {
+            "realm-name": self.connection.realm_name,
+            "organization_id": organization_id,
+        }
+
+        data_raw = self.connection.a_raw_delete(
+            urls_patterns.URL_ADMIN_ORGANIZATION_BY_ID.format(**params_path)
+        )
+
+        return raise_error_from_response(
+            data_raw, KeycloakDeleteError, expected_codes=[HTTP_NO_CONTENT]
+        )
+
+    def get_organization_idps(self, organization_id: str) -> list:
+        """
+        Get IDPs by organization id.
+
+        IdentityProviderRepresentation
+        https://www.keycloak.org/docs-api/26.1.4/rest-api/index.html#IdentityProviderRepresentation
+
+        :param organization_id: ID of the organization
+        :type organization_id: str
+        :return: List of IDPs in the organization
+        :rtype: list
+        """
+        params_path = {
+            "realm-name": self.connection.realm_name,
+            "organization_id": organization_id,
+        }
+
+        data_raw = self.connection.raw_get(
+            urls_patterns.URL_ADMIN_ORGANIZATION_IDPS.format(**params_path)
+        )
+        return raise_error_from_response(data_raw, KeycloakGetError)
+
+    async def a_get_organization_idps(self, organization_id: str) -> list:
+        """
+        Get IDPs by organization id asynchronously.
+
+        IdentityProviderRepresentation
+        https://www.keycloak.org/docs-api/26.1.4/rest-api/index.html#IdentityProviderRepresentation
+
+        :param organization_id: ID of the organization
+        :type organization_id: str
+        :return: List of IDPs in the organization
+        :rtype: list
+        """
+        params_path = {
+            "realm-name": self.connection.realm_name,
+            "organization_id": organization_id,
+        }
+
+        data_raw = self.connection.a_raw_get(
+            urls_patterns.URL_ADMIN_ORGANIZATION_IDPS.format(**params_path)
+        )
+        return raise_error_from_response(data_raw, KeycloakGetError)
+
+    def organization_idp_add(self, organization_id: str, idp_alias: str) -> dict | bytes:
+        """
+        Add an IDP to an organization.
+
+        :param organization_id: ID of the organization
+        :type organization_id: str
+        :param idp_alias: Alias of the IDP
+        :type idp_alias: str
+        :return: Response from Keycloak
+        :rtype: dict | bytes
+        """
+        params_path = {
+            "realm-name": self.connection.realm_name,
+            "organization_id": organization_id,
+        }
+
+        data_raw = self.connection.raw_post(
+            urls_patterns.URL_ADMIN_ORGANIZATION_IDPS.format(**params_path), data=idp_alias
+        )
+        return raise_error_from_response(
+            data_raw, KeycloakPostError, expected_codes=[HTTP_NO_CONTENT]
+        )
+
+    async def a_organization_idp_add(self, organization_id: str, idp_alias: str) -> dict | bytes:
+        """
+        Add an IDP to an organization asynchronously.
+
+        :param organization_id: ID of the organization
+        :type organization_id: str
+        :param idp_alias: Alias of the IDP
+        :type idp_alias: str
+        :return: Response from Keycloak
+        :rtype: dict | bytes
+        """
+        params_path = {
+            "realm-name": self.connection.realm_name,
+            "organization_id": organization_id,
+        }
+
+        data_raw = self.connection.a_raw_post(
+            urls_patterns.URL_ADMIN_ORGANIZATION_IDPS.format(**params_path), data=idp_alias
+        )
+        return raise_error_from_response(
+            data_raw, KeycloakPostError, expected_codes=[HTTP_NO_CONTENT]
+        )
+
+    def organization_idp_remove(self, organization_id: str, idp_alias: str) -> dict | bytes:
+        """
+        Remove an IDP from an organization.
+
+        :param organization_id: ID of the organization
+        :type organization_id: str
+        """
+        params_path = {
+            "realm-name": self.connection.realm_name,
+            "organization_id": organization_id,
+            "idp_alias": idp_alias,
+        }
+
+        data_raw = self.connection.raw_delete(
+            urls_patterns.URL_ADMIN_ORGANIZATION_IDP_BY_ALIAS.format(**params_path)
+        )
+
+        return raise_error_from_response(
+            data_raw,
+            KeycloakDeleteError,
+            expected_codes=[HTTP_NO_CONTENT],
+        )
+
+    async def a_organization_idp_remove(
+        self, organization_id: str, idp_alias: str
+    ) -> dict | bytes:
+        """
+        Remove an IDP from an organization asynchronously.
+
+        :param organization_id: ID of the organization
+        :type organization_id: str
+        """
+        params_path = {
+            "realm-name": self.connection.realm_name,
+            "organization_id": organization_id,
+            "idp_alias": idp_alias,
+        }
+
+        data_raw = self.connection.a_raw_delete(
+            urls_patterns.URL_ADMIN_ORGANIZATION_IDP_BY_ALIAS.format(**params_path)
+        )
+
+        return raise_error_from_response(
+            data_raw,
+            KeycloakDeleteError,
+            expected_codes=[HTTP_NO_CONTENT],
+        )
+
+    def get_user_organizations(self, user_id: str) -> list:
+        """
+        Get organizations by user id.
+
+        OrganizationRepresentation
+        https://www.keycloak.org/docs-api/26.1.4/rest-api/index.html#OrganizationRepresentation
+
+        :param user_id: ID of the user
+        :type user_id: str
+        :return: List of organizations the user is member of
+        :rtype: list
+        """
+        params_path = {"realm-name": self.connection.realm_name, "user_id": user_id}
+        data_raw = self.connection.raw_get(
+            urls_patterns.URL_ADMIN_USER_ORGANIZATIONS.format(**params_path)
+        )
+        return raise_error_from_response(data_raw, KeycloakGetError)
+
+    async def a_get_user_organizations(self, user_id: str) -> list:
+        """
+        Get organizations by user id asynchronously.
+
+        OrganizationRepresentation
+        https://www.keycloak.org/docs-api/26.1.4/rest-api/index.html#OrganizationRepresentation
+
+        :param user_id: ID of the user
+        :type user_id: str
+        :return: List of organizations the user is member of
+        :rtype: list
+        """
+        params_path = {"realm-name": self.connection.realm_name, "user_id": user_id}
+        data_raw = self.connection.a_raw_get(
+            urls_patterns.URL_ADMIN_USER_ORGANIZATIONS.format(**params_path)
+        )
+        return raise_error_from_response(data_raw, KeycloakGetError)
+
+    def get_organization_members(self, organization_id: str, query: dict | None = None) -> list:
+        """
+        Get members by organization id.
+
+        Returns organization members, filtered according to query parameters
+
+        MemberRepresentation
+        https://www.keycloak.org/docs-api/26.1.4/rest-api/index.html#MemberRepresentation
+
+        :param organization_id: ID of the organization
+        :type organization_id: str
+        :param query: Additional query parameters
+            (see https://www.keycloak.org/docs-api/26.1.4/rest-api/index.html#_organizations)
+        :type query: dict
+        :return: List of users in the organization
+        :rtype: list
+        """
+        query = query or {}
+        params_path = {
+            "realm-name": self.connection.realm_name,
+            "organization_id": organization_id,
+        }
+
+        url = urls_patterns.URL_ADMIN_ORGANIZATION_MEMBERS.format(**params_path)
+        if "first" in query or "max" in query:
+            return self.__fetch_paginated(url, query)
+
+        return self.__fetch_all(url, query)
+
+    async def a_get_organization_members(
+        self, organization_id: str, query: dict | None = None
+    ) -> list:
+        """
+        Get members by organization id asynchronously.
+
+        Returns organization members, filtered according to query parameters
+
+        MemberRepresentation
+        https://www.keycloak.org/docs-api/26.1.4/rest-api/index.html#MemberRepresentation
+
+        :param organization_id: ID of the organization
+        :type organization_id: str
+        :param query: Additional query parameters
+            (see https://www.keycloak.org/docs-api/26.1.4/rest-api/index.html#_organizations)
+        :type query: dict
+        :return: List of users in the organization
+        :rtype: list
+        """
+        query = query or {}
+        params_path = {
+            "realm-name": self.connection.realm_name,
+            "organization_id": organization_id,
+        }
+
+        url = urls_patterns.URL_ADMIN_ORGANIZATION_MEMBERS.format(**params_path)
+        if "first" in query or "max" in query:
+            return self.a___fetch_paginated(url, query)
+
+        return self.a___fetch_all(url, query)
+
+    def organization_user_add(self, user_id: str, organization_id: str) -> dict | bytes:
+        """
+        Add a user to an organization.
+
+        :param user_id: ID of the user to be added
+        :type user_id: str
+        :param organization_id: ID of the organization
+        :type organization_id: str
+        :return: Response from Keycloak
+        :rtype: dict | bytes
+        """
+        params_path = {
+            "realm-name": self.connection.realm_name,
+            "organization_id": organization_id,
+        }
+
+        data_raw = self.connection.raw_post(
+            urls_patterns.URL_ADMIN_ORGANIZATION_MEMBERS.format(**params_path), data=user_id
+        )
+        return raise_error_from_response(data_raw, KeycloakPostError, expected_codes=[HTTP_CREATED])
+
+    async def a_organization_user_add(self, user_id: str, organization_id: str) -> dict | bytes:
+        """
+        Add a user to an organization asynchronously.
+
+        :param user_id: ID of the user to be added
+        :type user_id: str
+        :param organization_id: ID of the organization
+        :type organization_id: str
+        :return: Response from Keycloak
+        :rtype: dict | bytes
+        """
+        params_path = {
+            "realm-name": self.connection.realm_name,
+            "organization_id": organization_id,
+        }
+
+        data_raw = self.connection.a_raw_post(
+            urls_patterns.URL_ADMIN_ORGANIZATION_MEMBERS.format(**params_path), data=user_id
+        )
+        return raise_error_from_response(data_raw, KeycloakPostError, expected_codes=[HTTP_CREATED])
+
+    def organization_user_remove(self, user_id: str, organization_id: str) -> dict | bytes:
+        """
+        Remove a user from an organization.
+
+        :param user_id: ID of the user to be removed
+        :type user_id: str
+        :param organization_id: ID of the organization
+        :type organization_id: str
+        :return: Response from Keycloak
+        :rtype: dict | bytes
+        """
+        params_path = {
+            "realm-name": self.connection.realm_name,
+            "organization_id": organization_id,
+            "user_id": user_id,
+        }
+
+        url = urls_patterns.URL_ADMIN_ORGANIZATION_DEL_MEMBER_BY_ID.format(**params_path)
+        data_raw = self.connection.raw_delete(url)
+        return raise_error_from_response(
+            data_raw,
+            KeycloakDeleteError,
+            expected_codes=[HTTP_NO_CONTENT],
+        )
+
+    async def a_organization_user_remove(self, user_id: str, organization_id: str) -> dict | bytes:
+        """
+        Remove a user from an organization asynchronously.
+
+        :param user_id: ID of the user to be removed
+        :type user_id: str
+        :param organization_id: ID of the organization
+        :type organization_id: str
+        :return: Response from Keycloak
+        :rtype: dict | bytes
+        """
+        params_path = {
+            "realm-name": self.connection.realm_name,
+            "organization_id": organization_id,
+            "user_id": user_id,
+        }
+
+        url = urls_patterns.URL_ADMIN_ORGANIZATION_DEL_MEMBER_BY_ID.format(**params_path)
+        data_raw = self.connection.a_raw_delete(url)
+        return raise_error_from_response(
+            data_raw,
+            KeycloakDeleteError,
+            expected_codes=[HTTP_NO_CONTENT],
+        )
+
     def get_users(self, query: dict | None = None) -> list:
         """
         Get all users.

src/keycloak/urls_patterns.py

@@ -238,3 +238,12 @@ URL_AUTHENTICATION_EXECUTION_LOWER_PRIORITY = (
 )
 
 URL_ADMIN_FLOWS_EXECUTION_CONFIG = URL_ADMIN_FLOWS_EXECUTION + "/config"
+
+# Organization API Endpoints
+URL_ADMIN_ORGANIZATIONS = URL_ADMIN_REALM + "/organizations"
+URL_ADMIN_ORGANIZATION_BY_ID = URL_ADMIN_ORGANIZATIONS + "/{organization_id}"
+URL_ADMIN_ORGANIZATION_MEMBERS = URL_ADMIN_ORGANIZATION_BY_ID + "/members"
+URL_ADMIN_ORGANIZATION_DEL_MEMBER_BY_ID = URL_ADMIN_ORGANIZATION_MEMBERS + "/{user_id}"
+URL_ADMIN_ORGANIZATION_IDPS = URL_ADMIN_ORGANIZATION_BY_ID + "/identity-providers"
+URL_ADMIN_ORGANIZATION_IDP_BY_ALIAS = URL_ADMIN_ORGANIZATION_IDPS + "/{idp_alias}"
+URL_ADMIN_USER_ORGANIZATIONS = URL_ADMIN_ORGANIZATIONS + "/members/{user_id}/organizations"

tests/test_keycloak_admin.py

@@ -329,6 +329,103 @@ def test_partial_import_realm(admin: KeycloakAdmin, realm: str) -> None:
     assert res["overwritten"] == 3
 
 
+def test_organizations(admin: KeycloakAdmin, realm: str) -> None:
+    """
+    Test organizations.
+
+    :param admin: Keycloak Admin client
+    :type admin: KeycloakAdmin
+    :param realm: Keycloak realm
+    :type realm: str
+    """
+    admin.change_current_realm(realm)
+    admin.update_realm(realm_name=realm, payload={"organizationsEnabled": True})
+
+    org_payload = {"name": "test-org01", "alias": "test-org01", "domains": [{"name": "org1.com"}]}
+    org_id = admin.create_organization(payload=org_payload)
+    assert org_id is not None, org_id
+
+    org = admin.get_organization(org_id)
+    assert org["name"] == "test-org01", org["name"]
+    assert org["alias"] == "test-org01", org["alias"]
+    assert org["domains"][0]["name"] == "org1.com", org["domains"][0]["name"]
+
+    orgs = admin.get_organizations()
+    assert len(orgs) == 1, orgs
+    assert orgs[0]["name"] == "test-org01", orgs[0]["name"]
+
+    user_id = admin.create_user(payload={"username": "test", "email": "test@test.test"})
+    admin.organization_user_add(user_id, org_id)
+
+    users = admin.get_organization_members(org_id)
+    assert len(users) == 1, users
+    assert users[0]["id"] == user_id, users[0]["id"]
+
+    user_orgs = admin.get_user_organizations(user_id)
+    assert len(user_orgs) == 1, user_orgs
+    assert user_orgs[0]["name"] == "test-org01", user_orgs[0]["name"]
+
+    admin.organization_user_remove(user_id, org_id)
+    users = admin.get_organization_members(org_id)
+    assert len(users) == 0, users
+
+    for i in range(admin.PAGE_SIZE + 50):
+        user_id = admin.create_user(
+            payload={"username": f"test-user{i:02d}", "email": f"test-user{i:02d}@test.test"}
+        )
+
+        admin.organization_user_add(user_id, org_id)
+
+    users = admin.get_organization_members(org_id)
+    assert len(users) == admin.PAGE_SIZE + 50, users
+
+    users = admin.get_organization_members(org_id, query={"first": 100, "max": -1, "search": ""})
+    assert len(users) == 50, len(users)
+
+    users = admin.get_organization_members(org_id, query={"max": 20, "first": -1, "search": ""})
+    assert len(users) == 20, len(users)
+
+    _ = admin.create_idp(
+        payload={
+            "providerId": "github",
+            "alias": "github",
+            "config": {"clientId": "test-client-id", "clientSecret": "test-client-secret"},
+        }
+    )
+
+    admin.organization_idp_add(org_id, "github")
+
+    idps = admin.get_organization_idps(org_id)
+    assert len(idps) == 1, idps
+    assert idps[0]["alias"] == "github", idps[0]["alias"]
+
+    admin.organization_idp_remove(org_id, "github")
+    idps = admin.get_organization_idps(org_id)
+    assert len(idps) == 0, idps
+
+    admin.delete_organization(org_id)
+    orgs = admin.get_organizations()
+    assert len(orgs) == 0, orgs
+
+    for i in range(admin.PAGE_SIZE + 50):
+        admin.create_organization(
+            payload={
+                "name": f"test-org{i:02d}",
+                "alias": f"org{i:02d}",
+                "domains": [{"name": f"org{i:02d}.com"}],
+            }
+        )
+
+    orgs = admin.get_organizations()
+    assert len(orgs) == admin.PAGE_SIZE + 50, len(orgs)
+
+    orgs = admin.get_organizations(query={"first": 100, "max": -1, "search": ""})
+    assert len(orgs) == 50, len(orgs)
+
+    orgs = admin.get_organizations(query={"first": -1, "max": 20, "search": ""})
+    assert len(orgs) == 20, len(orgs)
+
+
 def test_users(admin: KeycloakAdmin, realm: str) -> None:
     """
     Test users.
@@ -3632,6 +3729,108 @@ async def test_a_partial_import_realm(admin: KeycloakAdmin, realm: str) -> None:
     assert res["overwritten"] == 3
 
 
+@pytest.mark.asyncio
+async def a_test_organizations(admin: KeycloakAdmin, realm: str) -> None:
+    """
+    Test organizations.
+
+    :param admin: Keycloak Admin client
+    :type admin: KeycloakAdmin
+    :param realm: Keycloak realm
+    :type realm: str
+    """
+    await admin.a_change_current_realm(realm)
+    await admin.a_update_realm(realm_name=realm, payload={"organizationsEnabled": True})
+
+    org_payload = {"name": "test-org01", "alias": "test-org01", "domains": [{"name": "org1.com"}]}
+    org_id = await admin.a_create_organization(payload=org_payload)
+    assert org_id is not None, org_id
+
+    org = await admin.a_get_organization(org_id)
+    assert org["name"] == "test-org01", org["name"]
+    assert org["alias"] == "test-org01", org["alias"]
+    assert org["domains"][0]["name"] == "org1.com", org["domains"][0]["name"]
+
+    orgs = await admin.a_get_organizations()
+    assert len(orgs) == 1, orgs
+    assert orgs[0]["name"] == "test-org01", orgs[0]["name"]
+
+    user_id = await admin.a_create_user(payload={"username": "test", "email": "test@test.test"})
+    await admin.a_organization_user_add(user_id, org_id)
+
+    users = await admin.a_get_organization_members(org_id)
+    assert len(users) == 1, users
+    assert users[0]["id"] == user_id, users[0]["id"]
+
+    user_orgs = await admin.a_get_user_organizations(user_id)
+    assert len(user_orgs) == 1, user_orgs
+    assert user_orgs[0]["name"] == "test-org01", user_orgs[0]["name"]
+
+    await admin.a_organization_user_remove(user_id, org_id)
+    users = await admin.a_get_organization_members(org_id)
+    assert len(users) == 0, users
+
+    for i in range(admin.PAGE_SIZE + 50):
+        user_id = await admin.a_create_user(
+            payload={"username": f"test-user{i:02d}", "email": f"test-user{i:02d}@test.test"}
+        )
+
+        await admin.a_organization_user_add(user_id, org_id)
+
+    users = await admin.a_get_organization_members(org_id)
+    assert len(users) == admin.PAGE_SIZE + 50, users
+
+    users = await admin.a_get_organization_members(
+        org_id, query={"first": 100, "max": -1, "search": ""}
+    )
+    assert len(users) == 50, len(users)
+
+    users = await admin.a_get_organization_members(
+        org_id, query={"max": 20, "first": -1, "search": ""}
+    )
+    assert len(users) == 20, len(users)
+
+    _ = await admin.a_create_idp(
+        payload={
+            "providerId": "github",
+            "alias": "github",
+            "config": {"clientId": "test-client-id", "clientSecret": "test-client-secret"},
+        }
+    )
+
+    await admin.a_organization_idp_add(org_id, "github")
+
+    idps = await admin.a_get_organization_idps(org_id)
+    assert len(idps) == 1, idps
+    assert idps[0]["alias"] == "github", idps[0]["alias"]
+
+    await admin.a_organization_idp_remove(org_id, "github")
+    idps = await admin.a_get_organization_idps(org_id)
+    assert len(idps) == 0, idps
+
+    await admin.a_delete_organization(org_id)
+    orgs = await admin.a_get_organizations()
+    assert len(orgs) == 0, orgs
+
+    for i in range(admin.PAGE_SIZE + 50):
+        await admin.a_create_organization(
+            payload={
+                "name": f"test-org{i:02d}",
+                "alias": f"org{i:02d}",
+                "domains": [{"name": f"org{i:02d}.com"}],
+            }
+        )
+
+    orgs = await admin.a_get_organizations()
+    assert len(orgs) == admin.PAGE_SIZE + 50, len(orgs)
+
+    orgs = await admin.a_get_organizations(query={"first": 100, "max": -1, "search": ""})
+    assert len(orgs) == 50, len(orgs)
+
+    orgs = await admin.a_get_organizations(query={"first": -1, "max": 20, "search": ""})
+    assert len(orgs) == 20, len(orgs)
+
+
 @pytest.mark.asyncio
 async def test_a_users(admin: KeycloakAdmin, realm: str) -> None:
     """