Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@testtools.skipIf(
not CONF.object_storage_feature_enabled.container_sync,
'Old-style container sync function is disabled')
def test_container_synchronization(self):
def make_headers(cont, cont_client):
# tell first container to synchronize to a second
client_proxy_ip = \
urlparse.urlparse(cont_client.base_url).netloc.split(':')[0]
client_base_url = \
cont_client.base_url.replace(client_proxy_ip,
self.local_ip)
headers = {'X-Container-Sync-Key': 'sync_key',
'X-Container-Sync-To': "%s/%s" %
(client_base_url, str(cont))}
return headers
self._test_container_synchronization(make_headers)
@testtools.skipIf(not dogtag_subcas_enabled,
"dogtag subcas are deprecated")
@depends_on_ca_plugins('dogtag')
def test_try_and_fail_to_add_to_proj_dogtag_subca_that_is_not_mine(self):
self._try_and_fail_to_add_to_proj_subca_that_is_not_mine(
self.get_dogtag_root_ca_ref()
)
@testtools.skipIf(
"nfs" not in CONF.share.enable_ro_access_level_for_protocols,
"RO access rule tests are disabled for NFS protocol.")
def test_create_delete_ro_access_rule(self):
_create_delete_ro_access_rule(self)
@testtools.skipIf(not dogtag_imports_ok, "Dogtag imports not available")
def test_create_stored_key_order_with_non_approved_dogtag_profile(self):
test_model = order_models.OrderModel(**self.stored_key_data)
test_model.meta['container_ref'] = (
self.create_asymmetric_key_container())
test_model.meta['profile'] = "caTPSCert"
test_model.meta['ca_id'] = self.get_dogtag_ca_id()
create_resp, order_ref = self.behaviors.create_order(test_model)
self.assertEqual(202, create_resp.status_code)
self.assertIsNotNone(order_ref)
order_resp = self.wait_for_order(order_ref)
self.verify_pending_waiting_for_ca(order_resp)
@testtools.skipIf(not CONF.share.multitenancy_enabled,
"Only for multitenancy.")
def test_create_share_from_snapshot_share_network_not_provided(self):
# We expect usage of share network from parent's share
# when creating share from snapshot using multitenant driver.
# get parent share
parent = self.shares_client.get_share(self.share["id"])
# create snapshot
snap = self.create_snapshot_wait_for_active(
self.share["id"], cleanup_in_class=False)
# create share from snapshot
child = self.create_share(
self.protocol, snapshot_id=snap["id"], cleanup_in_class=False)
@testtools.skipIf(not dogtag_subcas_enabled,
"dogtag subcas are deprecated")
@depends_on_ca_plugins('dogtag')
def test_fail_to_create_subca_of_dogtag_not_owned_subca(self):
self._fail_to_create_subca_of_not_owned_subca(
self.get_dogtag_root_ca_ref())
@testtools.skipIf(opts is None, "Options not importable")
def setUp(self):
super(OptsTestCase, self).setUp()
@testtools.skipIf(sys.version_info < (2, 7),
"assertRaises as context not supported")
def test_assert_with_custom_message(self):
class A(functional.FunctionalMixin):
def __init__(self):
super(A, self).__init__()
a = A()
custom_message = "A custom message"
assert_message = "Assertion error: .+\\. " + custom_message
a.assertEqual(1, 1, "It's equal")
message = self._catch_exception_message(a.assertEqual,
"a", "b", custom_message)
self.assertRegex(message, assert_message)
a.assertNotEqual(1, 2)
@testtools.skipIf(not CONF.identity.uri_v3,
'v3 auth client not configured')
def test_v3_token(self):
# Can get a token using v3 of the identity API and use that to perform
# an operation on the compute service.
# Doesn't matter which compute API is used,
# picking list_servers because it's easy.
self.servers_v3.list_servers()
@testtools.skipIf(not dogtag_imports_ok, "Dogtag imports not available")
def test_create_simple_cmc_order_with_non_approved_dogtag_profile(self):
test_model = order_models.OrderModel(**self.simple_cmc_data)
test_model.meta['request_data'] = base64.b64encode(
certutil.create_good_csr())
test_model.meta['profile'] = 'caTPSCert'
test_model.meta['ca_id'] = self.get_dogtag_ca_id()
create_resp, order_ref = self.behaviors.create_order(test_model)
self.assertEqual(202, create_resp.status_code)
self.assertIsNotNone(order_ref)
order_resp = self.wait_for_order(order_ref)
self.verify_pending_waiting_for_ca(order_resp)