mirror of
https://github.com/apache/cloudstack.git
synced 2025-10-26 08:42:29 +01:00
5ff932eb86
* pass domainid for list users
* passing arg in wizzard
* adding userfilter to list ldap users and usersource to response
port of list ldap users tests to java
* assertion of differnt junit ldap methods
* broken test for directory server (and others)
* embedded context loading
* add user and query test
* UI: filter options passing filter and domain and onchange trigger
* disable tests that only work in ide
prereqs for domain-linkage fixed
move trigger to the right location in code
trigger for changing domain
* logging, comments and refactor
implement search users per domain
retrieve appropriate list of users to filter
get domain specific ldap provider
* query cloudstack users with now db filter
* recreate ldap linked account should succeed
* disable auto import users that don't exist
* ui choice and text
* import filter and potential remove from list bug fixed
* fix rights for domain admins
* list only member of linked groups not of principle group
* Do not show ldap user filter if not importing from ldap
do not delete un-needed items from dialog permanently
delete from temp object not from global one
* localdomain should not filterout users not imported from ldap
* several types of authentication handling errors fixed and unit tested
* conflict in output name
* add conflict source field to generic import dialog
* replace reflextion by enum member call
* conflict is now called conflict 🎉
477 lines
20 KiB
Python
477 lines
20 KiB
Python
# Licensed to the Apache Software Foundation (ASF) under one
|
|
# or more contributor license agreements. See the NOTICE file
|
|
# distributed with this work for additional information
|
|
# regarding copyright ownership. The ASF licenses this file
|
|
# to you under the Apache License, Version 2.0 (the
|
|
# "License"); you may not use this file except in compliance
|
|
# with the License. You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing,
|
|
# software distributed under the License is distributed on an
|
|
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
|
# KIND, either express or implied. See the License for the
|
|
# specific language governing permissions and limitations
|
|
# under the License.
|
|
|
|
#Import Local Modules
|
|
from .ldap_test_data import LdapTestData
|
|
from marvin.cloudstackTestCase import cloudstackTestCase
|
|
from marvin.lib.utils import (cleanup_resources)
|
|
from marvin.lib.base import (listLdapUsers,
|
|
ldapCreateAccount,
|
|
importLdapUsers,
|
|
User,
|
|
Domain,
|
|
Account,
|
|
addLdapConfiguration,
|
|
deleteLdapConfiguration,
|
|
linkAccountToLdap,
|
|
linkDomainToLdap,
|
|
updateConfiguration)
|
|
from marvin.lib.common import (get_domain,
|
|
get_zone)
|
|
from nose.plugins.attrib import attr
|
|
|
|
# for login validation
|
|
import requests
|
|
|
|
import logging
|
|
|
|
class TestLDAP(cloudstackTestCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
'''
|
|
needs to
|
|
- create the applicable ldap accounts in the directory server
|
|
- create three domains:
|
|
-- LDAP/manual
|
|
-- LDAP/import
|
|
-- LDAP/sync
|
|
'''
|
|
cls.logger = logging.getLogger(__name__)
|
|
stream_handler = logging.StreamHandler()
|
|
logger_formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
|
|
stream_handler.setFormatter(logger_formatter)
|
|
cls.logger.setLevel(logging.DEBUG)
|
|
cls.logger.addHandler(stream_handler)
|
|
|
|
cls.logger.info("Setting up Class")
|
|
testClient = super(TestLDAP, cls).getClsTestClient()
|
|
cls.apiclient = testClient.getApiClient()
|
|
|
|
try:
|
|
# Setup test data
|
|
cls.testdata = LdapTestData()
|
|
if cls.config.TestData and cls.config.TestData.Path:
|
|
cls.logger.debug("reading extra config from '" + cls.config.TestData.Path + "'")
|
|
cls.testdata.update(cls.config.TestData.Path)
|
|
cls.logger.debug(cls.testdata)
|
|
|
|
cls.services = testClient.getParsedTestDataConfig()
|
|
cls.services["configurableData"]["ldap_configuration"] = cls.testdata.testdata["ldap_configuration"]
|
|
cls.logger.debug(cls.services["configurableData"]["ldap_configuration"])
|
|
|
|
# Get Zone, Domain
|
|
cls.domain = get_domain(cls.apiclient)
|
|
cls.logger.debug("standard domain: %s" % cls.domain.id)
|
|
cls.zone = get_zone(cls.apiclient, cls.testClient.getZoneForTests())
|
|
|
|
cls._cleanup = []
|
|
|
|
# Build the test env
|
|
cls.create_domains(cls.testdata)
|
|
cls.configure_ldap_for_domains(cls.testdata)
|
|
|
|
cls.test_user = [
|
|
cls.testdata.testdata[LdapTestData.users][0][LdapTestData.uid],
|
|
cls.testdata.testdata[LdapTestData.users][1][LdapTestData.uid],
|
|
cls.testdata.testdata[LdapTestData.users][2][LdapTestData.uid]
|
|
]
|
|
except Exception as e:
|
|
cls.logger.debug("Exception in setUpClass(cls): %s" % e)
|
|
cls.tearDownClass()
|
|
raise Exception("setup failed due to %s", e)
|
|
|
|
return
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.logger.info("Tearing Down Class")
|
|
try:
|
|
cleanup_resources(cls.apiclient, reversed(cls._cleanup))
|
|
cls.remove_ldap_configuration_for_domains()
|
|
cls.logger.debug("done cleaning up resources in tearDownClass(cls) %s")
|
|
except Exception as e:
|
|
cls.logger.debug("Exception in tearDownClass(cls): %s" % e)
|
|
|
|
def setUp(self):
|
|
self.cleanup = []
|
|
|
|
self.server_details = self.config.__dict__["mgtSvr"][0].__dict__
|
|
self.server_url = "http://%s:8080/client/api" % self.server_details['mgtSvrIp']
|
|
|
|
return
|
|
|
|
def tearDown(self):
|
|
try:
|
|
cleanup_resources(self.apiclient, self.cleanup)
|
|
except Exception as e:
|
|
raise Exception("Warning: Exception during cleanup : %s" % e)
|
|
return
|
|
|
|
@attr(tags=["smoke", "advanced"], required_hardware="false")
|
|
def test_01_manual(self):
|
|
'''
|
|
test if an account can be imported
|
|
|
|
prerequisite
|
|
a ldap host is configured
|
|
a domain is linked to cloudstack
|
|
'''
|
|
cmd = listLdapUsers.listLdapUsersCmd()
|
|
cmd.domainid = self.manualDomain.id
|
|
cmd.userfilter = "LocalDomain"
|
|
response = self.apiclient.listLdapUsers(cmd)
|
|
self.logger.info("users found for linked domain %s" % response)
|
|
self.assertEqual(len(response), len(self.testdata.testdata[LdapTestData.users]), "unexpected number (%d) of ldap users" % len(self.testdata.testdata[LdapTestData.users]))
|
|
|
|
cmd = ldapCreateAccount.ldapCreateAccountCmd()
|
|
cmd.domainid = self.manualDomain.id
|
|
cmd.accounttype = 0
|
|
cmd.username = self.test_user[1]
|
|
create_response = self.apiclient.ldapCreateAccount(cmd)
|
|
|
|
# cleanup
|
|
# last results id should be the account
|
|
list_response = Account.list(self.apiclient, id=create_response.id)
|
|
account_created = Account(list_response[0].__dict__)
|
|
self.cleanup.append(account_created)
|
|
|
|
self.assertEqual(len(create_response.user), 1, "only one user %s should be present" % self.test_user[1])
|
|
|
|
self.assertEqual(len(list_response),
|
|
1,
|
|
"only one account (for user %s) should be present" % self.test_user[1])
|
|
|
|
return
|
|
|
|
@attr(tags=["smoke", "advanced"], required_hardware="false")
|
|
def test_02_import(self):
|
|
'''
|
|
test if components are synced
|
|
|
|
prerequisite
|
|
a ldap host is configured
|
|
a domain is linked to cloudstack
|
|
'''
|
|
domainid = self.importDomain.id
|
|
|
|
cmd = importLdapUsers.importLdapUsersCmd()
|
|
cmd.domainid = domainid
|
|
cmd.accounttype = 0
|
|
import_response = self.apiclient.importLdapUsers(cmd)
|
|
|
|
# this is needed purely for cleanup:
|
|
# cleanup
|
|
list_response = Account.list(self.apiclient, domainid=domainid)
|
|
for account in list_response:
|
|
account_created = Account(account.__dict__)
|
|
self.logger.debug("account to clean: %s (id: %s)" % (account_created.name, account_created.id))
|
|
self.cleanup.append(account_created)
|
|
|
|
self.assertEqual(len(import_response), len(self.testdata.testdata[LdapTestData.users]), "unexpected number of ldap users")
|
|
|
|
self.assertEqual(len(list_response), len(self.testdata.testdata[LdapTestData.users]), "only one account (for user %s) should be present" % self.test_user[1])
|
|
|
|
return
|
|
|
|
@attr(tags=["smoke", "advanced"], required_hardware="false")
|
|
def test_03_sync(self):
|
|
'''
|
|
test if components are synced
|
|
|
|
prerequisite
|
|
a ldap host is configured
|
|
a domain is linked to cloudstack
|
|
some accounts in that domain are linked to groups in ldap
|
|
'''
|
|
domainid = self.syncDomain.id
|
|
username = self.test_user[1]
|
|
|
|
# validate the user doesn't exist
|
|
response = User.list(self.apiclient,domainid=domainid,username=username)
|
|
self.assertEqual(response, None, "user should not exist yet")
|
|
|
|
self.logon_test_user(username)
|
|
|
|
# now validate the user exists in domain
|
|
response = User.list(self.apiclient,domainid=domainid,username=username)
|
|
for user in response:
|
|
user_created = User(user.__dict__)
|
|
self.logger.debug("user to clean: %s (id: %s)" % (user_created.username, user_created.id))
|
|
self.cleanup.append(user_created)
|
|
|
|
# now verify the creation of the user
|
|
self.assertEqual(len(response), 1, "user should exist by now")
|
|
|
|
return
|
|
|
|
@attr(tags=["smoke", "advanced"], required_hardware="false")
|
|
def test_04_filtered_list_of_users(self):
|
|
'''
|
|
test if we can get a filtered list of ldap users
|
|
|
|
prerequisite
|
|
a ldap host is configured
|
|
a couple of ldapdomains are linked to cloudstack domains
|
|
some accounts in those domain are linked to groups in ldap
|
|
some ldap accounts are linked and present with the same uid
|
|
some ldap accounts are not yet linked but present at other locations in cloudstack
|
|
|
|
NOTE 1: if this test is run last only the explicitely imported test user from test_03_sync
|
|
is in the system. The accounts from test_01_manual and test_02_import should have been cleared
|
|
by the test tearDown(). We can not depend on test_03_sync having run so the test must avoid
|
|
depending on it either being available or not.
|
|
|
|
NOTE 2: this test will not work if the ldap users UIDs are already present in the ACS instance
|
|
against which is being tested
|
|
'''
|
|
cmd = listLdapUsers.listLdapUsersCmd()
|
|
cmd.userfilter = "NoFilter"
|
|
cmd.domainid = self.manualDomain.id
|
|
response = self.apiclient.listLdapUsers(cmd)
|
|
self.logger.debug(cmd.userfilter + " : " + str(response))
|
|
self.assertEqual(len(response), len(self.testdata.testdata[LdapTestData.users]), "unexpected number of ldap users")
|
|
|
|
# create a non ldap user with the uid of cls.test_user[0] in parentDomain
|
|
# create a manual import of a cls.test_user[1] in manualDomain
|
|
# log on with test_user[2] in an syncDomain
|
|
|
|
# we can now test all four filtertypes in syncDomain and inspect the respective outcomes for validity
|
|
|
|
self.logon_test_user(self.test_user[2])
|
|
|
|
cmd.userfilter = "LocalDomain"
|
|
cmd.domainid = self.syncDomain.id
|
|
response = self.apiclient.listLdapUsers(cmd)
|
|
self.logger.debug(cmd.userfilter + " : " + str(response))
|
|
self.assertEqual(len(response),
|
|
len(self.testdata.testdata[LdapTestData.users]) - 1,
|
|
"unexpected number of ldap users")
|
|
|
|
@attr(tags=["smoke", "advanced"], required_hardware="false")
|
|
def test_05_relink_account_and_reuse_user(self):
|
|
'''
|
|
test if an account and thus a user can be removed and re-added
|
|
|
|
test if components still are synced
|
|
|
|
prerequisite
|
|
a ldap host is configured
|
|
a domain is linked to cloudstack
|
|
some accounts in that domain are linked to groups in ldap
|
|
'''
|
|
domainid = self.syncDomain.id
|
|
username = self.test_user[1]
|
|
|
|
# validate the user doesn't exist
|
|
response = User.list(self.apiclient,domainid=domainid,username=username)
|
|
self.assertEqual(response, None, "user should not exist yet")
|
|
|
|
self.logon_test_user(username)
|
|
|
|
# now validate the user exists in domain
|
|
response = User.list(self.apiclient,domainid=domainid,username=username)
|
|
# for user in response:
|
|
# user_created = User(user.__dict__)
|
|
# self.debug("user to clean: %s (id: %s)" % (user_created.username, user_created.id))
|
|
# # we don't cleanup to test if re-adding fails
|
|
# self.cleanup.append(user_created)
|
|
|
|
# now verify the creation of the user
|
|
self.assertEqual(len(response), 1, "user should exist by now")
|
|
|
|
# delete the account - quick implementation: user[1] happens to be a junior
|
|
self.junior_account.delete(self.apiclient)
|
|
|
|
# add the account with the same ldap group
|
|
self.bind_account_to_ldap(
|
|
account=self.testdata.testdata[LdapTestData.syncAccounts][0]["name"],
|
|
ldapdomain=self.testdata.testdata[LdapTestData.syncAccounts][0]["group"],
|
|
accounttype=self.testdata.testdata[LdapTestData.syncAccounts][0]["accounttype"])
|
|
|
|
# logon the user - should succeed - reported to fail
|
|
self.logon_test_user(username)
|
|
|
|
# now verify the creation of the user
|
|
response = User.list(self.apiclient,domainid=domainid,username=username)
|
|
for user in response:
|
|
user_created = User(user.__dict__)
|
|
self.debug("user to clean: %s (id: %s)" % (user_created.username, user_created.id))
|
|
# we don't cleanup to test if re-adding fails
|
|
# self.cleanup.append(user_created)
|
|
self.assertEqual(len(response), 1, "user should exist again")
|
|
return
|
|
|
|
|
|
def logon_test_user(self, username, domain = None):
|
|
# login of dahn should create a user in account juniors
|
|
args = {}
|
|
args["command"] = 'login'
|
|
args["username"] = username
|
|
args["password"] = 'password'
|
|
if domain == None:
|
|
args["domain"] = "/" + self.parentDomain.name + "/" + self.syncDomain.name
|
|
else:
|
|
args["domain"] = domain
|
|
args["response"] = "json"
|
|
session = requests.Session()
|
|
try:
|
|
resp = session.post(self.server_url, params=args, verify=False)
|
|
except requests.exceptions.ConnectionError as e:
|
|
self.fail("Failed to attempt login request to mgmt server")
|
|
|
|
|
|
@classmethod
|
|
def create_domains(cls, td):
|
|
# create a parent domain
|
|
cls.parentDomain = cls.create_domain(td.testdata["domains"][0], parent_domain=cls.domain.id)
|
|
cls.manualDomain = cls.create_domain(td.testdata["domains"][1], parent_domain=cls.parentDomain.id)
|
|
cls.importDomain = cls.create_domain(td.testdata["domains"][2], parent_domain=cls.parentDomain.id)
|
|
cls.syncDomain = cls.create_domain(td.testdata["domains"][3], parent_domain=cls.parentDomain.id)
|
|
|
|
@classmethod
|
|
def create_domain(cls, domain_to_create, parent_domain = None):
|
|
cls.logger.debug("Creating domain: %s under %s" % (domain_to_create[LdapTestData.name], parent_domain))
|
|
if parent_domain:
|
|
domain_to_create["parentdomainid"] = parent_domain
|
|
tmpDomain = Domain.create(cls.apiclient, domain_to_create)
|
|
cls.logger.debug("Created domain %s with id %s " % (tmpDomain.name, tmpDomain.id))
|
|
cls._cleanup.append(tmpDomain)
|
|
return tmpDomain
|
|
|
|
@classmethod
|
|
def configure_ldap_for_domains(cls, td) :
|
|
cmd = addLdapConfiguration.addLdapConfigurationCmd()
|
|
cmd.hostname = td.testdata[LdapTestData.configuration][LdapTestData.hostname]
|
|
cmd.port = td.testdata[LdapTestData.configuration][LdapTestData.port]
|
|
|
|
cls.logger.debug("configuring ldap server for domain %s" % LdapTestData.manualDomain)
|
|
cmd.domainid = cls.manualDomain.id
|
|
response = cls.apiclient.addLdapConfiguration(cmd)
|
|
cls.manualLdap = response
|
|
|
|
cls.logger.debug("configuring ldap server for domain %s" % LdapTestData.importDomain)
|
|
cmd.domainid = cls.importDomain.id
|
|
response = cls.apiclient.addLdapConfiguration(cmd)
|
|
cls.importLdap = response
|
|
|
|
cls.logger.debug("configuring ldap server for domain %s" % LdapTestData.syncDomain)
|
|
cmd.domainid = cls.syncDomain.id
|
|
response = cls.apiclient.addLdapConfiguration(cmd)
|
|
cls.syncLdap = response
|
|
|
|
cls.set_ldap_settings_on_domain(domainid=cls.manualDomain.id)
|
|
cls.bind_domain_to_ldap(domainid=cls.manualDomain.id, ldapdomain=cls.testdata.admins)
|
|
|
|
cls.set_ldap_settings_on_domain(domainid=cls.importDomain.id)
|
|
cls.bind_domain_to_ldap(domainid=cls.importDomain.id, ldapdomain=cls.testdata.admins,
|
|
accounttype=2, type="Group") # just to be testing different types
|
|
|
|
cls.set_ldap_settings_on_domain(domainid=cls.syncDomain.id)
|
|
cls.create_sync_accounts()
|
|
|
|
@classmethod
|
|
def remove_ldap_configuration_for_domains(cls) :
|
|
cls.logger.debug("deleting configurations for ldap server")
|
|
cmd = deleteLdapConfiguration.deleteLdapConfigurationCmd()
|
|
|
|
cmd.hostname = cls.manualLdap.hostname
|
|
cmd.port = cls.manualLdap.port
|
|
cmd.domainid = cls.manualLdap.domainid
|
|
response = cls.apiclient.deleteLdapConfiguration(cmd)
|
|
cls.logger.debug("configuration deleted for %s" % response)
|
|
|
|
cmd.hostname = cls.importLdap.hostname
|
|
cmd.port = cls.importLdap.port
|
|
cmd.domainid = cls.importLdap.domainid
|
|
response = cls.apiclient.deleteLdapConfiguration(cmd)
|
|
cls.logger.debug("configuration deleted for %s" % response)
|
|
|
|
cmd.hostname = cls.syncLdap.hostname
|
|
cmd.port = cls.syncLdap.port
|
|
cmd.domainid = cls.syncLdap.domainid
|
|
cls.logger.debug("deleting configuration %s" % cmd)
|
|
response = cls.apiclient.deleteLdapConfiguration(cmd)
|
|
cls.logger.debug("configuration deleted for %s" % response)
|
|
|
|
|
|
@classmethod
|
|
def create_sync_accounts(cls):
|
|
cls.logger.debug("creating account: %s" % LdapTestData.seniors)
|
|
cls.senior_account = cls.bind_account_to_ldap(
|
|
account=cls.testdata.testdata[LdapTestData.syncAccounts][1]["name"],
|
|
ldapdomain=cls.testdata.testdata[LdapTestData.syncAccounts][1]["group"],
|
|
accounttype=cls.testdata.testdata[LdapTestData.syncAccounts][1]["accounttype"])
|
|
cls.junior_account = cls.bind_account_to_ldap(
|
|
account=cls.testdata.testdata[LdapTestData.syncAccounts][0]["name"],
|
|
ldapdomain=cls.testdata.testdata[LdapTestData.syncAccounts][0]["group"],
|
|
accounttype=cls.testdata.testdata[LdapTestData.syncAccounts][0]["accounttype"])
|
|
|
|
@classmethod
|
|
def bind_account_to_ldap(cls, account, ldapdomain, type="Group", accounttype=0):
|
|
cmd = linkAccountToLdap.linkAccountToLdapCmd()
|
|
|
|
cmd.domainid = cls.syncDomain.id
|
|
cmd.account = account
|
|
cmd.ldapdomain = ldapdomain
|
|
cmd.type = type
|
|
cmd.accounttype = accounttype
|
|
|
|
response = cls.apiclient.linkAccountToLdap(cmd)
|
|
cls.logger.info("account linked to ladp %s" % response)
|
|
|
|
# this is needed purely for cleanup:
|
|
response = Account.list(cls.apiclient, id=response.accountid)
|
|
account_created = Account(response[0].__dict__)
|
|
cls._cleanup.append(account_created)
|
|
return account_created
|
|
|
|
@classmethod
|
|
def bind_domain_to_ldap(cls, domainid, ldapdomain, type="OU", accounttype=0):
|
|
cmd = linkDomainToLdap.linkDomainToLdapCmd()
|
|
cmd.domainid = domainid
|
|
cmd.type = type
|
|
cmd.accounttype = accounttype
|
|
cmd.ldapdomain = ldapdomain
|
|
response = cls.apiclient.linkDomainToLdap(cmd)
|
|
cls.logger.info("domain linked to ladp %s" % response)
|
|
|
|
@classmethod
|
|
def set_ldap_settings_on_domain(cls, domainid):
|
|
cmd = updateConfiguration.updateConfigurationCmd()
|
|
cmd.domainid = domainid
|
|
cmd.name = LdapTestData.basednConfig
|
|
cmd.value = cls.testdata.testdata[LdapTestData.configuration][LdapTestData.basedn]
|
|
response = cls.apiclient.updateConfiguration(cmd)
|
|
cls.logger.debug("set the basedn: %s" % response)
|
|
cmd.name = LdapTestData.ldapPwConfig
|
|
cmd.value = cls.testdata.testdata[LdapTestData.configuration][LdapTestData.ldapPw]
|
|
response = cls.apiclient.updateConfiguration(cmd)
|
|
cls.logger.debug("set the pw: %s" % response)
|
|
cmd.name = LdapTestData.principalConfig
|
|
cmd.value = cls.testdata.testdata[LdapTestData.configuration][LdapTestData.principal]
|
|
response = cls.apiclient.updateConfiguration(cmd)
|
|
cls.logger.debug("set the id: %s" % response)
|
|
if cls.testdata.testdata[LdapTestData.configuration].has_key(LdapTestData.groupPrinciple) :
|
|
cmd.name = LdapTestData.groupPrinciple
|
|
cmd.value = cls.testdata.testdata[LdapTestData.configuration][LdapTestData.groupPrinciple]
|
|
response = cls.apiclient.updateConfiguration(cmd)
|
|
cls.logger.debug("set the id: %s" % response)
|
|
|
|
|
|
## python ldap utility functions
|