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

How to use install_async method in localstack

Best Python code snippet using localstack_python

test_es.py

Source:test_es.py

...12COMMON_HEADERS = {"content-type": "application/json", "Accept-encoding": "identity"}13# Lock and event to ensure that the installation is executed before the tests14INIT_LOCK = threading.Lock()15installed = threading.Event()16def install_async():17    """18    Installs the default elasticsearch version in a worker thread. Used by conftest.py to make19    sure elasticsearch is downloaded once the tests arrive here.20    """21    if installed.is_set():22        return23    def run_install(*args):24        with INIT_LOCK:25            if installed.is_set():26                return27            LOG.info("installing elasticsearch default version")28            install_elasticsearch()29            LOG.info("done installing elasticsearch default version")30            LOG.info("installing opensearch default version")31            install_opensearch()32            LOG.info("done installing opensearch default version")33            installed.set()34    start_worker_thread(run_install)35@pytest.fixture(autouse=True)36def elasticsearch():37    if not installed.is_set():38        install_async()39    assert installed.wait(timeout=5 * 60), "gave up waiting for elasticsearch to install"40    yield41def try_cluster_health(cluster_url: str):42    response = requests.get(cluster_url)43    assert response.ok, f"cluster endpoint returned an error: {response.text}"44    response = requests.get(f"{cluster_url}/_cluster/health")45    assert response.ok, f"cluster health endpoint returned an error: {response.text}"46    assert response.json()["status"] in [47        "orange",48        "yellow",49        "green",50    ], "expected cluster state to be in a valid state"51class TestElasticsearchProvider:52    def test_list_versions(self, es_client):...

config.py

Source:config.py

