Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use on_load_exception method in localstack

Best Python code snippet using localstack_python

manager.py

Source:manager.py

...71            self.listener.on_load_after,72            (plugin_spec, plugin, result),73            "error while calling on_load_after",74        )75    def _fire_on_load_exception(self, plugin_spec, plugin, exception):76        _call_safe(77            self.listener.on_load_exception,78            (plugin_spec, plugin, exception),79            "error while calling on_load_exception",80        )81class PluginContainer(Generic[P]):82    """83    Object to pass around the plugin state inside a PluginManager.84    """85    name: str86    lock: threading.RLock87    plugin_spec: PluginSpec88    plugin: P = None89    load_value: Any = None90    is_init: bool = False91    is_loaded: bool = False92    init_error: Exception = None93    load_error: Exception = None94class PluginManager(PluginLifecycleNotifierMixin, Generic[P]):95    """96    Manages Plugins within a namespace discovered by a PluginFinder. The default mechanism is to resolve plugins from97    entry points using a StevedorePluginFinder.98    A Plugin that is managed by a PluginManager can be in three states:99        * resolved: the entrypoint pointing to the PluginSpec was imported and the PluginSpec instance was created100        * init: the PluginFactory of the PluginSpec was successfully invoked101        * loaded: the load method of the Plugin was successfully invoked102    Internally, the PluginManager uses PluginContainer instances to keep the state of Plugin instances.103    """104    namespace: str105    load_args: Union[List, Tuple]106    load_kwargs: Dict[str, Any]107    listener: PluginLifecycleListener108    def __init__(109        self,110        namespace: str,111        load_args: Union[List, Tuple] = None,112        load_kwargs: Dict = None,113        listener: PluginLifecycleListener = None,114        finder: PluginFinder = None,115    ):116        self.namespace = namespace117        self.load_args = load_args or list()118        self.load_kwargs = load_kwargs or dict()119        self.listener = listener or PluginLifecycleListener()120        self.finder = finder or StevedorePluginFinder(121            self.namespace, self._fire_on_resolve_exception122        )123        self._plugin_index = None124        self._init_mutex = threading.RLock()125    def load(self, name: str) -> P:126        """127        Loads the Plugin with the given name using the load args and kwargs set in the plugin manager constructor.128        If at any point in the lifecycle the plugin loading fails, the load method will raise the respective exception.129        Load is idempotent, so once the plugin is loaded, load will return the same instance again.130        """131        container = self._require_plugin(name)132        if not container.is_loaded:133            self._load_plugin(container)134        if container.init_error:135            raise container.init_error136        if container.load_error:137            raise container.load_error138        if not container.is_loaded:139            raise PluginException(140                "plugin did not load correctly", namespace=self.namespace, name=name141            )142        return container.plugin143    def load_all(self, propagate_exceptions=False) -> List[P]:144        """145        Attempts to load all plugins found in the namespace, and returns those that were loaded successfully. If146        propagate_exception is set to True, then the method will re-raise any errors as soon as it encouters them.147        """148        plugins = list()149        for name, container in self._plugins.items():150            if container.is_loaded:151                plugins.append(container.plugin)152                continue153            try:154                plugin = self.load(name)155                plugins.append(plugin)156            except PluginDisabled as e:157                LOG.debug("%s", e)158            except Exception as e:159                if propagate_exceptions:160                    raise161                else:162                    LOG.error("exception while loading plugin %s:%s: %s", self.namespace, name, e)163        return plugins164    def list_plugin_specs(self) -> List[PluginSpec]:165        return [container.plugin_spec for container in self._plugins.values()]166    def list_names(self) -> List[str]:167        return [spec.name for spec in self.list_plugin_specs()]168    def list_containers(self) -> List[PluginContainer[P]]:169        return list(self._plugins.values())170    def get_container(self, name: str) -> PluginContainer[P]:171        return self._require_plugin(name)172    def exists(self, name: str) -> bool:173        return name in self._plugins174    def is_loaded(self, name: str) -> bool:175        return self._require_plugin(name).is_loaded176    @property177    def _plugins(self) -> Dict[str, PluginContainer[P]]:178        if self._plugin_index is None:179            with self._init_mutex:180                if self._plugin_index is None:181                    self._plugin_index = self._init_plugin_index()182        return self._plugin_index183    def _require_plugin(self, name: str) -> PluginContainer[P]:184        if name not in self._plugins:185            raise ValueError("no plugin named %s in namespace %s" % (name, self.namespace))186        return self._plugins[name]187    def _load_plugin(self, container: PluginContainer):188        with container.lock:189            plugin_spec = container.plugin_spec190            # instantiate Plugin from spec if necessary191            if not container.is_init:192                try:193                    LOG.debug("instantiating plugin %s", plugin_spec)194                    container.plugin = self._plugin_from_spec(plugin_spec)195                    container.is_init = True196                    self._fire_on_init_after(plugin_spec, container.plugin)197                except Exception as e:198                    # TODO: maybe we should move these logging blocks to `load_all`, since this is the only instance199                    #  where exceptions messages may get lost.200                    if LOG.isEnabledFor(logging.DEBUG):201                        LOG.exception("error instantiating plugin %s", plugin_spec)202                    self._fire_on_init_exception(plugin_spec, e)203                    container.init_error = e204                    return205            plugin = container.plugin206            if not plugin.should_load():207                raise PluginDisabled(namespace=self.namespace, name=container.plugin_spec.name)208            args = self.load_args209            kwargs = self.load_kwargs210            self._fire_on_load_before(plugin_spec, plugin, args, kwargs)211            try:212                LOG.debug("loading plugin %s:%s", self.namespace, plugin_spec.name)213                result = plugin.load(*args, *kwargs)214                self._fire_on_load_after(plugin_spec, plugin, result)215                container.load_value = result216                container.is_loaded = True217            except Exception as e:218                if LOG.isEnabledFor(logging.DEBUG):219                    LOG.exception("error loading plugin %s", plugin_spec)220                self._fire_on_load_exception(plugin_spec, plugin, e)221                container.load_error = e222    def _plugin_from_spec(self, plugin_spec: PluginSpec) -> P:223        factory = plugin_spec.factory224        # functional decorators can overwrite the spec factory (pointing to the decorator) with a custom factory225        spec = getattr(factory, "__pluginspec__", None)226        if spec:227            factory = spec.factory228        return factory()229    def _init_plugin_index(self) -> Dict[str, PluginContainer]:230        return {plugin.name: plugin for plugin in self._import_plugins() if plugin}231    def _import_plugins(self) -> Iterable[PluginContainer]:232        for spec in self.finder.find_plugins():233            self._fire_on_resolve_after(spec)234            if spec.namespace != self.namespace:...

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.