blob: 8c7cdd1c6001f12ce06bc1b64c4ea375b303d8a3 [file] [log] [blame]
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
#Import Local Modules
from marvin.cloudstackTestCase import cloudstackTestCase
from marvin.cloudstackAPI import (createVolume,
createTemplate)
from marvin.lib.base import (Volume,
Iso,
VirtualMachine,
Template,
Snapshot,
SecurityGroup,
Account,
Zone,
Network,
NetworkOffering,
DiskOffering,
ServiceOffering,
VmSnapshot,
SnapshotPolicy,
SSHKeyPair,
Resources,
Configurations,
VpnCustomerGateway,
Hypervisor,
VpcOffering,
VPC,
NetworkACL)
from marvin.lib.common import (get_zone,
get_domain,
get_template,
list_os_types)
from marvin.lib.utils import (validateList,
cleanup_resources,
random_gen)
from marvin.codes import (PASS, FAIL, EMPTY_LIST)
from nose.plugins.attrib import attr
import time
class TestVpnCustomerGateways(cloudstackTestCase):
@classmethod
def setUpClass(cls):
try:
cls._cleanup = []
cls.testClient = super(TestVpnCustomerGateways, cls).getClsTestClient()
cls.api_client = cls.testClient.getApiClient()
cls.services = cls.testClient.getParsedTestDataConfig()
# Get Domain, Zone, Template
cls.domain = get_domain(cls.api_client)
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
cls.template = get_template(
cls.api_client,
cls.zone.id,
cls.services["ostype"]
)
cls.services['mode'] = cls.zone.networktype
cls.account = Account.create(
cls.api_client,
cls.services["account"],
domainid=cls.domain.id
)
# Getting authentication for user in newly created Account
cls.user = cls.account.user[0]
cls.userapiclient = cls.testClient.getUserApiClient(cls.user.username, cls.domain.name)
cls._cleanup.append(cls.account)
except Exception as e:
cls.tearDownClass()
raise Exception("Warning: Exception in setup : %s" % e)
return
def setUp(self):
self.apiClient = self.testClient.getApiClient()
self.cleanup = []
def tearDown(self):
#Clean up, terminate the created resources
cleanup_resources(self.apiClient, self.cleanup)
return
@classmethod
def tearDownClass(cls):
try:
cleanup_resources(cls.api_client, cls._cleanup)
except Exception as e:
raise Exception("Warning: Exception during cleanup : %s" % e)
return
def __verify_values(self, expected_vals, actual_vals):
"""
@Desc: Function to verify expected and actual values
@Steps:
Step1: Initializing return flag to True
Step1: Verifying length of expected and actual dictionaries is matching.
If not matching returning false
Step2: Listing all the keys from expected dictionary
Step3: Looping through each key from step2 and verifying expected and actual dictionaries have same value
If not making return flag to False
Step4: returning the return flag after all the values are verified
"""
return_flag = True
if len(expected_vals) != len(actual_vals):
return False
keys = list(expected_vals.keys())
for i in range(0, len(expected_vals)):
exp_val = expected_vals[keys[i]]
act_val = actual_vals[keys[i]]
if exp_val == act_val:
return_flag = return_flag and True
else:
return_flag = return_flag and False
self.debug("expected Value: %s, is not matching with actual value: %s" % (
exp_val,
act_val
))
return return_flag
@attr(tags=["advanced", "basic"], required_hardware="true")
def test_01_list_vpncustomergateways_pagination(self):
"""
@Desc: Test to List VPN Customer Gateways pagination
@steps:
Step1: Listing all the VPN Customer Gateways for a user
Step2: Verifying that no VPN Customer Gateways are listed
Step3: Creating (page size + 1) number of VPN Customer Gateways
Step4: Listing all the VPN Customer Gateways again for a user
Step5: Verifying that list size is (page size + 1)
Step6: Listing all the VPN Customer Gateways in page1
Step7: Verifying that list size is (page size)
Step8: Listing all the VPN Customer Gateways in page2
Step9: Verifying that list size is 1
Step10: Deleting the VPN Customer Gateways present in page 2
Step11: Listing all the VPN Customer Gateways in page2
Step12: Verifying that no VPN Customer Gateways are listed
"""
# Listing all the VPN Customer Gateways for a User
list_vpncustomergateways_before = VpnCustomerGateway.list(
self.userapiclient,
listall=self.services["listall"]
)
# Verifying that no VPN Customer Gateways are listed
self.assertIsNone(
list_vpncustomergateways_before,
"VPN Customer Gateways listed for newly created User"
)
# Creating pagesize + 1 number of VPN Customer Gateways
for i in range(0, (self.services["pagesize"] + 1)):
vpncustomergateway_created = VpnCustomerGateway.create(
self.userapiclient,
self.services["vpncustomergateway"],
name="VPNCustGateway"+str(i+1),
gateway="10.102.153." + str(i+1),
cidrlist="10.0.0.0/24",
account=self.account.name,
domainid=self.domain.id
)
self.assertIsNotNone(
vpncustomergateway_created,
"VPN Customer Gateway creation failed"
)
if (i < self.services["pagesize"] + 1):
self.cleanup.append(vpncustomergateway_created)
# Listing all the VPN Customer Gateways for a User
list_vpncustomergateways_after = VpnCustomerGateway.list(
self.userapiclient,
listall=self.services["listall"]
)
status = validateList(list_vpncustomergateways_after)
self.assertEqual(
PASS,
status[0],
"VPN Customer Gateway creation failed"
)
# Verifying that list size is pagesize + 1
self.assertEqual(
self.services["pagesize"] + 1,
len(list_vpncustomergateways_after),
"Failed to create pagesize + 1 number of VPN Customer Gateways"
)
# Listing all the VPN Customer Gateways in page 1
list_vpncustomergateways_page1 = VpnCustomerGateway.list(
self.userapiclient,
listall=self.services["listall"],
page=1,
pagesize=self.services["pagesize"]
)
status = validateList(list_vpncustomergateways_page1)
self.assertEqual(
PASS,
status[0],
"Failed to list VPN Customer Gateways in page 1"
)
# Verifying the list size to be equal to pagesize
self.assertEqual(
self.services["pagesize"],
len(list_vpncustomergateways_page1),
"Size of VPN Customer Gateways in page 1 is not matching"
)
# Listing all the VPN Customer Gateways in page 2
list_vpncustomergateways_page2 = VpnCustomerGateway.list(
self.userapiclient,
listall=self.services["listall"],
page=2,
pagesize=self.services["pagesize"]
)
status = validateList(list_vpncustomergateways_page2)
self.assertEqual(
PASS,
status[0],
"Failed to list VPN Customer Gateways in page 2"
)
# Verifying the list size to be equal to 1
self.assertEqual(
1,
len(list_vpncustomergateways_page2),
"Size of VPN Customer Gateways in page 2 is not matching"
)
# Deleting the VPM Customer Gateway present in page 2
VpnCustomerGateway.delete(
vpncustomergateway_created,
self.userapiclient
)
self.cleanup.remove(vpncustomergateway_created)
# Listing all the VPN Customer Gateways in page 2 again
list_vpncustomergateways_page2 = VpnCustomerGateway.list(
self.userapiclient,
listall=self.services["listall"],
page=2,
pagesize=self.services["pagesize"]
)
# Verifying that there are no VPN Customer Gateways listed
self.assertIsNone(
list_vpncustomergateways_page2,
"VPN Customer Gateways not deleted from page 2"
)
return
@attr(tags=["advanced", "basic"], required_hardware="true")
def test_02_update_vpncustomergateways(self):
"""
@Desc: Test to update VPN Customer Gateways pagination
@steps:
Step1: Listing all the VPN Customer Gateways for a user
Step2: Verifying that no VPN Customer Gateways are listed
Step3: Creating a VPN Customer Gateways
Step4: Listing all the VPN Customer Gateways again for a user
Step5: Verifying that list size is 1
Step6: Updating the VPN Customer Gateways created in step3
Step7: Listing the VPN customer gateway by id
Step8: Verifying that list size is 1
Step9: Verifying the details of the listed VPN customer gateway are same as updated in step6
"""
# Listing all the VPN Customer Gateways for a User
list_vpncustomergateways_before = VpnCustomerGateway.list(
self.userapiclient,
listall=self.services["listall"]
)
# Verifying that no VPN Customer Gateways are listed
self.assertIsNone(
list_vpncustomergateways_before,
"VPN Customer Gateways listed for newly created User"
)
# Creating A VPN Customer Gateways
vpncustomergateway_created = VpnCustomerGateway.create(
self.userapiclient,
self.services["vpncustomergateway"],
name="VPNCustGateway",
gateway="10.102.153.90",
cidrlist="10.0.0.0/24",
account=self.account.name,
domainid=self.domain.id
)
self.assertIsNotNone(
vpncustomergateway_created,
"VPN Customer Gateway creation failed"
)
self.cleanup.append(vpncustomergateway_created)
# Listing all the VPN Customer Gateways for a User
list_vpncustomergateways_after = VpnCustomerGateway.list(
self.userapiclient,
listall=self.services["listall"]
)
status = validateList(list_vpncustomergateways_after)
self.assertEqual(
PASS,
status[0],
"VPN Customer Gateway creation failed"
)
# Verifying that list size is 1
self.assertEqual(
1,
len(list_vpncustomergateways_after),
"Failed to create VPN Customer Gateways"
)
# Updating the VPN Customer gateway
vpncustomergateway_updated = VpnCustomerGateway.update(
vpncustomergateway_created,
self.userapiclient,
self.services["vpncustomergateway"],
name="NewVPNCustGateway",
gateway="10.102.153.90",
cidrlist="10.0.0.0/24",
)
self.assertIsNotNone(
vpncustomergateway_updated,
"Updation of VPN Customer Gateway failed"
)
# Listing the VPN Customer Gateways by Id
list_vpncustomergateway = VpnCustomerGateway.list(
self.userapiclient,
listall=self.services["listall"],
id=vpncustomergateway_created.id
)
status = validateList(list_vpncustomergateway)
self.assertEqual(
PASS,
status[0],
"Failed to list VPN Customer Gateways by Id"
)
# Verifying the list size to be equal to 1
self.assertEqual(
1,
len(list_vpncustomergateway),
"Size of VPN Customer Gateways by id is not matching"
)
# Verifying the details of the listed VPN Customer Gateway are same as updated
#Creating expected and actual values dictionaries
expected_dict = {
"name":vpncustomergateway_updated.name,
"id":vpncustomergateway_updated.id,
"account":vpncustomergateway_updated.account,
"domainid":vpncustomergateway_updated.domainid,
"gateway":vpncustomergateway_updated.gateway,
"cidrlist":vpncustomergateway_updated.cidrlist,
"seckey":vpncustomergateway_updated.ipsecpsk,
"ikepolicy":vpncustomergateway_updated.ikepolicy,
"ikelifetime":vpncustomergateway_updated.ikelifetime,
"esppolicy":vpncustomergateway_updated.esppolicy,
"esplifetime":vpncustomergateway_updated.esplifetime,
}
actual_dict = {
"name":list_vpncustomergateway[0].name,
"id":list_vpncustomergateway[0].id,
"account":list_vpncustomergateway[0].account,
"domainid":list_vpncustomergateway[0].domainid,
"gateway":list_vpncustomergateway[0].gateway,
"cidrlist":list_vpncustomergateway[0].cidrlist,
"seckey":list_vpncustomergateway[0].ipsecpsk,
"ikepolicy":list_vpncustomergateway[0].ikepolicy,
"ikelifetime":list_vpncustomergateway[0].ikelifetime,
"esppolicy":list_vpncustomergateway[0].esppolicy,
"esplifetime":list_vpncustomergateway[0].esplifetime,
}
vpncustomergateway_status = self.__verify_values(
expected_dict,
actual_dict
)
self.assertEqual(
True,
vpncustomergateway_status,
"Listed VPN Customer Gateway details are not as Updated"
)
return