Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@ddt.data(
(requests.delete, client._call_requests_delete),
(requests.delete, client.delete),
(requests.get, client._call_requests_get),
(requests.get, client.get),
(requests.head, client._call_requests_head),
(requests.head, client.head),
(requests.patch, client._call_requests_patch),
(requests.patch, client.patch),
(requests.post, client._call_requests_post),
(requests.post, client.post),
(requests.put, client._call_requests_put),
(requests.put, client.put),
(requests.request, client._call_requests_request),
(requests.request, client.request),
)
def test_identical_args(self, functions):
@ddt.data(fakes.MoverTestData().interface_name1,
fakes.MoverTestData().long_interface_name)
def test_get_mover_interface(self, interface_name):
self.hook.append(self.mover.resp_get_ref_succeed())
self.hook.append(self.mover.resp_get_succeed())
self.hook.append(self.mover.resp_get_succeed())
xml_connector = self.mover_interface_manager.xml_connector
xml_connector.post = utils.EMCMock(side_effect=self.hook)
interface = self.mover_interface_manager.get(
name=interface_name,
mover_name=self.mover.mover_name)
property_map = [
'name',
'mover_name',
'device',
@ddt.data((reporters.HTMLReporter, False),
(reporters.HTMLStaticReporter, True))
@ddt.unpack
def test_generate(self, cls, include_libs, mock_dumps, mock_ui_utils):
mock_render = mock_ui_utils.get_template.return_value.render
reporter = cls(get_verifications(), None)
self.assertEqual({"print": mock_render.return_value},
reporter.generate())
mock_render.assert_called_once_with(data=mock_dumps.return_value,
include_libs=include_libs)
mock_ui_utils.get_template.assert_called_once_with(
"verification/report.html")
self.assertEqual(1, mock_dumps.call_count)
args, kwargs = mock_dumps.call_args
@data('https://github.com/werwolfby/monitorrent/releases/tag/1.0.2',
'https://github.com/werwolfby/monitorrent/releases/tag/1.0.1',
'https://github.com/werwolfby/monitorrent/releases/tag/1.0.0')
def test_get_url(self, url):
new_version_checker = NewVersionChecker(Mock(), False)
new_version_resource = NewVersion(new_version_checker)
new_version_checker.new_version_url = url
self.api.add_route('/api/new_version', new_version_resource)
body = self.simulate_request("/api/new_version", decode='utf-8')
self.assertEqual(self.srmock.status, falcon.HTTP_OK)
self.assertTrue('application/json' in self.srmock.headers_dict['Content-Type'])
result = json.loads(body)
self.assertEqual(result, {'url': url})
@ddt.data(
{"new_size": 5},
{"new_size": 10}
)
def test__extend_share(self, new_size):
fake_share = mock.MagicMock()
self.scenario._extend_share(fake_share, new_size)
fake_share.extend.assert_called_with(new_size)
self.mock_wait_for_status.mock.assert_called_once_with(
fake_share,
ready_statuses=["available"],
update_resource=self.mock_get_from_manager.mock.return_value,
timeout=300, check_interval=3)
self.mock_get_from_manager.mock.assert_called_once_with()
@data(('off', True), ('json', True), ('html', True), ('foo', False))
def test_report(self, item):
"""Testing missing mandatory parameter."""
try:
options = {'definition': 'fake.yml'}
options.update({'report': item[0]})
ApplicationOptions(**options)
if not item[1]:
self.assertFalse("RuntimeError expected")
except RuntimeError as exception:
if item[1]:
self.assertFalse("Unexpected exception %s" % exception)
@ddt.data(({"max": 0, "min_iterations": 5, "sigmas": 2.5}, True),
({"max": -1}, False),
({"max": 0, "min_iterations": 2}, False),
({"max": 0, "sigmas": 0}, False),
({"foo": "bar"}, False))
@ddt.unpack
def test_validate(self, config, valid):
results = sla.SLA.validate("outliers", None, None, config)
if valid:
self.assertEqual([], results)
else:
self.assertEqual(1, len(results))
@data(ROOT_DOMAIN_ADMIN, CHILD_DOMAIN_ADMIN)
@attr(tags=["advanced"], required_hardware="true")
def test_02_create_template_snapshot(self, value):
"""Test create snapshot and templates from volume
# Validate the following
1. Create root domain/child domain admin account
2. Deploy VM in the account
3. Create snapshot from the virtual machine root volume
4. Create template from the snapshot
5. Verify that the secondary storage count of the account equals
the size of the template"""
response = self.setupAccount(value)
self.assertEqual(response[0], PASS, response[1])
self.virtualMachine = VirtualMachine.create(self.api_client, self.services["virtual_machine"],
@data("AUTHORIZED", "DENIED", "FAILED")
def test_authorization_response_response_jwe_response_context_fraud(
self,
resp_type):
self.json_loads_patch.return_value['type'] = resp_type
self.json_loads_patch.return_value['reason'] = "FRAUDULENT"
response = AuthorizationResponse(self.data, self.transport)
self.assertTrue(response.fraud, True)
@data(ISOLATED_NETWORK, SHARED_NETWORK, VPC_NETWORK)
@attr(tags=["advanced", "dvs"])
def test_add_static_nat_rule(self, value):
""" Add secondary IP to NIC of a VM"""
# Steps:
# 1. Create Account and create network in it (isoalted/ shared/ vpc)
# 2. Deploy a VM in this network and account
# 3. Add 2 secondary IPs to the default nic of VM
# 4. Acquire public IP, open firewall for it, and
# create static NAT rule for this public IP to the 1st secondary IP
# 5. Repeat step 4 for another public IP
# 6. Repeat step 4 for 2nd secondary IP
# 7. Repeat step 4 for invalid secondary IP
# 8. Try to remove 1st secondary IP (with active static nat rule)
# Validations: