Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

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.