port ldap tests from iqe repo

No-Issue

galaxy_ng/tests/integration/api/test_ldap.py

@@ -13,7 +13,7 @@
 import logging
 
 from ..utils import get_client
-
+from galaxykit.users import get_me
 
 log = logging.getLogger(__name__)
 
@@ -26,61 +26,137 @@ def settings(ansible_config):
     return api_client(f"{api_prefix}/_ui/v1/settings/")
 
 
+def is_present(group, groups):
+    """looks for a given group in the groups list
+
+    Args:
+        group: The group to be found.
+        groups: List of groups to iterate over.
+
+    Returns:
+        True of group is found in groups, False otherwise
+    """
+    group_found = False
+    for _group in groups:
+        if _group["name"] == group:
+            group_found = True
+    return group_found
+
+
 @pytest.mark.ldap
-def test_ldap_is_enabled(ansible_config, settings):
+def test_ldap_is_enabled(skip_if_ldap_disabled, ansible_config):
     """test whether ldap user can login"""
-    if not settings.get("GALAXY_AUTH_LDAP_ENABLED"):
-        pytest.skip("GALAXY_AUTH_LDAP_ENABLED is not enabled")
-
     config = ansible_config("admin")
     api_prefix = config.get("api_prefix").rstrip("/")
     api_client = get_client(config, request_token=False, require_auth=True)
     assert api_client(f"{api_prefix}/_ui/v1/settings/")["GALAXY_AUTH_LDAP_ENABLED"] is True
 
 
+
 @pytest.mark.ldap
-def test_ldap_login(ansible_config, settings):
-    """test whether ldap user can login"""
+def test_ldap_user_can_log_in(skip_if_ldap_disabled, galaxy_client, ldap_user):
+    """
+    Verifies that a user on LDAP server can log into automation hub
+    """
+    username = "awong"
+    user = ldap_user(username)
+    gc = galaxy_client(user)
+    resp = get_me(gc)
+    assert resp["username"] == username
 
-    if not settings.get("GALAXY_AUTH_LDAP_ENABLED"):
-        pytest.skip("GALAXY_AUTH_LDAP_ENABLED is not enabled")
 
-    config = ansible_config("admin")
-    api_prefix = config.get("api_prefix").rstrip("/")
-    api_client = get_client(config, request_token=False, require_auth=True)
+@pytest.mark.ldap
+def test_ldap_admin_user_is_superuser_in_ahub(skip_if_ldap_disabled, galaxy_client, ldap_user):
+    """
+    Verifies that a user from an admin group on LDAP server is a superuser in ahub
+    PULP_AUTH_LDAP_USER_FLAGS_BY_GROUP__is_superuser="cn=bobsburgers_admins,cn=groups,cn=accounts,dc=testing,dc=ansible,dc=com"
+    """
+    username = "bbelcher"
+    user = ldap_user(username)
+    gc = galaxy_client(user)
+    resp = get_me(gc)
+    assert resp["username"] == username
+    assert resp["is_superuser"] is True
 
-    # This test assumes the running ldap server is the
-    # testing image from: rroemhild/test-openldap
-    data = api_client(f"{api_prefix}/_ui/v1/me/")
-    assert data["username"] == "professor"
-    assert data["email"] == "[email protected]"
-    assert data["first_name"] == "Hubert"
-    assert data["last_name"] == "Farnsworth"
-    # This group is pre-created on hub
-    assert data["groups"][0]["name"] == "admin_staff"
+
+@pytest.mark.ldap
+def test_ldap_personal_information_synced(skip_if_ldap_disabled, galaxy_client, ldap_user):
+    """
+    Verifies that personal information is correctly imported to ahub
+    PULP_AUTH_LDAP_USER_ATTR_MAP = {first_name = "givenName", last_name = "sn", email = "mail"}
+    """
+    username = "brodriguez"
+    user = ldap_user(username)
+    gc = galaxy_client(user)
+    resp = get_me(gc)
+    assert resp["username"] == username
+    assert resp["is_superuser"] is False
+    assert resp["first_name"] == "Bender"
+    assert resp["last_name"] == "Rodriguez"
+    assert resp["email"] == "[email protected]"
 
 
 @pytest.mark.ldap
-def test_ldap_mirror_only_existing_groups(ansible_config, settings):
-    """Ensure that GALAXY_LDAP_MIRROR_ONLY_EXISTING_GROUPS works as expected."""
+def test_ldap_groups_synced(skip_if_ldap_disabled, settings, galaxy_client, ldap_user):
+    """
+    Verifies that groups are correctly created in ahub
+    PULP_AUTH_LDAP_MIRROR_GROUPS=true
+    """
+    if settings.get("GALAXY_LDAP_MIRROR_ONLY_EXISTING_GROUPS"):
+        pytest.skip("GALAXY_LDAP_MIRROR_ONLY_EXISTING_GROUPS is enabled, so new groups will not be synced")
+
+    username = "bstrickland"
+    # bstrickland belongs to groups stricklandpropane, stricklandpropane_admins
+    user = ldap_user(username)
+    gc = galaxy_client(user, ignore_cache=True)
+    resp = get_me(gc)
+    assert resp["username"] == username
+    groups = resp["groups"]
+    assert is_present("stricklandpropane", groups)
+    assert is_present("stricklandpropane_admins", groups)
 
