KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _get_instance_config method in molecule

Best Python code snippet using molecule_python

snomed_buildserver.py

Source:snomed_buildserver.py

...11from oauth2client.client import GoogleCredentials12from googleapiclient import errors13from deployment import settings14from deployment.settings import LOGGER15def _get_instance_config(key):16    """17    Retrieve the Google Cloud instance config.18    Instance config file looks like this:19    {20     'instance_name': 'foo',21     'ip_address': '130.211.100.33'22    }23    """24    if os.path.isfile(settings.BUILDSERVER_CONFIG_FILE):25        with open(settings.BUILDSERVER_CONFIG_FILE) as build_config_file:26            instance_config = json.load(build_config_file)27            if key in ["instance_name", "ip_address"]:28                return instance_config[key]29    return None30def _fail_loudly(sarge_obj):31    """32    Throw an exit(1) error when the return code from sarge runs command is33    not zero. Delete buildserver instance to prevent resource wastage.34    """35    buildserver_exists = run("which buildserver").returncode36    os.chdir(settings.BASE_DIR)37    if buildserver_exists == 1:38        delete_command = "{} delete --instance-name {}".format(39            "build/buildserver", _get_instance_config("instance_name")40        )41    elif buildserver_exists == 0:42        delete_command = "{} delete --instance-name {}".format(43            "buildserver", _get_instance_config("instance_name")44        )45    if not settings.DELETE_ON_FAILED_BUILD:46        LOGGER.debug(47            "DELETE_ON_FAILED_BUILD is set to False in settings."48            "This can cause expensive build servers to linger on."49        )50    if sarge_obj.returncode:51        if settings.DELETE_ON_FAILED_BUILD:52            LOGGER.debug(53                "Deploy failed. Deleting instance to preserve future runs."54            )55            run(delete_command)56            sys.exit(1)57    else:58        if settings.DELETE_ON_SUCCESSFUL_BUILD:59            run(delete_command)60        LOGGER.debug(61            """62            Build successful!63            The full contents of SNOMED CT UK64            Clinical Release and Drug Extension65            are now available on Google Cloud Storage."""66        )67        sys.exit(0)68def _get_default_instance_name():69    """Return an name from configuration OR a random 'heroku style' name."""70    instance_name = _get_instance_config("instance_name")71    if instance_name:72        return instance_name73    command_return_val = get_stdout(74        "{}/commands/get_name".format(settings.BASE_DIR)75    )76    return command_return_val.strip()77def call_ansible(server, playbook, extra_vars):78    """Deploy the SNOMED build server."""79    deployment_dir = os.path.join(settings.BASE_DIR, "deployment")80    os.chdir(deployment_dir)81    env = {82        "ANSIBLE_CONFIG": "{}/deployment/ansible.cfg".format(settings.BASE_DIR)83    }84    ansible_command = """85    ansible-playbook -i'{server},' {playbook} --extra-vars='{extra_vars}'86    --vault-id {home}/.vaultpass -vvvv""".format(87        playbook=playbook,88        extra_vars=extra_vars,89        server=server.strip(),90        home=settings.HOME,91    )92    LOGGER.debug(ansible_command)93    retval = run(ansible_command, env=env)94    _fail_loudly(retval)95@click.group()96def instance():97    """Group commands to create and interact with the SNOMED build instance."""98    pass99def create_instance(compute, name, zone, project):100    """101    Add a new instance to a gcloud project and zone.102    This instance is a custom build, with:103    - 4 cores104    - 16GB Ram105    - 512GB SSD106    A list of the available source disk images can be obtained using:107    `gcloud compute images list --uri`108    """109    # use an Ubuntu 18.04 base image110    source_disk_image = (111        "projects/ubuntu-os-cloud/global/images/ubuntu-1804-bionic-v20181222"112    )113    # use an SSD, for performance114    source_disk_type = "projects/%s/zones/%s/diskTypes/pd-ssd" % (115        project,116        zone,117    )118    # use a custom machine shape, with 4 cores and 16GB RAM119    machine_type = "zones/%s/machineTypes/custom-4-16384" % zone120    # custom instance config121    config = {122        "name": name,123        "machineType": machine_type,124        "scheduling": {"preemptible": True},125        "disks": [126            {127                "boot": True,128                "autoDelete": True,129                "initializeParams": {130                    "sourceImage": source_disk_image,131                    "diskType": source_disk_type,132                    "diskSizeGb": 512,133                },134            }135        ],136        "tags": {"items": ["http-server", "https-server"]},137        # Specify a network interface with NAT to access the public138        # internet.139        "networkInterfaces": [140            {141                "network": "global/networks/default",142                "accessConfigs": [143                    {"type": "ONE_TO_ONE_NAT", "name": "External NAT"}144                ],145            }146        ],147        # Allow the instance to access cloud storage and logging.148        "serviceAccounts": [149            {150                "email": "default",151                "scopes": [152                    "https://www.googleapis.com/auth/devstorage.read_write",153                    "https://www.googleapis.com/auth/logging.write",154                ],155            }156        ],157    }158    instances = compute.instances()159    LOGGER.debug(compute.instances())160    LOGGER.debug(zone)161    return instances.insert(project=project, zone=zone, body=config).execute()162def delete_instance(compute, project, zone, name):163    """Remove a Google Compute Engine instance."""164    return (165        compute.instances()166        .delete(project=project, zone=zone, instance=name)167        .execute()168    )169def wait_for_operation(compute, project, zone, operation):170    LOGGER.debug("Waiting for operation to finish...")171    while True:172        result = (173            compute.zoneOperations()174            .get(project=project, zone=zone, operation=operation)175            .execute()176        )177        if result["status"] == "DONE":178            LOGGER.debug("Done.")179            if "error" in result:180                raise Exception(result["error"])181            return result182        time.sleep(1)183@instance.command()184@click.option(185    "--instance-name",186    help="The name of the instance to be created",187    default=_get_default_instance_name(),188)189@click.option(190    "--zone",191    help="The GCP zone where the machine will be hosted",192    default="europe-west1-d",193)194def create(instance_name, zone):195    """Creates a preemptible instance optimized for a SNOMED CT Full build."""196    project = settings.PROJECT197    LOGGER.debug("Creating instance.")198    operation = create_instance(settings.COMPUTE, instance_name, zone, project)199    wait_for_operation(settings.COMPUTE, project, zone, operation["name"])200    unicode_ip = (201        settings.COMPUTE.instances()202        .get(project=project, zone=zone, instance=instance_name)203        .execute()["networkInterfaces"][0]["accessConfigs"][0]["natIP"]204    )205    call("gcloud compute config-ssh", shell=True)206    ip_address = str(unicode_ip)207    instance_details = {208        "instance_name": instance_name,209        "ip_address": ip_address,210    }211    click.echo(json.dumps(instance_details, sort_keys=True, indent=4))212    with open(settings.BUILDSERVER_CONFIG_FILE, "w") as config_file:213        config_file.write(json.dumps(instance_details))214@instance.command()215@click.option(216    "--instance-name",217    help="The name of the instance to be deleted",218    default=_get_instance_config("instance_name"),219)220@click.option(221    "--zone",222    help="The GCP zone where the machine is currently hosted",223    default="europe-west1-d",224)225def delete(instance_name, zone):226    """Deletes the currently active build server.227    Provide the flag --instance-name with the short-form name of your instance,228    in order to delete a machine created under another name.229    """230    LOGGER.debug("Terminating instance.")231    operation = delete_instance(232        settings.COMPUTE, settings.PROJECT, zone, instance_name233    )234    wait_for_operation(235        settings.COMPUTE, settings.PROJECT, zone, operation["name"]236    )237@instance.command()238@click.option(239    "--server",240    default="{}.europe-west1-d.savannah-emr".format(241        _get_instance_config("instance_name")242    ),243    help="The hostname of the server you're deploying to.",244)245@click.option(246    "--buildserver-version",247    help="The version of the BuildServer you'd like to deploy.",248    default=settings.BUILDSERVER_VERSION,249)250def deploy(server, buildserver_version):251    """Deploys the buildserver code and runs a full SNOMED CT build."""252    extra_vars = {253        "ansible_host": server,254        "buildserver_version": buildserver_version,255        "ansible_ssh_user": settings.SSH_USER,...

driver.py

Source:driver.py

...29    def default_ssh_connection_options(self):30        return self._get_ssh_connection_options()31    def login_options(self, instance_name):32        d = {"instance": instance_name}33        return util.merge_dicts(d, self._get_instance_config(instance_name))34    def ansible_connection_options(self, instance_name):35        try:36            d = self._get_instance_config(instance_name)37            return {38                "ansible_user": d["user"],39                "ansible_host": d["address"],40                "ansible_port": d["port"],41                "ansible_private_key_file": d["identity_file"],42                "connection": "ssh",43                "ansible_ssh_common_args": " ".join(self.ssh_connection_options),44            }45        except StopIteration:46            return {}47        except IOError:48            # Instance has yet to be provisioned , therefore the49            # instance_config is not on disk.50            return {}51    def _get_instance_config(self, instance_name):52        instance_config_dict = util.safe_load_file(self._config.driver.instance_config)53        return next(54            item for item in instance_config_dict if item["instance"] == instance_name55        )56    def template_dir(self):57        """Return path to its own cookiecutterm templates. It is used by init58        command in order to figure out where to load the templates from.59        """...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.