test_assemble.py 文件源码

def test_bwa_mem_mapping_sambamba():
    # Test single end, no rg
    toy_fasta = reftools.RefFasta(os.path.join(dir, "toy.fasta"))
    test_bam = assemble.bwa_mem_mapping_sambamba(
        "bwa", "samtools", "sambamba", toy_fasta,
        os.path.join(dir, "assemble_test"), [os.path.join(dir, "toy_1.fastq")], 1,
        "None", [""], cram=False)
    assert os.path.exists(os.path.join(dir, "assemble_test.sorted.bam"))
    assert pysam.idxstats(
        os.path.join(
            dir, "assemble_test.sorted.bam")) == 'seq1\t20\t0\t0\nseq2\t40\t0\t0\n*\t0\t0\t3\n'
    # Test single end, with rg
    toy_fasta = reftools.RefFasta(os.path.join(dir, "toy.fasta"))
    test_bam = assemble.bwa_mem_mapping_sambamba(
        "bwa", "samtools", "sambamba", toy_fasta,
        os.path.join(dir, "test_assemble.rg"), [os.path.join(dir, "toy_1.fastq")], 1,
        "@RG\tID:{}".format("test"), [""], cram=False)
    assert os.path.exists(os.path.join(dir, "test_assemble.rg.sorted.bam"))
    assert pysam.idxstats(
        os.path.join(
            dir, "test_assemble.rg.sorted.bam")) == 'seq1\t20\t0\t0\nseq2\t40\t0\t0\n*\t0\t0\t3\n'
    header = pysam.view("-H", os.path.join(dir, "test_assemble.rg.sorted.bam"))
    assert header.find("@RG\\tID:test") != -1
    # Test raising error for inaccessible reference
    with pytest.raises(RuntimeError):
        fake_fasta = reftools.RefFasta(os.path.join(
            dir, "DOES_NOT_EXIST.fasta"), no_initial_index=True)
        test_bam = assemble.bwa_mem_mapping_sambamba(
            "bwa", "samtools", "sambamba", fake_fasta,
            os.path.join(dir, "test_assemble"),
            [os.path.join(dir, "toy_1.fastq")], 1,
            "None", [""], cram=False)
    # Test raising error for no fastqs
    toy_fasta = reftools.RefFasta(os.path.join(dir, "toy.fasta"))
    with pytest.raises(RuntimeError):
        test_bam = assemble.bwa_mem_mapping_sambamba(
            "bwa", "samtools", "sambamba", toy_fasta,
            os.path.join(dir, "test_assemble"), [], 1,
            "None", [""], cram=False)
    # Test raising error for inaccessible fastq
    with pytest.raises(RuntimeError):
        toy_fasta = reftools.RefFasta(os.path.join(dir, "toy.fasta"))
        test_bam = assemble.bwa_mem_mapping_sambamba(
            "bwa", "samtools", "sambamba", toy_fasta,
            os.path.join(dir, "test_assemble"),
            [os.path.join(dir, "toy_FOO_fake.fastq")], 1,
            "None", [""], cram=False)