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

How to use _is_installed method in lisa

Best Python code snippet using lisa_python

remove_v_5_1.py

Source:remove_v_5_1.py

...126    write_to_file(stream.getvalue(), yaml_path)127def _is_premium_installed():128    installed = run(['rpm', '-q', 'cloudify-premium'], ignore_failures=True)129    return installed.returncode == 0130def _is_installed(config, service):131    return service in config['services_to_install']132def _get_components(config):133    _components = []134    if _is_installed(config, 'database_service'):135        _components.append('postgresqlserver')136    if _is_installed(config, 'queue_service'):137        _components.append('rabbitmq')138    if _is_installed(config, 'manager_service'):139        _components += [140            'manager',141            'postgresqlclient',142            'restservice',143            'manageripsetter',144            'nginx',145            'cli',146            'amqppostgres',147            'mgmtworker',148            'stage',149        ]150        if (_is_premium_installed()151                and not config.get('composer', {}).get('skip_installation')):152            _components.append('composer')153        _components.append('usagecollector')154        if not config.get('sanity', {}).get('skip_sanity'):155            _components.append('sanity')156    if _is_installed(config, 'monitoring_service'):157        _components.append('prometheus')158        if not _is_installed(config, 'manager_service'):159            _components.append('nginx')160    if _is_installed(config, 'entropy_service'):161        _components.append('haveged')162    return _components163def _get_packages(config):164    """Yum packages to install/uninstall, based on the current config"""165    packages = []166    # Adding premium components on all, even if we're on community, because167    # yum will return 0 (success) if any packages install successfully even if168    # some of the specified packages don't exist.169    if _is_installed(config, 'manager_service'):170        packages += SOURCES['manager']171        # Premium components172        packages += SOURCES['manager_cluster'] + SOURCES['manager_premium']173    if _is_installed(config, 'database_service'):174        packages += SOURCES['db']175        # Premium components176        packages += SOURCES['db_cluster']177    if _is_installed(config, 'queue_service'):178        packages += SOURCES['queue']179        # Premium components180        packages += SOURCES['queue_cluster']181    if _is_installed(config, 'monitoring_service'):182        packages += SOURCES['prometheus']183        # Premium components184        packages += SOURCES['prometheus_cluster']185    if _is_installed(config, 'entropy_service'):186        packages += SOURCES['haveged']187    return packages188def create_installation_files(config_path):189    logger.info('Creating necessary files for Cloudify removal')190    config = read_yaml_file(config_path)191    for service_name in config['services_to_install']:192        touch(join(INITIAL_INSTALL_DIR, service_name))193        if (service_name in194                ['database_service', 'queue_service', 'manager_service']):195            update_yaml_file(INSTALLED_COMPONENTS_FILE,196                             {service_name: _get_components(config)})197            update_yaml_file(INSTALLED_PACKAGES,198                             {service_name: _get_packages(config)})199def remove_cloudify(config_path, verbose):...

flows.py

Source:flows.py

