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

How to use _create_from_yaml method in avocado

Best Python code snippet using avocado_python

yaml_to_mux.py

Source:yaml_to_mux.py

...43    """44    Used to mark list as list of objects from whose node is going to be created45    """46    pass47def _create_from_yaml(path, cls_node=tree.TreeNode):48    """ Create tree structure from yaml stream """49    def tree_node_from_values(name, values):50        """ Create `name` node and add values  """51        node = cls_node(str(name))52        using = ''53        for value in values:54            if isinstance(value, tree.TreeNode):55                node.add_child(value)56            elif isinstance(value[0], tree.Control):57                if value[0].code == YAML_INCLUDE:58                    # Include file59                    ypath = value[1]60                    if not os.path.isabs(ypath):61                        ypath = os.path.join(os.path.dirname(path), ypath)62                    if not os.path.exists(ypath):63                        raise ValueError("File '%s' included from '%s' does not "64                                         "exist." % (ypath, path))65                    node.merge(_create_from_yaml('/:' + ypath, cls_node))66                elif value[0].code == YAML_USING:67                    if using:68                        raise ValueError("!using can be used only once per "69                                         "node! (%s:%s)" % (path, name))70                    using = value[1]71                    if using[0] == '/':72                        using = using[1:]73                    if using[-1] == '/':74                        using = using[:-1]75                elif value[0].code == YAML_REMOVE_NODE:76                    value[0].value = value[1]   # set the name77                    node.ctrl.append(value[0])    # add "blue pill" of death78                elif value[0].code == YAML_REMOVE_VALUE:79                    value[0].value = value[1]   # set the name80                    node.ctrl.append(value[0])81                elif value[0].code == YAML_MUX:82                    node.multiplex = True83            else:84                node.value[value[0]] = value[1]85        if using:86            if name is not '':87                for name in using.split('/')[::-1]:88                    node = cls_node(name, children=[node])89            else:90                using = using.split('/')[::-1]91                node.name = using.pop()92                while True:93                    if not using:94                        break95                    name = using.pop()  # 'using' is list pylint: disable=E110196                    node = cls_node(name, children=[node])97                node = cls_node('', children=[node])98        return node99    def mapping_to_tree_loader(loader, node):100        """ Maps yaml mapping tag to TreeNode structure """101        _value = []102        for key_node, value_node in node.value:103            if key_node.tag.startswith('!'):    # reflect tags everywhere104                key = loader.construct_object(key_node)105            else:106                key = loader.construct_python_str(key_node)107            value = loader.construct_object(value_node)108            _value.append((key, value))109        objects = ListOfNodeObjects()110        for name, values in _value:111            if isinstance(values, ListOfNodeObjects):   # New node from list112                objects.append(tree_node_from_values(name, values))113            elif values is None:            # Empty node114                objects.append(cls_node(str(name)))115            else:                           # Values116                objects.append(Value((name, values)))117        return objects118    def mux_loader(loader, obj):119        """120        Special !mux loader which allows to tag node as 'multiplex = True'.121        """122        if not isinstance(obj, yaml.ScalarNode):123            objects = mapping_to_tree_loader(loader, obj)124        else:   # This means it's empty node. Don't call mapping_to_tree_loader125            objects = ListOfNodeObjects()126        objects.append((tree.Control(YAML_MUX), None))127        return objects128    Loader.add_constructor(u'!include',129                           lambda loader, node: tree.Control(YAML_INCLUDE))130    Loader.add_constructor(u'!using',131                           lambda loader, node: tree.Control(YAML_USING))132    Loader.add_constructor(u'!remove_node',133                           lambda loader, node: tree.Control(YAML_REMOVE_NODE))134    Loader.add_constructor(u'!remove_value',135                           lambda loader, node: tree.Control(YAML_REMOVE_VALUE))136    Loader.add_constructor(u'!mux', mux_loader)137    Loader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,138                           mapping_to_tree_loader)139    # Parse file name ([$using:]$path)140    path = __RE_FILE_SPLIT.split(path, 1)141    if len(path) == 1:142        path = __RE_FILE_SUBS.sub(':', path[0])143        using = ["run"]144    else:145        nodes = __RE_FILE_SUBS.sub(':', path[0]).strip('/').split('/')146        using = [node for node in nodes if node]147        if not path[0].startswith('/'):  # relative path, put into /run148            using.insert(0, 'run')149        path = __RE_FILE_SUBS.sub(':', path[1])150    # Load the tree151    with open(path) as stream:152        loaded_tree = yaml.load(stream, Loader)153        if loaded_tree is None:154            return155        loaded_tree = tree_node_from_values('', loaded_tree)156    # Add prefix157    if using:158        loaded_tree.name = using.pop()159        while True:160            if not using:161                break162            loaded_tree = cls_node(using.pop(), children=[loaded_tree])163        loaded_tree = cls_node('', children=[loaded_tree])164    return loaded_tree165def create_from_yaml(paths, debug=False):166    """167    Create tree structure from yaml-like file168    :param fileobj: File object to be processed169    :raise SyntaxError: When yaml-file is corrupted170    :return: Root of the created tree structure171    """172    def _merge(data, path):173        """ Normal run """174        tmp = _create_from_yaml(path)175        if tmp:176            data.merge(tmp)177    def _merge_debug(data, path):178        """ Use NamedTreeNodeDebug magic """179        node_cls = tree.get_named_tree_cls(path)180        tmp = _create_from_yaml(path, node_cls)181        if tmp:182            data.merge(tmp)183    if not debug:184        data = tree.TreeNode()185        merge = _merge186    else:187        data = tree.TreeNodeDebug()188        merge = _merge_debug189    path = None190    try:191        for path in paths:192            merge(data, path)193    # Yaml can raise IndexError on some files194    except (yaml.YAMLError, IndexError) as details:...

