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

How to use _import_name method in tempest

Best Python code snippet using tempest_python

new_object.py

Source:new_object.py

1from logging import getLogger, LoggerAdapter2logger = getLogger(__name__)3from thewired import NamespaceConfigParser4from thewired import NamespaceConfigParsingError5import operator6import collections7from kaleidoscope.spec import AttributeSpec8import copy9import functools10class ObjectSpecConfigParser(NamespaceConfigParser):11    object_spec_keys = ['colors', 'description', 'attributes']12    def __init__(self):13        super().__init__(prefix="spec.object")14    @staticmethod15    def validate_formatter_args(formatter_args):16        """17        Description:18            make sure the formatter argument config values can become objects19        Input:20            formatter_args: a dict of argument name --> values21        Output:22            formatter_args dict with values config strings replaced with runtime objects23        """24        import logging25        import importlib26        #importlib.import_module('cush')27        name_ext = {'name_ext' : 'ObjectSpecConfigParser.validate_formatter_args'}28        #log = LoggerAdapter(logger, name_ext)29        log = logging.getLogger(__name__)30        log.setLevel(logging.DEBUG)31        final_args = dict()32        print('Verifiying and Instantiating formatter args...')33        for k,v in formatter_args.items():34            print('Checking value: "{}"'.format(v))35            #- figure out how to import prefixes to be able to eval36            _import_name = ''37            for n,prefix in enumerate(v.split('.')):38                print('Checking prefix "{}" of value "{}"'.format(prefix,v))39                _import_name += prefix40                try:41                    importlib.import_module(_import_name)42                except (ImportError, NameError) as err:43                    msg = 'Can not import formatter value prefix: "{}"'.format(_import_name)44                    print(msg)45                else:46                    print('Successfully imported "{}"'.format(_import_name))47                finally:48                    #- append the dot at the end for next iteration49                    if n < len(v.split('.')) - 1:50                        _import_name += '.'51                #- if we're at the end, try to eval52                if n == len(v.split('.')) - 1:53                    print('Trying final eval of "{}"'.format(_import_name))54                    try:55                        final_args[k] = eval(_import_name)56                    except (AttributeError, NameError, TypeError, ValueError) as err:57                        msg = 'Can not instantiate formatter value: "{}"'.format(_import_name)58                        msg += ': {}'.format(str(err))59                        raise NameSpaceConfigParsingError(msg)60                        61                    else:62                        #- if we are here, the eval succeeded63                        #- break the inner loop and continue to the next argument64                        print('Successfully instantiated formatter argument value')65        return final_args66    def parse_submap(self, dictConfig, cur_ns, prev_ns=None):67        log = LoggerAdapter(logger, {'name_ext' : 'ObjectSpecConfigParser.parse_submap'})68        log.debug("Entered: cur_ns: {} | prev_ns{}".format(cur_ns, prev_ns))69        cur_ns._add_item('specmap', dict(), iter=True)70        reduced_dictconfig = dict(dictConfig)71        for key in dictConfig.keys():72            if key in self.object_spec_keys:73                msg = list()74                msg.append("Found object spec key: key: {}".format(key))75                msg.append(" | cur_ns:{}".format(cur_ns))76                msg.append(" | prev_ns:{}".format(prev_ns))77                log.debug(''.join(msg))78                if key == 'attributes':79                    attributes = list()80                    for attribute in dictConfig[key]:81                        #- AttributeSpec can be a dict or a simple value82                        #- dict overrides the formatter callable and display length83                        if isinstance(attribute, collections.Mapping):84                            #- only 1 top-level key, the attribute name85                            _name = list(attribute.keys())[0]86                            _length = attribute[_name].get('length', None)87                            #- formatter can also be simple value (callable name) or a88                            #- dict if there are args to pass in when it is called89                            if isinstance(attribute[_name].get('formatter', None), collections.Mapping):90                                _formatter = attribute[_name]['name']91                                _formatter_args = copy.copy(attribute[_name]).pop('name')92                            else:93                                _formatter = attribute[_name].get('formatter', None)94                                _formatter_args = None95                            attributes.append(AttributeSpec(_name, _length, _formatter))96                        else:97                            attributes.append(AttributeSpec(attribute, None, None))98                    cur_ns.specmap['attributes'] = attributes99                else:100                    cur_ns.specmap[key] = dictConfig[key]101                #- don't create namespace nodes for the keys we process102                reduced_dictconfig.pop(key)103        log.debug("calling super().parse_submap...")...