-    if not settings.get("GALAXY_AUTH_LDAP_ENABLED"):
-        pytest.skip("GALAXY_AUTH_LDAP_ENABLED is not enabled")
 
+@pytest.mark.ldap
+def test_ldap_mirror_only_existing_groups(skip_if_ldap_disabled, settings, galaxy_client, ldap_user):
+    """Ensure that GALAXY_LDAP_MIRROR_ONLY_EXISTING_GROUPS works as expected."""
     if not settings.get("GALAXY_LDAP_MIRROR_ONLY_EXISTING_GROUPS"):
         pytest.skip("GALAXY_LDAP_MIRROR_ONLY_EXISTING_GROUPS is not enabled")
-
-    config = ansible_config("ldap_non_admin")
-    api_prefix = config.get("api_prefix").rstrip("/")
-    api_client = get_client(config, request_token=False, require_auth=True)
-
-    # This test assumes the running ldap server is the
-    # testing image from: rroemhild/test-openldap
-    data = api_client(f"{api_prefix}/_ui/v1/me/")
-    assert data["username"] == "fry"
-    assert data["email"] == "[email protected]"
-    assert data["first_name"] == "Philip"
-    assert data["last_name"] == "Fry"
+    # bstrickland belongs to groups stricklandpropane, stricklandpropane_admins
+    username = "bstrickland"
+    user = ldap_user(username)
+    gc = galaxy_client(user, ignore_cache=True)
+    resp = get_me(gc)
+    assert resp["username"] == username
     # This user is member only of "ships_crew" group that doesnt exist
     # so this user will not get groups mirrored.
-    assert len(data["groups"]) == 0
+    assert len(resp["groups"]) == 0
+
+
+@pytest.mark.ldap
+def test_ldap_ignored_groups(skip_if_ldap_disabled, galaxy_client, ldap_user):
+    """
+    Verifies that groups can be ignored and not created in ahub
+    PULP_AUTH_LDAP_MIRROR_GROUPS_EXCEPT=['dreamland']
+    """
+
+    username = "marcher"
+    user = ldap_user(username)
+    gc = galaxy_client(user)
+    resp = get_me(gc)
+    assert resp["username"] == username
+    groups = resp["groups"]
+    assert not is_present("dreamland", groups)
+
+
+@pytest.mark.ldap
+def test_ldap_user_with_no_group(skip_if_ldap_disabled, galaxy_client, ldap_user):
+    """
+    Verifies that users that does not belong to any group are also synced
+    """
+    username = "saml_user"
+    user = ldap_user(username)
+    gc = galaxy_client(user)
+    resp = get_me(gc)
+    assert resp["username"] == username
+    assert resp["groups"] == []
+

galaxy_ng/tests/integration/conftest.py

@@ -11,6 +11,7 @@
 from galaxykit.groups import get_group_id
 from galaxykit.namespaces import create_namespace
 from galaxykit.utils import GalaxyClientError
+from galaxykit.users import get_user, get_user_list
 from .constants import USERNAME_PUBLISHER, GALAXY_STAGE_ANSIBLE_PROFILES
 from .utils import (
     ansible_galaxy,
@@ -582,3 +583,42 @@ def pytest_collection_modifyitems(items, config):
     for item in items:
         if not any(item.iter_markers()):
             item.add_marker("all")
+
+
+@pytest.fixture(scope="session")
+def skip_if_ldap_disabled(ansible_config):
+    config = ansible_config("admin")
+    client = get_client(config)
+    resp = client("_ui/v1/settings/")
+    try:
+        ldap_enabled = resp["GALAXY_AUTH_LDAP_ENABLED"]
+        if not ldap_enabled:
+            pytest.skip("This test can only be run if LDAP is enabled")
+    except KeyError:
+        pytest.skip("This test can only be run if LDAP is enabled")
+
+
+@pytest.fixture
+def ldap_user(galaxy_client, request):
+    def _(name):
+        ldap_password = "Th1sP4ssd"
+        user = {"username": name, "password": ldap_password}
+
+        def clean_test_user_and_groups():
+            gc = galaxy_client("admin")
+            user_list = get_user_list(gc)
+            _user = get_user(gc, name)
+            for group in _user["groups"]:
+                gc.delete_group(group["name"])
+            try:
+                gc.delete_user(name)
+            except GalaxyClientError as e:
+                if e.args[0] == 403:
+                    logger.debug(f"user {name} is superuser and can't be deleted")
+                else:
+                    raise e
+
+        request.addfinalizer(clean_test_user_and_groups)
+        return user
+
+    return _