kube.py

Source:kube.py

...18        self.logger.debug(f"Rendering template {template_name}")19        template = self.templateEnv.get_template(template_name)20        rendered = template.render(**kwargs)21        self.logger.debug(f"\n{rendered}")22        self._create_from_yaml(rendered)23    def _create_from_yaml(self, rendered_yaml: str):24        for obj in yaml.safe_load_all(rendered_yaml):25            try:26                kubernetes.utils.create_from_dict(27                    self.api_client,28                    obj,29                    namespace=self.namespace30                )31            except kubernetes.utils.FailToCreateError as e:32                body = json.loads(e.api_exceptions[0].body)33                if body['reason'] == "AlreadyExists":34                    self.logger.info(body['message'])35                    continue36                else:37                    raise...

test_flow.py

Source:test_flow.py

...40    assert response.value.detail == 'Flow couldn\'t get started'41@pytest.mark.asyncio42async def test_create_from_yaml_success(monkeypatch):43    monkeypatch.setattr(flow.flow_store, '_create', mock_create_success)44    response = await flow._create_from_yaml(yamlspec=UploadFile(filename='abc.yaml'),45                                            uses_files=[UploadFile(filename='abcd.yaml')],46                                            pymodules_files=[UploadFile(filename='abc.py')])47    assert response['status_code'] == 20048    assert response['flow_id'] == _temp_id49    assert response['host'] == '0.0.0.0'50    assert response['port'] == 1234551    assert response['status'] == 'started'52@pytest.mark.asyncio53async def test_create_from_yaml_parse_exception(monkeypatch):54    monkeypatch.setattr(flow.flow_store, '_create', mock_flow_parse_exception)55    with pytest.raises(flow.HTTPException) as response:56        await flow._create_from_yaml(yamlspec=UploadFile(filename='abc.yaml'),57                                     uses_files=[UploadFile(filename='abcd.yaml')],58                                     pymodules_files=[UploadFile(filename='abc.py')])59    assert response.value.status_code == 40460    assert response.value.detail == 'Invalid yaml file.'61@pytest.mark.asyncio62async def test_create_from_yaml_flow_start_exception(monkeypatch):63    monkeypatch.setattr(flow.flow_store, '_create', mock_flow_start_exception)64    with pytest.raises(flow.HTTPException) as response:65        await flow._create_from_yaml(yamlspec=UploadFile(filename='abc.yaml'),66                                     uses_files=[UploadFile(filename='abcd.yaml')],67                                     pymodules_files=[UploadFile(filename='abc.py')])68    assert response.value.status_code == 40469    assert 'Flow couldn\'t get started' in response.value.detail70@pytest.mark.asyncio71async def test_fetch_flow_success(monkeypatch):72    monkeypatch.setattr(flow.flow_store, '_get', mock_fetch_success)73    response = await flow._fetch(_temp_id)74    assert response['status_code'] == 20075    assert response['host'] == '0.0.0.0'76    assert response['port'] == 1234577    assert response['yaml'] == '!Flow\npods:\n   pod1:\n      uses:_pass'78@pytest.mark.asyncio79async def test_fetch_flow_success_yaml_only(monkeypatch):...

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.