...30        for m in self.modules:31            if m.name == name:32                return m33        raise Error('No module with name {}'.format(name))34    async def install_async(self):35        futures = map(Connection.establish, self.connections)36        await asyncio.gather(*futures)37    def install(self):38        asyncio.get_event_loop().run_until_complete(self.install_async())39    async def deploy_modules_ordered_async(self):40        for module in self.modules:41            await module.deploy()42    def deploy_modules_ordered(self):43        asyncio.get_event_loop().run_until_complete(44                                self.deploy_modules_ordered_async())45def load(file_name):46    with open(file_name, 'r') as f:47        contents = json.load(f)48    config = Config(file_name)49    config.nodes = _load_list(contents['nodes'], _load_node)50    config.modules = _load_list(contents['modules'],51                                lambda m: _load_module(m, config))52    config.connections = _load_list(contents['connections'],...

conftest.py

Source:conftest.py

1"""2Pytest configuration that spins up a single localstack instance that is shared across test modules.3See: https://docs.pytest.org/en/6.2.x/fixture.html#conftest-py-sharing-fixtures-across-multiple-files4It is thread/process safe to run with pytest-parallel, however not for pytest-xdist.5"""6import logging7import multiprocessing as mp8import os9import threading10import pytest11from localstack import config12from localstack.config import is_env_true13from localstack.constants import ENV_INTERNAL_TEST_RUN14from localstack.services import infra15from localstack.utils.common import safe_requests16from tests.integration.test_es import install_async as es_install_async17from tests.integration.test_opensearch import install_async as opensearch_install_async18from tests.integration.test_terraform import TestTerraform19logger = logging.getLogger(__name__)20localstack_started = mp.Event()  # event indicating whether localstack has been started21localstack_stop = mp.Event()  # event that can be triggered to stop localstack22localstack_stopped = mp.Event()  # event indicating that localstack has been stopped23startup_monitor_event = mp.Event()  # event that can be triggered to start localstack24# collection of functions that should be executed to initialize tests25test_init_functions = set()26@pytest.hookimpl()27def pytest_configure(config):28    # first pytest lifecycle hook29    _start_monitor()30def pytest_runtestloop(session):31    # second pytest lifecycle hook (before test runner starts)32    # collect test classes33    test_classes = set()34    for item in session.items:35        if item.parent and item.parent.cls:36            test_classes.add(item.parent.cls)37        # OpenSearch/Elasticsearch are pytests, not unit test classes, so we check based on the item parent's name.38        # Any pytests that rely on opensearch/elasticsearch must be special-cased by adding them to the list below39        parent_name = str(item.parent).lower()40        if any(opensearch_test in parent_name for opensearch_test in ["opensearch", "firehose"]):41            test_init_functions.add(opensearch_install_async)42        if any(opensearch_test in parent_name for opensearch_test in ["es", "firehose"]):43            test_init_functions.add(es_install_async)44    # add init functions for certain tests that download/install things45    for test_class in test_classes:46        # set flag that terraform will be used47        if TestTerraform is test_class:48            logger.info("will initialize TestTerraform")49            test_init_functions.add(TestTerraform.init_async)50            continue51    if not session.items:52        return53    if session.config.option.collectonly:54        return55    # trigger localstack startup in startup_monitor and wait until it becomes ready56    startup_monitor_event.set()57    localstack_started.wait()58@pytest.hookimpl()59def pytest_unconfigure(config):60    # last pytest lifecycle hook (before pytest exits)61    _trigger_stop()62def _start_monitor():63    threading.Thread(target=startup_monitor).start()64def _trigger_stop():65    localstack_stop.set()66    startup_monitor_event.set()67def startup_monitor() -> None:68    """69    The startup monitor is a thread that waits for the startup_monitor_event and, once the event is true, starts a70    localstack instance in it's own thread context.71    """72    logger.info("waiting on localstack_start signal")73    startup_monitor_event.wait()74    if localstack_stop.is_set():75        # this is called if _trigger_stop() is called before any test has requested the localstack_runtime fixture.76        logger.info("ending startup_monitor")77        localstack_stopped.set()78        return79    if is_env_true("TEST_SKIP_LOCALSTACK_START") or os.environ.get("TEST_TARGET") == "AWS_CLOUD":80        logger.info("TEST_SKIP_LOCALSTACK_START is set, not starting localstack")81        localstack_started.set()82        return83    logger.info("running localstack")84    run_localstack()85def run_localstack():86    """87    Start localstack and block until it terminates. Terminate localstack by calling _trigger_stop().88    """89    # configure90    os.environ[ENV_INTERNAL_TEST_RUN] = "1"91    safe_requests.verify_ssl = False92    config.FORCE_SHUTDOWN = False93    config.EDGE_BIND_HOST = "0.0.0.0"94    def watchdog():95        logger.info("waiting stop event")96        localstack_stop.wait()  # triggered by _trigger_stop()97        logger.info("stopping infra")98        infra.stop_infra()99    monitor = threading.Thread(target=watchdog)100    monitor.start()101    logger.info("starting localstack infrastructure")102    infra.start_infra(asynchronous=True)103    for fn in test_init_functions:104        try:105            # asynchronous init functions106            fn()107        except Exception:108            logger.exception("exception while running init function for test")109    logger.info("waiting for infra to be ready")110    infra.INFRA_READY.wait()  # wait for infra to start (threading event)111    localstack_started.set()  # set conftest inter-process Event112    logger.info("waiting for shutdown")113    try:114        logger.info("waiting for watchdog to join")115        monitor.join()116    finally:117        logger.info("ok bye")118        localstack_stopped.set()119@pytest.fixture(scope="session", autouse=True)120def localstack_runtime():121    """122    This is a dummy fixture. Each test requests the fixture, but it actually just makes sure that localstack is running,123    blocks until localstack is running, or starts localstack the first time the fixture is requested.124    It doesn't actually do anything but signal to the `startup_monitor` function.125    """126    if localstack_started.is_set():127        # called by all tests after the startup has completed and the initial tests are unblocked128        yield129        return130    startup_monitor_event.set()131    localstack_started.wait()132    yield...

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.