Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_delete_certificate(lb_client, delete_lb_resources_and_wait_patch):
module = get_module(dict())
certificate = get_certificate(dict())
delete_lb_resources_and_wait_patch.return_value = dict(
certificate=to_dict(certificate), changed=True
)
result = oci_load_balancer_certificate.delete_certificate(lb_client, module)
assert result["changed"] is True
def test_create_data_guard_association(db_client, create_and_wait_patch):
module = get_module(dict())
data_guard_association = get_data_guard_association("PRIMARY", "STANDBY")
create_and_wait_patch.return_value = {
"data_guard_association": to_dict(data_guard_association),
"changed": True,
}
result = oci_data_guard_association.create_data_guard_association(db_client, module)
assert result["data_guard_association"]["role"] is data_guard_association.role
def test_delete_file_system(file_storage_client, delete_and_wait_patch):
module = get_module(dict(file_system_id="ocid1.mounttarget.aaa"))
file_system = get_file_system()
delete_and_wait_patch.return_value = {
"file_system": to_dict(file_system),
"changed": True,
}
result = oci_file_system.delete_file_system(file_storage_client, module)
assert result["file_system"]["display_name"] is file_system.display_name
def list_customer_secret_keys(identity_client, user_id, csk_id, module):
try:
customer_secret_keys = oci_utils.list_all_resources(
identity_client.list_customer_secret_keys,
user_id=user_id,
display_name=module.params["display_name"],
)
if csk_id:
return next(
(
to_dict([secret_key])
for secret_key in customer_secret_keys
if secret_key.id == csk_id
),
{},
)
return to_dict(customer_secret_keys)
except ServiceError as ex:
module.fail_json(msg=ex.message)
module, ComputeManagementClient
)
compute_client = oci_utils.create_service_client(module, ComputeClient)
try:
compartment_id = module.params["compartment_id"]
instance_pool_id = module.params["instance_pool_id"]
instance_summaries = to_dict(
oci_utils.list_all_resources(
compute_management_client.list_instance_pool_instances,
compartment_id=compartment_id,
instance_pool_id=instance_pool_id,
)
)
# Get model from summaries returned by `list_instance_pools_instances`
result = to_dict(
[
oci_utils.call_with_backoff(
compute_client.get_instance, instance_id=inst_summ["id"]
).data
for inst_summ in instance_summaries
]
)
except ServiceError as ex:
module.fail_json(msg=ex.message)
module.exit_json(instance_pool_instances=result)
def are_lists_equal(s, t):
if s is None and t is None:
return True
if (s is None and len(t) >= 0) or (t is None and len(s) >= 0) or (len(s) != len(t)):
return False
if len(s) == 0:
return True
s = to_dict(s)
t = to_dict(t)
if type(s[0]) == dict:
# Handle list of dicts. Dictionary returned by the API may have additional keys. For example, a get call on
# service gateway has an attribute `services` which is a list of `ServiceIdResponseDetails`. This has a key
# `service_name` which is not provided in the list of `services` by a user while making an update call; only
# `service_id` is provided by the user in the update call.
sorted_s = sort_list_of_dictionary(s)
sorted_t = sort_list_of_dictionary(t)
for index, d in enumerate(sorted_s):
if not is_dictionary_subset(d, sorted_t[index]):
return False
return True
else:
# Handle lists of primitive types.
try:
def list_swift_passwords(identity_client, user_id, sw_pass_id, module):
try:
swift_passwords = oci_utils.call_with_backoff(
identity_client.list_swift_passwords, user_id=user_id
).data
if sw_pass_id:
return next(
(
to_dict([sw_pass])
for sw_pass in swift_passwords
if sw_pass.id == sw_pass_id
),
{},
)
return to_dict(swift_passwords)
except ServiceError as ex:
module.fail_json(msg=ex.message)
if not HAS_OCI_PY_SDK:
module.fail_json(msg="oci python sdk required for this module.")
virtual_network_client = oci_utils.create_service_client(
module, VirtualNetworkClient
)
service_gateway_id = module.params["service_gateway_id"]
compartment_id = module.params["compartment_id"]
result = []
try:
if service_gateway_id is not None:
result = [
to_dict(
oci_utils.call_with_backoff(
virtual_network_client.get_service_gateway,
service_gateway_id=service_gateway_id,
).data
)
]
else:
optional_list_method_params = ["display_name", "lifecycle_state", "vcn_id"]
optional_kwargs = {
param: module.params[param]
for param in optional_list_method_params
if module.params.get(param) is not None
}
result = to_dict(
oci_utils.list_all_resources(
virtual_network_client.list_service_gateways,
db_client.list_db_versions,
compartment_id=compartment_id,
**optional_kwargs
)
else:
get_logger().debug(
"Listing all DB Versions under Compartment %s", compartment_id
)
db_versions = oci_utils.list_all_resources(
db_client.list_db_versions, compartment_id=compartment_id
)
except ServiceError as ex:
get_logger().error("Unable to list DB Versions due to %s", ex.message)
module.fail_json(msg=ex.message)
result["db_versions"] = to_dict(db_versions)
return result
cpe_id = module.params["cpe_id"]
compartment_id = module.params["compartment_id"]
result = []
try:
if cpe_id is not None:
result = [
to_dict(
oci_utils.call_with_backoff(
virtual_network_client.get_cpe, cpe_id=cpe_id
).data
)
]
else:
result = to_dict(
oci_utils.list_all_resources(
virtual_network_client.list_cpes,
display_name=module.params["display_name"],
compartment_id=compartment_id,
)
)
except ServiceError as ex:
module.fail_json(msg=ex.message)
module.exit_json(cpes=result)