mirror of https://github.com/apache/cloudstack.git
marvin_refactor: improve unittests for additional data factoreis
Signed-off-by: Prasanna Santhanam <tsp@apache.org>
This commit is contained in:
Prasanna Santhanam
parent
2bb1533a8e
commit
939c4d2efc
|
|
@ -22,19 +22,22 @@ from should_dsl import should, should_not
|
|||
|
||||
from marvin.cloudstackTestClient import cloudstackTestClient
|
||||
|
||||
from marvin.factory.data.account import UserAccountFactory, AdminAccountFactory, DomainAdminFactory
|
||||
from marvin.factory.data.account import UserAccount, AdminAccount, DomainAdmin
|
||||
from marvin.factory.data.serviceoffering import *
|
||||
from marvin.factory.data.diskoffering import *
|
||||
from marvin.factory.data.template import *
|
||||
from marvin.factory.data.user import *
|
||||
from marvin.factory.data.networkoffering import *
|
||||
from marvin.factory.data.network import *
|
||||
from marvin.factory.data.vm import *
|
||||
from marvin.factory.data.firewallrule import *
|
||||
from marvin.factory.data.firewallrule import SshFirewallRule
|
||||
from marvin.factory.data.vpc import DefaultVpc
|
||||
|
||||
from marvin.factory.virtualmachine import *
|
||||
|
||||
from marvin.entity.firewall import Firewall
|
||||
from marvin.entity.serviceoffering import ServiceOffering
|
||||
from marvin.entity.diskoffering import DiskOffering
|
||||
from marvin.entity.networkoffering import NetworkOffering
|
||||
from marvin.entity.zone import Zone
|
||||
from marvin.entity.account import Account
|
||||
|
|
@ -54,11 +57,11 @@ class BuildVsCreateStrategyTest(unittest.TestCase):
|
|||
pass
|
||||
|
||||
def test_buildUserAccountFactory(self):
|
||||
af = UserAccountFactory.build()
|
||||
af = UserAccount.build()
|
||||
self.assert_(af is not None, msg="Account factory didn't initialize")
|
||||
|
||||
def test_createAccountFactory(self):
|
||||
af = UserAccountFactory(apiclient=self.apiClient)
|
||||
af = UserAccount(apiclient=self.apiClient)
|
||||
self.assert_(isinstance(af, Account))
|
||||
self.assert_(af.id is not None, msg="Account creation failed")
|
||||
self.assert_(af.domain is not None, msg="Account belongs to no domain")
|
||||
|
|
@ -69,21 +72,21 @@ class AccountFactoryTest(unittest.TestCase):
|
|||
self.apiClient = cloudstackTestClient(mgtSvr='localhost', logging=logging.getLogger('factory.cloudstack')).getApiClient()
|
||||
|
||||
def test_adminAccountFactory(self):
|
||||
accnt = AdminAccountFactory(apiclient=self.apiClient)
|
||||
accnt = AdminAccount(apiclient=self.apiClient)
|
||||
self.assert_(accnt is not None, msg="no account created by factory")
|
||||
self.assert_(accnt.name is not None)
|
||||
|
||||
def test_userAccountFactoryCustomArgs(self):
|
||||
accnt = UserAccountFactory(apiclient=self.apiClient, firstname='test', lastname='test')
|
||||
accnt = UserAccount(apiclient=self.apiClient, firstname='test', lastname='test')
|
||||
a = accnt.list(apiclient=self.apiClient, account=accnt.name, domainid=accnt.domainid)
|
||||
self.assert_(accnt is not None, msg="no account created by factory")
|
||||
self.assert_(accnt.name is not None)
|
||||
|
||||
def test_disableAccountPostFactoryGeneration(self):
|
||||
domadmin = DomainAdminFactory(apiclient=self.apiClient)
|
||||
domadmin = DomainAdmin(apiclient=self.apiClient)
|
||||
a = Account.list(apiclient=self.apiClient, id=domadmin.id)
|
||||
self.assert_(domadmin is not None, msg="no account was created")
|
||||
domadmin.disable(self.apiClient, lock=True, account=domadmin.name, domainid=domadmin.domainid)
|
||||
domadmin.disable(lock=True, account=domadmin.name, domainid=domadmin.domainid)
|
||||
|
||||
def tearDown(self):
|
||||
pass
|
||||
|
|
@ -94,7 +97,7 @@ class ServiceOfferingFactoryTest(unittest.TestCase):
|
|||
self.apiClient = cloudstackTestClient(mgtSvr='localhost', logging=logging.getLogger('factory.cloudstack')).getApiClient()
|
||||
|
||||
def test_serviceOfferingFactory(self):
|
||||
soffering = SmallServiceOfferingFactory(apiclient=self.apiClient)
|
||||
soffering = SmallServiceOffering(apiclient=self.apiClient)
|
||||
self.assert_(soffering is not None, msg="no service offering was created")
|
||||
self.assert_(soffering.name is not None, msg="error in service offering factory creation")
|
||||
|
||||
|
|
@ -108,13 +111,13 @@ class NetworkOfferingFactoryTest(unittest.TestCase):
|
|||
self.apiClient = cloudstackTestClient(mgtSvr='localhost', logging=logging.getLogger('factory.cloudstack')).getApiClient()
|
||||
|
||||
def test_defaultSourceNatOfferingFactory(self):
|
||||
snatOffering = DefaultIsolatedNetworkOfferingWithSourceNatServiceFactory(apiclient=self.apiClient)
|
||||
snatOffering = DefaultIsolatedNetworkOfferingWithSourceNatService(apiclient=self.apiClient)
|
||||
self.assert_(snatOffering is not None, msg = "no network offering was created")
|
||||
self.assert_(snatOffering.name is not None, msg="error in network offering creation")
|
||||
|
||||
@attr(tags='offering')
|
||||
def test_defaultSGOfferingEnable(self):
|
||||
DefaultSharedNetworkOfferingWithSGServiceFactory(apiclient=self.apiClient)
|
||||
DefaultSharedNetworkOfferingWithSGService(apiclient=self.apiClient)
|
||||
|
||||
def tearDown(self):
|
||||
pass
|
||||
|
|
@ -137,7 +140,7 @@ class UserFactorySubFactoryTest(unittest.TestCase):
|
|||
caller is not to create the user before creating the account
|
||||
@return:
|
||||
"""
|
||||
uf = UserFactory(apiclient=self.apiClient)
|
||||
uf = User(apiclient=self.apiClient)
|
||||
user = User.list(apiclient=self.apiClient, username=uf.username)
|
||||
self.assert_(uf.username == user[0].username, msg="Usernames don't match")
|
||||
|
||||
|
|
@ -148,16 +151,13 @@ class NetworkFactoryTest(unittest.TestCase):
|
|||
logging=logging.getLogger('factory.cloudstack')).getApiClient()
|
||||
|
||||
def tearDown(self):
|
||||
self.accnt.delete(apiclient=self.apiClient)
|
||||
self.accnt.delete()
|
||||
|
||||
@attr(tags='network')
|
||||
def test_isolatedGuestNetwork(self):
|
||||
"""Test to create a network within a guest account
|
||||
@return:
|
||||
"""
|
||||
self.accnt = UserAccountFactory(apiclient=self.apiClient)
|
||||
self.accnt = UserAccount(apiclient=self.apiClient)
|
||||
zones = Zone.list(apiclient=self.apiClient)
|
||||
network = GuestIsolatedNetworkFactory(
|
||||
network = GuestIsolatedNetwork(
|
||||
apiclient=self.apiClient,
|
||||
zoneid=zones[0].id
|
||||
)
|
||||
|
|
@ -171,7 +171,7 @@ class NetworkOfferingFactoryWithMultiplePostHooksTest(unittest.TestCase):
|
|||
|
||||
@attr(tags='post')
|
||||
def test_multiplePostHooksNetworkOffering(self):
|
||||
sharedOffering = DefaultSharedNetworkOfferingFactory(apiclient=self.apiClient)
|
||||
sharedOffering = DefaultSharedNetworkOffering(apiclient=self.apiClient)
|
||||
sharedOffering |should| be_instance_of(NetworkOffering)
|
||||
sharedOffering |should_not| equal_to(None)
|
||||
sharedOffering.state |should| equal_to('Enabled')
|
||||
|
|
@ -179,17 +179,17 @@ class NetworkOfferingFactoryWithMultiplePostHooksTest(unittest.TestCase):
|
|||
%(sharedOffering.id, sharedOffering.name, sharedOffering.state))
|
||||
|
||||
|
||||
class VirtualMachineFactoryTest(unittest.TestCase):
|
||||
class VirtualMachineTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.apiClient = cloudstackTestClient(mgtSvr='localhost', logging=logging.getLogger('factory.cloudstack')).getApiClient()
|
||||
|
||||
def tearDown(self):
|
||||
self.vm.destroy(apiclient=self.apiClient)
|
||||
self.vm.destroy()
|
||||
|
||||
def test_virtualMachineDeploy(self):
|
||||
accnt = UserAccountFactory(apiclient=self.apiClient)
|
||||
service = SmallServiceOfferingFactory(apiclient=self.apiClient)
|
||||
tf = DefaultBuiltInTemplateFactory.build()
|
||||
accnt = UserAccount(apiclient=self.apiClient)
|
||||
service = SmallServiceOffering(apiclient=self.apiClient)
|
||||
tf = DefaultBuiltInTemplate.build()
|
||||
zones = Zone.list(apiclient=self.apiClient)
|
||||
template = Template.list(apiclient=self.apiClient,
|
||||
templatefilter="featured",
|
||||
|
|
@ -209,11 +209,11 @@ class IpAddressFactoryTest(unittest.TestCase):
|
|||
logging=logging.getLogger('factory.cloudstack')).getApiClient()
|
||||
|
||||
def tearDown(self):
|
||||
self.vm.destroy(apiclient=self.apiClient)
|
||||
self.vm.destroy()
|
||||
|
||||
def test_associateIpAddressToNetwork(self):
|
||||
# user account where we run test
|
||||
accnt = UserAccountFactory(apiclient=self.apiClient)
|
||||
accnt = UserAccount(apiclient=self.apiClient)
|
||||
self.assert_(isinstance(accnt, Account))
|
||||
|
||||
# get required arguments - templates, service offerings, zone
|
||||
|
|
@ -240,11 +240,11 @@ class FirewallRuleFactoryTest(unittest.TestCase):
|
|||
logging=logging.getLogger('factory.cloudstack')).getApiClient()
|
||||
|
||||
def tearDown(self):
|
||||
self.account.delete(apiclient=self.apiClient)
|
||||
self.account.delete()
|
||||
|
||||
@attr(tags='firewall')
|
||||
def test_firewallRuleFactoryTest(self):
|
||||
self.account = UserAccountFactory(apiclient=self.apiClient)
|
||||
self.account = UserAccount(apiclient=self.apiClient)
|
||||
domainid = get_domain(self.apiClient).id
|
||||
self.account |should| be_instance_of(Account)
|
||||
vm = VirtualMachineFactory(
|
||||
|
|
@ -260,13 +260,6 @@ class FirewallRuleFactoryTest(unittest.TestCase):
|
|||
vm.nic |should_not| equal_to(None)
|
||||
vm.nic |should| be_instance_of(list)
|
||||
|
||||
ipaddresses = IpAddress.listPublic(
|
||||
apiclient=self.apiClient,
|
||||
networkid=vm.nic[0].networkid
|
||||
)
|
||||
ipaddresses |should_not| equal_to(None)
|
||||
ipaddresses |should| be_instance_of(list)
|
||||
|
||||
ipaddress = IpAddress(
|
||||
apiclient=self.apiClient,
|
||||
account=self.account.name,
|
||||
|
|
@ -276,9 +269,173 @@ class FirewallRuleFactoryTest(unittest.TestCase):
|
|||
ipaddress |should_not| be(None)
|
||||
ipaddress |should| be_instance_of(IpAddress)
|
||||
|
||||
fwrule = SshFirewallRuleFactory(
|
||||
fwrule = SshFirewallRule(
|
||||
apiclient=self.apiClient,
|
||||
ipaddressid=ipaddress.id
|
||||
)
|
||||
fwrule |should_not| be(None)
|
||||
fwrule |should| be_instance_of(Firewall)
|
||||
|
||||
|
||||
@attr(tags='disk')
|
||||
class DiskOfferingFactoryTest(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.apiClient = cloudstackTestClient(mgtSvr='localhost',
|
||||
logging=logging.getLogger('factory.cloudstack')).getApiClient()
|
||||
|
||||
def test_createSharedDiskOffering(self):
|
||||
shared_disk_offering = SharedDiskOffering(apiclient=self.apiClient)
|
||||
shared_disk_offering |should_not| be(None)
|
||||
shared_disk_offering.name |should_not| equal_to(None)
|
||||
shared_disk_offering.delete()
|
||||
|
||||
def test_createLocalDiskOffering(self):
|
||||
local_disk_offering = LocalDiskOffering(apiclient=self.apiClient)
|
||||
local_disk_offering | should_not | be(None)
|
||||
local_disk_offering.name | should_not | equal_to(None)
|
||||
local_disk_offering.delete()
|
||||
|
||||
def test_listDeletedDiskOfferings(self):
|
||||
local_disk_offering = LocalDiskOffering(apiclient=self.apiClient)
|
||||
local_disk_offering | should_not | be(None)
|
||||
local_disk_offering.name | should_not | equal_to(None)
|
||||
local_disk_offering.delete()
|
||||
|
||||
listed_offering = DiskOffering.list(apiclient=self.apiClient, name=local_disk_offering.name)
|
||||
listed_offering | should | be(None)
|
||||
|
||||
|
||||
@attr(tags='gc')
|
||||
class DeleteAllNonAdminAccounts(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.apiClient = cloudstackTestClient(mgtSvr='localhost',
|
||||
logging=logging.getLogger('factory.cloudstack')).getApiClient()
|
||||
|
||||
def test_deleteAllNonAdminAccounts(self):
|
||||
accounts = Account.list(apiclient=self.apiClient, listall=True)
|
||||
for account in accounts:
|
||||
if account.accounttype == 0 or account.accounttype == 2 or account.accounttype == 1:
|
||||
if account.name == 'admin':
|
||||
continue
|
||||
account.delete(apiclient=self.apiClient)
|
||||
|
||||
|
||||
@attr(tags='post')
|
||||
class StaticNatPostVirtualMachineDeployHook(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.apiClient = cloudstackTestClient(mgtSvr='localhost',
|
||||
logging=logging.getLogger('factory.cloudstack')).getApiClient()
|
||||
|
||||
def test_staticNatPostVmCreation(self):
|
||||
self.account = UserAccount(apiclient=self.apiClient)
|
||||
domainid = get_domain(self.apiClient).id
|
||||
self.account |should| be_instance_of(Account)
|
||||
vm = VirtualMachineWithStaticNat(
|
||||
apiclient=self.apiClient,
|
||||
account=self.account.name,
|
||||
domainid=domainid,
|
||||
templateid=get_template(self.apiClient).id,
|
||||
serviceofferingid=get_service_offering(self.apiClient).id,
|
||||
zoneid=get_zone(self.apiClient).id
|
||||
)
|
||||
vm |should| be_instance_of(VirtualMachine)
|
||||
vm.state |should| equal_to('Running')
|
||||
vm.nic |should_not| equal_to(None)
|
||||
vm.nic |should| be_instance_of(list)
|
||||
|
||||
def tearDown(self):
|
||||
self.account.delete()
|
||||
|
||||
class VpcCreationTest(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.apiClient = cloudstackTestClient(mgtSvr='localhost',
|
||||
logging=logging.getLogger('factory.cloudstack')).getApiClient()
|
||||
|
||||
def test_vpcCreation(self):
|
||||
self.account = UserAccount(apiclient=self.apiClient)
|
||||
self.vpc = DefaultVpc(
|
||||
apiclient=self.apiClient,
|
||||
account=self.account.name,
|
||||
domainid=get_domain(apiclient=self.apiClient).id,
|
||||
zoneid=get_zone(apiclient=self.apiClient).id
|
||||
)
|
||||
self.vpc |should_not| be(None)
|
||||
ntwk = DefaultVpcNetwork(
|
||||
apiclient=self.apiClient,
|
||||
account=self.account.name,
|
||||
domainid=get_domain(apiclient=self.apiClient).id,
|
||||
vpcid=self.vpc.id,
|
||||
zoneid=get_zone(apiclient=self.apiClient).id
|
||||
)
|
||||
ntwk | should_not | be(None)
|
||||
ntwk.state | should | equal_to('Allocated')
|
||||
ntwk.delete()
|
||||
|
||||
def tearDown(self):
|
||||
self.vpc.delete()
|
||||
self.account.delete()
|
||||
|
||||
@attr(tags='vpc')
|
||||
class VpcVmDeployTest(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.apiClient = cloudstackTestClient(mgtSvr='localhost',
|
||||
logging=logging.getLogger('factory.cloudstack')).getApiClient()
|
||||
|
||||
def test_vpcVmDeploy(self):
|
||||
self.account = UserAccount(apiclient=self.apiClient)
|
||||
self.vpc = DefaultVpc(
|
||||
apiclient=self.apiClient,
|
||||
account=self.account.name,
|
||||
domainid=self.account.domainid,
|
||||
zoneid=get_zone(apiclient=self.apiClient).id
|
||||
)
|
||||
self.vpc |should_not| be(None)
|
||||
ntwk = DefaultVpcNetwork(
|
||||
apiclient=self.apiClient,
|
||||
account=self.account.name,
|
||||
domainid=self.account.domainid,
|
||||
vpcid=self.vpc.id,
|
||||
zoneid=get_zone(apiclient=self.apiClient).id
|
||||
)
|
||||
ntwk | should_not | be(None)
|
||||
ntwk.state | should | equal_to('Allocated')
|
||||
vm = VirtualMachineFactory(
|
||||
apiclient=self.apiClient,
|
||||
account=self.account.name,
|
||||
domainid=self.account.domainid,
|
||||
templateid=get_template(self.apiClient).id,
|
||||
serviceofferingid=get_service_offering(self.apiClient).id,
|
||||
networkid=ntwk.id,
|
||||
zoneid=get_zone(self.apiClient).id
|
||||
)
|
||||
vm | should_not | be(None)
|
||||
vm.state | should | equal_to("Running")
|
||||
|
||||
def tearDown(self):
|
||||
self.account.delete()
|
||||
|
||||
|
||||
@attr(tags='single')
|
||||
class VpcVmFactoryTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.apiClient = cloudstackTestClient(mgtSvr='localhost',
|
||||
logging=logging.getLogger('factory.cloudstack')).getApiClient()
|
||||
|
||||
def test_vpcVmDeployShort(self):
|
||||
self.account = UserAccount(apiclient=self.apiClient)
|
||||
vm = VpcVirtualMachine(
|
||||
apiclient=self.apiClient,
|
||||
account=self.account.name,
|
||||
domainid=self.account.domainid,
|
||||
templateid=get_template(self.apiClient).id,
|
||||
serviceofferingid=get_service_offering(self.apiClient).id,
|
||||
zoneid=get_zone(self.apiClient).id
|
||||
)
|
||||
vm | should_not | be(None)
|
||||
vm.state | should | equal_to("Running")
|
||||
|
||||
def tearDown(self):
|
||||
self.account.delete()
|
||||
|
|
|
|||
Loading…
Reference in New Issue