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

How to use on_before_stop method in localstack

Best Python code snippet using localstack_python

provider.py

Source:provider.py

...65    lock: threading.RLock66    def __init__(self) -> None:67        self.lambda_service = LambdaService()68        self.lock = threading.RLock()69    def on_before_stop(self) -> None:70        self.lambda_service.stop()71    def _map_config_out(self, version: FunctionVersion) -> FunctionConfiguration:72        return FunctionConfiguration(73            RevisionId=version.config_meta.revision_id,74            FunctionName=version.id.function_name,75            FunctionArn=version.id.qualified_arn(),76            LastModified=version.config_meta.last_modified,77            LastUpdateStatus=LastUpdateStatus.Successful,78            State=State.Active,79            Version=version.id.qualifier,80            Description=version.config.description,81            Role=version.config.role,82            Timeout=version.config.timeout,83            Runtime=version.config.runtime,...

app.py

Source:app.py

...151    def on_key(self, k):152        pass153    def on_mouse(self, m):154        pass155    def on_before_stop(self):156        pass157    def on_after_stop(self):...

database.py

Source:database.py

1from weakref import WeakKeyDictionary2import datetime3from pymongo import MongoClient4from nameko.extensions import DependencyProvider5from nameko.exceptions import safe_for_serialization6class MongoDatabase(DependencyProvider):7    default_connection_uri = 'mongodb://localhost:27017'8    def __init__(self, result_backend=False,9                 on_before_setup=None, on_after_setup=None,10                 on_before_stop=None, on_after_stop=None,11                 on_before_worker_setup=None, on_after_worker_setup=None,12                 on_before_worker_result=None, on_after_worker_result=None):13        self.result_backend = result_backend14        self.logs = WeakKeyDictionary()15        self.client = None16        self.database = None17        # Callbacks18        self.on_before_setup = on_before_setup19        self.on_after_setup = on_after_setup20        self.on_before_stop = on_before_stop21        self.on_after_stop = on_after_stop22        self.on_before_worker_setup = on_before_worker_setup23        self.on_after_worker_setup = on_after_worker_setup24        self.on_before_worker_result = on_before_worker_result25        self.on_after_worker_result = on_after_worker_result26    @property27    def db(self):28        return self.database29    def _run_callback(self, name, **kwargs):30        assert hasattr(self, name), "Callback {} does not exist".format(name)31        callback = getattr(self, name)32        if callback is not None:33            callback(self, **kwargs)34        35    def setup(self):36        self._run_callback('on_before_setup')37        config = self.container.config38        db_name = config.get('MONGODB_DB_NAME', self.container.service_name)39        conn_uri = config.get('MONGODB_CONNECTION_URL', self.default_connection_uri)40        params = {}41        if config.get('MONGODB_USER'):42            if config.get('MONGODB_USER'):43                params['username'] = config.get('MONGODB_USER')44            if config.get('MONGODB_PASSWORD'):45                params['password'] = config.get('MONGODB_PASSWORD')46            if config.get('MONGODB_AUTHENTICATION_BASE'):47                params['authSource'] = config.get('MONGODB_AUTHENTICATION_BASE')48            if config.get('MONGODB_AUTH_MECHANISM'):49                params['authSource'] = config.get('MONGODB_AUTH_MECHANISM')50        self.client = MongoClient(conn_uri, **params)51        self.database = self.client[db_name]52        if self.result_backend:53            self.db.logging.create_index('start', expireAfterSeconds=24*60*60)54            self.db.logging.create_index('call_id')55        self._run_callback('on_after_setup')56    def stop(self):57        self._run_callback('on_before_stop')58        self.client.close()59        self.client = None60        self._run_callback('on_after_stop')61    def get_dependency(self, worker_ctx):62        return self.db63    64    def worker_setup(self, worker_ctx):65        self._run_callback('on_before_worker_setup', worker_ctx=worker_ctx)66        if self.result_backend:67            self.logs[worker_ctx] = datetime.datetime.now()68            service_name = worker_ctx.service_name69            method_name = worker_ctx.entrypoint.method_name70            call_id = worker_ctx.call_id71            self.db.logging.insert_one(72                {73                    'call_id': call_id,74                    'service_name': service_name,75                    'method_name': method_name,76                    'status': 'PENDING',77                    'start': self.logs[worker_ctx]78                }79            )80        self._run_callback('on_after_worker_setup', worker_ctx=worker_ctx)81    def worker_result(self, worker_ctx, result=None, exc_info=None):82        self._run_callback('on_before_worker_result', worker_ctx=worker_ctx, result=result, exc_info=exc_info)83        if self.result_backend:84            call_id = worker_ctx.call_id85            if exc_info is None:86                status = 'SUCCESS'87            else:88                status = 'FAILED'89            now = datetime.datetime.now()90            start = self.logs.pop(worker_ctx)91            self.db.logging.update_one(92                {'call_id': call_id},93                {94                    '$set': {95                        'status': status,96                        'end': now,97                        'elapsed': (now - start).seconds,98                        'exception': safe_for_serialization(exc_info) if exc_info is not None else None99                    }100                }101            )...

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.