core.py

Source:core.py

1"""2Core module, responsible of creating the Flask and Celery application3objects.4"""5from celery import Celery6from flask import Flask, current_app7from flask.config import Config8from datacat.utils.plugin_loading import import_object9from datacat.web.blueprints.admin import admin_bp10from datacat.web.blueprints.public import public_bp11def make_flask_app(config=None):12    app = Flask('datacat')13    app.register_blueprint(admin_bp, url_prefix='/api/1/admin')14    app.register_blueprint(public_bp, url_prefix='/api/1/data')15    app.config.update(make_config())16    if config is not None:17        app.config.update(config)18    return app19def make_celery(config):20    # celery_app = Celery('datacat',21    #                     broker=config['CELERY_BROKER_URL'],22    #                     backend=config['CELERY_RESULT_BACKEND'])23    celery_app = celery_placeholder_app24    celery_app.broker = config['CELERY_BROKER_URL']25    # celery_app.backend = config['CELERY_RESULT_BACKEND']26    celery_app.conf.update(config)27    TaskBase = celery_app.Task28    class AppContextTask(TaskBase):29        abstract = True30        def __call__(self, *args, **kwargs):31            with current_app.app_context():32                return TaskBase.__call__(self, *args, **kwargs)33    celery_app.Task = AppContextTask34    return celery_app35def make_config():36    cfg = Config('')37    cfg.from_object('datacat.settings.default')38    cfg.from_envvar('DATACAT_SETTINGS', silent=True)39    return cfg40def load_plugins(app):41    from datacat.utils.plugin_manager import PluginManager42    plugins = []43    for name in app.config['PLUGINS']:44        # Instantiate plugin class45        plugin = import_object(name)46        plugin._import_name = name47        plugins.append(plugin)48        # Setup the plugin49        plugin.setup(app)50    return PluginManager(plugins)51def finalize_app(app):52    """Prepare application for running"""53    from datacat.db import db_info54    with app.app_context():55        app.plugins = load_plugins(app)56        previously_enabled_plugins = set(db_info.get('core.plugins_enabled', []))  # noqa57        previously_installed_plugins = set(db_info.get('core.plugins_installed', []))  # noqa58        enabled_plugins = set(app.config['PLUGINS'])59        # ------------------------------------------------------------60        # Run the ``install()`` method for all the plugins that61        # were not previously installed62        plugins_to_install = enabled_plugins - previously_installed_plugins63        # ------------------------------------------------------------64        # Run the ``enable()`` method for all the plugins that65        # were not previously enabled66        plugins_to_enable = enabled_plugins - previously_enabled_plugins67        # ------------------------------------------------------------68        # Run the ``disable()`` method for all the plugins that69        # were previously enabled (and aren't anymore)70        plugins_to_disable = previously_enabled_plugins - enabled_plugins71        # Nothing will be uninstalled implicitly!72        # ------------------------------------------------------------73        # Perform the operations from above74        # ------------------------------------------------------------75        for plugin in app.plugins:76            if plugin._import_name in plugins_to_install:77                plugin.install()78            if plugin._import_name in plugins_to_enable:79                plugin.enable()80            if plugin._import_name in plugins_to_disable:81                plugin.disable()82            if plugin._import_name in enabled_plugins:83                plugin.upgrade()84        # ------------------------------------------------------------85        # Register new information about plugins86        db_info['core.plugins_enabled'] = list(enabled_plugins)87        db_info['core.plugins_installed'] = list(88            previously_installed_plugins | enabled_plugins)89def make_app(config=None):90    from datacat.db import create_tables, connect91    app = make_flask_app(config)92    celery_app = make_celery(app.config)93    celery_app.set_current()94    _adm_conn = connect(**app.config['DATABASE'])95    _adm_conn.autocommit = True96    create_tables(_adm_conn)97    finalize_app(app)98    return app...

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.