How to use merge_family method in Pytest

Best Python code snippet using pytest

junitxml.py

Source:junitxml.py

...48            return u"#x%02X" % i49        else:50            return u"#x%04X" % i51    return py.xml.raw(illegal_xml_re.sub(repl, py.xml.escape(arg)))52def merge_family(left, right):53    result = {}54    for kl, vl in left.items():55        for kr, vr in right.items():56            if not isinstance(vl, list):57                raise TypeError(type(vl))58            result[kl] = vl + vr59    left.update(result)60families = {}61families["_base"] = {"testcase": ["classname", "name"]}62families["_base_legacy"] = {"testcase": ["file", "line", "url"]}63# xUnit 1.x inherits legacy attributes64families["xunit1"] = families["_base"].copy()65merge_family(families["xunit1"], families["_base_legacy"])66# xUnit 2.x uses strict base attributes67families["xunit2"] = families["_base"]68class _NodeReporter(object):69    def __init__(self, nodeid, xml):70        self.id = nodeid71        self.xml = xml72        self.add_stats = self.xml.add_stats73        self.family = self.xml.family74        self.duration = 075        self.properties = []76        self.nodes = []77        self.testcase = None78        self.attrs = {}79    def append(self, node):...

main.py

Source:main.py

1import os2import argparse3IS_HIC=False4def execute_cmd(cmd):5    print("Info: ", cmd)6    os.system(cmd)7def bgzip_and_index(vcf, bgzip, tabix):8    bgzip_cmd = "{} -f {}".format(bgzip, vcf)9    tabix_cmd = "{} -f {}".format(tabix, vcf+".gz")10    execute_cmd(bgzip_cmd)11    execute_cmd(tabix_cmd)12    return vcf+".gz"13def bgunzip(vcf):14    if "gz" in vcf:15        unzip_cmd = "gunzip {}".format(vcf)16        execute_cmd(unzip_cmd)17        return vcf.replace(".gz","")18    else:19        return vcf20def sc_file(d,df):21    # f_name = f.split("/")[-1]22    ds = os.path.join(d, df)23    # mv_cmd = "mv {} {}".format(f,ds)24    # execute_cmd(mv_cmd)25    return ds26def i_phase(spechap,extract,bgzip,tabix,bam, vcf, out_dir, name, ref):27    vcf = bgunzip(vcf)28    lst_out = os.path.join(out_dir, name+".lst")29    lst_sorted_out = os.path.join(out_dir, name+".sorted.lst")30    phased_vcf = os.path.join(out_dir, name+".phased.vcf")31    sort_cmd = "sort -n -k6 {} > {}".format(lst_out,lst_sorted_out)32    if IS_HIC:33        e_cmd = "{} --vcf {} --bam {} -o {} --ref {} --hic 1 --maxfragments 300000000".format(extract, vcf, bam, lst_out, ref)34        execute_cmd(e_cmd)35        vcf = bgzip_and_index(vcf, bgzip, tabix)36        execute_cmd(sort_cmd)37        s_cmd = "{} -f {} -v {} -o {} --hic".format(spechap, lst_sorted_out, vcf, phased_vcf)38        execute_cmd(s_cmd)39    else:40        e_cmd = "{} --vcf {} --bam {} -o {} --ref {}".format(extract, vcf, bam, lst_out, ref)41        execute_cmd(e_cmd)42        vcf = bgzip_and_index(vcf, bgzip, tabix)43        execute_cmd(sort_cmd)44        s_cmd = "{} -f {} -v {} -o {}".format(spechap, lst_sorted_out, vcf, phased_vcf)45        execute_cmd(s_cmd)46    # execute_cmd(e_cmd)47    # execute_cmd(sort_cmd)48    # execute_cmd(s_cmd)49    return phased_vcf50def e_lst(extract, bams, vcf, out_dir, name, ref):51    lst_a = ""52    lst_l = os.path.join(out_dir,name+".all.lst")53    lst_l_sorted = os.path.join(out_dir,name+".all.sorted.lst")54    for i in range(len(bams)):55        b = bams[i]56        lst = os.path.join(out_dir, name+".tmp"+str(i)+".lst")57        e_cmd = "{} --vcf {} --bam {} -o {} --ref {}".format(extract, vcf, b, lst,ref)58        if IS_HIC:59            e_cmd = "{} --vcf {} --bam {} -o {} --ref {} --hic 1 --maxfragments 300000000".format(extract, vcf, b, lst,ref)60        lst_a = lst_a+" "+lst61        execute_cmd(e_cmd)62    cat_cmd = "cat {} > {}".format(lst_a, lst_l)63    sort_cmd = "sort -n -k6 {} > {}".format(lst_l,lst_l_sorted)64    execute_cmd(cat_cmd)65    execute_cmd(sort_cmd)66    return lst_l_sorted67def phase_with_lst(spechap, lst, vcf, out_file, bgzip, tabix):68    vcf = bgzip_and_index(vcf,bgzip,tabix)69    s_cmd = "{} -f {} -v {} -o {} --keep_phasing_info".format(spechap, lst, vcf, out_file)70    if IS_HIC:71        s_cmd = "{} -f {} -v {} -o {} --keep_phasing_info --hic".format(spechap, lst, vcf, out_file)72    execute_cmd(s_cmd)73    bgzip_and_index(out_file,bgzip,tabix)74    return out_file+".gz"75def extract_hete(vcf,out, bcftools):76    out_vcf = os.path.join(out,vcf.split("/")[-1].replace(".vcf",".hete.vcf").replace(".bcf",".hete.vcf").replace(".gz",""))77    cmd = "{} view -g het {} > {}".format(bcftools,vcf,out_vcf)78    execute_cmd(cmd)79    return out_vcf80def main():81    parser = argparse.ArgumentParser("trio phase")82    parser.add_argument(83        '--father_v', help='Father VCF file not in gz format', required=False)84    parser.add_argument(85        '--mother_v', help='Mother VCF file not in gz format', required=False)86    parser.add_argument(87        '--child_v', help='Child VCF file not in gz format', required=True)88    parser.add_argument(89        '--father_b', help='Father bam file indexed', required=False)90    parser.add_argument(91        '--mother_b', help='Mother bam file indexed', required=False)92    parser.add_argument(93        '--child_b', help='Child bam file indexed', required=True)94    parser.add_argument(95        '--spechap', help='spechap path', required=True)96    parser.add_argument(97        '--extractHairs', help='extractHairs path', required=True)98    parser.add_argument(99        '--bgzip', help='bgzip path', required=True)100    parser.add_argument(101        '--tabix', help='tabix path', required=True)102    parser.add_argument(103        '--bcftools', help='bcftools path', required=True)104    parser.add_argument('-o', '--out_dir', help='Out dir', required=True)105    parser.add_argument('--ref', help='reference', required=True)106    parser.add_argument('--data_type', help='Data type: hic, ngs, ', required=False, default="ngs")107    parser.add_argument(108        '--step', help='which step', required=False)109    parser.add_argument(110        '--hic', help='Is hic data',action='store_true', required=False, default=False)111    parser.add_argument(112        '--script_root', help='script root', required=True)113    args = parser.parse_args()114    IS_HIC = args.hic115    if not os.path.exists(args.out_dir):116        os.mkdir(args.out_dir)117    if args.step == "2":118        print("skip individual phase")119        m_phased_v1= bgunzip(args.mother_v)120        f_phased_v1= bgunzip(args.father_v)121        c_phased_v1= bgunzip(args.child_v)122        c_phased_v2 = sc_file(args.out_dir,"child.vcf")123        m_phased_v2 = sc_file(args.out_dir,"mother.vcf")124        f_phased_v2 = sc_file(args.out_dir,"father.vcf")125    else:126        print("extract hete..")127        if args.mother_v:128            m_hete = extract_hete(args.mother_v,args.out_dir,args.bcftools)129        if args.father_v:130            f_hete = extract_hete(args.father_v,args.out_dir,args.bcftools)131        if args.child_v:132            c_hete = extract_hete(args.child_v,args.out_dir,args.bcftools)133        print("individual phase..")134        m_phased_v1=""135        f_phased_v1=""136        c_phased_v1 = i_phase(args.spechap, args.extractHairs, args.bgzip, args.tabix, args.child_b, c_hete, args.out_dir, "child", args.ref)137        if args.mother_v and args.mother_b:138            m_phased_v1 = i_phase(args.spechap, args.extractHairs,args.bgzip, args.tabix, args.mother_b, m_hete, args.out_dir, "mother", args.ref)139        if args.father_v and args.father_b:140            f_phased_v1 = i_phase(args.spechap, args.extractHairs,args.bgzip, args.tabix, args.father_b, f_hete, args.out_dir, "father", args.ref)141        c_phased_v2 = c_phased_v1+".trio.vcf"142        m_phased_v2 = m_phased_v1+".trio.vcf"143        f_phased_v2 = f_phased_v1+".trio.vcf"144    # print("Raw phase with only vcf")145    # raw_cmd = ""146    # if not args.mother_v:147    #     raw_cmd = "python merge_family.py -f {} -c {} -o {}".format(f_phased_v1, c_phased_v1, args.out_dir)148    # if not args.father_v:149    #     raw_cmd = "python merge_family.py -m {} -c {} -o {}".format(m_phased_v1, c_phased_v1, args.out_dir)150    # else:151    #     raw_cmd = "python merge_family.py -m {} -f {} -c {} -o {}".format(m_phased_v1, f_phased_v1, c_phased_v1, args.out_dir)152    # if execute_cmd(raw_cmd):153    #     print(raw_cmd,"running error")154    #     exit155    print("phasing child ...")156    bams = []157    if args.mother_b:158        bams.append(args.mother_b)159    if args.father_b:160        bams.append(args.father_b)161    c_lst = e_lst(args.extractHairs, bams, c_phased_v1, args.out_dir, "child", args.ref)162    c_out = phase_with_lst(args.spechap, c_lst, c_phased_v1, c_phased_v2, args.bgzip, args.tabix)163    print("phasing parent...")164    if args.mother_b and args.mother_v:165        bams = [args.child_b]166        m_lst = e_lst(args.extractHairs, bams, m_phased_v1, args.out_dir, "mother", args.ref)167        m_out = phase_with_lst(args.spechap, m_lst, m_phased_v1, m_phased_v2, args.bgzip, args.tabix)168    if args.father_b and args.father_v:169        bams = [args.child_b]170        f_lst = e_lst(args.extractHairs, bams, f_phased_v1, args.out_dir, "father", args.ref)171        f_out = phase_with_lst(args.spechap, f_lst, f_phased_v1, f_phased_v2, args.bgzip, args.tabix)172    print("Reflect...")173    m_re = os.path.join(args.out_dir,'mother.reflect.vcf')174    f_re = os.path.join(args.out_dir,'father.reflect.vcf')175    c_re = os.path.join(args.out_dir,'child.reflect.vcf')176    if args.mother_v:177        re_cmd = "python {}/reflect_back.py --hete {} --orig {} --out {}".format(args.script_root,m_phased_v2+".gz",args.mother_v, m_re)178        execute_cmd(re_cmd)179    if args.father_v:180        re_cmd = "python {}/reflect_back.py --hete {} --orig {} --out {}".format(args.script_root,f_phased_v2+".gz",args.father_v, f_re)181        execute_cmd(re_cmd)182    if args.child_v:183        re_cmd = "python {}/reflect_back.py --hete {} --orig {} --out {}".format(args.script_root,c_phased_v2+".gz",args.child_v, c_re)184        execute_cmd(re_cmd)185    print("Phase with only vcf")186    m_re_gz = bgzip_and_index(m_re, args.bgzip, args.tabix)187    f_re_gz = bgzip_and_index(f_re, args.bgzip, args.tabix)188    c_re_gz = bgzip_and_index(c_re, args.bgzip, args.tabix)189    if not args.mother_v:190        raw_cmd = "python {}/merge_family.py -f {} -c {} -o {}".format(args.script_root,f_re_gz, c_re_gz, args.out_dir)191    if not args.father_v:192        raw_cmd = "python {}/merge_family.py -m {} -c {} -o {}".format(args.script_root,m_re_gz, c_re_gz, args.out_dir)193    else:194        raw_cmd = "python {}/merge_family.py -m {} -f {} -c {} -o {}".format(args.script_root,m_re_gz, f_re_gz, c_re_gz, args.out_dir)195    if execute_cmd(raw_cmd):196        print(raw_cmd,"running error")197        exit198if __name__ == "__main__":...

Pytest Tutorial

Looking for an in-depth tutorial around pytest? LambdaTest covers the detailed pytest tutorial that has everything related to the pytest, from setting up the pytest framework to automation testing. Delve deeper into pytest testing by exploring advanced use cases like parallel testing, pytest fixtures, parameterization, executing multiple test cases from a single file, and more.

Chapters

What is pytest
Pytest installation: Want to start pytest from scratch? See how to install and configure pytest for Python automation testing.
Run first test with pytest framework: Follow this step-by-step tutorial to write and run your first pytest script.
Parallel testing with pytest: A hands-on guide to parallel testing with pytest to improve the scalability of your test automation.
Generate pytest reports: Reports make it easier to understand the results of pytest-based test runs. Learn how to generate pytest reports.
Pytest Parameterized tests: Create and run your pytest scripts while avoiding code duplication and increasing test coverage with parameterization.
Pytest Fixtures: Check out how to implement pytest fixtures for your end-to-end testing needs.
Execute Multiple Test Cases: Explore different scenarios for running multiple test cases in pytest from a single file.
Stop Test Suite after N Test Failures: See how to stop your test suite after n test failures in pytest using the @pytest.mark.incremental decorator and maxfail command-line option.