apache/cloudstack
1
0
Fork 0
mirror of https://github.com/apache/cloudstack.git synced 2025-10-26 08:42:29 +01:00
Code Issues Packages Projects Releases Wiki Activity

test: Add e2e tests for listing resources (#8410)

Browse Source 
* Use dualzones for ci github actions

* Update advdualzone.cfg to be similar to advanced.cfg & fixup test_metrics_api.py

* Fixup e2e tests for running with multiple zones

* Add e2e tests for listing of accounts, disk_offerings, domains, hosts, service_offerings, storage_pools, volumes

* Fixup

* another fixup

* Add test for listing volumes with tags filter

* Add check for existing volumes in test_list_volumes

* Wait for volumes to be deleted on cleanup

* Filter out volumes in Destroy state before checking the count of volumes
This commit is contained in:
Vishesh 2024-02-07 12:55:55 +05:30 committed by GitHub
parent b0ac78705d
commit df412a99d2
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
14 changed files with 2929 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
.github/workflows/ci.yml vendored
View File

@ -194,8 +194,15 @@ jobs:
component/test_vpc_network
component/test_vpc_offerings
component/test_vpc_routers
component/test_vpn_users",
"component/test_vpc_network_lbrules" ]
component/test_vpn_users
component/test_vpc_network_lbrules",
"smoke/test_list_accounts
smoke/test_list_disk_offerings
smoke/test_list_domains
smoke/test_list_hosts
smoke/test_list_service_offerings
smoke/test_list_storage_pools
smoke/test_list_volumes"]
steps:
- uses: actions/checkout@v4
@ -278,7 +285,7 @@ jobs:
while ! nc -vzw 5 localhost 8096 2>&1 > /dev/null; do grep Exception /tmp/jetty-log; sleep 10; done
set -e
echo -e "\nStarting Advanced Zone DataCenter deployment"
python3 tools/marvin/marvin/deployDataCenter.py -i setup/dev/advanced.cfg 2>&1 || true
python3 tools/marvin/marvin/deployDataCenter.py -i setup/dev/advdualzone.cfg 2>&1 || true
- name: Run Integration Tests with Simulator
run: |
@ -291,7 +298,7 @@ jobs:
TESTS=($(echo $TESTS | tr -d '\n' | tr -s ' '))
for suite in "${TESTS[@]}" ; do
echo -e "Currently running test: $suite\n"
time nosetests-3.4 --with-xunit --xunit-file=integration-test-results/$suite.xml --with-marvin --marvin-config=setup/dev/advanced.cfg test/integration/$suite.py -s -a tags=advanced,required_hardware=false --zone=Sandbox-simulator --hypervisor=simulator || true ;
time nosetests-3.4 --with-xunit --xunit-file=integration-test-results/$suite.xml --with-marvin --marvin-config=setup/dev/advdualzone.cfg test/integration/$suite.py -s -a tags=advanced,required_hardware=false --zone=zim1 --hypervisor=simulator || true ;
done
echo -e "Stopping Simulator, integration tests run completed\n"

60
setup/dev/advdualzone.cfg
View File

