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 Github

copy

Full Screen

...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:...

Full Screen

Full Screen

kube.py

Source:kube.py Github

copy

Full Screen

...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...

Full Screen

Full Screen

test_flow.py

Source:test_flow.py Github

copy

Full Screen

...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):...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run avocado automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful