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_get_possible_subtype_based_on_payload(self):
payload = {
'instanceType': 'compute',
'instanceDetails': {}
}
subtype = cli_util.get_possible_subtype_based_on_payload(oci.core.models.InstanceConfigurationInstanceDetails, 'core', payload)
assert subtype.__class__.__name__ == 'ComputeInstanceDetails'
def _get_stopped(inst):
stopped = oci.core.models.Instance()
stopped.id = inst.id
stopped.lifecycle_state = "STOPPED"
stopped_resp = get_response(200, None, stopped, None)
return stopped_resp
def create_vcn_and_subnet(virtual_network, compartment_id, availability_domain):
vcn_name = 'py_sdk_fs_example_vcn'
cidr_block = "10.0.0.0/16"
vcn_dns_label = 'pysdkfs'
result = virtual_network.create_vcn(
oci.core.models.CreateVcnDetails(
cidr_block=cidr_block,
display_name=vcn_name,
compartment_id=compartment_id,
dns_label=vcn_dns_label
)
)
vcn = oci.wait_until(
virtual_network,
virtual_network.get_vcn(result.data.id),
'lifecycle_state',
'AVAILABLE',
max_wait_seconds=300
).data
subnet_name = 'py_sdk_fs_example_subnet'
subnet_dns_label = 'pyfssub'
def create_ipv6(virtual_network_client, vnic_id, display_name):
create_ipv6_response = virtual_network_client.create_ipv6(
oci.core.models.CreateIpv6Details(
display_name=display_name,
vnic_id=vnic_id
)
).data
return create_ipv6_response
def stop_resource(instance_id, region):
base_compute = oci.core.compute_client.ComputeClient(config)
base_compute.base_client.set_region(region)
# Stop a running instance, but it is possible for it to fail so putting it in a try catch for common errors.
try:
if base_compute.get_instance(instance_id).data.lifecycle_state in 'RUNNING':
try:
print('\t\tStopping instance. Stop response code: {1}'
.format(instance_id, str(base_compute.instance_action(instance_id, 'STOP').status)))
except oci.exceptions.ServiceError as e:
print('\t\tStopping instance failed. {0}' .format(e))
else:
print('\t\tThe instance was in the incorrect state to stop' .format(instance_id))
except oci.exceptions.ServiceError as e:
print('\t\tStopping instance failed. {0}'.format(e))
# We can assign freeform and defined tags at creation time. Freeform tags are a dictionary of string-to-string,
# where the key is the tag name and the value is the tag value.
#
# Defined tags are a dictionary where the key is the tag namespace (string) and the value is another dictionary. In
# this second dictionary, the key is the tag name (string) and the value is the tag value. The tag names have to
# correspond to the name of a tag within the specified namespace (and the namespace must exist).
#
# Resources where we can create/update tags will have the freeform_tags and defined_tags attributes. Consult the API
# documentation to see what these are (https://oracle-cloud-infrastructure-python-sdk.readthedocs.io/en/latest/api/landing.html)
num_tries = 0
while True:
# You may get a 400 if you create/reactivate a tag and try and use it straight away. If you have a delay/sleep between
# creating the tag and then using it (or alternatively retry the 400) that may resolve the issue.
try:
create_vcn_response = virtual_network.create_vcn(
oci.core.models.CreateVcnDetails(
cidr_block='10.0.0.0/16',
compartment_id=compartment_id,
display_name='Python SDK tagging example VCN',
dns_label='vcn{}'.format(random.randint(0, 1000000)),
freeform_tags={'free': 'form', 'another': 'item'},
defined_tags={tag_namespace.name: {tag_one_name: 'hello', tag_two_name: 'world'}}
)
)
vcn_id = create_vcn_response.data.id
vcn_after_wait_response = oci.wait_until(virtual_network, virtual_network.get_vcn(vcn_id), 'lifecycle_state', 'AVAILABLE', max_wait_seconds=300)
print('Created VCN with tags: {}'.format(vcn_after_wait_response.data))
break
except oci.exceptions.ServiceError as e:
if e.status == 400:
print('Retrying on 400: {}'.format(e))
num_tries += 1
def delete_vcn_and_subnets(virtual_network, vcn_and_subnets):
vcn = vcn_and_subnets['vcn']
subnet_one = vcn_and_subnets['subnets'][0]
subnet_two = vcn_and_subnets['subnets'][1]
composite_virtual_network = oci.core.VirtualNetworkClientCompositeOperations(virtual_network)
composite_virtual_network.delete_subnet_and_wait_for_state(
subnet_one.id,
[oci.core.models.Subnet.LIFECYCLE_STATE_TERMINATED]
)
composite_virtual_network.delete_subnet_and_wait_for_state(
subnet_two.id,
[oci.core.models.Subnet.LIFECYCLE_STATE_TERMINATED]
)
composite_virtual_network.delete_vcn_and_wait_for_state(
vcn.id,
[oci.core.models.Vcn.LIFECYCLE_STATE_TERMINATED]
)
def DeleteVolumeGroups(config, Compartments):
AllItems = []
object = oci.core.BlockstorageClient(config)
identity = oci.identity.IdentityClient(config)
print ("Getting all Block Volume Groups objects")
for compartment in Compartments:
ads = identity.list_availability_domains(compartment_id=compartment.id).data
for ad in ads:
items = oci.pagination.list_call_get_all_results(object.list_volume_groups, availability_domain=ad.name, compartment_id=compartment.id).data
for item in items:
if (item.lifecycle_state != "TERMINATED"):
AllItems.append(item)
print("- {} - {}".format(item.display_name, item.lifecycle_state))
itemsPresent = True
while itemsPresent:
count = 0
'TERMINATED',
max_wait_seconds=600,
succeed_on_not_found=True
)
if len(sys.argv) != 4:
raise RuntimeError('This script needs to be provided a compartment ID, an availability domain and an image OCID')
compartment_id = sys.argv[1]
availability_domain = sys.argv[2]
image_id = sys.argv[3]
# Default config file and profile
config = oci.config.from_file()
virtual_network = oci.core.VirtualNetworkClient(config)
compute = oci.core.ComputeClient(config)
vcn_and_subnet = create_vcn_and_subnet(virtual_network, compartment_id, availability_domain)
instance = launch_instance(compute, compartment_id, vcn_and_subnet['subnet'], image_id)
# First we'll create a reserved public IP. This does not have to be assigned to a private IP at create time, although
# it can be by specifying a private_ip_id in CreatePublicIpDetails.
create_public_ip_response = virtual_network.create_public_ip(
oci.core.models.CreatePublicIpDetails(
compartment_id=compartment_id,
display_name='py_sdk_example_res_ip',
lifetime='RESERVED'
)
)
print('Created Public IP: {}'.format(create_public_ip_response.data))
print('\n================================\n')
def delete_vcn_and_subnets(virtual_network_client_composite_ops, vcn_and_subnets):
vcn = vcn_and_subnets['vcn']
subnet_one = vcn_and_subnets['subnets'][0]
subnet_two = vcn_and_subnets['subnets'][1]
virtual_network_client_composite_ops.delete_subnet_and_wait_for_state(
subnet_one.id,
[oci.core.models.Subnet.LIFECYCLE_STATE_TERMINATED]
)
print('Deleted Subnet 1')
virtual_network_client_composite_ops.delete_subnet_and_wait_for_state(
subnet_two.id,
[oci.core.models.Subnet.LIFECYCLE_STATE_TERMINATED]
)
print('Deleted Subnet 2')
virtual_network_client_composite_ops.delete_vcn_and_wait_for_state(
vcn.id,
[oci.core.models.Vcn.LIFECYCLE_STATE_TERMINATED]
)
print('Deleted VCN')