@ -18,12 +18,12 @@
"zones": [
{
"name": "zim1",
"guestcidraddress": "10.100.1.0/24",
"dns1": "10.147.100.6",
"guestcidraddress": "10.1.1.0/24",
"dns1": "10.147.28.6",
"physical_networks": [
{
"broadcastdomainrange": "Zone",
"vlan": "1100-1200",
"vlan": "100-200",
"name": "z1-pnet",
"traffictypes": [
{
@ -63,19 +63,19 @@
},
"ipranges": [
{
"startip": "192.168.100.2",
"endip": "192.168.100.200",
"startip": "192.168.2.2",
"endip": "192.168.2.200",
"netmask": "255.255.255.0",
"vlan": "50",
"gateway": "192.168.100.1"
"gateway": "192.168.2.1"
}
],
"networktype": "Advanced",
"pods": [
{
"endip": "172.16.100.200",
"endip": "172.16.15.200",
"name": "Z1P1",
"startip": "172.16.100.2",
"startip": "172.16.15.2",
"netmask": "255.255.255.0",
"clusters": [
{
@ -96,11 +96,11 @@
"clustertype": "CloudManaged",
"primaryStorages": [
{
"url": "nfs://10.147.100.6:/export/home/sandbox/z1p1",
"url": "nfs://10.147.28.6:/export/home/sandbox/z1p1",
"name": "Z1PS1"
},
{
"url": "nfs://10.147.100.6:/export/home/sandbox/z1p2",
"url": "nfs://10.147.28.6:/export/home/sandbox/z1p2",
"name": "Z1PS2"
}
]
@ -123,35 +123,35 @@
"clustertype": "CloudManaged",
"primaryStorages": [
{
"url": "nfs://10.147.100.6:/export/home/sandbox/z1p3",
"url": "nfs://10.147.28.6:/export/home/sandbox/z1p3",
"name": "Z1PS3"
},
{
"url": "nfs://10.147.100.6:/export/home/sandbox/z1p4",
"url": "nfs://10.147.28.6:/export/home/sandbox/z1p4",
"name": "Z1PS4"
}
]
}
],
"gateway": "172.16.100.1"
"gateway": "172.16.15.1"
}
],
"internaldns1": "10.147.100.6",
"internaldns1": "10.147.28.6",
"secondaryStorages": [
{
"url": "nfs://10.147.100.6:/export/home/sandbox/z1secondary",
"url": "nfs://10.147.28.6:/export/home/sandbox/z1secondary",
"provider" : "NFS"
}
]
},
{
"name": "zim2",
"guestcidraddress": "10.200.1.0/24",
"dns1": "10.147.200.6",
"guestcidraddress": "10.1.2.0/24",
"dns1": "10.147.29.6",
"physical_networks": [
{
"broadcastdomainrange": "Zone",
"vlan": "2100-2200",
"vlan": "300-400",
"name": "z2-pnet",
"traffictypes": [
{
@ -191,19 +191,19 @@
},
"ipranges": [
{
"startip": "192.168.200.2",
"endip": "192.168.200.200",
"startip": "192.168.3.2",
"endip": "192.168.3.200",
"netmask": "255.255.255.0",
"vlan": "50",
"gateway": "192.168.200.1"
"vlan": "51",
"gateway": "192.168.3.1"
}
],
"networktype": "Advanced",
"pods": [
{
"endip": "172.16.200.200",
"endip": "172.16.16.200",
"name": "Z2P1",
"startip": "172.16.200.2",
"startip": "172.16.16.2",
"netmask": "255.255.255.0",
"clusters": [
{
@ -224,11 +224,11 @@
"clustertype": "CloudManaged",
"primaryStorages": [
{
"url": "nfs://10.147.200.6:/export/home/sandbox/z2p1",
"url": "nfs://10.147.29.6:/export/home/sandbox/z2p1",
"name": "Z2PS1"
},
{
"url": "nfs://10.147.200.6:/export/home/sandbox/z2p2",
"url": "nfs://10.147.29.6:/export/home/sandbox/z2p2",
"name": "Z2PS2"
}
]
@ -251,20 +251,20 @@
"clustertype": "CloudManaged",
"primaryStorages": [
{
"url": "nfs://10.147.200.6:/export/home/sandbox/z2p3",
"url": "nfs://10.147.29.6:/export/home/sandbox/z2p3",
"name": "Z2PS3"
},
{
"url": "nfs://10.147.200.6:/export/home/sandbox/z2p4",
"url": "nfs://10.147.29.6:/export/home/sandbox/z2p4",
"name": "Z2PS4"
}
]
}
],
"gateway": "172.16.200.1"
"gateway": "172.16.16.1"
}
],
"internaldns1": "10.147.200.6",
"internaldns1": "10.147.29.6",
"secondaryStorages": [
{
"url": "nfs://10.147.200.6:/export/home/sandbox/z2secondary",

2
test/integration/component/test_affinity_groups_projects.py
View File

@ -1065,7 +1065,7 @@ class TestDeployVMAffinityGroups(cloudstackTestCase):
"""
test DeployVM in anti-affinity groups with more vms than hosts.
"""
hosts = list_hosts(self.api_client, type="routing")
hosts = list_hosts(self.api_client, type="routing", zoneid=self.zone.id)
aff_grp = self.create_aff_grp(self.account_api_client)
vms = []
for host in hosts:

379
test/integration/smoke/test_list_accounts.py Normal file
View File

@ -0,0 +1,379 @@
# 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.
""" Tests for API listing of accounts with different filters
"""
# Import Local Modules
from marvin.cloudstackTestCase import cloudstackTestCase
from marvin.lib.base import (Account,
Domain)
from marvin.lib.common import (get_domain, list_accounts)
# Import System modules
from nose.plugins.attrib import attr
_multiprocess_shared_ = True
class TestListAccounts(cloudstackTestCase):
@classmethod
def setUpClass(cls):
testClient = super(TestListAccounts, cls).getClsTestClient()
cls.apiclient = testClient.getApiClient()
cls.services = testClient.getParsedTestDataConfig()
cls.domain = get_domain(cls.apiclient)
cls.account = list_accounts(cls.apiclient, name="admin")[0]
cls._cleanup = []
cls.accounts = list_accounts(cls.apiclient, listall=True)
cls.child_domain_1 = Domain.create(
cls.apiclient,
cls.services["domain"],
parentdomainid=cls.domain.id
)
cls._cleanup.append(cls.child_domain_1)
cls.services["account"]["username"] = "child_account_admin"
cls.child_account_admin = Account.create(
cls.apiclient,
cls.services["account"],
admin=True,
domainid=cls.child_domain_1.id
)
cls._cleanup.append(cls.child_account_admin)
cls.services["username"] = "child_account_user"
cls.child_account_user = Account.create(
cls.apiclient,
cls.services["account"],
admin=0,
domainid=cls.child_domain_1.id
)
cls.child_account_user.disable(cls.apiclient)
cls._cleanup.append(cls.child_account_user)
cls.child_domain_2 = Domain.create(
cls.apiclient,
cls.services["domain"],
parentdomainid=cls.domain.id
)
cls._cleanup.append(cls.child_domain_2)
@classmethod
def tearDownClass(cls):
super(TestListAccounts, cls).tearDownClass()
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_01_list_accounts_accounttype_filter(self):
"""Test listing accounts with accounttype filter
"""
list_account_response = Account.list(
self.apiclient,
accounttype=0,
domainid=self.child_domain_1.id
)
self.assertTrue(
isinstance(list_account_response, list),
"List Account response is not a valid list"
)
self.assertEqual(
len(list_account_response),
1,
"List Account response has incorrect length"
)
self.assertEqual(
list_account_response[0].name,
self.child_account_user.name,
"Check for list response return valid data"
)
self.assertEqual(
list_account_response[0].accounttype,
0,
"Check for list response return valid data"
)
list_account_response = Account.list(
self.apiclient,
accounttype=2,
domainid=self.child_domain_1.id
)
self.assertTrue(
isinstance(list_account_response, list),
"List Account response is not a valid list"
)
self.assertEqual(
len(list_account_response),
1,
"List Account response has incorrect length"
)
self.assertEqual(
list_account_response[0].name,
self.child_account_admin.name,
"Check for list response return valid data"
)
self.assertEqual(
list_account_response[0].accounttype,
2,
"Check for list response return valid data"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_02_list_accounts_domainid_filter(self):
"""Test listing accounts with domainid filter
"""
list_account_response = Account.list(
self.apiclient,
domainid=self.child_domain_1.id
)
self.assertTrue(
isinstance(list_account_response, list),
"List Account response is not a valid list"
)
self.assertEqual(
len(list_account_response),
2,
"List Account response has incorrect length"
)
self.assertEqual(
self.child_domain_1.id,
list_account_response[0].domainid,
"Check for list response return valid data"
)
self.assertEqual(
self.child_domain_1.id,
list_account_response[1].domainid,
"Check for list response return valid data"
)
list_account_response = Account.list(
self.apiclient,
domainid=self.child_domain_2.id
)
self.assertIsNone(list_account_response, "Check for list response return valid data")
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_03_list_accounts_id_filter(self):
"""Test listing accounts with id filter
"""
list_account_response = Account.list(
self.apiclient,
id=self.child_account_user.id
)
self.assertTrue(
isinstance(list_account_response, list),
"List Account response is not a valid list"
)
self.assertEqual(
len(list_account_response),
1,
"List Account response has incorrect length"
)
self.assertEqual(
list_account_response[0].name,
self.child_account_user.name,
"Expected account name and actual account name should be same"
)
list_account_response = Account.list(
self.apiclient,
id=self.child_account_admin.id
)
self.assertTrue(
isinstance(list_account_response, list),
"List Account response is not a valid list"
)
self.assertEqual(
len(list_account_response),
1,
"List Account response has incorrect length"
)
self.assertEqual(
list_account_response[0].name,
self.child_account_admin.name,
"Expected account name and actual account name should be same"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_04_list_accounts_name_filter(self):
"""Test listing accounts with name filter
"""
list_account_response = Account.list(
self.apiclient,
name=self.child_account_user.name,
domainid=self.child_domain_1.id
)
self.assertTrue(
isinstance(list_account_response, list),
"List Account response is not a valid list"
)
self.assertEqual(
len(list_account_response),
1,
"List Account response has incorrect length"
)
self.assertEqual(
list_account_response[0].name,
self.child_account_user.name,
"Expected account name and actual account name should be same"
)
list_account_response = Account.list(
self.apiclient,
name=self.child_account_admin.name,
domainid=self.child_domain_1.id
)
self.assertTrue(
isinstance(list_account_response, list),
"List Account response is not a valid list"
)
self.assertEqual(
len(list_account_response),
1,
"List Account response has incorrect length"
)
self.assertEqual(
list_account_response[0].name,
self.child_account_admin.name,
"Expected account name and actual account name should be same"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_05_list_accounts_state_filter(self):
"""Test listing accounts with state filter
"""
list_account_response = Account.list(
self.apiclient,
state="enabled",
domainid=self.child_domain_1.id
)
self.assertTrue(
isinstance(list_account_response, list),
"List Account response is not a valid list"
)
self.assertEqual(
len(list_account_response),
1,
"List Account response has incorrect length"
)
self.assertEqual(
list_account_response[0].name,
self.child_account_admin.name,
"Expected account name and actual account name should be same"
)
list_account_response = Account.list(
self.apiclient,
state="disabled",
domainid=self.child_domain_1.id
)
self.assertTrue(
isinstance(list_account_response, list),
"List Account response is not a valid list"
)
self.assertEqual(
len(list_account_response),
1,
"List Account response has incorrect length"
)
self.assertEqual(
list_account_response[0].name,
self.child_account_user.name,
"Expected account name and actual account name should be same"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_06_list_accounts_keyword_filter(self):
"""Test listing accounts with keyword filter
"""
list_account_response = Account.list(
self.apiclient,
keyword=self.child_account_user.name,
domainid=self.child_domain_1.id
)
self.assertTrue(
isinstance(list_account_response, list),
"List Account response is not a valid list"
)
self.assertEqual(
self.child_account_user.name,
list_account_response[0].name,
"Expected account name and actual account name should be same"
)
list_account_response = Account.list(
self.apiclient,
keyword=self.child_account_admin.name,
domainid=self.child_domain_1.id
)
self.assertTrue(
isinstance(list_account_response, list),
"List Account response is not a valid list"
)
self.assertEqual(
self.child_account_admin.name,
list_account_response[0].name,
"Expected account name and actual account name should be same"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_07_list_accounts_with_listall_filters(self):
"""Test listing accounts with listall filters
"""
list_account_response = Account.list(
self.apiclient,
listall=False
)
self.assertTrue(
isinstance(list_account_response, list),
"List Account response is not a valid list"
)
self.assertEqual(
1,
len(list_account_response),
"List Account response has incorrect length"
)
list_account_response = Account.list(
self.apiclient,
listall=True
)
self.assertTrue(
isinstance(list_account_response, list),
"List Account response is not a valid list"
)
self.assertEqual(
2,
len(list_account_response) - len(self.accounts),
"List Account response has incorrect length"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_08_list_accounts_with_no_filters(self):
"""Test listing accounts with no filters
"""
list_account_response = Account.list(
self.apiclient
)
self.assertTrue(
isinstance(list_account_response, list),
"List Account response is not a valid list"
)
self.assertEqual(
1,
len(list_account_response),
"List Account response has incorrect length"
)

319
test/integration/smoke/test_list_disk_offerings.py Normal file
View File

@ -0,0 +1,319 @@
# 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.
""" Tests for API listing of disk offerings with different filters
"""
# Import Local Modules
from marvin.cloudstackTestCase import cloudstackTestCase
from marvin.codes import FAILED
from marvin.lib.base import (Account,
Domain,
Volume,
ServiceOffering,
DiskOffering,
VirtualMachine)
from marvin.lib.common import (get_domain, list_accounts,
list_zones, list_clusters, list_hosts)
# Import System modules
from nose.plugins.attrib import attr
_multiprocess_shared_ = True
class TestListDiskOfferings(cloudstackTestCase):
@classmethod
def setUpClass(cls):
testClient = super(TestListDiskOfferings, cls).getClsTestClient()
cls.apiclient = testClient.getApiClient()
cls.services = testClient.getParsedTestDataConfig()
cls.hypervisor = testClient.getHypervisorInfo()
cls.domain = get_domain(cls.apiclient)
cls.zones = list_zones(cls.apiclient)
cls.zone = cls.zones[0]
cls.clusters = list_clusters(cls.apiclient)
cls.cluster = cls.clusters[0]
cls.hosts = list_hosts(cls.apiclient)
cls.account = list_accounts(cls.apiclient, name="admin")[0]
cls._cleanup = []
cls.disk_offerings = DiskOffering.list(cls.apiclient, listall=True)
cls.disk_offering = DiskOffering.create(cls.apiclient,
cls.services["disk_offering"],
domainid=cls.domain.id)
cls._cleanup.append(cls.disk_offering)
cls.child_domain_1 = Domain.create(
cls.apiclient,
cls.services["domain"],
parentdomainid=cls.domain.id
)
cls._cleanup.append(cls.child_domain_1)
cls.account_1 = Account.create(
cls.apiclient,
cls.services["account"],
admin=True,
domainid=cls.domain.id
)
cls._cleanup.append(cls.account_1)
cls.domainadmin_api_client = testClient.getUserApiClient(
UserName=cls.account_1.user[0].username,
DomainName=cls.domain.name,
type=2
)
cls.disk_offering_child_domain = DiskOffering.create(cls.apiclient,
cls.services["disk_offering"],
domainid=cls.child_domain_1.id,
zoneid=cls.zone.id,
encrypt=True)
cls._cleanup.append(cls.disk_offering_child_domain)
@classmethod
def tearDownClass(cls):
super(TestListDiskOfferings, cls).tearDownClass()
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_01_list_disk_offerings_id_filter(self):
""" Test list disk offerings with id filter
"""
# List all disk offerings
disk_offerings = DiskOffering.list(self.apiclient, id=self.disk_offering.id)
self.assertTrue(
isinstance(disk_offerings, list),
"List disk offerings response is not a valid list"
)
self.assertEqual(
len(disk_offerings),
1,
"List disk offerings response has incorrect length"
)
# Verify the id of the disk offering returned is the same as the one requested
self.assertEqual(
disk_offerings[0].id,
self.disk_offering.id,
"List disk offerings should return the disk offering requested"
)
disk_offerings = DiskOffering.list(self.apiclient, id=-1)
self.assertIsNone(disk_offerings, "List disk offerings response is not None")
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_02_list_disk_offerings_name_filter(self):
""" Test list disk offerings with name filter
"""
disk_offerings = DiskOffering.list(self.apiclient, name=self.services["disk_offering"]["name"])
self.assertTrue(
isinstance(disk_offerings, list),
"List disk offerings response is not a valid list"
)
self.assertEqual(
len(disk_offerings),
2,
"List disk offerings response has incorrect length"
)
# Verify the name of the disk offering returned is the same as the one requested
self.assertEqual(
disk_offerings[0].name,
self.services["disk_offering"]["name"],
"List disk offerings should return the disk offering requested"
)
self.assertEqual(
disk_offerings[1].name,
self.services["disk_offering"]["name"],
"List disk offerings should return the disk offering requested"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_03_list_disk_offerings_zoneid_filter(self):
""" Test list disk offerings with zoneid filter
"""
disk_offerings_zone_1 = DiskOffering.list(self.apiclient, zoneid=self.zone.id)
self.assertTrue(
isinstance(disk_offerings_zone_1, list),
"List disk offerings response is not a valid list"
)
self.assertEqual(
len(disk_offerings_zone_1) - len(self.disk_offerings),
2,
"List disk offerings response has incorrect length"
)
for disk_offering in disk_offerings_zone_1:
self.assertTrue(
disk_offering.zoneid is None or disk_offering.zoneid == self.zone.id,
"List disk offerings should return the disk offering requested"
)
if len(self.zones) > 1:
disk_offerings_zone_2 = DiskOffering.list(self.apiclient, zoneid=self.zones[1].id)
self.assertTrue(
isinstance(disk_offerings_zone_2, list),
"List disk offerings response is not a valid list"
)
for disk_offering in disk_offerings_zone_2:
self.assertTrue(
disk_offering.zoneid is None or disk_offering.zoneid == self.zones[1].id,
"List disk offerings should return the disk offering requested"
)
self.assertEqual(len(disk_offerings_zone_1) - len(disk_offerings_zone_2), 1)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_04_list_disk_offerings_domainid_filter(self):
""" Test list disk offerings with domainid filter
"""
disk_offerings = DiskOffering.list(self.apiclient, domainid=self.domain.id)
self.assertTrue(
isinstance(disk_offerings, list),
"List disk offerings response is not a valid list"
)
self.assertEqual(
len(disk_offerings),
1,
"List disk offerings response has incorrect length"
)
self.assertEqual(
disk_offerings[0].domainid,
self.domain.id,
"List disk offerings should return the disk offering requested"
)
disk_offerings = DiskOffering.list(self.apiclient, domainid=self.child_domain_1.id)
self.assertTrue(
isinstance(disk_offerings, list),
"List disk offerings response is not a valid list"
)
self.assertEqual(
len(disk_offerings),
1,
"List disk offerings response has incorrect length"
)
self.assertEqual(
disk_offerings[0].domainid,
self.child_domain_1.id,
"List disk offerings should return the disk offering requested"
)
disk_offerings = DiskOffering.list(self.apiclient, domainid=-1)
self.assertIsNone(disk_offerings, "List disk offerings response is not None")
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_05_list_disk_offerings_encrypted_filter(self):
""" Test list disk offerings with encrypted filter
"""
disk_offerings = DiskOffering.list(self.apiclient, encrypt=True)
self.assertTrue(
isinstance(disk_offerings, list),
"List disk offerings response is not a valid list"
)
self.assertEqual(
len(disk_offerings),
1,
"List disk offerings response has incorrect length"
)
self.assertTrue(
disk_offerings[0].encrypt,
"List disk offerings should return the disk offering requested"
)
disk_offerings = DiskOffering.list(self.apiclient, encrypt=False)
self.assertTrue(
isinstance(disk_offerings, list),
"List disk offerings response is not a valid list"
)
self.assertEqual(
len(disk_offerings) - len(self.disk_offerings),
1,
"List disk offerings response has incorrect length"
)
for disk_offering in disk_offerings:
self.assertFalse(
disk_offering.encrypt,
"List disk offerings should return the disk offering requested"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_06_list_disk_offerings_keyword_filter(self):
""" Test list disk offerings with keyword filter
"""
disk_offerings = DiskOffering.list(self.apiclient, keyword=self.disk_offering.name)
self.assertTrue(
isinstance(disk_offerings, list),
"List disk offerings response is not a valid list"
)
self.assertEqual(
len(disk_offerings),
2,
"List disk offerings response has incorrect length"
)
self.assertEqual(
disk_offerings[0].name,
self.disk_offering.name,
"List disk offerings should return the disk offering requested"
)
self.assertEqual(
disk_offerings[1].name,
self.disk_offering.name,
"List disk offerings should return the disk offering requested"
)
disk_offerings = DiskOffering.list(self.apiclient, keyword="random")
self.assertIsNone(disk_offerings, "List disk offerings response is not None")
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_07_list_disk_offering_isrecursive_filter(self):
""" Test list disk offerings with isrecursive parameter
"""
disk_offerings = DiskOffering.list(self.domainadmin_api_client, isrecursive=True)
self.assertTrue(
isinstance(disk_offerings, list),
"List disk offerings response is not a valid list"
)
self.assertEqual(
len(disk_offerings) - len(self.disk_offerings),
2,
"List disk offerings response has incorrect length"
)
disk_offerings = DiskOffering.list(self.domainadmin_api_client, isrecursive=False)
self.assertTrue(
isinstance(disk_offerings, list),
"List disk offerings response is not a valid list"
)
self.assertEqual(
len(disk_offerings) - len(self.disk_offerings),
1,
"List disk offerings response has incorrect length"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_08_list_disk_offering_no_filter(self):
""" Test list disk offerings with no filters
"""
disk_offerings = DiskOffering.list(self.apiclient)
self.assertTrue(
isinstance(disk_offerings, list),
"List disk offerings response is not a valid list"
)
self.assertEqual(
len(disk_offerings) - len(self.disk_offerings),
2,
"List disk offerings response has incorrect length"
)

216
test/integration/smoke/test_list_domains.py Normal file
View File

@ -0,0 +1,216 @@
# 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.
""" Tests for API listing of domains with different filters
"""
# Import Local Modules
from marvin.cloudstackTestCase import cloudstackTestCase
from marvin.lib.base import (Account,
Domain)
from marvin.lib.common import (get_domain, list_accounts)
# Import System modules
from nose.plugins.attrib import attr
_multiprocess_shared_ = True
class TestListDomains(cloudstackTestCase):
@classmethod
def setUpClass(cls):
testClient = super(TestListDomains, cls).getClsTestClient()
cls.apiclient = testClient.getApiClient()
cls.services = testClient.getParsedTestDataConfig()
cls.domain = get_domain(cls.apiclient)
cls.account = list_accounts(cls.apiclient, name="admin")[0]
cls._cleanup = []
cls.child_domain_1 = Domain.create(
cls.apiclient,
cls.services["domain"],
parentdomainid=cls.domain.id
)
cls._cleanup.append(cls.child_domain_1)
cls.child_account_1 = Account.create(
cls.apiclient,
cls.services["account"],
admin=True,
domainid=cls.child_domain_1.id
)
cls._cleanup.append(cls.child_account_1)
cls.child_account_apiclient = testClient.getUserApiClient(cls.child_account_1.user[0]['username'], cls.child_domain_1.name, type=2)
cls.child_domain_2 = Domain.create(
cls.apiclient,
cls.services["domain"],
parentdomainid=cls.child_domain_1.id
)
cls._cleanup.append(cls.child_domain_2)
@classmethod
def tearDownClass(cls):
super(TestListDomains, cls).tearDownClass()
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_01_list_domains_id_filter(self):
""" Test list domains with id filter
"""
# List all domains
domains = Domain.list(self.apiclient, id=self.domain.id)
self.assertEqual(
isinstance(domains, list),
True,
"List Domain response is not a valid list"
)
self.assertEqual(
len(domains),
1,
"List Domain response has incorrect length"
)
self.assertEqual(
domains[0].id,
self.domain.id,
"Check if list domains returns valid domain"
)
# List all domains with a non-existent id
with self.assertRaises(Exception):
Domain.list(self.apiclient, id=-1)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_02_list_domains_name_filter(self):
""" Test list domains with name filter
"""
# List all domains
domains = Domain.list(self.apiclient, name=self.domain.name)
self.assertEqual(
isinstance(domains, list),
True,
"List Domain response is not a valid list"
)
self.assertEqual(
len(domains),
1,
"List Domain response has incorrect length"
)
self.assertEqual(
domains[0].name,
self.domain.name,
"Check if list domains returns valid domain"
)
domains = Domain.list(self.apiclient, name="non-existent-domain")
self.assertIsNone(domains, "List Domain response is not None")
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_03_list_domains_listall_filter(self):
""" Test list domains with listall parameter
"""
# List all domains
domains = Domain.list(self.child_account_apiclient, listall=True)
self.assertEqual(
isinstance(domains, list),
True,
"List Domain response is not a valid list"
)
self.assertEqual(
len(domains),
2,
"List Domain response has incorrect length"
)
domains = Domain.list(self.child_account_apiclient, listall=False)
self.assertEqual(
isinstance(domains, list),
True,
"List Domain response is not a valid list"
)
self.assertEqual(
len(domains),
1,
"List Domain response has incorrect length"
)
self.assertEqual(
domains[0].id,
self.child_domain_1.id,
"Check if list domains returns valid domain"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_04_list_domains_level_filter(self):
""" Test list domains with level filter
"""
# List all domains
domains = Domain.list(self.apiclient, level=0)
self.assertEqual(
isinstance(domains, list),
True,
"List Domain response is not a valid list"
)
self.assertEqual(
len(domains),
1,
"List Domain response has incorrect length"
)
self.assertEqual(
domains[0].id,
self.domain.id,
"Check if list domains returns valid domain"
)
domains = Domain.list(self.apiclient, level=1)
self.assertEqual(
isinstance(domains, list),
True,
"List Domain response is not a valid list"
)
self.assertEqual(
len(domains),
1,
"List Domain response has incorrect length"
)
domains = Domain.list(self.apiclient, level=2)
self.assertEqual(
isinstance(domains, list),
True,
"List Domain response is not a valid list"
)
self.assertEqual(
len(domains),
1,
"List Domain response has incorrect length"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_05_list_domains_no_filter(self):
""" Test list domains with no filter
"""
# List all domains
domains = Domain.list(self.apiclient)
self.assertEqual(
isinstance(domains, list),
True,
"List Domain response is not a valid list"
)
self.assertEqual(
len(domains),
3,
"List Domain response has incorrect length"
)

372
test/integration/smoke/test_list_hosts.py Normal file
View File

@ -0,0 +1,372 @@
# 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.
""" Tests for API listing of hosts with different filters
"""
# Import Local Modules
from marvin.cloudstackTestCase import cloudstackTestCase
from marvin.codes import FAILED
from marvin.lib.base import (Configurations, Host)
from marvin.lib.common import (get_domain, list_accounts,
list_zones, list_clusters)
# Import System modules
from nose.plugins.attrib import attr
_multiprocess_shared_ = True
class TestListHosts(cloudstackTestCase):
@classmethod
def setUpClass(cls):
testClient = super(TestListHosts, cls).getClsTestClient()
cls.apiclient = testClient.getApiClient()
cls.services = testClient.getParsedTestDataConfig()
cls.hypervisor = testClient.getHypervisorInfo()
cls.zones = list_zones(cls.apiclient)
cls.zone = cls.zones[0]
cls.clusters = list_clusters(cls.apiclient)
cls.cluster = cls.clusters[0]
cls.hosts = Host.list(cls.apiclient)
@classmethod
def tearDownClass(cls):
super(TestListHosts, cls).tearDownClass()
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_01_list_hosts_no_filter(self):
"""Test list hosts with no filter"""
hosts = Host.list(self.apiclient)
self.assertTrue(
isinstance(hosts, list),
"Host response type should be a valid list"
)
self.assertGreater(
len(hosts),
0,
"Length of host response should greater than 0"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_02_list_hosts_clusterid_filter(self):
"""Test list hosts with clusterid filter"""
hosts = Host.list(self.apiclient, clusterid=self.cluster.id)
self.assertTrue(
isinstance(hosts, list),
"Host response type should be a valid list"
)
self.assertGreater(
len(hosts),
0,
"Length of host response should greater than 0"
)
for host in hosts:
self.assertEqual(
host.clusterid,
self.cluster.id,
"Host should be in the cluster %s" % self.cluster.id
)
with self.assertRaises(Exception):
hosts = Host.list(self.apiclient, clusterid="invalidclusterid")
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_03_list_hosts_hahost_filter(self):
"""Test list hosts with hahost filter"""
configs = Configurations.list(
self.apiclient,
name='ha.tag'
)
if isinstance(configs, list) and configs[0].value != "" and configs[0].value is not None:
hosts = Host.list(self.apiclient, hahost=True)
if hosts is not None:
self.assertTrue(
isinstance(hosts, list),
"Host response type should be a valid list"
)
self.assertGreater(
len(hosts),
0,
"Length of host response should greater than 0"
)
for host in hosts:
self.assertEqual(
host.hahost,
True,
"Host should be a HA host"
)
hosts = Host.list(self.apiclient, hahost=False)
if hosts is not None:
self.assertTrue(
isinstance(hosts, list),
"Host response type should be a valid list"
)
self.assertGreater(
len(hosts),
0,
"Length of host response should greater than 0"
)
for host in hosts:
self.assertTrue(
host.hahost is None or host.hahost is False,
"Host should not be a HA host"
)
else:
self.debug("HA is not enabled in the setup")
hosts = Host.list(self.apiclient, hahost="invalidvalue")
self.assertTrue(
isinstance(hosts, list),
"Host response type should be a valid list"
)
self.assertGreater(
len(hosts),
0,
"Length of host response should greater than 0"
)
self.assertEqual(
len(hosts),
len(self.hosts),
"Length of host response should be equal to the length of hosts"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_04_list_hosts_hypervisor_filter(self):
"""Test list hosts with hypervisor filter"""
hosts = Host.list(self.apiclient, hypervisor=self.hypervisor)
self.assertTrue(
isinstance(hosts, list),
"Host response type should be a valid list"
)
self.assertGreater(
len(hosts),
0,
"Length of host response should greater than 0"
)
for host in hosts:
self.assertEqual(
host.hypervisor.lower(),
self.hypervisor.lower(),
"Host should be a %s hypervisor" % self.hypervisor
)
hosts = Host.list(self.apiclient, hypervisor="invalidhypervisor")
self.assertTrue(
isinstance(hosts, list),
"Host response type should be a valid list"
)
self.assertEqual(
len(hosts),
len(self.hosts),
"Length of host response should be equal to the length of hosts"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_05_list_hosts_id_filter(self):
"""Test list hosts with id filter"""
hosts = Host.list(self.apiclient, id=self.hosts[0].id)
self.assertTrue(
isinstance(hosts, list),
"Host response type should be a valid list"
)
self.assertEqual(
len(hosts),
1,
"Length of host response should be 1"
)
self.assertEqual(
hosts[0].id,
self.hosts[0].id,
"Host id should match with the host id in the list"
)
with self.assertRaises(Exception):
hosts = Host.list(self.apiclient, id="invalidid")
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_06_list_hosts_keyword_filter(self):
"""Test list hosts with keyword filter"""
hosts = Host.list(self.apiclient, keyword=self.hosts[0].name)
self.assertTrue(
isinstance(hosts, list),
"Host response type should be a valid list"
)
self.assertGreater(
len(hosts),
0,
"Length of host response should be greater than 0"
)
for host in hosts:
self.assertIn(
host.name,
self.hosts[0].name,
"Host name should match with the host name in the list"
)
hosts = Host.list(self.apiclient, keyword="invalidkeyword")
self.assertIsNone(
hosts,
"Host response should be None"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_07_list_hosts_name_filter(self):
"""Test list hosts with name filter"""
hosts = Host.list(self.apiclient, name=self.hosts[0].name)
self.assertTrue(
isinstance(hosts, list),
"Host response type should be a valid list"
)
self.assertGreater(
len(hosts),
0,
"Length of host response should be greater than 0"
)
for host in hosts:
self.assertIn(
host.name,
self.hosts[0].name,
"Host name should match with the host name in the list"
)
hosts = Host.list(self.apiclient, name="invalidname")
self.assertIsNone(
hosts,
"Host response should be None"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_08_list_hosts_podid_filter(self):
"""Test list hosts with podid filter"""
hosts = Host.list(self.apiclient, podid=self.hosts[0].podid)
self.assertTrue(
isinstance(hosts, list),
"Host response type should be a valid list"
)
self.assertGreater(
len(hosts),
0,
"Length of host response should be greater than 0"
)
for host in hosts:
self.assertEqual(
host.podid,
self.hosts[0].podid,
"Host podid should match with the host podid in the list"
)
with self.assertRaises(Exception):
hosts = Host.list(self.apiclient, podid="invalidpodid")
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_09_list_hosts_resourcestate_filter(self):
"""Test list hosts with resourcestate filter"""
hosts = Host.list(self.apiclient, resourcestate=self.hosts[0].resourcestate)
self.assertTrue(
isinstance(hosts, list),
"Host response type should be a valid list"
)
self.assertGreater(
len(hosts),
0,
"Length of host response should be greater than 0"
)
for host in hosts:
self.assertEqual(
host.resourcestate,
self.hosts[0].resourcestate,
"Host resourcestate should match with the host resourcestate in the list"
)
hosts = Host.list(self.apiclient, resourcestate="invalidresourcestate")
self.assertIsNone(
hosts,
"Host response should be None"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_10_list_hosts_state_filter(self):
"""Test list hosts with state filter"""
hosts = Host.list(self.apiclient, state=self.hosts[0].state)
self.assertTrue(
isinstance(hosts, list),
"Host response type should be a valid list"
)
self.assertGreater(
len(hosts),
0,
"Length of host response should be greater than 0"
)
for host in hosts:
self.assertEqual(
host.state,
self.hosts[0].state,
"Host state should match with the host state in the list"
)
hosts = Host.list(self.apiclient, state="invalidstate")
self.assertIsNone(
hosts,
"Host response should be None"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_11_list_hosts_type_filter(self):
"""Test list hosts with type filter"""
hosts = Host.list(self.apiclient, type=self.hosts[0].type)
self.assertTrue(
isinstance(hosts, list),
"Host response type should be a valid list"
)
self.assertGreater(
len(hosts),
0,
"Length of host response should be greater than 0"
)
for host in hosts:
self.assertEqual(
host.type,
self.hosts[0].type,
"Host type should match with the host type in the list"
)
hosts = Host.list(self.apiclient, type="invalidtype")
self.assertIsNone(
hosts,
"Host response should be None"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_12_list_hosts_zoneid_filter(self):
"""Test list hosts with zoneid filter"""
hosts = Host.list(self.apiclient, zoneid=self.zone.id)
self.assertTrue(
isinstance(hosts, list),
"Host response type should be a valid list"
)
self.assertGreater(
len(hosts),
0,
"Length of host response should be greater than 0"
)
for host in hosts:
self.assertEqual(
host.zoneid,
self.zone.id,
"Host zoneid should match with the host zoneid in the list"
)
with self.assertRaises(Exception):
hosts = Host.list(self.apiclient, zoneid="invalidzoneid")

559
test/integration/smoke/test_list_service_offerings.py Normal file
View File

@ -0,0 +1,559 @@
# 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.
""" Tests for API listing of service offerings with different filters
"""
# Import Local Modules
from marvin.cloudstackTestCase import cloudstackTestCase
from marvin.codes import FAILED
from marvin.lib.base import (Account,
Domain,
Volume,
ServiceOffering,
DiskOffering,
VirtualMachine)
from marvin.lib.common import (get_domain, list_accounts,
list_zones, list_clusters, list_hosts)
# Import System modules
from nose.plugins.attrib import attr
_multiprocess_shared_ = True
class TestListServiceOfferings(cloudstackTestCase):
@classmethod
def setUpClass(cls):
testClient = super(TestListServiceOfferings, cls).getClsTestClient()
cls.apiclient = testClient.getApiClient()
cls.services = testClient.getParsedTestDataConfig()
cls.hypervisor = testClient.getHypervisorInfo()
cls.domain = get_domain(cls.apiclient)
cls.zones = list_zones(cls.apiclient)
cls.zone = cls.zones[0]
cls.clusters = list_clusters(cls.apiclient)
cls.cluster = cls.clusters[0]
cls.hosts = list_hosts(cls.apiclient)
cls.account = list_accounts(cls.apiclient, name="admin")[0]
cls._cleanup = []
cls.service_offerings = ServiceOffering.list(cls.apiclient)
cls.system_service_offerings = ServiceOffering.list(cls.apiclient, issystem=True)
cls.child_domain_1 = Domain.create(
cls.apiclient,
cls.services["domain"],
parentdomainid=cls.domain.id
)
cls._cleanup.append(cls.child_domain_1)
cls.account_1 = Account.create(
cls.apiclient,
cls.services["account"],
admin=True,
domainid=cls.domain.id
)
cls._cleanup.append(cls.account_1)
cls.domainadmin_api_client = testClient.getUserApiClient(
UserName=cls.account_1.user[0].username,
DomainName=cls.domain.name,
type=2
)
cls.system_offering = ServiceOffering.create(
cls.apiclient,
cls.services["service_offerings"]["tiny"],
issystem=True,
name="custom_system_offering",
systemvmtype="domainrouter"
)
cls._cleanup.append(cls.system_offering)
cls.service_offering_1 = ServiceOffering.create(
cls.apiclient,
cls.services["service_offerings"]["small"],
cpunumber=2,
cpuspeed=2000,
domainid=cls.child_domain_1.id,
encryptroot=True,
name="custom_offering_1",
zoneid=cls.zone.id
)
cls._cleanup.append(cls.service_offering_1)
@classmethod
def tearDownClass(cls):
super(TestListServiceOfferings, cls).tearDownClass()
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_01_list_service_offerings_cpunumber_filter(self):
"""Test list service offerings with cpunumber filter
"""
# List all service offerings with cpunumber 1
service_offerings = ServiceOffering.list(
self.apiclient,
cpunumber=1
)
self.assertTrue(
isinstance(service_offerings, list),
"List ServiceOfferings response is not a valid list"
)
self.assertEqual(
len(service_offerings) - len(self.service_offerings),
1,
"List ServiceOfferings response is empty"
)
for service_offering in service_offerings:
self.assertGreaterEqual(
service_offering.cpunumber,
1,
"List ServiceOfferings response has incorrect cpunumber"
)
service_offerings = ServiceOffering.list(
self.apiclient,
cpunumber=99999
)
self.assertIsNone(service_offerings, "List ServiceOfferings response is not None")
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_02_list_service_offerings_cpuspeed_filter(self):
"""Test list service offerings with cpuspeed filter
"""
# List all service offerings with cpuspeed 1000
service_offerings = ServiceOffering.list(
self.apiclient,
cpuspeed=1000
)
self.assertTrue(
isinstance(service_offerings, list),
"List ServiceOfferings response is not a valid list"
)
self.assertGreaterEqual(
len(service_offerings),
1,
"List ServiceOfferings response is empty"
)
for service_offering in service_offerings:
self.assertGreaterEqual(
service_offering.cpuspeed,
1000,
"List ServiceOfferings response has incorrect cpuspeed"
)
service_offerings = ServiceOffering.list(
self.apiclient,
cpuspeed=99999
)
self.assertIsNone(service_offerings, "List ServiceOfferings response is not None")
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_03_list_service_offerings_memory_filter(self):
"""Test list service offerings with memory filter
"""
# List all service offerings with memory 256
service_offerings = ServiceOffering.list(
self.apiclient,
memory=256
)
self.assertTrue(
isinstance(service_offerings, list),
"List ServiceOfferings response is not a valid list"
)
self.assertGreaterEqual(
len(service_offerings),
1,
"List ServiceOfferings response is empty"
)
for service_offering in service_offerings:
self.assertGreaterEqual(
service_offering.memory,
256,
"List ServiceOfferings response has incorrect memory"
)
service_offerings = ServiceOffering.list(
self.apiclient,
memory=99999
)
self.assertIsNone(service_offerings, "List ServiceOfferings response is not None")
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_04_list_service_offerings_domainid_filter(self):
"""Test list service offerings with domainid filter
"""
# List all service offerings with domainid
service_offerings = ServiceOffering.list(
self.apiclient,
domainid=self.domain.id
)
self.assertIsNone(
service_offerings,
"List ServiceOfferings response is not None"
)
service_offerings = ServiceOffering.list(
self.apiclient,
domainid=self.child_domain_1.id
)
self.assertTrue(
isinstance(service_offerings, list),
"List ServiceOfferings response is not a valid list"
)
self.assertEqual(
len(service_offerings),
1,
"List ServiceOfferings response is empty"
)
for service_offering in service_offerings:
self.assertEqual(
service_offering.domainid,
self.child_domain_1.id,
"List ServiceOfferings response has incorrect domainid"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_05_list_service_offerings_encryptroot_filter(self):
"""Test list service offerings with encryptroot filter
"""
# List all service offerings with encryptroot True
service_offerings = ServiceOffering.list(
self.apiclient,
encryptroot=True
)
self.assertTrue(
isinstance(service_offerings, list),
"List ServiceOfferings response is not a valid list"
)
self.assertGreaterEqual(
len(service_offerings),
1,
"List ServiceOfferings response is empty"
)
for service_offering in service_offerings:
self.assertTrue(
service_offering.encryptroot,
"List ServiceOfferings response has incorrect encryptroot"
)
service_offerings = ServiceOffering.list(
self.apiclient,
encryptroot=False
)
self.assertTrue(
isinstance(service_offerings, list),
"List ServiceOfferings response is not a valid list"
)
self.assertGreaterEqual(
len(service_offerings),
1,
"List ServiceOfferings response is empty"
)
for service_offering in service_offerings:
self.assertFalse(
service_offering.encryptroot,
"List ServiceOfferings response has incorrect encryptroot"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_06_list_service_offerings_id_filter(self):
"""Test list service offerings with id filter
"""
# List all service offerings with id
service_offerings = ServiceOffering.list(
self.apiclient,
id=self.system_offering.id
)
self.assertIsNone(
service_offerings,
"List ServiceOfferings response is not None"
)
service_offerings = ServiceOffering.list(
self.apiclient,
id=self.service_offering_1.id
)
self.assertTrue(
isinstance(service_offerings, list),
"List ServiceOfferings response is not a valid list"
)
self.assertEqual(
len(service_offerings),
1,
"List ServiceOfferings response is empty"
)
self.assertEqual(
service_offerings[0].id,
self.service_offering_1.id,
"List ServiceOfferings response has incorrect id"
)
service_offerings = ServiceOffering.list(
self.apiclient,
id=-1
)
self.assertIsNone(service_offerings, "List ServiceOfferings response is not None")
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_07_list_service_offerings_isrecursive_filter(self):
"""Test list service offerings with isrecursive filter
"""
# List all service offerings with listall True
service_offerings = ServiceOffering.list(
self.domainadmin_api_client,
isrecursive=True
)
self.assertTrue(
isinstance(service_offerings, list),
"List ServiceOfferings response is not a valid list"
)
self.assertEqual(
len(service_offerings),
len(self.service_offerings) + 1,
"List ServiceOfferings response is empty"
)
# List all service offerings with isrecursive False
service_offerings = ServiceOffering.list(
self.domainadmin_api_client,
isrecursive=False
)
self.assertTrue(
isinstance(service_offerings, list),
"List ServiceOfferings response is not a valid list"
)
self.assertGreaterEqual(
len(service_offerings),
len(self.service_offerings),
"List ServiceOfferings response is empty"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_08_list_service_offerings_issystem_filter(self):
"""Test list service offerings with issystem filter
"""
# List all service offerings with issystem True
service_offerings = ServiceOffering.list(
self.apiclient,
issystem=True
)
self.assertTrue(
isinstance(service_offerings, list),
"List ServiceOfferings response is not a valid list"
)
self.assertEqual(
len(service_offerings),
len(self.system_service_offerings) + 1,
"List ServiceOfferings response is empty"
)
for service_offering in service_offerings:
self.assertTrue(
service_offering.issystem,
"List ServiceOfferings response has incorrect issystem"
)
# List all service offerings with issystem False
service_offerings = ServiceOffering.list(
self.apiclient,
issystem=False
)
self.assertTrue(
isinstance(service_offerings, list),
"List ServiceOfferings response is not a valid list"
)
self.assertEqual(
len(service_offerings),
len(self.service_offerings) + 1,
"List ServiceOfferings response is empty"
)
for service_offering in service_offerings:
self.assertFalse(
service_offering.issystem,
"List ServiceOfferings response has incorrect issystem"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_09_list_service_offerings_keyword_filter(self):
"""Test list service offerings with keyword filter
"""
# List all service offerings with keyword
service_offerings = ServiceOffering.list(
self.apiclient,
keyword=self.system_offering.name
)
self.assertIsNone(
service_offerings,
"List ServiceOfferings response is not None"
)
service_offerings = ServiceOffering.list(
self.apiclient,
keyword=self.service_offering_1.name
)
self.assertTrue(
isinstance(service_offerings, list),
"List ServiceOfferings response is not a valid list"
)
self.assertEqual(
len(service_offerings),
1,
"List ServiceOfferings response is empty"
)
self.assertEqual(
service_offerings[0].name,
self.service_offering_1.name,
"List ServiceOfferings response has incorrect name"
)
service_offerings = ServiceOffering.list(
self.apiclient,
keyword="invalid"
)
self.assertIsNone(service_offerings, "List ServiceOfferings response is not None")
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_10_list_service_offerings_name_filter(self):
"""Test list service offerings with name filter
"""
# List all service offerings with name
service_offerings = ServiceOffering.list(
self.apiclient,
name=self.system_offering.name
)
self.assertIsNone(
service_offerings,
"List ServiceOfferings response is not None"
)
service_offerings = ServiceOffering.list(
self.apiclient,
name=self.system_offering.name,
issystem=True
)
self.assertTrue(
isinstance(service_offerings, list),
"List ServiceOfferings response is not a valid list"
)
self.assertEqual(
len(service_offerings),
1,
"List ServiceOfferings response is empty"
)
self.assertEqual(
service_offerings[0].name,
self.system_offering.name,
"List ServiceOfferings response has incorrect name"
)
service_offerings = ServiceOffering.list(
self.apiclient,
name=self.service_offering_1.name
)
self.assertTrue(
isinstance(service_offerings, list),
"List ServiceOfferings response is not a valid list"
)
self.assertEqual(
len(service_offerings),
1,
"List ServiceOfferings response is empty"
)
self.assertEqual(
service_offerings[0].name,
self.service_offering_1.name,
"List ServiceOfferings response has incorrect name"
)
service_offerings = ServiceOffering.list(
self.apiclient,
name="invalid"
)
self.assertIsNone(service_offerings, "List ServiceOfferings response is not None")
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_11_list_service_offerings_systemvmtype_filter(self):
"""Test list service offerings with systemvmtype filter
"""
# List all service offerings with systemvmtype domainrouter
service_offerings = ServiceOffering.list(
self.apiclient,
systemvmtype="domainrouter"
)
self.assertIsNone(
service_offerings,
"List ServiceOfferings response is not None"
)
service_offerings = ServiceOffering.list(
self.apiclient,
systemvmtype="domainrouter",
issystem=True
)
self.assertTrue(
isinstance(service_offerings, list),
"List ServiceOfferings response is not a valid list"
)
self.assertGreaterEqual(
len(service_offerings),
1,
"List ServiceOfferings response is empty"
)
for service_offering in service_offerings:
self.assertEqual(
service_offering.systemvmtype,
"domainrouter",
"List ServiceOfferings response has incorrect systemvmtype"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_12_list_service_offerings_zoneid_filter(self):
"""Test list service offerings with zoneid filter
"""
service_offerings = ServiceOffering.list(
self.apiclient,
zoneid=self.zone.id
)
self.assertTrue(
isinstance(service_offerings, list),
"List ServiceOfferings response is not a valid list"
)
self.assertEqual(
len(service_offerings),
len(self.service_offerings) + 1,
"List ServiceOfferings response is empty"
)
for service_offering in service_offerings:
self.assertTrue(
service_offering.zoneid is None or service_offering.zoneid == self.zone.id,
"List ServiceOfferings response has incorrect zoneid"
)
if len(self.zones) > 1:
service_offerings = ServiceOffering.list(
self.apiclient,
zoneid=self.zones[1].id
)
if service_offerings is not None:
self.assertTrue(
isinstance(service_offerings, list),
"List ServiceOfferings response is not a valid list"
)
self.assertEqual(
len(service_offerings),
len(self.service_offerings),
"List ServiceOfferings response is empty"
)

396
test/integration/smoke/test_list_storage_pools.py Normal file
View File

@ -0,0 +1,396 @@
# 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.
""" Tests for API listing of storage pools with different filters
"""
# Import Local Modules
from marvin.cloudstackTestCase import cloudstackTestCase
from marvin.codes import FAILED
from marvin.lib.base import (StoragePool)
from marvin.lib.common import (get_domain, list_accounts,
list_zones, list_clusters, list_hosts)
# Import System modules
from nose.plugins.attrib import attr
_multiprocess_shared_ = True
class TestListStoragePools(cloudstackTestCase):
@classmethod
def setUpClass(cls):
testClient = super(TestListStoragePools, cls).getClsTestClient()
cls.apiclient = testClient.getApiClient()
cls.services = testClient.getParsedTestDataConfig()
cls.hypervisor = testClient.getHypervisorInfo()
cls.domain = get_domain(cls.apiclient)
cls.zones = list_zones(cls.apiclient)
cls.zone = cls.zones[0]
cls.clusters = list_clusters(cls.apiclient)
cls.cluster = cls.clusters[0]
cls.hosts = list_hosts(cls.apiclient)
cls.account = list_accounts(cls.apiclient, name="admin")[0]
cls.storage_pools = StoragePool.list(cls.apiclient)
@classmethod
def tearDownClass(cls):
super(TestListStoragePools, cls).tearDownClass()
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_01_list_storage_pools_clusterid_filter(self):
""" Test list storage pools by clusterid filter
"""
storage_pools = StoragePool.list(
self.apiclient,
clusterid=self.cluster.id
)
self.assertTrue(
isinstance(storage_pools, list),
"Storage pool response type should be a list"
)
self.assertGreater(
len(storage_pools),
0,
"Length of storage pools should greater than 0"
)
for storage_pool in storage_pools:
self.assertEqual(
storage_pool.clusterid,
self.cluster.id,
"Cluster id should be equal to the cluster id passed in the filter"
)
storage_pools = StoragePool.list(
self.apiclient,
clusterid="-1"
)
self.assertIsNone(
storage_pools,
"Response should be empty when invalid cluster id is passed"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_02_list_storage_pools_id_filter(self):
""" Test list storage pools by id filter
"""
valid_id = self.storage_pools[0].id
storage_pools = StoragePool.list(
self.apiclient,
id=valid_id
)
self.assertTrue(
isinstance(storage_pools, list),
"Storage pool response type should be a list"
)
self.assertEqual(
len(storage_pools),
1,
"Length of storage pools should be equal to 1"
)
self.assertEqual(
storage_pools[0].id,
valid_id,
"Cluster id should be equal to the cluster id passed in the filter"
)
storage_pools = StoragePool.list(
self.apiclient,
id="-1"
)
self.assertIsNone(
storage_pools,
"Response should be empty when invalid cluster id is passed"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_03_list_storage_pools_ipaddress_filter(self):
""" Test list storage pools by ipaddress filter
"""
valid_ipaddress = self.storage_pools[0].ipaddress
storage_pools = StoragePool.list(
self.apiclient,
ipaddress=valid_ipaddress
)
self.assertTrue(
isinstance(storage_pools, list),
"Storage pool response type should be a list"
)
self.assertGreater(
len(storage_pools),
0,
"Length of storage pools should greater than 0"
)
for storage_pool in storage_pools:
self.assertEqual(
storage_pool.ipaddress,
valid_ipaddress,
"IP address should be equal to the ip address passed in the filter"
)
storage_pools = StoragePool.list(
self.apiclient,
ipaddress="1.1.1.1"
)
self.assertIsNone(
storage_pools,
"Response should be empty when invalid ip address is passed"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_04_list_storage_pools_keyword_filter(self):
""" Test list storage pools by keyword filter
"""
valid_keyword = self.storage_pools[0].name
storage_pools = StoragePool.list(
self.apiclient,
keyword=valid_keyword
)
self.assertTrue(
isinstance(storage_pools, list),
"Storage pool response type should be a list"
)
self.assertGreater(
len(storage_pools),
0,
"Length of storage pools should greater than 0"
)
for storage_pool in storage_pools:
self.assertIn(
valid_keyword,
storage_pool.name,
"Keyword should be present in the storage pool name"
)
storage_pools = StoragePool.list(
self.apiclient,
keyword="invalid"
)
self.assertIsNone(
storage_pools,
"Response should be empty when invalid keyword is passed"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_05_list_storage_pools_name_filter(self):
""" Test list storage pools by name filter
"""
valid_name = self.storage_pools[0].name
storage_pools = StoragePool.list(
self.apiclient,
name=valid_name
)
self.assertTrue(
isinstance(storage_pools, list),
"Storage pool response type should be a list"
)
self.assertGreater(
len(storage_pools),
0,
"Length of storage pools should greater than 0"
)
for storage_pool in storage_pools:
self.assertEqual(
storage_pool.name,
valid_name,
"Name should be equal to the name passed in the filter"
)
storage_pools = StoragePool.list(
self.apiclient,
name="invalid"
)
self.assertIsNone(
storage_pools,
"Response should be empty when invalid name is passed"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_06_list_storage_pools_path_filter(self):
""" Test list storage pools by path filter
"""
valid_path = self.storage_pools[0].path
storage_pools = StoragePool.list(
self.apiclient,
path=valid_path
)
self.assertTrue(
isinstance(storage_pools, list),
"Storage pool response type should be a list"
)
self.assertGreater(
len(storage_pools),
0,
"Length of storage pools should greater than 0"
)
for storage_pool in storage_pools:
self.assertEqual(
storage_pool.path,
valid_path,
"Path should be equal to the path passed in the filter"
)
storage_pools = StoragePool.list(
self.apiclient,
path="invalid"
)
self.assertIsNone(
storage_pools,
"Response should be empty when invalid path is passed"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_07_list_storage_pools_podid_filter(self):
""" Test list storage pools by podid filter
"""
storage_pools = StoragePool.list(
self.apiclient,
podid=self.cluster.podid
)
self.assertTrue(
isinstance(storage_pools, list),
"Storage pool response type should be a list"
)
self.assertGreater(
len(storage_pools),
0,
"Length of storage pools should greater than 0"
)
for storage_pool in storage_pools:
self.assertEqual(
storage_pool.podid,
self.cluster.podid,
"Pod id should be equal to the pod id passed in the filter"
)
storage_pools = StoragePool.list(
self.apiclient,
podid="-1"
)
self.assertIsNone(
storage_pools,
"Response should be empty when invalid pod id is passed"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_08_list_storage_pools_scope_filter(self):
""" Test list storage pools by scope filter
"""
valid_scope = self.storage_pools[0].scope
storage_pools = StoragePool.list(
self.apiclient,
scope=valid_scope
)
self.assertTrue(
isinstance(storage_pools, list),
"Storage pool response type should be a list"
)
self.assertGreater(
len(storage_pools),
0,
"Length of storage pools should greater than 0"
)
for storage_pool in storage_pools:
self.assertEqual(
storage_pool.scope,
valid_scope,
"Scope should be equal to the scope passed in the filter"
)
with self.assertRaises(Exception):
storage_pools = StoragePool.list(
self.apiclient,
scope="invalid"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_09_list_storage_pools_status_filter(self):
""" Test list storage pools by status filter
"""
valid_status = self.storage_pools[0].status
storage_pools = StoragePool.list(
self.apiclient,
status=valid_status
)
self.assertTrue(
isinstance(storage_pools, list),
"Storage pool response type should be a list"
)
self.assertGreater(
len(storage_pools),
0,
"Length of storage pools should greater than 0"
)
for storage_pool in storage_pools:
self.assertEqual(
storage_pool.status,
valid_status,
"State should be equal to the status passed in the filter"
)
with self.assertRaises(Exception):
storage_pools = StoragePool.list(
self.apiclient,
status="invalid"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_10_list_storage_pools_zoneid_filter(self):
""" Test list storage pools by zoneid filter
"""
storage_pools = StoragePool.list(
self.apiclient,
zoneid=self.zone.id
)
self.assertTrue(
isinstance(storage_pools, list),
"Storage pool response type should be a list"
)
self.assertGreater(
len(storage_pools),
0,
"Length of storage pools should greater than 0"
)
for storage_pool in storage_pools:
self.assertEqual(
storage_pool.zoneid,
self.zone.id,
"Zone id should be equal to the zone id passed in the filter"
)
storage_pools = StoragePool.list(
self.apiclient,
zoneid="-1"
)
self.assertIsNone(
storage_pools,
"Response should be empty when invalid zone id is passed"
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_11_list_storage_pools_no_filter(self):
""" Test list storage pools with no filter
"""
storage_pools = StoragePool.list(
self.apiclient
)
self.assertTrue(
isinstance(storage_pools, list),
"Storage pool response type should be a list"
)
self.assertGreater(
len(storage_pools),
0,
"Length of storage pools should greater than 0"
)

615
test/integration/smoke/test_list_volumes.py Normal file
View File

@ -0,0 +1,615 @@
# 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.
""" Tests for API listing of volumes with different filters
"""
# Import Local Modules
from marvin.cloudstackTestCase import cloudstackTestCase
from marvin.codes import FAILED
from marvin.lib.base import (Account,
Domain,
Volume,
ServiceOffering,
Tag,
DiskOffering,
VirtualMachine)
from marvin.lib.common import (get_domain, list_accounts,
list_zones, list_clusters, list_hosts, get_suitable_test_template)
# Import System modules
from nose.plugins.attrib import attr
_multiprocess_shared_ = True
class TestListVolumes(cloudstackTestCase):
@classmethod
def setUpClass(cls):
testClient = super(TestListVolumes, cls).getClsTestClient()
cls.apiclient = testClient.getApiClient()
cls.services = testClient.getParsedTestDataConfig()
cls.hypervisor = testClient.getHypervisorInfo()
cls.domain = get_domain(cls.apiclient)
cls.zones = list_zones(cls.apiclient)
cls.zone = cls.zones[0]
cls.clusters = list_clusters(cls.apiclient)
cls.cluster = cls.clusters[0]
cls.hosts = list_hosts(cls.apiclient)
cls.account = list_accounts(cls.apiclient, name="admin")[0]
cls._cleanup = []
cls.service_offering = ServiceOffering.create(
cls.apiclient,
cls.services["service_offerings"]["tiny"]
)
cls._cleanup.append(cls.service_offering)
template = get_suitable_test_template(
cls.apiclient,
cls.zone.id,
cls.services["ostype"],
cls.hypervisor
)
if template == FAILED:
assert False, "get_test_template() failed to return template"
cls.services["template"]["ostypeid"] = template.ostypeid
cls.services["template_2"]["ostypeid"] = template.ostypeid
cls.services["ostypeid"] = template.ostypeid
cls.services["virtual_machine"]["zoneid"] = cls.zone.id
cls.services["mode"] = cls.zone.networktype
cls.disk_offering = DiskOffering.create(cls.apiclient,
cls.services["disk_offering"])
cls._cleanup.append(cls.disk_offering)
# Create VM
cls.virtual_machine = VirtualMachine.create(
cls.apiclient,
cls.services["virtual_machine"],
templateid=template.id,
accountid=cls.account.name,
domainid=cls.account.domainid,
clusterid=cls.cluster.id,
serviceofferingid=cls.service_offering.id,
mode=cls.services["mode"]
)
cls.child_domain = Domain.create(
cls.apiclient,
cls.services["domain"])
cls._cleanup.append(cls.child_domain)
cls.child_account = Account.create(
cls.apiclient,
cls.services["account"],
admin=True,
domainid=cls.child_domain.id)
cls._cleanup.append(cls.child_account)
cls.vol_1 = Volume.create(cls.apiclient,
cls.services["volume"],
zoneid=cls.zone.id,
account=cls.account.name,
domainid=cls.account.domainid,
diskofferingid=cls.disk_offering.id)
cls._cleanup.append(cls.vol_1)
cls.vol_1 = cls.virtual_machine.attach_volume(
cls.apiclient,
cls.vol_1
)
cls._cleanup.append(cls.virtual_machine)
Tag.create(cls.apiclient, cls.vol_1.id, "Volume", {"abc": "xyz"})
cls.vol_2 = Volume.create(cls.apiclient,
cls.services["volume"],
zoneid=cls.zone.id,
account=cls.account.name,
domainid=cls.account.domainid,
diskofferingid=cls.disk_offering.id)
cls._cleanup.append(cls.vol_2)
cls.vol_3 = Volume.create(cls.apiclient,
cls.services["volume"],
zoneid=cls.zone.id,
account=cls.child_account.name,
domainid=cls.child_account.domainid,
diskofferingid=cls.disk_offering.id)
cls._cleanup.append(cls.vol_3)
@classmethod
def tearDownClass(cls):
super(TestListVolumes, cls).tearDownClass()
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_01_list_volumes_account_domain_filter(self):
"""Test listing Volumes with account & domain filter
"""
list_volume_response = Volume.list(
self.apiclient,
zoneid=self.zone.id,
account=self.account.name,
domainid=self.account.domainid
)
self.assertTrue(
isinstance(list_volume_response, list),
"List Volume response is not a valid list"
)
self.assertEqual(
len(list_volume_response),
3,
"ListVolumes response expected 3 Volumes, received %s" % len(list_volume_response)
)
list_volume_response = Volume.list(
self.apiclient,
zoneid=self.zone.id,
account=self.child_account.name,
domainid=self.child_account.domainid
)
self.assertTrue(
isinstance(list_volume_response, list),
"List Volume response is not a valid list"
)
self.assertEqual(
len(list_volume_response),
1,
"ListVolumes response expected 1 Volume, received %s" % len(list_volume_response)
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_02_list_volumes_diskofferingid_filter(self):
"""Test listing Volumes with diskofferingid filter
"""
list_volume_response = Volume.list(
self.apiclient,
zoneid=self.zone.id,
account=self.account.name,
domainid=self.account.domainid,
diskofferingid=self.disk_offering.id
)
self.assertTrue(
isinstance(list_volume_response, list),
"List Volume response is not a valid list"
)
self.assertEqual(
len(list_volume_response),
2,
"ListVolumes response expected 2 Volumes, received %s" % len(list_volume_response)
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_03_list_volumes_id_filter(self):
"""Test listing Volumes with id filter
"""
list_volume_response = Volume.list(
self.apiclient,
zoneid=self.zone.id,
id=self.vol_1.id
)
self.assertTrue(
isinstance(list_volume_response, list),
"List Volume response is not a valid list"
)
self.assertEqual(
len(list_volume_response),
1,
"ListVolumes response expected 1 Volume, received %s" % len(list_volume_response)
)
self.assertEqual(
list_volume_response[0].id,
self.vol_1.id,
"ListVolumes response expected Volume with id %s, received %s" % (self.vol_1.id, list_volume_response[0].id)
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_04_list_volumes_ids_filter(self):
"""Test listing Volumes with ids filter
"""
list_volume_response = Volume.list(
self.apiclient,
zoneid=self.zone.id,
ids=[self.vol_1.id, self.vol_2.id, self.vol_3.id]
)
self.assertTrue(
isinstance(list_volume_response, list),
"List Volume response is not a valid list"
)
self.assertEqual(
len(list_volume_response),
2,
"ListVolumes response expected 2 Volumes, received %s" % len(list_volume_response)
)
self.assertIn(list_volume_response[0].id, [self.vol_1.id, self.vol_2.id],
"ListVolumes response Volume 1 not in list")
self.assertIn(list_volume_response[1].id, [self.vol_1.id, self.vol_2.id],
"ListVolumes response Volume 2 not in list")
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_05_list_volumes_isrecursive(self):
"""Test listing Volumes with isrecursive filter
"""
list_volume_response = Volume.list(
self.apiclient,
zoneid=self.zone.id,
isrecursive=True,
domainid=self.account.domainid
)
self.assertTrue(
isinstance(list_volume_response, list),
"List Volume response is not a valid list"
)
self.assertEqual(
len([v for v in list_volume_response if v.state != "Destroy"]),
4,
"ListVolumes response expected 4 Volumes, received %s" % len(list_volume_response)
)
list_volume_response = Volume.list(
self.apiclient,
zoneid=self.zone.id,
isrecursive=False,
domainid=self.account.domainid
)
self.assertTrue(
isinstance(list_volume_response, list),
"List Volume response is not a valid list"
)
self.assertEqual(
len([v for v in list_volume_response if v.state != "Destroy"]),
3,
"ListVolumes response expected 3 Volumes, received %s" % len(list_volume_response)
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_06_list_volumes_keyword_filter(self):
"""Test listing Volumes with keyword filter
"""
list_volume_response = Volume.list(
self.apiclient,
zoneid=self.zone.id,
keyword=self.services["volume"]["diskname"]
)
self.assertTrue(
isinstance(list_volume_response, list),
"List Volume response is not a valid list"
)
self.assertEqual(
len(list_volume_response),
2,
"ListVolumes response expected 2 Volumes, received %s" % len(list_volume_response)
)
self.assertIn(
list_volume_response[0].id, [self.vol_1.id, self.vol_2.id],
"ListVolumes response Volume 1 not in list")
self.assertIn(list_volume_response[1].id, [self.vol_1.id, self.vol_2.id],
"ListVolumes response Volume 2 not in list")
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_07_list_volumes_listall(self):
"""Test listing Volumes with listall filter
"""
list_volume_response = Volume.list(
self.apiclient,
zoneid=self.zone.id,
listall=True
)
self.assertTrue(
isinstance(list_volume_response, list),
"List Volume response is not a valid list"
)
self.assertEqual(
len([v for v in list_volume_response if v.state != "Destroy"]),
4,
"ListVolumes response expected 4 Volumes, received %s" % len(list_volume_response)
)
list_volume_response = Volume.list(
self.apiclient,
zoneid=self.zone.id,
listall=False
)
self.assertTrue(
isinstance(list_volume_response, list),
"List Volume response is not a valid list"
)
self.assertEqual(
len([v for v in list_volume_response if v.state != "Destroy"]),
3,
"ListVolumes response expected 3 Volumes, received %s" % len(list_volume_response)
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_08_listsystemvms(self):
list_volumes_response = Volume.list(
self.apiclient,
zoneid=self.zone.id,
listsystemvms=True
)
self.assertEqual(
isinstance(list_volumes_response, list),
True,
"List Volume response is not a valid list"
)
self.assertGreater(
len(list_volumes_response),
3,
"ListVolumes response expected more than 3 Volumes, received %s" % len(list_volumes_response)
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_09_list_volumes_name_filter(self):
"""Test listing Volumes with name filter
"""
list_volume_response = Volume.list(
self.apiclient,
zoneid=self.zone.id,
name=self.vol_1.name
)
self.assertTrue(
isinstance(list_volume_response, list),
"List Volume response is not a valid list"
)
self.assertEqual(
len(list_volume_response),
1,
"ListVolumes response expected 1 Volumes, received %s" % len(list_volume_response)
)
self.assertEqual(
list_volume_response[0].id,
self.vol_1.id,
"ListVolumes response expected Volume with id %s, received %s" % (self.vol_1.id, list_volume_response[0].id)
)
self.assertEqual(
list_volume_response[0].name,
self.vol_1.name,
"ListVolumes response expected Volume with name %s, received %s" % (
self.vol_1.name, list_volume_response[0].name)
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_10_list_volumes_podid_filter(self):
"""Test listing Volumes with podid filter
"""
list_volume_response = Volume.list(
self.apiclient,
zoneid=self.zone.id,
podid=self.vol_1.podid
)
self.assertTrue(
isinstance(list_volume_response, list),
"List Volume response is not a valid list"
)
self.assertGreater(
len(list_volume_response),
1,
"ListVolumes response expected more than 1 Volume, received %s" % len(list_volume_response)
)
self.assertIn(self.vol_1.id, [volume.id for volume in list_volume_response],
"ListVolumes response expected Volume with id %s" % self.vol_1.id)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_11_list_volumes_state_filter(self):
"""Test listing Volumes with state filter
"""
list_volume_response = Volume.list(
self.apiclient,
zoneid=self.zone.id,
state="Ready"
)
self.assertTrue(
isinstance(list_volume_response, list),
"List Volume response is not a valid list"
)
self.assertEqual(
len(list_volume_response),
2,
"ListVolumes response expected 2 Volumes, received %s" % len(list_volume_response)
)
self.assertIn(self.vol_1.id, [volume.id for volume in list_volume_response],
"ListVolumes response expected Volume with id %s" % self.vol_1.id)
list_volume_response = Volume.list(
self.apiclient,
zoneid=self.zone.id,
state="Allocated"
)
self.assertTrue(
isinstance(list_volume_response, list),
"List Volume response is not a valid list"
)
self.assertEqual(
len(list_volume_response),
1,
"ListVolumes response expected 1 Volumes, received %s" % len(list_volume_response)
)
self.assertEqual(self.vol_2.id, list_volume_response[0].id,
"ListVolumes response expected Volume with id %s" % self.vol_3.id)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_12_list_volumes_storageid_filter(self):
"""Test listing Volumes with storageid filter
"""
list_volume_response = Volume.list(
self.apiclient,
zoneid=self.zone.id,
storageid=self.vol_1.storageid
)
self.assertTrue(
isinstance(list_volume_response, list),
"List Volume response is not a valid list"
)
self.assertGreaterEqual(
len(list_volume_response),
1,
"ListVolumes response expected 1 or more Volumes, received %s" % len(list_volume_response)
)
self.assertIn(self.vol_1.id, [volume.id for volume in list_volume_response],
"ListVolumes response expected Volume with id %s" % self.vol_1.id)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_13_list_volumes_type_filter(self):
"""Test listing Volumes with type filter
"""
list_volume_response = Volume.list(
self.apiclient,
zoneid=self.zone.id,
type="DATADISK"
)
self.assertTrue(
isinstance(list_volume_response, list),
"List Volume response is not a valid list"
)
self.assertEqual(
len(list_volume_response),
2,
"ListVolumes response expected 2 Volumes, received %s" % len(list_volume_response)
)
self.assertIn(self.vol_1.id, [volume.id for volume in list_volume_response],
"ListVolumes response expected Volume with id %s" % self.vol_1.id)
list_volume_response = Volume.list(
self.apiclient,
zoneid=self.zone.id,
type="ROOT"
)
self.assertTrue(
isinstance(list_volume_response, list),
"List Volume response is not a valid list"
)
self.assertEqual(
len(list_volume_response),
1,
"ListVolumes response expected 1 Volumes, received %s" % len(list_volume_response)
)
self.assertNotIn(list_volume_response[0].id, [self.vol_1.id, self.vol_2.id],
"ListVolumes response expected ROOT Volume")
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_14_list_volumes_virtualmachineid_filter(self):
"""Test listing Volumes with virtualmachineid filter
"""
list_volume_response = Volume.list(
self.apiclient,
zoneid=self.zone.id,
virtualmachineid=self.vol_1.virtualmachineid
)
self.assertTrue(
isinstance(list_volume_response, list),
"List Volume response is not a valid list"
)
self.assertEqual(
len(list_volume_response),
2,
"ListVolumes response expected 2 Volumes, received %s" % len(list_volume_response)
)
self.assertIn(self.vol_1.id, [volume.id for volume in list_volume_response],
"ListVolumes response expected Volume with id %s" % self.vol_1.id)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_15_list_volumes_zoneid_filter(self):
"""Test listing Volumes with zoneid filter
"""
list_volume_response = Volume.list(
self.apiclient,
zoneid=self.zones[0].id
)
self.assertTrue(
isinstance(list_volume_response, list),
"List Volume response is not a valid list"
)
self.assertEqual(
len(list_volume_response),
3,
"ListVolumes response expected 3 Volumes, received %s" % len(list_volume_response)
)
if len(self.zones) > 1:
list_volume_response = Volume.list(
self.apiclient,
zoneid=self.zones[1].id
)
self.assertIsNone(list_volume_response, "List Volume response is not None")
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_16_list_volumes_tags_filter(self):
"""Test listing Volumes with tags filter
"""
list_volume_response = Volume.list(
self.apiclient,
tags=[{"key": "abc", "value": "xyz"}]
)
self.assertTrue(
isinstance(list_volume_response, list),
"List Volume response is not a valid list"
)
self.assertEqual(
len(list_volume_response),
1,
"ListVolumes response expected 1 or more Volumes, received %s" % len(list_volume_response)
)
self.assertEqual(
list_volume_response[0].id,
self.vol_1.id,
"ListVolumes response expected Volume with id %s, received %s" % (self.vol_1.id, list_volume_response[0].id)
)
self.assertEqual(
list_volume_response[0].tags[0]["key"],
"abc",
"ListVolumes response expected Volume with tag key abc, received %s" % list_volume_response[0].tags[0]["key"]
)
self.assertEqual(
list_volume_response[0].tags[0]["value"],
"xyz",
"ListVolumes response expected Volume with tag value xyz, received %s" % list_volume_response[0].tags[0]["value"]
)
list_volume_response = Volume.list(
self.apiclient,
tags=[{"key": "abc", "value": "xyz1"}]
)
self.assertIsNone(list_volume_response, "List Volume response is not None")
with self.assertRaises(Exception):
list_volume_response = Volume.list(
self.apiclient,
tags=[{"key": None, "value": None}]
)
@attr(tags=["advanced", "advancedns", "smoke", "basic"], required_hardware="false")
def test_17_list_volumes_no_filter(self):
"""Test listing Volumes with no filter
"""
list_volume_response = Volume.list(
self.apiclient,
zoneid=self.zone.id
)
self.assertTrue(
isinstance(list_volume_response, list),
"List Volume response is not a valid list"
)
self.assertGreaterEqual(
len(list_volume_response),
3,
"ListVolumes response expected 3 or more Volumes, received %s" % len(list_volume_response)
)
self.assertIn(self.vol_1.id, [volume.id for volume in list_volume_response],
"ListVolumes response expected Volume with id %s" % self.vol_1.id)

1
test/integration/smoke/test_metrics_api.py
View File

@ -289,7 +289,6 @@ class TestMetrics(cloudstackTestCase):
self.assertTrue(hasattr(li, 'hosts'))
self.assertEqual(li.hosts, len(list_hosts(self.apiclient,
zoneid=self.zone.id,
type='Routing')))
self.assertTrue(hasattr(li, 'imagestores'))

2
test/integration/smoke/test_secondary_storage.py
View File

@ -340,7 +340,7 @@ class TestSecStorageServices(cloudstackTestCase):
# 1. Try complete migration from a storage with more (or equal) free space - migration should be refused
storages = self.list_secondary_storages(self.apiclient)
if (len(storages)) < 2:
if (len(storages)) < 2 or (storages[0]['zoneid'] != storages[1]['zoneid']):
self.skipTest(
"This test requires more than one secondary storage")

16
test/integration/smoke/test_templates.py
View File

@ -1024,17 +1024,17 @@ class TestCopyAndDeleteTemplatesAcrossZones(cloudstackTestCase):
cls.services["disk_offering"]
)
cls._cleanup.append(cls.disk_offering)
template = get_template(
cls.template = get_template(
cls.apiclient,
cls.zone.id,
cls.services["ostype"]
)
if template == FAILED:
if cls.template == FAILED:
assert False, "get_template() failed to return template with description %s" % cls.services["ostype"]
cls.services["template"]["ostypeid"] = template.ostypeid
cls.services["template_2"]["ostypeid"] = template.ostypeid
cls.services["ostypeid"] = template.ostypeid
cls.services["template"]["ostypeid"] = cls.template.ostypeid
cls.services["template_2"]["ostypeid"] = cls.template.ostypeid
cls.services["ostypeid"] = cls.template.ostypeid
cls.services["virtual_machine"]["zoneid"] = cls.zone.id
cls.services["volume"]["diskoffering"] = cls.disk_offering.id
@ -1055,7 +1055,7 @@ class TestCopyAndDeleteTemplatesAcrossZones(cloudstackTestCase):
cls.virtual_machine = VirtualMachine.create(
cls.apiclient,
cls.services["virtual_machine"],
templateid=template.id,
templateid=cls.template.id,
accountid=cls.account.name,
domainid=cls.account.domainid,
serviceofferingid=cls.service_offering.id,
@ -1104,7 +1104,7 @@ class TestCopyAndDeleteTemplatesAcrossZones(cloudstackTestCase):
raise Exception("Warning: Exception during cleanup : %s" % e)
return
@attr(tags=["advanced", "advancedns"], required_hardware="false")
@attr(tags=["advanced", "advancedns"], required_hardware="true")
def test_09_copy_delete_template(self):
cmd = listZones.listZonesCmd()
zones = self.apiclient.listZones(cmd)
@ -1156,7 +1156,7 @@ class TestCopyAndDeleteTemplatesAcrossZones(cloudstackTestCase):
list_template_response = Template.list(
self.apiclient,
templatefilter=self.services["template"]["templatefilter"],
templatefilter=self.services["templatefilter"],
id=self.template.id,
zoneid=self.destZone.id
)

23
tools/marvin/marvin/cloudstackTestCase.py
View File

@ -23,7 +23,8 @@ from marvin.lib.base import (
Network,
NetworkACL,
NetworkOffering,
VirtualMachine
VirtualMachine,
Volume
)
@ -98,12 +99,32 @@ class cloudstackTestCase(unittest.case.TestCase):
"""
Delete resources (created during tests)
"""
volume_list = []
for obj in resources:
if isinstance(obj, VirtualMachine):
obj.delete(api_client, expunge=True)
elif isinstance(obj, Volume):
obj.destroy(api_client, expunge=True)
volume_list.append(obj)
else:
obj.delete(api_client)
cls.wait_for_volumes_cleanup(api_client, volume_list)
def wait_for_volumes_cleanup(cls, api_client, volume_list=[]):
"""Wait for volumes to be deleted"""
for volume in volume_list:
max_retries = 24 # Max wait time will be 5 * 24 = 120 seconds
while max_retries > 0:
volumes = Volume.list(
api_client,
id=volume.id
)
if volumes is None or len(volumes) == 0:
break
max_retries = max_retries - 1
time.sleep(5)
def check_wget_from_vm(self, vm, public_ip, network=None, testnegative=False, isVmAccessible=True):
import urllib.request, urllib.error
self.debug(f"Checking if we can wget from a VM={vm.name} http server on public_ip={public_ip.ipaddress.ipaddress}, expecting failure == {testnegative} and vm is acceccible == {isVmAccessible}")