KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use yaml_block method in tappy
Best Python code snippet using tappy_python
taptool
Source:taptool 
...114 print(entry.description)115 else:116 print(117 str(entry),118 dump_yaml_block(yaml_block, path_mappings=path_mappings) if yaml_block is not None else None,119 sep="\n",120 end="\n",121 )122 else:123 raise ValueError(f"Unsupported tap entry {entry.category!r}")124 assert num_tests_found == num_tests_expected, (num_tests_found, num_tests_expected)125 print(f"Total selected: {num_tests_selected}", file=sys.stderr)126def load_vstest_playlist(playlist_file_path):127 root = xml.etree.ElementTree.parse(playlist_file_path).getroot()128 assert root.tag == "Playlist"129 assert root.attrib == {"Version": "1.0"}130 assert not root.text or not root.text.strip()131 assert not root.tail or not root.tail.strip()132 for e in root:133 assert e.tag == "Add"134 assert not e.text or not e.text.strip()135 assert not e.tail or not e.tail.strip()136 assert not list(e)137 assert e.keys() == ["Test"]138 yield e.get("Test")139def parse_duration(duration):140 m = re.match(r"(?P<hours>\d+):(?P<minutes>\d+):(?P<seconds>\d+\.\d+)", duration)141 assert m, duration142 return datetime.timedelta(143 hours=int(m.group("hours")),144 minutes=int(m.group("minutes")),145 seconds=float(m.group("seconds")),146 )147def test_predicate_description_contains(text, ignore_case, entry, selected):148 description = entry.description149 if ignore_case:150 description = description.lower()151 text = text.lower()152 return selected and text in description153def test_predicate_description_not_contains(text, ignore_case, entry, selected):154 description = entry.description155 if ignore_case:156 description = description.lower()157 text = text.lower()158 return selected and text not in description159def test_predicate_not_in(tests_to_exclude, entry, selected):160 return selected and entry.description not in tests_to_exclude161def test_predicate_longer_than(duration, entry, selected):162 reported_duration = entry.yaml_block["duration"] if entry.yaml_block is not None else None163 return selected and (not reported_duration or parse_duration(reported_duration) >= duration)164def test_predicate_shorter_than(duration, entry, selected):165 reported_duration = entry.yaml_block["duration"] if entry.yaml_block is not None else None166 return selected and (not reported_duration or parse_duration(reported_duration) <= duration)167class folded_str(str): pass168class literal_str(str): pass169def folded_str_representer(dumper, data):170 return dumper.represent_scalar("tag:yaml.org,2002:str", data, style=">")171def literal_str_representer(dumper, data):172 return dumper.represent_scalar("tag:yaml.org,2002:str", data, style="|")173yaml.add_representer(datetime.timedelta, lambda dumper, data: dumper.represent_scalar("tag:yaml.org,2002:str", str(data), style="\""))174yaml.add_representer(folded_str, folded_str_representer)175yaml.add_representer(literal_str, literal_str_representer)176def dump_yaml_block(data, *, indent=4, path_mappings=None):177 data["output"] = literal_str(data["output"].replace("\r\n", "\n").replace("\t", " " * indent))178 yaml_block = yaml.dump(179 data,180 indent=indent,181 explicit_start=True,182 explicit_end=True,183 default_flow_style=False,184 )185 def replacement(m):186 return f"""{apply_path_mappings(path_mappings, m.group("file"))}:{m.group("line")}"""187 yaml_block = re.sub(r"(?<= in )(?P<file>.*):line (?P<line>\d+)\s*$", replacement, yaml_block, flags=re.MULTILINE)188 return textwrap.indent(yaml_block, prefix=" " * indent, predicate=None)189def apply_path_mappings(path_mappings, path):190 path = pathlib.Path(path.replace("\\", "/"))...
tools.py
Source:tools.py
1import os, re, yaml2def extractFunctions(searchText):3 return re.findall("((void|bool|int|float|vec\\d|mat\\d)\\s(((?!=).)*)(\\(.*\\)))", searchText)4def isDocumentationIn(yaml_block):5 return 'doc' in yaml_block6def isDescriptionIn(yaml_block):7 if isDocumentationIn(yaml_block):8 return 'description' in yaml_block['doc']9 return False10def getDescriptionOf(yaml_block):11 if isDescriptionIn(yaml_block):12 return yaml_block['doc']['description'] + '\n'13 else:14 return ''15def isShaderIn(yaml_block):16 return 'shaders' in yaml_block17def isShaderBlocksIn(yaml_block):18 if isShaderIn(yaml_block):19 return 'blocks' in yaml_block['shaders']20 return False21def isGlobalBlockIn(yaml_block):22 if isShaderBlocksIn(yaml_block):23 return 'global' in yaml_block['shaders']['blocks']24def getGlobalBlockIn(yaml_block):25 if isGlobalBlockIn(yaml_block):26 return "\n"+yaml_block['shaders']['blocks']['global']27 return ""28def getGlobalFunctionsIn(yaml_block):29 return extractFunctions(getGlobalBlockIn(yaml_block))30def isNormalBlockIn(yaml_block):31 if isShaderBlocksIn(yaml_block):32 return 'normal' in yaml_block['shaders']['blocks']33def getNormalBlockIn(yaml_block):34 if isNormalBlockIn(yaml_block):35 return "\n"+yaml_block['shaders']['blocks']['normal']36 return ""37def isColorBlockIn(yaml_block):38 if isShaderBlocksIn(yaml_block):39 return 'color' in yaml_block['shaders']['blocks']40def getColorBlockIn(yaml_block):41 if isColorBlockIn(yaml_block):42 return "\n"+yaml_block['shaders']['blocks']['color']43 return ""44def isFilterBlockIn(yaml_block):45 if isShaderBlocksIn(yaml_block):46 return 'filter' in yaml_block['shaders']['blocks']47def getFilterBlockIn(yaml_block):48 if isFilterBlockIn(yaml_block):49 return "\n"+yaml_block['shaders']['blocks']['filter']50 return ""51def isUniformsIn(yaml_block):52 return 'uniforms' in yaml_block['shaders']53def isDefinesIn(yaml_block):54 return 'defines' in yaml_block['shaders']55def getDefinesIn(yaml_block):56 if isDefinesIn(yaml_block):57 return yaml_block['shaders']['defines'].keys()58 return []59def isUniformsIn(yaml_block):60 return 'uniforms' in yaml_block['shaders']61def getUniformsIn(yaml_block):62 if isUniformsIn(yaml_block):63 return yaml_block['shaders']['uniforms'].keys()64 return []65def getUniformType(uniform_value):66 if type(uniform_value) is str:67 return 'sampler2D'68 elif type(uniform_value) is float:69 return 'float'70 elif isinstance(uniform_value, list):71 return 'vec'+len(uniform_value)72def uniformHaveMetadata(uniform_name, yaml_block):73 if 'ui' in yaml_block:74 if 'shaders' in yaml_block['ui']:75 if 'uniforms' in yaml_block['ui']['shaders']:76 if uniform_name in yaml_block['ui']['shaders']['uniforms']:77 return True78 return False79def defineHaveMetadata(define_name, yaml_block):80 if 'ui' in yaml_block:81 if 'shaders' in yaml_block['ui']:82 if 'defines' in yaml_block['ui']['shaders']:83 if define_name in yaml_block['ui']['shaders']['defines']:84 return True85 return False86def isTestIn(yaml_block):87 return 'test' in yaml_block88def isExamplesIn(yaml_block):89 if isDocumentationIn(yaml_block):90 return 'examples' in yaml_block['doc']91 return False92def isDependenceIn(yaml_block):93 return 'mix' in yaml_block94def getDependencesIn(yaml_block):95 if isDependenceIn(yaml_block):96 if isinstance(yaml_block['mix'], (str, unicode)):97 return [yaml_block['mix']]98 else:99 return yaml_block['mix']100 return []101def collectDefines(yaml_file, block_name, defines_dict):102 block = yaml_file['styles'][block_name]103 depts = getDependencesIn(block)104 for dep_block_name in depts:105 collectDefines(yaml_file, dep_block_name, defines_dict)106 defines = getDefinesIn(block)107 for define_name in defines:108 defines_dict[define_name] = block['shaders']['defines'][define_name]109def collectUniforms(yaml_file, block_name, uniforms_dict):110 block = yaml_file['styles'][block_name]111 depts = getDependencesIn(block)112 for dep_block_name in depts:113 collectUniforms(yaml_file, dep_block_name, uniforms_dict)114 uniforms = getUniformsIn(block)115 for uniform_name in uniforms:116 uniforms_dict[uniform_name] = block['shaders']['uniforms'][uniform_name]117def getAllGlobals(yaml_file, block_name):118 rta = ""119 depts = getDependencesIn(yaml_file['styles'][block_name])120 for dep_block_name in depts:121 rta += getAllGlobals(yaml_file, dep_block_name)122 rta += getGlobalBlockIn(yaml_file['styles'][block_name])123 return rta124def getAllNormals(yaml_file, block_name):125 rta = ""126 depts = getDependencesIn(yaml_file['styles'][block_name])127 for dep_block_name in depts:128 rta += getAllNormals(yaml_file, dep_block_name)129 rta += getNormalBlockIn(yaml_file['styles'][block_name])130 return rta131def getAllColors(yaml_file, block_name):132 rta = ""133 depts = getDependencesIn(yaml_file['styles'][block_name])134 for dep_block_name in depts:135 rta += getAllColors(yaml_file, dep_block_name)136 rta += getColorBlockIn(yaml_file['styles'][block_name])137 return rta138def getAllFilters(yaml_file, block_name):139 rta = ""140 depts = getDependencesIn(yaml_file['styles'][block_name])141 for dep_block_name in depts:142 rta += getAllFilters(yaml_file, dep_block_name)143 rta += getFilterBlockIn(yaml_file['styles'][block_name])144 return rta145# Recursive dict merge (From https://gist.github.com/angstwad/bf22d1822c38a92ec0a9)146def dict_merge(dct, merge_dct):147 """ Recursive dict merge. Inspired by :meth:``dict.update()``, instead of148 updating only top-level keys, dict_merge recurses down into dicts nested149 to an arbitrary depth, updating keys. The ``merge_dct`` is merged into150 ``dct``.151 :param dct: dict onto which the merge is executed152 :param merge_dct: dct merged into dct153 :return: None154 """155 for k, v in merge_dct.iteritems():156 if (k in dct and isinstance(dct[k], dict)157 and isinstance(merge_dct[k], dict)):158 dict_merge(dct[k], merge_dct[k])159 else:160 dct[k] = merge_dct[k]161# Append yaml dependences in yaml_file ('import' files) to another yaml file (full_yaml_file)162def appendDependencies(full_yaml, filename):163 # print "Append all dependences of " + filename164 folder = os.path.dirname(filename)165 yaml_file = yaml.safe_load(open(filename))166 dict_merge(full_yaml, yaml_file)167 if 'import' in yaml_file:168 if (type(yaml_file['import']) is str):169 dep = folder + '/' + yaml_file['import']170 # print "\tMerging " + dep171 appendDependencies(full_yaml, dep)172 else:173 for file in yaml_file['import']:174 dep = folder + '/' + file175 # print "\tMerging " + dep176 appendDependencies(full_yaml, dep)177# From https://github.com/adafruit/Adafruit_Python_GPIO/blob/master/Adafruit_GPIO/Platform.py178def isRPi():179 """Detect the version of the Raspberry Pi. Returns either 1, 2 or180 None depending on if it's a Raspberry Pi 1 (model A, B, A+, B+),181 Raspberry Pi 2 (model B+), or not a Raspberry Pi.182 """183 # Check /proc/cpuinfo for the Hardware field value.184 # 2708 is pi 1185 # 2709 is pi 2186 # Anything else is not a pi.187 try:188 with open('/proc/cpuinfo', 'r') as infile:189 cpuinfo = infile.read()190 # Match a line like 'Hardware : BCM2709'191 match = re.search('^Hardware\s+:\s+(\w+)$', cpuinfo, flags=re.MULTILINE | re.IGNORECASE)192 if not match:193 # Couldn't find the hardware, assume it isn't a pi.194 return None195 if match.group(1) == 'BCM2708':196 # Pi 1197 return True198 elif match.group(1) == 'BCM2709':199 # Pi 2200 return True201 else:202 # Something else, not a pi.203 return False204 except:...
helm_helpers.py
Source:helm_helpers.py
1import argparse2from dataclasses import dataclass3import glob4import os5import re6import shutil7import subprocess8from typing import Dict9import yaml10def to_string(obj):11 return obj.__class__.__name__ + "/" + obj.name12def find(element, dictionary):13 keys = element.split('/')14 current_dictionary = dictionary15 for key in keys:16 if re.search(r'[\d+]', key):17 key = int(re.search(r'\d+', key).group())18 elif key not in current_dictionary.keys():19 return None20 current_dictionary = current_dictionary[key]21 return current_dictionary22@dataclass23class FluxObject:24 name: str = None25 def __str__(self):26 return to_string(self)27 def __hash__(self):28 return hash(self.__str__())29@dataclass30class GitRepository(FluxObject):31 url: str = None32 tag: str = None33@dataclass34class HelmConfigValues(FluxObject):35 values: str = None36@dataclass37class HelmRelease(FluxObject):38 chart: str = None39 repo: GitRepository = None40 repo_name: str = None41 values: HelmConfigValues = None42 values_config_map_name: str = None43def build_git_repository(yaml_block) -> GitRepository:44 repo = GitRepository(name=find("metadata/name", yaml_block), url=find("spec/url", yaml_block))45 repo.tag = get_git_repository_tag(yaml_block)46 return repo47def get_git_repository_tag(yaml_block) -> str:48 ref = find("spec/ref", yaml_block)49 if "tag" in ref:50 return ref['tag']51 return ref['branch']52def build_helm_release(yaml_block) -> HelmRelease:53 return HelmRelease(name=find("metadata/name", yaml_block), chart=find("spec/chart/spec/chart", yaml_block),54 repo_name=find("spec/chart/spec/sourceRef/name", yaml_block),55 values_config_map_name=find("spec/valuesFrom/[0]/name", yaml_block))56def build_helm_values(yaml_block) -> HelmConfigValues | None:57 if "values.yaml" not in yaml_block["data"]:58 return None59 return HelmConfigValues(find("metadata/name", yaml_block), find("data/values.yaml", yaml_block))60Kind2Builder = {"GitRepository": build_git_repository, "HelmRelease": build_helm_release,61 "ConfigMap": build_helm_values}62def create_flux_objects_from_files(glob_pattern) -> Dict[str, object]:63 created_objects = {}64 files = glob.glob(glob_pattern)65 for file in files:66 with open(file, 'r') as file_stream:67 yaml_docs = yaml.load_all(file_stream, Loader=yaml.FullLoader)68 create_flux_objects_from_yaml_doc(created_objects, yaml_docs)69 return created_objects70def create_flux_objects_from_yaml_doc(created_objects, yaml_docs):71 for yaml_doc in yaml_docs:72 kind = find("kind", yaml_doc)73 if not kind:74 print(f"Could not determine kind from {yaml_doc!s:200.200}...")75 elif kind not in Kind2Builder.keys():76 print(f"Could not find builder for kind {kind} in {yaml_doc!s:200.200}...")77 else:78 create_flux_object_from_yaml_doc(created_objects, kind, yaml_doc)79def create_flux_object_from_yaml_doc(created_objects, kind, yaml_doc):80 builder = Kind2Builder[kind]81 flux_object = builder(yaml_doc)82 if not flux_object:83 print(f"Could not build flux object from {yaml_doc!s:200.200}...")84 else:85 created_objects[str(flux_object)] = flux_object86def compose_helm_releases(flux_objects):87 for release in {name: flux_object for name, flux_object in flux_objects.items() if88 isinstance(flux_object, HelmRelease)}.values(): # type: HelmRelease89 release.repo = flux_objects[GitRepository.__name__ + "/" + release.repo_name]90 release.values = flux_objects[HelmConfigValues.__name__ + "/" + release.values_config_map_name]91 yield release92def parse_args():93 parser = argparse.ArgumentParser(description='Render k8s manifests from flux helm releases')94 parser.add_argument('--base-dir', '-b', nargs='?', dest="base_path", required=True,95 help='Path to folder containing the flux manifests')96 parser.add_argument('--work-dir', '-w', nargs='?', dest="work_dir", required=True, help='Path to working directory')97 arguments = parser.parse_args()98 return arguments99def recreate_working_dir():100 try:101 shutil.rmtree(working_dir)102 except FileNotFoundError:103 pass104 os.mkdir(working_dir)105if __name__ == '__main__':106 args = parse_args()107 base_path = args.base_path108 working_dir = args.work_dir109 path_to_git_repos = f"{base_path}/sources"110 path_to_helm_releases = f"{base_path}/helmreleases"111 path_to_config_maps = f"{base_path}/configmaps"112 output_dir = working_dir + "/generated"113 all_flux_objects = create_flux_objects_from_files(f"{base_path}/**/*.yaml")114 recreate_working_dir()115 os.mkdir(output_dir)116 for helm_release in compose_helm_releases(all_flux_objects):117 git_clone_target_folder = f"{working_dir}/{helm_release.repo.name}"118 subprocess.run(['git', 'clone', '--depth', '1', '--branch', helm_release.repo.tag, helm_release.repo.url,119 git_clone_target_folder], check=True)120 release_value_file_name = f'{working_dir}/{helm_release.name}-values.yaml'121 with open(release_value_file_name, 'w') as value_file:122 value_file.write(helm_release.values.values)123 path_to_chart = git_clone_target_folder + "/" + helm_release.chart124 generated_manifests_file = output_dir + "/" + helm_release.name + ".yaml"125 with open(generated_manifests_file, "w") as helm_output:126 subprocess.run(['helm', '-f', release_value_file_name, 'template', '--debug', path_to_chart],127 stdout=helm_output, check=True)128 assert os.path.exists(generated_manifests_file)...
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:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run tappy 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!!
Was this article helpful?