1"""Flows and submodules controller"""23# Std4import os5import re67# Core8from core.logger import Logger9import flows101112class Flow(object):13    def __init__(self, name=None, url=None) -> None:14        self._name = name15        self._url = url16        self._is_installed = False17        self._flow_dir_path = os.path.join(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "flows"), self._name)18        if self._check_installation():19            self._grep_commands()20            self._grep_configs()21            self._grep_modules()22            self._grep_projects()23            self._grep_templates()2425    def _check_installation(self) -> bool:26        if self._name == "common":27            self._is_installed = True28        else:29            if os.path.isdir(self._flow_dir_path) and len(os.listdir(self._flow_dir_path)) > 0:30                self._is_installed = True31            else:32                self._is_installed = False33        return self._is_installed3435    def _grep_commands(self):36        commands_dir_path = os.path.join(self._flow_dir_path, "commands")37        if os.path.isdir(commands_dir_path) and len(os.listdir(commands_dir_path)) > 0:38            pass3940    def _grep_configs(self):41        pass4243    def _grep_modules(self):44        pass4546    def _grep_projects(self):47        pass4849    def _grep_templates(self):50        pass5152    def is_installed(self):53        return self._is_installed5455    def get_name(self):56        return self._name5758    def get_path(self):59        return self._flow_dir_path6061    def get_info_tuple(self):62        return self._name, self._url, self._is_installed6364    def get_info(self) -> str:65        return "Name: {}, URL: {}, Installed: {}".format(self._name, self._url, "yes" if self._is_installed else "no")666768class Flows(object):69    def __init__(self) -> None:70        self._root_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))71        self._flows = []72        self._flows.append(Flow(name="common", url="https://code.onsemi.com/scm/qcsav/odin.git"))73        self._init_flows()7475    def _init_flows(self):76        gitmodules_file_path = os.path.join(self._root_dir, ".gitmodules")77        if os.path.isfile(gitmodules_file_path):78            gitmodules_file = open(gitmodules_file_path, 'r')79            flow_name = None80            flow_url = None81            for index, line in enumerate(gitmodules_file.readlines(), 0):82                if line.strip().startswith("[") and flow_name is None:83                    flow_name = re.findall(r"\s*\[\s*submodule\s*\"flows/([A-Za-z0-9_]+)\"\s*\]", line)84                    if len(flow_name) == 1:85                        flow_name = flow_name[0]86                    else:87                        Logger.fatal("Parsing error: \".gitmodules\":{}: {}".format(index, line))88                if line.strip().startswith("url") and flow_url is None:89                    flow_url = re.findall(r"url\s*=\s*(.*)", line.strip())90                    if len(flow_url) == 1:91                        flow_url = flow_url[0]92                        self._flows.append(Flow(name=flow_name, url=flow_url))93                        flow_name, flow_url = None, None94                    else:95                        Logger.fatal("Parsing error: \".gitmodules\":{}: {}".format(index, line))9697    def get_list(self) -> list:98        return self._flows99100    def get_commands(self):  # TODO101        commands = []102        for flow in self._flows:103            if flow.is_installed():104                flow_name = flow.get_name()105                print(flow_name)106
...

tools.py

Source:tools.py

...7        self._proc_object   = None8        self._proc_cmdline  = []9        self._output_path   = os.path.join(os.path.dirname(__file__),10                                           self.__class__.__name__.lower() + '.log')11        if not self._is_installed():12            raise Exception('error: %s not installed' % self.__class__.__name__)13    def _is_installed(self):14        raise NotImplementedError()15    def set_cmdline(self):16        raise NotImplementedError()17    def run(self):18        if not len(self._proc_cmdline):19            raise Exception('error: please use set_cmdline first!')20        if self.status() != 'running':21            self._proc_object = subprocess.Popen(self._proc_cmdline, stdout=open(self._output_path, 'wb'), stderr=subprocess.STDOUT)22    def status(self):23        if not self._proc_object:24            return 'not started'25        # set returncode attr26        self._proc_object.poll()27        return 'running' if self._proc_object.returncode == None else self._parse_status(self._proc_object.returncode)28    def _parse_status(self, status):29        ''' override this function to handle different exit codes '''30        return 'done(%d)' % status31    def results(self):32        if not os.path.isfile(self._output_path):33            return '-'34        with open(self._output_path, 'rb') as fh:35            data = fh.read()36        return data[:10 * 1024] # handle this limit37    def stop(self):38        if self._proc_object:39            self._proc_object.kill()40class YARA(Tool):41    def _is_installed(self):42        return os.path.isfile('yara/yara')43    def set_cmdline(self, rule_file, dir='/', recursive=True, pid=None):44        if pid:45            self._proc_cmdline = ['yara/yara', rule_file, pid]46        else:47            self._proc_cmdline = ['yara/yara'] + ['-r', rule_file, dir] if recursive else [rule_file, dir]48class Chkrootkit(Tool):49    def _is_installed(self):50        return os.path.isfile('chkrootkit/chkrootkit')51    def set_cmdline(self):52        self._proc_cmdline = ['chkrootkit/chkrootkit']53class Find(Tool):54    def _is_installed(self):55        return True56    def set_cmdline(self, dir, name):...

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.