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

How to use lookup_config_file method in molecule

Best Python code snippet using molecule_python

shell.py

Source:shell.py

...32from molecule.command.base import click_group_ex33from click_help_colors import _colorize34click_completion.init()35colorama.init(autoreset=True, strip=not should_do_markup())36LOCAL_CONFIG = lookup_config_file(".config/molecule/config.yml")37ENV_FILE = ".env.yml"38def _version_string():39    v = pkg_resources.parse_version(molecule.__version__)40    color = "bright_yellow" if v.is_prerelease else "green"41    msg = "molecule %s\n" % _colorize(molecule.__version__, color)42    msg += _colorize(43        "   ansible==%s python==%s.%s"44        % (45            pkg_resources.parse_version(ansible_version),46            sys.version_info[0],47            sys.version_info[1],48        ),49        "bright_black",50    )...

lookup.py

Source:lookup.py

1import asyncio2from typing import List, Optional, Tuple3import common4import lookup5from lookup.database.domains import DomainState6from runners.constants import LOOKUP_CONFIG_FILE, LOOKUP_LOG_FILE, prepare_start7class LookupRunner:8    def __init__(self, log_level=None):9        if log_level is not None:10            lookup.logger.setLevel(log_level)11        self.database: Optional[lookup.Database] = None12        self.crawler: Optional[lookup.Crawler] = None13        self.server: Optional[lookup.WebServer] = None14        self.fetcher: Optional[common.Fetcher] = None15    async def _prepare(self):16        if self.database:17            return18        prepare_start(LOOKUP_CONFIG_FILE, LOOKUP_LOG_FILE, lookup.Config, lookup.logger)19        self.database = lookup.Database()20        await self.database.setup()21    async def start(22        self, start_crawler: List[str] = None, start_server: bool = True23    ) -> None:24        """25        Start lookup server and/or crawler.26        :param start_crawler: None if don't start crawler. Else a list of starting urls.27        :param start_server: True if web server should be started else False.28        """29        await self._prepare()30        lookup.event_counter.all_time_fetched = (31            await self.database.queue.get_count_by_state(lookup.QueueState.Fetched)32        )33        lookup.event_counter.queue_size = await self.database.queue.get_size()34        lookup.event_counter.actor_count = await self.database.objects.get_object_cnt(35            lookup.AsObjectType.Actor36        )37        if start_crawler is not None:38            self.fetcher = common.Fetcher(39                lookup.logger, lookup.Config.parallel_fetches, lookup.Config.debug40            )41            await self.fetcher.setup()42            self.crawler = lookup.Crawler(self.database, self.fetcher)43            await self.crawler.run(start_crawler)44        if start_server:45            self.server = lookup.WebServer(self.database, self.crawler)46            await self.server.run()47    async def spin_and_log(self):48        while True:49            await asyncio.sleep(10)50            stats = lookup.event_counter.reset_stats()51            stats["queue_size"] = await self.database.queue.get_size()52            if self.crawler:53                stats["waiting_reachable"] = sum(54                    155                    for d in self.crawler.domains.values()56                    if d.state <= DomainState.Unknown57                    and d.fail_streak == 058                    and d.has_waiting_elements59                )60            await self.database.stats.insert(stats)61    async def add_verifier(self, verifier_uri) -> Tuple[int, str]:62        await self._prepare()63        fetcher = common.Fetcher(lookup.logger, debug=lookup.Config.debug)64        await fetcher.setup()65        verifier = await fetcher.fetch_ap(verifier_uri)66        key_pem = verifier["publicKey"]["publicKeyPem"]67        item = await self.database.verifiers.add(verifier["id"], key_pem)68        await fetcher.shutdown()69        return item["id"], item["uri"]70    async def cleanup(self):71        if self.crawler:72            await self.fetcher.shutdown()73        if self.server:74            await self.server.shutdown()75        if self.fetcher:76            await self.crawler.stop()77        if self.database:...

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.