mirror of
https://github.com/apache/cloudstack.git
synced 2025-10-26 08:42:29 +01:00
2270 lines
72 KiB
Python
2270 lines
72 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.
|
|
""" P1 tests for Account
|
|
"""
|
|
# Import Local Modules
|
|
from marvin.cloudstackTestCase import cloudstackTestCase
|
|
from marvin.lib.utils import (random_gen,
|
|
cleanup_resources,
|
|
validateList)
|
|
from marvin.cloudstackAPI import *
|
|
from marvin.lib.base import (Domain,
|
|
Account,
|
|
ServiceOffering,
|
|
VirtualMachine,
|
|
Network,
|
|
User,
|
|
NATRule,
|
|
Template,
|
|
PublicIPAddress, Role)
|
|
from marvin.lib.common import (get_domain,
|
|
get_zone,
|
|
get_test_template,
|
|
list_accounts,
|
|
list_virtual_machines,
|
|
list_service_offering,
|
|
list_templates,
|
|
list_users,
|
|
get_builtin_template_info,
|
|
wait_for_cleanup)
|
|
from nose.plugins.attrib import attr
|
|
from marvin.cloudstackException import CloudstackAPIException
|
|
from marvin.codes import PASS
|
|
import time
|
|
|
|
from pyVmomi.VmomiSupport import GetVersionFromVersionUri
|
|
|
|
class Services:
|
|
|
|
"""Test Account Services
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.services = {
|
|
"domain": {
|
|
"name": "Domain",
|
|
},
|
|
"account": {
|
|
"email": "test@test.com",
|
|
"firstname": "Test",
|
|
"lastname": "User",
|
|
"username": "test",
|
|
# Random characters are appended for unique
|
|
# username
|
|
"password": "fr3sca",
|
|
},
|
|
"role": {
|
|
"name": "MarvinFake Role",
|
|
"type": "User",
|
|
"description": "Fake Role created by Marvin test"
|
|
},
|
|
"user": {
|
|
"email": "user@test.com",
|
|
"firstname": "User",
|
|
"lastname": "User",
|
|
"username": "User",
|
|
# Random characters are appended for unique
|
|
# username
|
|
"password": "fr3sca",
|
|
},
|
|
"service_offering": {
|
|
"name": "Tiny Instance",
|
|
"displaytext": "Tiny Instance",
|
|
"cpunumber": 1,
|
|
"cpuspeed": 100,
|
|
# in MHz
|
|
"memory": 128,
|
|
# In MBs
|
|
},
|
|
"virtual_machine": {
|
|
"displayname": "Test VM",
|
|
"username": "root",
|
|
"password": "password",
|
|
"ssh_port": 22,
|
|
"hypervisor": 'XenServer',
|
|
# Hypervisor type should be same as
|
|
# hypervisor type of cluster
|
|
"privateport": 22,
|
|
"publicport": 22,
|
|
"protocol": 'TCP',
|
|
},
|
|
"template": {
|
|
"displaytext": "Public Template",
|
|
"name": "Public template",
|
|
"ostype": 'CentOS 5.6 (64-bit)',
|
|
"url": "",
|
|
"hypervisor": '',
|
|
"format": '',
|
|
"isfeatured": True,
|
|
"ispublic": True,
|
|
"isextractable": True,
|
|
"templatefilter": "self"
|
|
},
|
|
"natrule": {
|
|
"publicport": 22,
|
|
"privateport": 22,
|
|
"protocol": 'TCP',
|
|
},
|
|
"ostype": 'CentOS 5.6 (64-bit)',
|
|
"sleep": 60,
|
|
"timeout": 10,
|
|
}
|
|
|
|
|
|
class TestAccounts(cloudstackTestCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.testClient = super(TestAccounts, cls).getClsTestClient()
|
|
cls.api_client = cls.testClient.getApiClient()
|
|
|
|
cls.services = Services().services
|
|
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
|
|
cls.hypervisor = cls.testClient.getHypervisorInfo()
|
|
cls.services['mode'] = cls.zone.networktype
|
|
cls.template = get_test_template(
|
|
cls.api_client,
|
|
cls.zone.id,
|
|
cls.hypervisor
|
|
)
|
|
cls.services["virtual_machine"]["zoneid"] = cls.zone.id
|
|
cls.services["virtual_machine"]["template"] = cls.template.id
|
|
|
|
cls.service_offering = ServiceOffering.create(
|
|
cls.api_client,
|
|
cls.services["service_offering"]
|
|
)
|
|
cls._cleanup = [cls.service_offering]
|
|
return
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
super(TestAccounts,cls).tearDownClass()
|
|
|
|
def setUp(self):
|
|
self.apiclient = self.testClient.getApiClient()
|
|
self.dbclient = self.testClient.getDbConnection()
|
|
self.cleanup = []
|
|
return
|
|
|
|
def tearDown(self):
|
|
super(TestAccounts,self).tearDown()
|
|
|
|
@attr(
|
|
tags=[
|
|
"advanced",
|
|
"basic",
|
|
"eip",
|
|
"advancedns",
|
|
"sg"],
|
|
required_hardware="false")
|
|
def test_01_create_account(self):
|
|
"""Test Create Account and user for that account
|
|
"""
|
|
|
|
# Validate the following
|
|
# 1. Create an Account. Verify the account is created.
|
|
# 2. Create User associated with that account. Verify the created user
|
|
|
|
# Create an account
|
|
account = Account.create(
|
|
self.apiclient,
|
|
self.services["account"]
|
|
)
|
|
self.debug("Created account: %s" % account.name)
|
|
self.cleanup.append(account)
|
|
list_accounts_response = list_accounts(
|
|
self.apiclient,
|
|
id=account.id
|
|
)
|
|
self.assertEqual(
|
|
isinstance(list_accounts_response, list),
|
|
True,
|
|
"Check list accounts for valid data"
|
|
)
|
|
self.assertNotEqual(
|
|
len(list_accounts_response),
|
|
0,
|
|
"Check List Account response"
|
|
)
|
|
|
|
account_response = list_accounts_response[0]
|
|
self.assertEqual(
|
|
account.accounttype,
|
|
account_response.accounttype,
|
|
"Check Account Type of Created account"
|
|
)
|
|
self.assertEqual(
|
|
account.name,
|
|
account_response.name,
|
|
"Check Account Name of Created account"
|
|
)
|
|
# Create an User associated with account
|
|
user = User.create(
|
|
self.apiclient,
|
|
self.services["user"],
|
|
account=account.name,
|
|
domainid=account.domainid
|
|
)
|
|
self.cleanup.append(user)
|
|
self.debug("Created user: %s" % user.id)
|
|
list_users_response = list_users(
|
|
self.apiclient,
|
|
id=user.id
|
|
)
|
|
self.assertEqual(
|
|
isinstance(list_users_response, list),
|
|
True,
|
|
"Check list users for valid data"
|
|
)
|
|
|
|
self.assertNotEqual(
|
|
len(list_users_response),
|
|
0,
|
|
"Check List User response"
|
|
)
|
|
|
|
user_response = list_users_response[0]
|
|
self.assertEqual(
|
|
user.username,
|
|
user_response.username,
|
|
"Check username of Created user"
|
|
)
|
|
self.assertEqual(
|
|
user.state,
|
|
user_response.state,
|
|
"Check state of created user"
|
|
)
|
|
self.assertEqual(
|
|
"native",
|
|
user_response.usersource,
|
|
"Check user source of created user"
|
|
)
|
|
return
|
|
|
|
|
|
@attr(tags=["advanced", "basic", "eip", "advancedns", "sg"],
|
|
required_hardware="false")
|
|
def test_02_update_account(self):
|
|
"""
|
|
Tests that accounts can be updated with new name, network domain, dynamic role
|
|
:return:
|
|
"""
|
|
dynamic_roles_active = self.apiclient.listCapabilities(listCapabilities.listCapabilitiesCmd()).dynamicrolesenabled
|
|
if not dynamic_roles_active:
|
|
self.skipTest("Dynamic Role-Based API checker not enabled, skipping test")
|
|
|
|
ts = str(time.time())
|
|
network_domain = 'mycloud.com'
|
|
|
|
# role will be in cleanup list first so it won't be deleted before the account it is on.
|
|
role = Role.create(self.apiclient, self.services['role'])
|
|
self.cleanup.append(role)
|
|
|
|
account = Account.create(self.apiclient, self.services['account'])
|
|
self.cleanup.append(account)
|
|
|
|
account.update(self.apiclient, newname=account.name + ts)
|
|
account.update(self.apiclient, roleid=role.id)
|
|
account.update(self.apiclient, networkdomain=network_domain)
|
|
|
|
list_accounts_response = list_accounts(self.apiclient, id=account.id)
|
|
test_account = list_accounts_response[0]
|
|
|
|
self.assertEqual(
|
|
test_account.roleid, role.id,
|
|
"Check the role for the account is changed")
|
|
|
|
self.assertEqual(
|
|
test_account.networkdomain, network_domain,
|
|
"Check the domain for the account is changed")
|
|
|
|
self.assertEqual(
|
|
test_account.name, account.name + ts,
|
|
"Check the name for the account is changed")
|
|
|
|
try:
|
|
account.update(self.apiclient, newname="")
|
|
self.fail("Account name change to empty name succeeded. Must be error.")
|
|
except CloudstackAPIException:
|
|
pass
|
|
|
|
|
|
|
|
class TestRemoveUserFromAccount(cloudstackTestCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.testClient = super(
|
|
TestRemoveUserFromAccount,
|
|
cls).getClsTestClient()
|
|
cls.api_client = cls.testClient.getApiClient()
|
|
|
|
cls.services = Services().services
|
|
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
|
|
cls.hypervisor = cls.testClient.getHypervisorInfo()
|
|
cls.services['mode'] = cls.zone.networktype
|
|
cls.template = get_test_template(
|
|
cls.api_client,
|
|
cls.zone.id,
|
|
cls.hypervisor
|
|
)
|
|
cls.services["virtual_machine"]["zoneid"] = cls.zone.id
|
|
cls.services["virtual_machine"]["template"] = cls.template.id
|
|
cls._cleanup = []
|
|
|
|
cls.service_offering = ServiceOffering.create(
|
|
cls.api_client,
|
|
cls.services["service_offering"]
|
|
)
|
|
cls._cleanup.append(cls.service_offering)
|
|
# Create an account
|
|
cls.account = Account.create(
|
|
cls.api_client,
|
|
cls.services["account"]
|
|
)
|
|
cls._cleanup.append(cls.account)
|
|
|
|
return
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
super(TestRemoveUserFromAccount,cls).tearDownClass()
|
|
|
|
def setUp(self):
|
|
self.apiclient = self.testClient.getApiClient()
|
|
self.dbclient = self.testClient.getDbConnection()
|
|
self.cleanup = []
|
|
return
|
|
|
|
def tearDown(self):
|
|
super(TestRemoveUserFromAccount,self).tearDown()
|
|
|
|
@attr(
|
|
tags=[
|
|
"advanced",
|
|
"basic",
|
|
"eip",
|
|
"advancedns",
|
|
"sg"],
|
|
required_hardware="false")
|
|
def test_01_user_remove_VM_running(self):
|
|
"""Test Remove one user from the account
|
|
"""
|
|
|
|
# Validate the following
|
|
# 1. Create an account with 2 users.
|
|
# 2. Start 2 VMs; one for each user of the account
|
|
# 3. Remove one user from the account. Verify that account
|
|
# still exists.
|
|
# 4. Verify that VM started by the removed user are still running
|
|
|
|
# Create an User associated with account and VMs
|
|
user_1 = User.create(
|
|
self.apiclient,
|
|
self.services["user"],
|
|
account=self.account.name,
|
|
domainid=self.account.domainid
|
|
)
|
|
self.debug("Created user: %s" % user_1.id)
|
|
self.cleanup.append(user_1)
|
|
|
|
user_2 = User.create(
|
|
self.apiclient,
|
|
self.services["user"],
|
|
account=self.account.name,
|
|
domainid=self.account.domainid
|
|
)
|
|
self.debug("Created user: %s" % user_2.id)
|
|
self.cleanup.append(user_2)
|
|
|
|
vm_1 = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
accountid=self.account.name,
|
|
domainid=self.account.domainid,
|
|
serviceofferingid=self.service_offering.id
|
|
)
|
|
self.debug("Deployed VM in account: %s, ID: %s" % (
|
|
self.account.name,
|
|
vm_1.id
|
|
))
|
|
self.cleanup.append(vm_1)
|
|
|
|
vm_2 = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
accountid=self.account.name,
|
|
domainid=self.account.domainid,
|
|
serviceofferingid=self.service_offering.id
|
|
)
|
|
self.debug("Deployed VM in account: %s, ID: %s" % (
|
|
self.account.name,
|
|
vm_2.id
|
|
))
|
|
self.cleanup.append(vm_2)
|
|
|
|
# Remove one of the user
|
|
self.debug("Deleting user: %s" % user_1.id)
|
|
user_1.delete(self.apiclient)
|
|
self.cleanup.remove(user_1)
|
|
|
|
# Account should exist after deleting user
|
|
accounts_response = list_accounts(
|
|
self.apiclient,
|
|
id=self.account.id
|
|
)
|
|
self.assertEqual(
|
|
isinstance(accounts_response, list),
|
|
True,
|
|
"Check for valid list accounts response"
|
|
)
|
|
|
|
self.assertNotEqual(
|
|
len(accounts_response),
|
|
0,
|
|
"Check List Account response"
|
|
)
|
|
vm_response = list_virtual_machines(
|
|
self.apiclient,
|
|
account=self.account.name,
|
|
domainid=self.account.domainid
|
|
)
|
|
self.assertEqual(
|
|
isinstance(vm_response, list),
|
|
True,
|
|
"Check for valid list VM response"
|
|
)
|
|
|
|
self.assertNotEqual(
|
|
len(vm_response),
|
|
0,
|
|
"Check List VM response"
|
|
)
|
|
|
|
# VMs associated with that account should be running
|
|
for vm in vm_response:
|
|
self.assertEqual(
|
|
vm.state,
|
|
'Running',
|
|
"Check state of VMs associated with account"
|
|
)
|
|
return
|
|
|
|
|
|
class TestNonRootAdminsPrivileges(cloudstackTestCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.testClient = super(
|
|
TestNonRootAdminsPrivileges,
|
|
cls).getClsTestClient()
|
|
cls.api_client = cls.testClient.getApiClient()
|
|
|
|
cls.services = Services().services
|
|
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
|
|
cls.services['mode'] = cls.zone.networktype
|
|
cls._cleanup = []
|
|
# Create an account, domain etc
|
|
cls.domain = Domain.create(
|
|
cls.api_client,
|
|
cls.services["domain"],
|
|
)
|
|
cls._cleanup.append(cls.domain)
|
|
cls.account = Account.create(
|
|
cls.api_client,
|
|
cls.services["account"],
|
|
admin=True,
|
|
domainid=cls.domain.id
|
|
)
|
|
cls._cleanup.append(cls.account)
|
|
return
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
super(TestNonRootAdminsPrivileges,cls).tearDownClass()
|
|
|
|
def setUp(self):
|
|
self.apiclient = self.testClient.getApiClient()
|
|
self.dbclient = self.testClient.getDbConnection()
|
|
self.cleanup = []
|
|
return
|
|
|
|
def tearDown(self):
|
|
super(TestNonRootAdminsPrivileges,self).tearDown()
|
|
|
|
@attr(
|
|
tags=[
|
|
"advanced",
|
|
"basic",
|
|
"eip",
|
|
"advancedns",
|
|
"sg"],
|
|
required_hardware="false")
|
|
def test_01_non_root_admin_Privileges(self):
|
|
"""Test to verify Non Root admin previleges"""
|
|
|
|
# Validate the following
|
|
# 1. Create few accounts/users in ROOT domain
|
|
# 2. Verify listAccounts API gives only accounts associated with new
|
|
# domain.
|
|
|
|
# Create accounts for ROOT domain
|
|
account_1 = Account.create(
|
|
self.apiclient,
|
|
self.services["account"]
|
|
)
|
|
self.debug("Created account: %s" % account_1.name)
|
|
self.cleanup.append(account_1)
|
|
account_2 = Account.create(
|
|
self.apiclient,
|
|
self.services["account"]
|
|
)
|
|
self.debug("Created account: %s" % account_2.name)
|
|
self.cleanup.append(account_2)
|
|
|
|
accounts_response = list_accounts(
|
|
self.apiclient,
|
|
domainid=self.domain.id,
|
|
listall=True
|
|
)
|
|
|
|
self.assertEqual(
|
|
isinstance(accounts_response, list),
|
|
True,
|
|
"Check list accounts response for valid data"
|
|
)
|
|
|
|
self.assertEqual(
|
|
len(accounts_response),
|
|
1,
|
|
"Check List accounts response"
|
|
)
|
|
# Verify only account associated with domain is listed
|
|
for account in accounts_response:
|
|
self.assertEqual(
|
|
account.domainid,
|
|
self.domain.id,
|
|
"Check domain ID of account"
|
|
)
|
|
return
|
|
|
|
|
|
class TestServiceOfferingSiblings(cloudstackTestCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.api_client = super(
|
|
TestServiceOfferingSiblings,
|
|
cls
|
|
).getClsTestClient().getApiClient()
|
|
cls.services = Services().services
|
|
|
|
cls._cleanup = []
|
|
|
|
# Create Domains, accounts etc
|
|
cls.domain_1 = Domain.create(
|
|
cls.api_client,
|
|
cls.services["domain"]
|
|
)
|
|
cls._cleanup.append(cls.domain_1)
|
|
cls.domain_2 = Domain.create(
|
|
cls.api_client,
|
|
cls.services["domain"]
|
|
)
|
|
cls._cleanup.append(cls.domain_2)
|
|
cls.service_offering = ServiceOffering.create(
|
|
cls.api_client,
|
|
cls.services["service_offering"],
|
|
domainid=cls.domain_1.id
|
|
)
|
|
cls._cleanup.append(cls.service_offering)
|
|
# Create account for doamin_1
|
|
cls.account_1 = Account.create(
|
|
cls.api_client,
|
|
cls.services["account"],
|
|
admin=True,
|
|
domainid=cls.domain_1.id
|
|
)
|
|
cls._cleanup.append(cls.account_1)
|
|
|
|
# Create an account for domain_2
|
|
cls.account_2 = Account.create(
|
|
cls.api_client,
|
|
cls.services["account"],
|
|
admin=True,
|
|
domainid=cls.domain_2.id
|
|
)
|
|
cls._cleanup.append(cls.account_2)
|
|
|
|
return
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
super(TestServiceOfferingSiblings,cls).tearDownClass()
|
|
|
|
def setUp(self):
|
|
self.apiclient = self.testClient.getApiClient()
|
|
self.dbclient = self.testClient.getDbConnection()
|
|
self.cleanup = []
|
|
return
|
|
|
|
def tearDown(self):
|
|
super(TestServiceOfferingSiblings,self).tearDown()
|
|
|
|
@attr(
|
|
tags=[
|
|
"advanced",
|
|
"basic",
|
|
"eip",
|
|
"advancedns",
|
|
"sg"],
|
|
required_hardware="false")
|
|
def test_01_service_offering_siblings(self):
|
|
"""Test to verify service offerings at same level in hierarchy"""
|
|
|
|
# Validate the following
|
|
# 1. Verify service offering is visible for domain_1
|
|
# 2. Verify service offering is not visible for domain_2
|
|
|
|
service_offerings = list_service_offering(
|
|
self.apiclient,
|
|
domainid=self.domain_1.id
|
|
)
|
|
self.assertEqual(
|
|
isinstance(service_offerings, list),
|
|
True,
|
|
"Check if valid list service offerings response"
|
|
)
|
|
|
|
self.assertNotEqual(
|
|
len(service_offerings),
|
|
0,
|
|
"Check List Service Offerings response"
|
|
)
|
|
|
|
for service_offering in service_offerings:
|
|
self.debug("Validating service offering: %s" % service_offering.id)
|
|
self.assertEqual(
|
|
service_offering.id,
|
|
self.service_offering.id,
|
|
"Check Service offering ID for domain" +
|
|
str(self.domain_1.name)
|
|
)
|
|
# Verify private service offering is not visible to other domain
|
|
service_offerings = list_service_offering(
|
|
self.apiclient,
|
|
domainid=self.domain_2.id
|
|
)
|
|
self.assertEqual(
|
|
service_offerings,
|
|
None,
|
|
"Check List Service Offerings response for other domain"
|
|
)
|
|
return
|
|
|
|
|
|
class TestServiceOfferingHierarchy(cloudstackTestCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.api_client = super(
|
|
TestServiceOfferingHierarchy,
|
|
cls
|
|
).getClsTestClient().getApiClient()
|
|
cls.services = Services().services
|
|
|
|
cls._cleanup = []
|
|
|
|
# Create domain, service offerings etc
|
|
cls.domain_1 = Domain.create(
|
|
cls.api_client,
|
|
cls.services["domain"]
|
|
)
|
|
cls._cleanup.append(cls.domain_1)
|
|
cls.domain_2 = Domain.create(
|
|
cls.api_client,
|
|
cls.services["domain"],
|
|
parentdomainid=cls.domain_1.id
|
|
)
|
|
cls._cleanup.append(cls.domain_2)
|
|
cls.service_offering = ServiceOffering.create(
|
|
cls.api_client,
|
|
cls.services["service_offering"],
|
|
domainid=cls.domain_1.id
|
|
)
|
|
cls._cleanup.append(cls.service_offering)
|
|
# Create account for doamin_1
|
|
cls.account_1 = Account.create(
|
|
cls.api_client,
|
|
cls.services["account"],
|
|
admin=True,
|
|
domainid=cls.domain_1.id
|
|
)
|
|
cls._cleanup.append(cls.account_1)
|
|
# Create an account for domain_2
|
|
cls.account_2 = Account.create(
|
|
cls.api_client,
|
|
cls.services["account"],
|
|
admin=True,
|
|
domainid=cls.domain_2.id
|
|
)
|
|
cls._cleanup.append(cls.account_2)
|
|
|
|
return
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
super(TestServiceOfferingHierarchy,cls).tearDownClass()
|
|
|
|
def setUp(self):
|
|
self.apiclient = self.testClient.getApiClient()
|
|
self.dbclient = self.testClient.getDbConnection()
|
|
self.cleanup = []
|
|
return
|
|
|
|
def tearDown(self):
|
|
super(TestServiceOfferingHierarchy,self).tearDown()
|
|
|
|
@attr(
|
|
tags=[
|
|
"advanced",
|
|
"basic",
|
|
"eip",
|
|
"advancedns",
|
|
"sg"],
|
|
required_hardware="false")
|
|
def test_01_service_offering_hierarchy(self):
|
|
"""Test to verify service offerings at same level in hierarchy"""
|
|
|
|
# Validate the following
|
|
# 1. Verify service offering is visible for domain_1
|
|
# 2. Verify service offering is also visible for domain_2
|
|
|
|
service_offerings = list_service_offering(
|
|
self.apiclient,
|
|
domainid=self.domain_1.id
|
|
)
|
|
self.assertEqual(
|
|
isinstance(service_offerings, list),
|
|
True,
|
|
"Check List Service Offerings for a valid response"
|
|
)
|
|
self.assertNotEqual(
|
|
len(service_offerings),
|
|
0,
|
|
"Check List Service Offerings response"
|
|
)
|
|
|
|
for service_offering in service_offerings:
|
|
self.assertEqual(
|
|
service_offering.id,
|
|
self.service_offering.id,
|
|
"Check Service offering ID for domain" +
|
|
str(self.domain_1.name)
|
|
)
|
|
|
|
# Verify private service offering is not visible to other domain
|
|
service_offerings = list_service_offering(
|
|
self.apiclient,
|
|
domainid=self.domain_2.id
|
|
)
|
|
self.assertEqual(
|
|
service_offerings,
|
|
None,
|
|
"Check List Service Offerings for a valid response"
|
|
)
|
|
return
|
|
|
|
|
|
class TestTemplateHierarchy(cloudstackTestCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.testClient = super(TestTemplateHierarchy, cls).getClsTestClient()
|
|
cls.api_client = cls.testClient.getApiClient()
|
|
cls.hypervisor = cls.testClient.getHypervisorInfo()
|
|
|
|
cls.services = Services().services
|
|
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
|
|
cls.services['mode'] = cls.zone.networktype
|
|
|
|
cls._cleanup = []
|
|
|
|
# Create domains, accounts and template
|
|
cls.domain_1 = Domain.create(
|
|
cls.api_client,
|
|
cls.services["domain"]
|
|
)
|
|
cls._cleanup.append(cls.domain_1)
|
|
cls.domain_2 = Domain.create(
|
|
cls.api_client,
|
|
cls.services["domain"],
|
|
parentdomainid=cls.domain_1.id
|
|
)
|
|
cls._cleanup.append(cls.domain_2)
|
|
|
|
# Create account for doamin_1
|
|
cls.account_1 = Account.create(
|
|
cls.api_client,
|
|
cls.services["account"],
|
|
admin=True,
|
|
domainid=cls.domain_1.id
|
|
)
|
|
cls._cleanup.append(cls.account_1)
|
|
|
|
# Create an account for domain_2
|
|
cls.account_2 = Account.create(
|
|
cls.api_client,
|
|
cls.services["account"],
|
|
admin=True,
|
|
domainid=cls.domain_2.id
|
|
)
|
|
cls._cleanup.append(cls.account_2)
|
|
|
|
builtin_info = get_builtin_template_info(cls.api_client, cls.zone.id)
|
|
cls.services["template"]["url"] = builtin_info[0]
|
|
cls.services["template"]["hypervisor"] = builtin_info[1]
|
|
cls.services["template"]["format"] = builtin_info[2]
|
|
|
|
# Register new template
|
|
cls.template = Template.register(
|
|
cls.api_client,
|
|
cls.services["template"],
|
|
zoneid=cls.zone.id,
|
|
account=cls.account_1.name,
|
|
domainid=cls.domain_1.id,
|
|
hypervisor=cls.hypervisor
|
|
)
|
|
cls._cleanup.append(cls.template)
|
|
|
|
# Wait for template to download
|
|
cls.template.download(cls.api_client)
|
|
|
|
# Wait for template status to be changed across
|
|
time.sleep(60)
|
|
return
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
super(TestTemplateHierarchy,cls).tearDownClass()
|
|
|
|
def setUp(self):
|
|
self.apiclient = self.testClient.getApiClient()
|
|
self.dbclient = self.testClient.getDbConnection()
|
|
self.cleanup = []
|
|
return
|
|
|
|
def tearDown(self):
|
|
super(TestTemplateHierarchy,self).tearDown()
|
|
|
|
@attr(tags=["advanced", "basic", "eip", "advancedns", "sg"], required_hardware="true")
|
|
def test_01_template_hierarchy(self):
|
|
"""Test to verify template at same level in hierarchy"""
|
|
|
|
# Validate the following
|
|
# 1. Verify template is visible for domain_1
|
|
# 2. Verify template is also visible for domain_2
|
|
|
|
# Sleep to ensure that template state is reflected across
|
|
|
|
templates = list_templates(
|
|
self.apiclient,
|
|
templatefilter='self',
|
|
account=self.account_1.name,
|
|
domainid=self.domain_1.id
|
|
)
|
|
self.assertEqual(
|
|
isinstance(templates, list),
|
|
True,
|
|
"Template response %s is not a list" % templates
|
|
)
|
|
self.assertNotEqual(
|
|
len(templates),
|
|
0,
|
|
"No templates found"
|
|
)
|
|
|
|
for template in templates:
|
|
self.assertEqual(
|
|
template.id,
|
|
self.template.id,
|
|
"Check Template ID for domain" + str(self.domain_1.name)
|
|
)
|
|
|
|
# Verify private service offering is not visible to other domain
|
|
templates = list_templates(
|
|
self.apiclient,
|
|
id=self.template.id,
|
|
templatefilter='all',
|
|
account=self.account_2.name,
|
|
domainid=self.domain_2.id
|
|
)
|
|
self.assertEqual(
|
|
isinstance(templates, list),
|
|
True,
|
|
"Template response %s is not a list" % templates
|
|
)
|
|
self.assertNotEqual(
|
|
len(templates),
|
|
0,
|
|
"No templates found"
|
|
)
|
|
|
|
for template in templates:
|
|
self.assertEqual(
|
|
template.id,
|
|
self.template.id,
|
|
"Check Template ID for domain" + str(self.domain_2.name)
|
|
)
|
|
return
|
|
|
|
|
|
class TestAddVmToSubDomain(cloudstackTestCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.testClient = super(TestAddVmToSubDomain, cls).getClsTestClient()
|
|
cls.api_client = cls.testClient.getApiClient()
|
|
|
|
cls.services = Services().services
|
|
cls.domain = get_domain(cls.api_client)
|
|
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
|
|
cls.hypervisor = cls.testClient.getHypervisorInfo()
|
|
cls.services['mode'] = cls.zone.networktype
|
|
cls._cleanup = []
|
|
|
|
cls.sub_domain = Domain.create(
|
|
cls.api_client,
|
|
cls.services["domain"],
|
|
parentdomainid=cls.domain.id
|
|
)
|
|
cls._cleanup.append(cls.sub_domain)
|
|
|
|
# Create account for doamin_1
|
|
cls.account_1 = Account.create(
|
|
cls.api_client,
|
|
cls.services["account"],
|
|
admin=True,
|
|
domainid=cls.domain.id
|
|
)
|
|
cls._cleanup.append(cls.account_1)
|
|
|
|
# Create an account for domain_2
|
|
cls.account_2 = Account.create(
|
|
cls.api_client,
|
|
cls.services["account"],
|
|
admin=True,
|
|
domainid=cls.sub_domain.id
|
|
)
|
|
cls._cleanup.append(cls.account_2)
|
|
|
|
cls.service_offering = ServiceOffering.create(
|
|
cls.api_client,
|
|
cls.services["service_offering"],
|
|
domainid=cls.domain.id
|
|
)
|
|
cls._cleanup.append(cls.service_offering)
|
|
|
|
cls.template = get_test_template(
|
|
cls.api_client,
|
|
cls.zone.id,
|
|
cls.hypervisor
|
|
)
|
|
cls.services["virtual_machine"]["zoneid"] = cls.zone.id
|
|
cls.vm_1 = VirtualMachine.create(
|
|
cls.api_client,
|
|
cls.services["virtual_machine"],
|
|
templateid=cls.template.id,
|
|
accountid=cls.account_1.name,
|
|
domainid=cls.account_1.domainid,
|
|
serviceofferingid=cls.service_offering.id
|
|
)
|
|
cls._cleanup.append(cls.vm_1)
|
|
|
|
cls.vm_2 = VirtualMachine.create(
|
|
cls.api_client,
|
|
cls.services["virtual_machine"],
|
|
templateid=cls.template.id,
|
|
accountid=cls.account_2.name,
|
|
domainid=cls.account_2.domainid,
|
|
serviceofferingid=cls.service_offering.id
|
|
)
|
|
cls._cleanup.append(cls.vm_2)
|
|
return
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
super(TestAddVmToSubDomain,cls).tearDownClass()
|
|
|
|
def setUp(self):
|
|
self.apiclient = self.testClient.getApiClient()
|
|
self.dbclient = self.testClient.getDbConnection()
|
|
self.cleanup = []
|
|
return
|
|
|
|
def tearDown(self):
|
|
super(TestAddVmToSubDomain,self).tearDown()
|
|
|
|
@attr(
|
|
tags=[
|
|
"advanced",
|
|
"basic",
|
|
"eip",
|
|
"advancedns",
|
|
"sg"],
|
|
required_hardware="false")
|
|
def test_01_add_vm_to_subdomain(self):
|
|
""" Test Sub domain allowed to launch VM when a Domain
|
|
level zone is created"""
|
|
|
|
# Validate the following
|
|
# 1. Verify VM created by Account_1 is in Running state
|
|
# 2. Verify VM created by Account_2 is in Running state
|
|
|
|
vm_response = list_virtual_machines(
|
|
self.apiclient,
|
|
id=self.vm_1.id
|
|
)
|
|
self.assertEqual(
|
|
isinstance(vm_response, list),
|
|
True,
|
|
"Check List VM for a valid response"
|
|
)
|
|
self.assertNotEqual(
|
|
len(vm_response),
|
|
0,
|
|
"Check List Template response"
|
|
)
|
|
|
|
for vm in vm_response:
|
|
self.debug("VM ID: %s and state: %s" % (vm.id, vm.state))
|
|
self.assertEqual(
|
|
vm.state,
|
|
'Running',
|
|
"Check State of Virtual machine"
|
|
)
|
|
|
|
vm_response = list_virtual_machines(
|
|
self.apiclient,
|
|
id=self.vm_2.id
|
|
)
|
|
self.assertNotEqual(
|
|
len(vm_response),
|
|
0,
|
|
"Check List Template response"
|
|
)
|
|
|
|
for vm in vm_response:
|
|
self.debug("VM ID: %s and state: %s" % (vm.id, vm.state))
|
|
self.assertEqual(
|
|
vm.state,
|
|
'Running',
|
|
"Check State of Virtual machine"
|
|
)
|
|
return
|
|
|
|
|
|
class TestUserDetails(cloudstackTestCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.testClient = super(TestUserDetails, cls).getClsTestClient()
|
|
cls.api_client = cls.testClient.getApiClient()
|
|
|
|
cls.services = Services().services
|
|
cls.domain = get_domain(cls.api_client)
|
|
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
|
|
cls.services['mode'] = cls.zone.networktype
|
|
cls._cleanup = []
|
|
return
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
super(TestUserDetails,cls).tearDownClass()
|
|
|
|
def setUp(self):
|
|
self.apiclient = self.testClient.getApiClient()
|
|
self.dbclient = self.testClient.getDbConnection()
|
|
self.cleanup = []
|
|
return
|
|
|
|
def tearDown(self):
|
|
super(TestUserDetails,self).tearDown()
|
|
|
|
@attr(tags=[
|
|
"role",
|
|
"accounts",
|
|
"simulator",
|
|
"advanced",
|
|
"advancedns",
|
|
"basic",
|
|
"eip",
|
|
"sg"
|
|
])
|
|
def test_updateUserDetails(self):
|
|
"""Test user update API
|
|
"""
|
|
|
|
# Steps for test scenario
|
|
# 1. create a user account
|
|
# 2. update the user details (firstname, lastname, user) with
|
|
# updateUser API
|
|
# 3. listUsers in the account
|
|
# 4. delete the account
|
|
# Validate the following
|
|
# 1. listAccounts should show account created successfully
|
|
# 2. updateUser API should return valid response
|
|
# 3. user should be updated with new details
|
|
|
|
self.debug("Creating an user account..")
|
|
self.account = Account.create(
|
|
self.apiclient,
|
|
self.services["account"],
|
|
domainid=self.domain.id
|
|
)
|
|
self.cleanup.append(self.account)
|
|
|
|
# Fetching the user details of account
|
|
self.debug(
|
|
"Fetching user details for account: %s" %
|
|
self.account.name)
|
|
users = User.list(
|
|
self.apiclient,
|
|
account=self.account.name,
|
|
domainid=self.account.domainid
|
|
)
|
|
self.assertEqual(
|
|
isinstance(users, list),
|
|
True,
|
|
"List users should return a valid list for account"
|
|
)
|
|
user_1 = users[0]
|
|
self.debug("Updating the details of user: %s" % user_1.name)
|
|
firstname = random_gen()
|
|
lastname = random_gen()
|
|
|
|
self.debug("New firstname: %s, lastname: %s" % (firstname, lastname))
|
|
User.update(
|
|
self.apiclient,
|
|
user_1.id,
|
|
firstname=firstname,
|
|
lastname=lastname
|
|
)
|
|
|
|
# Fetching the user details of account
|
|
self.debug(
|
|
"Fetching user details for user: %s" % user_1.name)
|
|
users = User.list(
|
|
self.apiclient,
|
|
id=user_1.id,
|
|
listall=True
|
|
)
|
|
|
|
self.assertEqual(
|
|
isinstance(users, list),
|
|
True,
|
|
"List users should return a valid list for account"
|
|
)
|
|
user_1 = users[0]
|
|
self.assertEqual(
|
|
user_1.firstname,
|
|
firstname,
|
|
"User's first name should be updated with new one"
|
|
)
|
|
self.assertEqual(
|
|
user_1.lastname,
|
|
lastname,
|
|
"User's last name should be updated with new one"
|
|
)
|
|
return
|
|
|
|
@attr(tags=[
|
|
"role",
|
|
"accounts",
|
|
"simulator",
|
|
"advanced",
|
|
"advancedns",
|
|
"basic",
|
|
"eip",
|
|
"sg"
|
|
])
|
|
def test_updateAdminDetails(self):
|
|
"""Test update admin details
|
|
"""
|
|
|
|
# Steps for test scenario
|
|
# 1. create a admin account
|
|
# 2. update the user details (firstname, lastname, user) with
|
|
# updateUser API
|
|
# 3. listUsers in the account
|
|
# 4. delete the account
|
|
# Validate the following
|
|
# 1. listAccounts should show account created successfully
|
|
# 2. updateUser API should return valid response
|
|
# 3. user should be updated with new details
|
|
|
|
self.debug("Creating a ROOT admin account")
|
|
self.account = Account.create(
|
|
self.apiclient,
|
|
self.services["account"],
|
|
admin=True,
|
|
)
|
|
self.cleanup.append(self.account)
|
|
|
|
# Fetching the user details of account
|
|
self.debug(
|
|
"Fetching user details for account: %s" %
|
|
self.account.name)
|
|
users = User.list(
|
|
self.apiclient,
|
|
account=self.account.name,
|
|
domainid=self.account.domainid
|
|
)
|
|
self.assertEqual(
|
|
isinstance(users, list),
|
|
True,
|
|
"List users should return a valid list for account"
|
|
)
|
|
user_1 = users[0]
|
|
self.debug("Updating the details of user: %s" % user_1.name)
|
|
firstname = random_gen()
|
|
lastname = random_gen()
|
|
|
|
self.debug("New firstname: %s, lastname: %s" % (firstname, lastname))
|
|
User.update(
|
|
self.apiclient,
|
|
user_1.id,
|
|
firstname=firstname,
|
|
lastname=lastname
|
|
)
|
|
|
|
# Fetching the user details of account
|
|
self.debug(
|
|
"Fetching user details for user: %s" % user_1.name)
|
|
users = User.list(
|
|
self.apiclient,
|
|
id=user_1.id,
|
|
listall=True
|
|
)
|
|
|
|
self.assertEqual(
|
|
isinstance(users, list),
|
|
True,
|
|
"List users should return a valid list for account"
|
|
)
|
|
user_1 = users[0]
|
|
self.assertEqual(
|
|
user_1.firstname,
|
|
firstname,
|
|
"User's first name should be updated with new one"
|
|
)
|
|
self.assertEqual(
|
|
user_1.lastname,
|
|
lastname,
|
|
"User's last name should be updated with new one"
|
|
)
|
|
return
|
|
|
|
@attr(tags=[
|
|
"role",
|
|
"accounts",
|
|
"simulator",
|
|
"advanced",
|
|
"advancedns",
|
|
"basic",
|
|
"eip",
|
|
"sg"
|
|
])
|
|
def test_updateDomainAdminDetails(self):
|
|
"""Test update domain admin details
|
|
"""
|
|
|
|
# Steps for test scenario
|
|
# 2. update the user details (firstname, lastname, user) with
|
|
# updateUser API
|
|
# 3. listUsers in the account
|
|
# 4. delete the account
|
|
# Validate the following
|
|
# 1. listAccounts should show account created successfully
|
|
# 2. updateUser API should return valid response
|
|
# 3. user should be updated with new details
|
|
|
|
self.debug("Creating a domain admin account")
|
|
self.account = Account.create(
|
|
self.apiclient,
|
|
self.services["account"],
|
|
admin=True,
|
|
domainid=self.domain.id
|
|
)
|
|
self.cleanup.append(self.account)
|
|
|
|
# Fetching the user details of account
|
|
self.debug(
|
|
"Fetching user details for account: %s" %
|
|
self.account.name)
|
|
users = User.list(
|
|
self.apiclient,
|
|
account=self.account.name,
|
|
domainid=self.account.domainid
|
|
)
|
|
self.assertEqual(
|
|
isinstance(users, list),
|
|
True,
|
|
"List users should return a valid list for account"
|
|
)
|
|
user_1 = users[0]
|
|
self.debug("Updating the details of user: %s" % user_1.name)
|
|
firstname = random_gen()
|
|
lastname = random_gen()
|
|
|
|
self.debug("New firstname: %s, lastname: %s" % (firstname, lastname))
|
|
User.update(
|
|
self.apiclient,
|
|
user_1.id,
|
|
firstname=firstname,
|
|
lastname=lastname
|
|
)
|
|
|
|
# Fetching the user details of account
|
|
self.debug(
|
|
"Fetching user details for user: %s" % user_1.name)
|
|
users = User.list(
|
|
self.apiclient,
|
|
id=user_1.id,
|
|
listall=True
|
|
)
|
|
|
|
self.assertEqual(
|
|
isinstance(users, list),
|
|
True,
|
|
"List users should return a valid list for account"
|
|
)
|
|
user_1 = users[0]
|
|
self.assertEqual(
|
|
user_1.firstname,
|
|
firstname,
|
|
"User's first name should be updated with new one"
|
|
)
|
|
self.assertEqual(
|
|
user_1.lastname,
|
|
lastname,
|
|
"User's last name should be updated with new one"
|
|
)
|
|
return
|
|
|
|
|
|
class TestUserLogin(cloudstackTestCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.testClient = super(TestUserLogin, cls).getClsTestClient()
|
|
cls.api_client = cls.testClient.getApiClient()
|
|
|
|
cls.services = Services().services
|
|
cls.domain = get_domain(cls.api_client)
|
|
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
|
|
cls.services['mode'] = cls.zone.networktype
|
|
cls._cleanup = []
|
|
return
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
super(TestUserLogin,cls).tearDownClass()
|
|
|
|
def setUp(self):
|
|
self.apiclient = self.testClient.getApiClient()
|
|
self.dbclient = self.testClient.getDbConnection()
|
|
self.cleanup = []
|
|
return
|
|
|
|
def tearDown(self):
|
|
super(TestUserLogin,self).tearDown()
|
|
|
|
@attr(tags=["login", "accounts", "simulator", "advanced",
|
|
"advancedns", "basic", "eip", "sg"])
|
|
def test_LoginApiUuidResponse(self):
|
|
"""Test if Login API does not return UUID's
|
|
"""
|
|
|
|
# Steps for test scenario
|
|
# 1. create a user account
|
|
# 2. login to the user account with given credentials (loginCmd)
|
|
# 3. delete the user account
|
|
# Validate the following
|
|
# 1. listAccounts should return account created
|
|
# 2. loginResponse should have UUID only is response. Assert by
|
|
# checking database id is not same as response id
|
|
# Login also succeeds with non NULL sessionId in response
|
|
|
|
self.debug("Creating an user account..")
|
|
self.account = Account.create(
|
|
self.apiclient,
|
|
self.services["account"],
|
|
domainid=self.domain.id
|
|
)
|
|
self.cleanup.append(self.account)
|
|
|
|
self.debug("Logging into the cloudstack with login API")
|
|
response = User.login(
|
|
self.apiclient,
|
|
username=self.account.name,
|
|
password=self.services["account"]["password"]
|
|
)
|
|
|
|
self.debug("Login API response: %s" % response)
|
|
|
|
self.assertNotEqual(
|
|
response.sessionkey,
|
|
None,
|
|
"Login to the CloudStack should be successful" +
|
|
"response shall have non Null key"
|
|
)
|
|
return
|
|
|
|
@attr(tags=["simulator", "advanced",
|
|
"advancedns", "basic", "eip", "sg"])
|
|
def test_ApiListDomain(self):
|
|
"""Test case to check the correctness of List domain API, to make sure that no field is missed in the output.
|
|
"""
|
|
|
|
# Steps for test scenario
|
|
# 1. create a domain
|
|
# 2. Create a sub-domain with domain created in step 1 as parent.
|
|
# Validate the following
|
|
# 1. listDomains returns created domain and sub-domain
|
|
# 2. The list Domain response has all the expected 44 elements/fields in it.
|
|
|
|
listDomainResponseElements = ["id", "name", "level", "parentdomainid", "parentdomainname", "haschild", "path",
|
|
"state",
|
|
"vmlimit", "vmtotal", "vmavailable", "iplimit", "iptotal", "ipavailable",
|
|
"volumelimit",
|
|
"volumetotal", "volumeavailable", "snapshotlimit", "snapshottotal",
|
|
"snapshotavailable",
|
|
"templatelimit", "templatetotal", "templateavailable", "projectlimit",
|
|
"projecttotal", "projectavailable",
|
|
"networklimit", "networktotal", "networkavailable", "vpclimit", "vpctotal",
|
|
"vpcavailable",
|
|
"cpulimit", "cputotal", "cpuavailable", "memorylimit", "memorytotal",
|
|
"memoryavailable", "primarystoragelimit",
|
|
"primarystoragetotal", "primarystorageavailable", "secondarystoragelimit",
|
|
"secondarystoragetotal", "secondarystorageavailable"
|
|
]
|
|
|
|
self.debug("Creating a domain for testing list domain response")
|
|
domain = Domain.create(
|
|
self.apiclient,
|
|
self.services["domain"],
|
|
parentdomainid=self.domain.id
|
|
)
|
|
self.cleanup.append(domain)
|
|
self.debug("Domain: %s is created successfully." % domain.name)
|
|
|
|
self.debug("Validating the created domain")
|
|
list_domain = Domain.list(self.api_client, id=domain.id)
|
|
domain_list_validation_result = validateList(list_domain)
|
|
|
|
self.assertEqual(domain_list_validation_result[0], PASS,
|
|
"Domain list validation failed due to %s" %
|
|
domain_list_validation_result[2])
|
|
|
|
subDomain = Domain.create(
|
|
self.apiclient,
|
|
self.services["domain"],
|
|
parentdomainid=domain.id
|
|
)
|
|
self.cleanup.append(subDomain)
|
|
self.debug("Sub-Domain: %s is created successfully." % subDomain.name)
|
|
|
|
self.debug("Validating the created sub-domain")
|
|
list_sub_domain = Domain.list(self.api_client, id=subDomain.id)
|
|
subdomain_list_validation_result = validateList(list_sub_domain)
|
|
|
|
self.assertEqual(subdomain_list_validation_result[0], PASS,
|
|
"Sub-Domain list validation failed due to %s" %
|
|
subdomain_list_validation_result[2])
|
|
|
|
self.debug("Checking that the listDomain response has all the elements.")
|
|
|
|
domainOutputString = list_domain[0].__dict__
|
|
|
|
for element in listDomainResponseElements:
|
|
self.assertTrue((element.lower() in domainOutputString), element + " field is missing in list domain rsponse.")
|
|
|
|
self.debug("Verified that the listDomain response has all the elements.")
|
|
|
|
self.debug("Checking that the list sub-domain response has all the elements.")
|
|
|
|
subdomainOutputString = list_sub_domain[0].__dict__
|
|
|
|
for element in listDomainResponseElements:
|
|
self.assertTrue((element.lower() in subdomainOutputString), element + " field is missing in list domain rsponse.")
|
|
|
|
self.debug("Verified that the list sub-Domain response has all the elements.")
|
|
|
|
return
|
|
|
|
@attr(tags=["login", "accounts", "simulator", "advanced",
|
|
"advancedns", "basic", "eip", "sg"])
|
|
def test_LoginApiDomain(self):
|
|
"""Test login API with domain
|
|
"""
|
|
|
|
# Steps for test scenario
|
|
# 1. create a domain
|
|
# 2. create user in the domain
|
|
# 3. login to the user account above using UUID domain/user
|
|
# 4. delete the user account
|
|
# Validate the following
|
|
# 1. listDomains returns created domain
|
|
# 2. listAccounts returns created user
|
|
# 3. loginResponse should have UUID only in responses
|
|
# Login also succeeds with non NULL sessionId in response
|
|
|
|
self.debug("Creating a domain for login with API domain test")
|
|
domain = Domain.create(
|
|
self.apiclient,
|
|
self.services["domain"],
|
|
parentdomainid=self.domain.id
|
|
)
|
|
self.cleanup.append(domain)
|
|
self.debug("Domain: %s is created successfully." % domain.name)
|
|
self.debug(
|
|
"Checking if the created domain is listed in list domains API")
|
|
domains = Domain.list(self.apiclient, id=domain.id, listall=True)
|
|
|
|
self.assertEqual(
|
|
isinstance(domains, list),
|
|
True,
|
|
"List domains shall return a valid response"
|
|
)
|
|
self.debug("Creating an user account in domain: %s" % domain.name)
|
|
self.account = Account.create(
|
|
self.apiclient,
|
|
self.services["account"],
|
|
domainid=domain.id
|
|
)
|
|
self.cleanup.append(self.account)
|
|
|
|
accounts = Account.list(
|
|
self.apiclient,
|
|
name=self.account.name,
|
|
domainid=self.account.domainid,
|
|
listall=True
|
|
)
|
|
|
|
self.assertEqual(
|
|
isinstance(accounts, list),
|
|
True,
|
|
"List accounts should return a valid response"
|
|
)
|
|
|
|
self.debug("Logging into the cloudstack with login API")
|
|
response = User.login(
|
|
self.apiclient,
|
|
username=self.account.name,
|
|
password=self.services["account"]["password"],
|
|
domainid=domain.id)
|
|
self.debug("Login API response: %s" % response)
|
|
|
|
self.assertNotEqual(
|
|
response.sessionkey,
|
|
None,
|
|
"Login to the CloudStack should be successful" +
|
|
"response shall have non Null key"
|
|
)
|
|
return
|
|
|
|
|
|
class TestUserAPIKeys(cloudstackTestCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.testClient = super(TestUserAPIKeys, cls).getClsTestClient()
|
|
cls.api_client = cls.testClient.getApiClient()
|
|
|
|
cls.services = Services().services
|
|
cls.domain = get_domain(cls.api_client)
|
|
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
|
|
cls.services['mode'] = cls.zone.networktype
|
|
cls._cleanup = []
|
|
# Create an account, domain etc
|
|
cls.domain = Domain.create(
|
|
cls.api_client,
|
|
cls.services["domain"],
|
|
)
|
|
cls._cleanup.append(cls.domain)
|
|
cls.account = Account.create(
|
|
cls.api_client,
|
|
cls.services["account"],
|
|
admin=False,
|
|
domainid=cls.domain.id
|
|
)
|
|
cls._cleanup.append(cls.account)
|
|
cls.domain_2 = Domain.create(
|
|
cls.api_client,
|
|
cls.services["domain"],
|
|
)
|
|
cls._cleanup.append(cls.domain_2)
|
|
cls.account_2 = Account.create(
|
|
cls.api_client,
|
|
cls.services["account"],
|
|
admin=False,
|
|
domainid=cls.domain_2.id
|
|
)
|
|
cls._cleanup.append(cls.account_2)
|
|
return
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
super(TestUserAPIKeys,cls).tearDownClass()
|
|
|
|
def setUp(self):
|
|
self.apiclient = self.testClient.getApiClient()
|
|
self.dbclient = self.testClient.getDbConnection()
|
|
self.cleanup = []
|
|
return
|
|
|
|
def tearDown(self):
|
|
super(TestUserAPIKeys,self).tearDown()
|
|
|
|
@attr(tags=[
|
|
"role",
|
|
"accounts",
|
|
"simulator",
|
|
"advanced",
|
|
"advancedns",
|
|
"basic",
|
|
"eip",
|
|
"sg"
|
|
])
|
|
def test_user_key_renew_same_account(self):
|
|
# Create an User associated with the account
|
|
user_1 = User.create(
|
|
self.apiclient,
|
|
self.services["user"],
|
|
account=self.account.name,
|
|
domainid=self.domain.id
|
|
)
|
|
self.cleanup.append(user_1)
|
|
account_response = list_accounts(
|
|
self.apiclient,
|
|
id=self.account.id
|
|
)[0]
|
|
self.assertEqual(
|
|
hasattr(account_response, 'user'),
|
|
True,
|
|
"Users are included in account response")
|
|
|
|
account_users = account_response.user
|
|
self.assertEqual(
|
|
isinstance(account_users, list),
|
|
True,
|
|
"Check account for valid data"
|
|
)
|
|
self.assertNotEqual(
|
|
len(account_users),
|
|
0,
|
|
"Check number of User in Account")
|
|
[user] = [u for u in account_users if u.username == user_1.username]
|
|
self.assertEqual(
|
|
user.apikey,
|
|
None,
|
|
"Check that the user don't have an API key yet")
|
|
|
|
self.debug("Register API keys for user")
|
|
userkeys = User.registerUserKeys(self.apiclient, user_1.id)
|
|
users = list_accounts(
|
|
self.apiclient,
|
|
id=self.account.id
|
|
)[0].user
|
|
[user] = [u for u in users if u.id == user_1.id]
|
|
self.assertEqual(
|
|
user.apikey,
|
|
userkeys.apikey,
|
|
"Check User api key")
|
|
user.secretkey = self.get_secret_key(user.id)
|
|
self.assertEqual(
|
|
user.secretkey,
|
|
userkeys.secretkey,
|
|
"Check User having secret key")
|
|
|
|
self.debug("Get test client with user keys")
|
|
cs_api = self.testClient.getUserApiClient(
|
|
UserName=self.account.name,
|
|
DomainName=self.account.domain)
|
|
self.debug("Renew API keys for user using current keys")
|
|
new_keys = User.registerUserKeys(cs_api, user_1.id)
|
|
self.assertNotEqual(
|
|
userkeys.apikey,
|
|
new_keys.apikey,
|
|
"Check API key is different")
|
|
new_keys.secretkey = self.get_secret_key(user_1.id)
|
|
self.assertNotEqual(
|
|
userkeys.secretkey,
|
|
new_keys.secretkey,
|
|
"Check secret key is different")
|
|
|
|
def get_secret_key(self, id):
|
|
cmd = getUserKeys.getUserKeysCmd()
|
|
cmd.id = id
|
|
keypair = self.apiclient.getUserKeys(cmd)
|
|
return keypair.secretkey
|
|
|
|
@attr(tags=[
|
|
"role",
|
|
"accounts",
|
|
"simulator",
|
|
"advanced",
|
|
"advancedns",
|
|
"basic",
|
|
"eip",
|
|
"sg"
|
|
])
|
|
def test_user_cannot_renew_other_keys(self):
|
|
cs_api = self.testClient.getUserApiClient(
|
|
UserName=self.account.name,
|
|
DomainName=self.account.domain)
|
|
self.debug("Try to change API key of an account in another domain")
|
|
users = list_accounts(
|
|
self.apiclient,
|
|
id=self.account_2.id
|
|
)[0].user
|
|
with self.assertRaises(CloudstackAPIException) as e:
|
|
User.registerUserKeys(cs_api, users[0].id)
|
|
|
|
|
|
class TestDomainForceRemove(cloudstackTestCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.testClient = super(TestDomainForceRemove, cls).getClsTestClient()
|
|
cls.api_client = cls.testClient.getApiClient()
|
|
|
|
cls.services = Services().services
|
|
cls.domain = get_domain(cls.api_client)
|
|
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
|
|
cls.hypervisor = cls.testClient.getHypervisorInfo()
|
|
cls.services['mode'] = cls.zone.networktype
|
|
|
|
cls.template = get_test_template(
|
|
cls.api_client,
|
|
cls.zone.id,
|
|
cls.hypervisor
|
|
)
|
|
|
|
cls.services["virtual_machine"]["zoneid"] = cls.zone.id
|
|
cls._cleanup = []
|
|
return
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
super(TestDomainForceRemove,cls).tearDownClass()
|
|
|
|
def setUp(self):
|
|
self.apiclient = self.testClient.getApiClient()
|
|
self.dbclient = self.testClient.getDbConnection()
|
|
self.cleanup = []
|
|
return
|
|
|
|
def tearDown(self):
|
|
super(TestDomainForceRemove,self).tearDown()
|
|
|
|
@attr(
|
|
tags=[
|
|
"domains",
|
|
"advanced",
|
|
"advancedns",
|
|
"simulator",
|
|
"dvs"],
|
|
required_hardware="false")
|
|
def test_forceDeleteDomain(self):
|
|
""" Test delete domain with force option"""
|
|
|
|
# Steps for validations
|
|
# 1. create a domain DOM
|
|
# 2. create 2 users under this domain
|
|
# 3. deploy 1 VM into each of these user accounts
|
|
# 4. create PF / FW rules for port 22 on these VMs for their
|
|
# respective accounts
|
|
# 5. delete the domain with force=true option
|
|
# Validate the following
|
|
# 1. listDomains should list the created domain
|
|
# 2. listAccounts should list the created accounts
|
|
# 3. listvirtualmachines should show the Running VMs
|
|
# 4. PF and FW rules should be shown in listFirewallRules
|
|
# 5. domain should delete successfully and above three list calls
|
|
# should show all the resources now deleted. listRouters should
|
|
# not return any routers in the deleted accounts/domains
|
|
|
|
self.debug("Creating a domain for login with API domain test")
|
|
self.child_domain = Domain.create(
|
|
self.apiclient,
|
|
self.services["domain"],
|
|
parentdomainid=self.domain.id
|
|
)
|
|
self.cleanup.append(self.child_domain)
|
|
self.debug("Domain is created successfully.")
|
|
self.debug(
|
|
"Checking if the created domain is listed in list domains API")
|
|
domains = Domain.list(self.apiclient, id=self.child_domain.id, listall=True)
|
|
|
|
self.assertEqual(
|
|
isinstance(domains, list),
|
|
True,
|
|
"List domains shall return a valid response"
|
|
)
|
|
self.debug("Creating 2 user accounts in domain: %s" % self.child_domain.name)
|
|
self.account_1 = Account.create(
|
|
self.apiclient,
|
|
self.services["account"],
|
|
domainid=self.child_domain.id
|
|
)
|
|
self.cleanup.append(self.account_1)
|
|
|
|
self.account_2 = Account.create(
|
|
self.apiclient,
|
|
self.services["account"],
|
|
domainid=self.child_domain.id
|
|
)
|
|
self.cleanup.append(self.account_2)
|
|
|
|
vm_1 = None
|
|
vm_2 = None
|
|
try:
|
|
self.debug("Creating a tiny service offering for VM deployment")
|
|
self.service_offering = ServiceOffering.create(
|
|
self.apiclient,
|
|
self.services["service_offering"],
|
|
domainid=self.domain.id
|
|
)
|
|
self.cleanup.append(self.service_offering)
|
|
|
|
self.debug("Deploying virtual machine in account 1: %s" %
|
|
self.account_1.name)
|
|
self.vm_1 = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid=self.account_1.name,
|
|
domainid=self.account_1.domainid,
|
|
serviceofferingid=self.service_offering.id
|
|
)
|
|
self.cleanup.append(self.vm_1)
|
|
|
|
self.debug("Deploying virtual machine in account 2: %s" %
|
|
self.account_2.name)
|
|
self.vm_2 = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid=self.account_2.name,
|
|
domainid=self.account_2.domainid,
|
|
serviceofferingid=self.service_offering.id
|
|
)
|
|
self.cleanup.append(self.vm_2)
|
|
|
|
networks = Network.list(
|
|
self.apiclient,
|
|
account=self.account_1.name,
|
|
domainid=self.account_1.domainid,
|
|
listall=True
|
|
)
|
|
self.assertEqual(
|
|
isinstance(networks, list),
|
|
True,
|
|
"List networks should return a valid response"
|
|
)
|
|
network_1 = networks[0]
|
|
self.debug("Default network in account 1: %s is %s" % (
|
|
self.account_1.name,
|
|
network_1.name))
|
|
src_nat_list = PublicIPAddress.list(
|
|
self.apiclient,
|
|
associatednetworkid=network_1.id,
|
|
account=self.account_1.name,
|
|
domainid=self.account_1.domainid,
|
|
listall=True,
|
|
issourcenat=True,
|
|
)
|
|
self.assertEqual(
|
|
isinstance(src_nat_list, list),
|
|
True,
|
|
"List Public IP should return a valid source NAT"
|
|
)
|
|
self.assertNotEqual(
|
|
len(src_nat_list),
|
|
0,
|
|
"Length of response from listPublicIp should not be 0"
|
|
)
|
|
|
|
src_nat = src_nat_list[0]
|
|
|
|
self.debug(
|
|
"Trying to create a port forwarding rule in source NAT: %s" %
|
|
src_nat.ipaddress)
|
|
# Create NAT rule
|
|
self.nat_rule = NATRule.create(
|
|
self.apiclient,
|
|
self.vm_1,
|
|
self.services["natrule"],
|
|
ipaddressid=src_nat.id
|
|
)
|
|
self.cleanup.append(self.nat_rule)
|
|
self.debug("Created PF rule on source NAT: %s" % src_nat.ipaddress)
|
|
|
|
nat_rules = NATRule.list(self.apiclient, id=self.nat_rule.id)
|
|
|
|
self.assertEqual(
|
|
isinstance(nat_rules, list),
|
|
True,
|
|
"List NAT should return a valid port forwarding rules"
|
|
)
|
|
|
|
self.assertNotEqual(
|
|
len(nat_rules),
|
|
0,
|
|
"Length of response from listLbRules should not be 0"
|
|
)
|
|
except Exception as e:
|
|
self.fail(e)
|
|
|
|
self.debug("Deleting domain with force option")
|
|
try:
|
|
self.child_domain.delete(self.apiclient, cleanup=True)
|
|
self.cleanup.remove(self.child_domain)
|
|
self.cleanup.remove(self.account_1)
|
|
self.cleanup.remove(self.account_2)
|
|
self.cleanup.remove(self.vm_1)
|
|
self.cleanup.remove(self.vm_2)
|
|
self.cleanup.remove(self.nat_rule)
|
|
|
|
except Exception as e:
|
|
self.debug("Waiting for account.cleanup.interval" +
|
|
" to cleanup any remaining resources")
|
|
# Sleep 3*account.gc to ensure that all resources are deleted
|
|
wait_for_cleanup(self.apiclient, ["account.cleanup.interval"] * 3)
|
|
with self.assertRaises(CloudstackAPIException):
|
|
Domain.list(
|
|
self.apiclient,
|
|
id=self.child_domain.id,
|
|
listall=True
|
|
)
|
|
|
|
self.debug("Checking if the resources in domain are deleted")
|
|
with self.assertRaises(CloudstackAPIException):
|
|
Account.list(
|
|
self.apiclient,
|
|
name=self.account_1.name,
|
|
domainid=self.account_1.domainid,
|
|
listall=True
|
|
)
|
|
return
|
|
|
|
@attr(
|
|
tags=[
|
|
"domains",
|
|
"advanced",
|
|
"advancedns",
|
|
"simulator"],
|
|
required_hardware="false")
|
|
def test_DeleteDomain(self):
|
|
""" Test delete domain without force option"""
|
|
|
|
# Steps for validations
|
|
# 1. create a domain DOM
|
|
# 2. create 2 users under this domain
|
|
# 3. deploy 1 VM into each of these user accounts
|
|
# 4. create PF / FW rules for port 22 on these VMs for their
|
|
# respective accounts
|
|
# 5. delete the domain with force=false option
|
|
# Validate the following
|
|
# 1. listDomains should list the created domain
|
|
# 2. listAccounts should list the created accounts
|
|
# 3. listvirtualmachines should show the Running VMs
|
|
# 4. PF and FW rules should be shown in listFirewallRules
|
|
# 5. domain deletion should fail saying there are resources under use
|
|
|
|
self.debug("Creating a domain for login with API domain test")
|
|
self.child_domain = Domain.create(
|
|
self.apiclient,
|
|
self.services["domain"],
|
|
parentdomainid=self.domain.id
|
|
)
|
|
# in this test delete domain *should* fail so we need to housekeep:
|
|
self.cleanup.append(self.child_domain)
|
|
self.debug("Domain: %s is created successfully." % self.child_domain.name)
|
|
self.debug(
|
|
"Checking if the created domain is listed in list domains API")
|
|
domains = Domain.list(self.apiclient, id=self.child_domain.id, listall=True)
|
|
|
|
self.assertEqual(
|
|
isinstance(domains, list),
|
|
True,
|
|
"List domains shall return a valid response"
|
|
)
|
|
self.debug("Creating 2 user accounts in domain: %s" % self.child_domain.name)
|
|
self.account_1 = Account.create(
|
|
self.apiclient,
|
|
self.services["account"],
|
|
domainid=self.child_domain.id
|
|
)
|
|
self.cleanup.append(self.account_1)
|
|
|
|
self.account_2 = Account.create(
|
|
self.apiclient,
|
|
self.services["account"],
|
|
domainid=self.child_domain.id
|
|
)
|
|
self.cleanup.append(self.account_2)
|
|
|
|
self.debug("Creating a tiny service offering for VM deployment")
|
|
self.service_offering = ServiceOffering.create(
|
|
self.apiclient,
|
|
self.services["service_offering"],
|
|
domainid=self.domain.id
|
|
)
|
|
self.cleanup.append(self.service_offering)
|
|
|
|
self.debug("Deploying virtual machine in account 1: %s" %
|
|
self.account_1.name)
|
|
self.vm_1 = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid=self.account_1.name,
|
|
domainid=self.account_1.domainid,
|
|
serviceofferingid=self.service_offering.id
|
|
)
|
|
self.cleanup.append(self.vm_1)
|
|
|
|
self.debug("Deploying virtual machine in account 2: %s" %
|
|
self.account_2.name)
|
|
self.vm_2 = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid=self.account_2.name,
|
|
domainid=self.account_2.domainid,
|
|
serviceofferingid=self.service_offering.id
|
|
)
|
|
self.cleanup.append(self.vm_2)
|
|
|
|
networks = Network.list(
|
|
self.apiclient,
|
|
account=self.account_1.name,
|
|
domainid=self.account_1.domainid,
|
|
listall=True
|
|
)
|
|
self.assertEqual(
|
|
isinstance(networks, list),
|
|
True,
|
|
"List networks should return a valid response"
|
|
)
|
|
network_1 = networks[0]
|
|
self.debug("Default network in account 1: %s is %s" % (
|
|
self.account_1.name,
|
|
network_1.name))
|
|
src_nat_list = PublicIPAddress.list(
|
|
self.apiclient,
|
|
associatednetworkid=network_1.id,
|
|
account=self.account_1.name,
|
|
domainid=self.account_1.domainid,
|
|
listall=True,
|
|
issourcenat=True,
|
|
)
|
|
self.assertEqual(
|
|
isinstance(src_nat_list, list),
|
|
True,
|
|
"List Public IP should return a valid source NAT"
|
|
)
|
|
self.assertNotEqual(
|
|
len(src_nat_list),
|
|
0,
|
|
"Length of response from listPublicIp should not be 0"
|
|
)
|
|
|
|
src_nat = src_nat_list[0]
|
|
|
|
self.debug(
|
|
"Trying to create a port forwarding rule in source NAT: %s" %
|
|
src_nat.ipaddress)
|
|
# Create NAT rule
|
|
self.nat_rule = NATRule.create(
|
|
self.apiclient,
|
|
self.vm_1,
|
|
self.services["natrule"],
|
|
ipaddressid=src_nat.id
|
|
)
|
|
self.cleanup.append(self.nat_rule)
|
|
self.debug("Created PF rule on source NAT: %s" % src_nat.ipaddress)
|
|
|
|
nat_rules = NATRule.list(self.apiclient, id=self.nat_rule.id)
|
|
|
|
self.assertEqual(
|
|
isinstance(nat_rules, list),
|
|
True,
|
|
"List NAT should return a valid port forwarding rules"
|
|
)
|
|
|
|
self.assertNotEqual(
|
|
len(nat_rules),
|
|
0,
|
|
"Length of response from listLbRules should not be 0"
|
|
)
|
|
|
|
self.debug("Deleting domain without force option")
|
|
try:
|
|
self.child_domain.delete(self.apiclient, cleanup=False)
|
|
self.cleanup.remove(self.nat_rule)
|
|
self.cleanup.remove(self.vm_2)
|
|
self.cleanup.remove(self.vm_1)
|
|
self.cleanup.remove(self.account_2)
|
|
self.cleanup.remove(self.account_1)
|
|
self.cleanup.remove(self.child_domain)
|
|
self.fail("shouldn't be able to delete domain")
|
|
except:
|
|
pass
|
|
return
|
|
|
|
class TestMoveUser(cloudstackTestCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.testClient = super(TestMoveUser, cls).getClsTestClient()
|
|
cls.api_client = cls.testClient.getApiClient()
|
|
cls.testdata = cls.testClient.getParsedTestDataConfig()
|
|
|
|
cls.domain = get_domain(cls.api_client)
|
|
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
|
|
cls.testdata['mode'] = cls.zone.networktype
|
|
|
|
cls.template = get_test_template(
|
|
cls.api_client,
|
|
cls.zone.id,
|
|
cls.testdata["ostype"]
|
|
)
|
|
|
|
cls.testdata["virtual_machine"]["zoneid"] = cls.zone.id
|
|
cls._cleanup = []
|
|
return
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
super(TestMoveUser,cls).tearDownClass()
|
|
|
|
def setUp(self):
|
|
self.apiclient = self.testClient.getApiClient()
|
|
self.dbclient = self.testClient.getDbConnection()
|
|
self.cleanup = []
|
|
self.testdata = self.testClient.getParsedTestDataConfig()
|
|
self.account1 = Account.create(
|
|
self.apiclient,
|
|
self.testdata["acl"]["accountD1"],
|
|
domainid=self.domain.id
|
|
)
|
|
self.cleanup.append(self.account1)
|
|
|
|
self.account2 = Account.create(
|
|
self.apiclient,
|
|
self.testdata["acl"]["accountD1A"],
|
|
domainid=self.domain.id
|
|
)
|
|
self.cleanup.append(self.account2)
|
|
|
|
self.user = User.create(
|
|
self.apiclient,
|
|
self.testdata["user"],
|
|
account=self.account1.name,
|
|
domainid=self.account1.domainid
|
|
)
|
|
self.cleanup.append(self.user)
|
|
|
|
return
|
|
|
|
def tearDown(self):
|
|
super(TestMoveUser,self).tearDownClass()
|
|
|
|
@attr(tags=["domains", "advanced", "advancedns", "simulator","dvs"], required_hardware="false")
|
|
def test_move_user_to_accountID(self):
|
|
|
|
self.user.move(self.api_client, dest_accountid=self.account2.id)
|
|
|
|
self.assertEqual(
|
|
self.account2.name,
|
|
self.user.list(self.apiclient, id=self.user.id)[0].account,
|
|
"Check user source of created user"
|
|
)
|
|
return
|
|
|
|
@attr(tags=["domains", "advanced", "advancedns", "simulator","dvs"], required_hardware="false")
|
|
def test_move_user_to_account_name(self):
|
|
|
|
self.user.move(self.api_client, dest_account=self.account2.name)
|
|
|
|
self.assertEqual(
|
|
self.account2.name,
|
|
self.user.list(self.apiclient, id=self.user.id)[0].account,
|
|
"Check user source of created user"
|
|
)
|
|
return
|
|
|
|
@attr(tags=["domains", "advanced", "advancedns", "simulator","dvs"], required_hardware="false")
|
|
def test_move_user_to_different_domain(self):
|
|
domain2 = Domain.create(self.api_client,
|
|
self.testdata["domain"],
|
|
parentdomainid=self.domain.id
|
|
)
|
|
self.cleanup.append(domain2)
|
|
|
|
account_different_domain = Account.create(
|
|
self.apiclient,
|
|
self.testdata["acl"]["accountD1B"],
|
|
domainid=domain2.id
|
|
)
|
|
self.cleanup.append(account_different_domain)
|
|
try:
|
|
self.user.move(self.api_client, dest_account=account_different_domain.name)
|
|
except Exception:
|
|
pass
|
|
else:
|
|
self.fail("It should not be allowed to move users across accounts in different domains, failing")
|
|
|
|
return
|
|
|
|
@attr(tags=["domains", "advanced", "advancedns", "simulator","dvs"], required_hardware="false")
|
|
def test_move_user_incorrect_account_id(self):
|
|
|
|
try:
|
|
self.user.move(self.api_client, dest_accountid='incorrect-account-id')
|
|
except Exception:
|
|
pass
|
|
else:
|
|
self.fail("moving to non-existing account should not be possible, failing")
|
|
return
|
|
|
|
@attr(tags=["domains", "advanced", "advancedns", "simulator","dvs"], required_hardware="false")
|
|
def test_move_user_incorrect_account_name(self):
|
|
|
|
try:
|
|
self.user.move(self.api_client, dest_account='incorrect-account-name')
|
|
except Exception:
|
|
pass
|
|
else:
|
|
self.fail("moving to non-existing account should not be possible, failing")
|
|
return
|