Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@attr('config-disruptive')
def test_009_import_config_from_remote():
if remote_import is None:
skipped("Arguments missing in config file")(
test_009_import_config_from_remote)()
import_config_remote(handle, remote_import['file_dir'],
remote_import['file_name'],
hostname=remote_import['hostname'],
protocol=remote_import['protocol'],
username=remote_import['username'],
password=remote_import['password'])
@attr(hypervisor="xenserver")
@attr(tags=["advanced", "basic"], required_hardware="true")
def test_01_VRServiceFailureAlerting(self):
if self.zone.networktype == "Basic":
list_router_response = list_routers(
self.apiclient,
listall="true"
)
else:
list_router_response = list_routers(
self.apiclient,
account=self.account.name,
domainid=self.account.domainid
)
self.assertEqual(
isinstance(list_router_response, list),
@attr(tags=["advanced", "advancedns", "simulator"], required_hardware="false")
def test_02_pods(self):
"""Check the status of pods"""
# Validate the following
# 1. List pods
# 2. Check allocation state is "enabled" or not
pods = Pod.list(
self.apiclient,
zoneid=self.zone.id,
listall=True
)
self.assertEqual(
isinstance(pods, list),
True,
"Check if listPods returns a valid response"
@attr(operation='list')
@attr(assertion='non-empty public buckets via s3website return page for /, where page is private, missing errordoc')
@attr('s3website')
@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
def test_website_public_bucket_list_private_index_missingerrordoc():
bucket = get_new_bucket()
f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDocErrorDoc'])
bucket.make_public()
indexhtml = bucket.new_key(f['IndexDocument_Suffix'])
indexstring = choose_bucket_prefix(template=INDEXDOC_TEMPLATE, max_len=256)
indexhtml.set_contents_from_string(indexstring)
indexhtml.set_canned_acl('private')
#time.sleep(1)
while bucket.get_key(f['IndexDocument_Suffix']) is None:
time.sleep(SLEEP_INTERVAL)
res = _website_request(bucket.name, '')
@attr(operation='create w/empty user agent')
@attr(assertion='succeeds')
@nose.with_setup(teardown=_clear_custom_headers)
def test_object_create_bad_ua_empty():
key = _setup_bad_object({'User-Agent': ''})
key.set_contents_from_string('bar')
@attr(hypervisor="xenserver")
@attr(tags=["advanced", "basic"], required_hardware="false")
def test_01_reset_vm_on_reboot(self):
#TODO: SIMENH: add new test to check volume contents
"""Test reset virtual machine on reboot
"""
# Validate the following
# create vm and list the volume for that VM. Reboot vm and check if root volume is different as before.
volumelist_before_reboot = Volume.list(
self.apiclient,
virtualmachineid=self.virtual_machine.id,
type='ROOT',
listall=True
)
self.assertNotEqual(
@attr(all=True, integration=False)
def test3_delete_software(self):
"""
(TestBoM) delete_software returns 0 on success
"""
self.assertEqual(self.DUT.request_bom(self._dao, 1)[1], 0)
(_results,
_error_code) = self.DUT.delete_software(self.DUT._last_id)
self.assertTrue(_results)
self.assertEqual(_error_code, 0)
@attr(operation='list')
@attr(assertion='x-amz-website-redirect-location should fire websiteconf, absolute, private key')
@attr('s3website')
@attr('x-amz-website-redirect-location')
@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
def test_website_xredirect_private_abs():
bucket = get_new_bucket()
f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDoc'])
bucket.make_public()
k = bucket.new_key('page')
content = 'wrong-content'
redirect_dest = 'http://example.com/foo'
headers = {'x-amz-website-redirect-location': redirect_dest}
k.set_contents_from_string(content, headers=headers, policy='private')
redirect = k.get_redirect()
eq(k.get_redirect(), redirect_dest)
@attr(
tags=[
"advanced",
"eip",
"advancedns",
"basic",
"sg"],
required_hardware="false")
def test_upload_attach_volume(self):
"""Test Upload volume and attach to VM in stopped state
"""
# Validate the following
# 1. Upload the volume using uploadVolume API call
# 2. Deploy VM with startvm=false.
# 3. Attach the volume to the deployed VM in step 2
@attr(all=True, integration=True)
def test3_delete_hardware(self):
"""
(TestBoM) delete_hardware returns 0 on success
"""
self.assertEqual(self.DUT.request_bom(0)[1], 0)
self.DUT.add_hardware(0, 0, 0)
(_results,
_error_code) = self.DUT.delete_hardware(self.DUT._last_id)
self.assertTrue(_results)
self.assertEqual(_error_code, 0)