KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use read_b method in Airtest

Best Python code snippet using Airtest

test_fragment.py

Source:test_fragment.py

1#!/usr/bin/env python32# -*- coding: utf-8 -*-3import unittest4import singlecellmultiomics.molecule5import singlecellmultiomics.fragment6import pysam7import pysamiterators.iterators8import os9from singlecellmultiomics.fragment import Fragment, CHICFragment10from singlecellmultiomics.utils import create_MD_tag11from singlecellmultiomics.utils import complement12"""13These tests check if the Molecule module is working correctly14"""15class TestTAPs(unittest.TestCase):16    def test_random_priming(self):17        temp_folder = 'data'18        enable_ref_write=False19        ref_path = f'{temp_folder}/chic_ref.fa'20        alignments_path = f'{temp_folder}/chic_test_alignments.bam'21        if not os.path.exists(temp_folder):22            os.makedirs(temp_folder)23        # Create reference bam file24        refseq = 'TTAATCATGAAACCGTGGAGGCAAATCGGAGTGTAAGGCTTGACTGGATTCCTACGTTGCGTAGGTTCATGGGGGG'25        if enable_ref_write:26            with open(ref_path, 'w') as f:27                f.write(f">chr1\n{refseq}\n>chr2\n{complement(refseq)}\n""")28            # This command needs to finish, which is not working properly during testing29            pysam.faidx(ref_path)30        # CATG at base 531        # Create BAM file with NLA fragment:32        def get_reads():33            with pysam.AlignmentFile(alignments_path_unsorted,'wb',reference_names=['chr1'],reference_lengths=[len(refseq)]) as bam:34                ### Nla III mate pair example, containing 2 CpGs and 1 call on the wrong strand35                read_A = pysam.AlignedSegment(bam.header)36                read_A.reference_name = 'chr1'37                read_A.reference_start = 538                # Before last A is a bogus G>A conversion to test strandness:39                read_A.query_sequence = 'CATGAAACCGTGGAGGCAAATTGGAGTAT'40                read_A.cigarstring = f'{len(read_A.query_sequence)}M'41                read_A.qual = 'A'*len(read_A.query_sequence)42                read_A.mapping_quality = 6043                read_A.query_name = 'EX1_GA_CONV_2x_CpG_TAPS'44                read_A.set_tag('SM', 'Cell_A')45                read_A.is_read1 = True46                read_A.is_read2 = False47                read_A.set_tag('lh','TG')48                # Set substitution tag:49                read_A.set_tag('MD',50                               create_MD_tag(51                                       refseq[read_A.reference_start:read_A.reference_end], read_A.query_sequence))52                read_A.is_paired = True53                read_A.is_proper_pair = True54                # Create a second read which is a mate of the previous55                read_B = pysam.AlignedSegment(bam.header)56                read_B.reference_name = 'chr1'57                read_B.reference_start = 2558                read_B.query_sequence = refseq[25:60].replace('TGT','TAT').replace('CG', 'TG')59                read_B.cigarstring = f'{len(read_B.query_sequence)}M'60                read_B.qual = 'A'*len(read_B.query_sequence)61                read_B.mapping_quality = 6062                read_B.is_read2 = True63                read_B.is_read1 = False64                read_B.is_reverse = True65                read_B.query_name = 'EX1_GA_CONV_2x_CpG_TAPS'66                read_B.set_tag('SM', 'Cell_A')67                read_B.set_tag('lh','TG')68                read_B.set_tag('MD',69                           create_MD_tag(refseq[read_B.reference_start:read_B.reference_end],70                                         read_B.query_sequence,71                                   ))72                read_B.is_paired = True73                read_B.is_proper_pair = True74                read_A.next_reference_id = read_B.reference_id75                read_A.next_reference_start = read_B.reference_start76                read_B.next_reference_id = read_A.reference_id77                read_B.next_reference_start = read_A.reference_start78                read_A.mate_is_reverse = read_B.is_reverse79                read_B.mate_is_reverse = read_A.is_reverse80                return read_A, read_B81        alignments_path_unsorted = f'{alignments_path}.unsorted.bam'82        read_A, read_B = get_reads()83        read_B.set_tag('rS','AATTAA') # Set random primer sequence84        ## Act on reads with random primer set:85        frag = Fragment([read_A, read_B])86        self.assertTrue(frag.R2_primer_length==0)87        self.assertTrue(frag.unsafe_trimmed)88        ## Act on reads without random primer89        read_A, read_B = get_reads()90        frag = Fragment([read_A, read_B], R2_primer_length=10)91        self.assertEqual(frag.R2_primer_length, 10)92        self.assertFalse(frag.unsafe_trimmed)93        read_A, read_B = get_reads()94        frag = CHICFragment([read_A, read_B], R2_primer_length=0)95        self.assertEqual(frag.R2_primer_length, 0)96        self.assertFalse(frag.unsafe_trimmed)97        read_A, read_B = get_reads()98        read_B.set_tag('rS','AATTAA') # Set random primer sequence99        frag = CHICFragment([read_A, read_B], R2_primer_length=0)100        self.assertEqual(frag.R2_primer_length, 0)101        self.assertTrue(frag.unsafe_trimmed)102        read_A, read_B = get_reads()103        read_B.set_tag('MX','scCHIC384C8U3l')104        frag = CHICFragment([read_A, read_B])105        self.assertEqual(frag.R2_primer_length, 0)106        self.assertFalse(frag.unsafe_trimmed)107        read_A, read_B = get_reads()108        read_B.set_tag('MX','scCHIC384C8U3')109        frag = CHICFragment([read_A, read_B])110        self.assertEqual(frag.R2_primer_length,6)111        self.assertFalse(frag.unsafe_trimmed)112    def test_init(self):113        """Test if the fragment can be initialised"""114        with pysam.AlignmentFile('./data/mini_nla_test.bam') as f:115            for i,(R1,R2) in enumerate(pysamiterators.iterators.MatePairIterator(f)):116                frag = singlecellmultiomics.fragment.Fragment([R1,R2])117                self.assertIsNotNone(frag)118    def test_get(self):119        """Test if the fragment can be initialised"""120        with pysam.AlignmentFile('./data/mini_nla_test.bam') as f:121            for i,(R1,R2) in enumerate(pysamiterators.iterators.MatePairIterator(f)):122                frag = singlecellmultiomics.fragment.Fragment([R1,R2])123                if R1 is not None:124                    self.assertEqual(frag.get_R1(), R1 )125                    self.assertEqual(frag[0], R1 )126                if R2 is not None:127                    self.assertEqual(frag.get_R2(), R2 )128                    self.assertEqual(frag[1], R2 )129    def test_get_sample(self):130        """Test if the sample name of the fragment can be obtained"""131        with pysam.AlignmentFile('./data/mini_nla_test.bam') as f:132            for i,(R1,R2) in enumerate(pysamiterators.iterators.MatePairIterator(f)):133                frag = singlecellmultiomics.fragment.Fragment([R1,R2])134                self.assertTrue( frag.get_sample().startswith('A') )135    def test_set_sample(self):136        """Test if the sample name of the fragment can be changed"""137        with pysam.AlignmentFile('./data/mini_nla_test.bam') as f:138            for i,(R1,R2) in enumerate(pysamiterators.iterators.MatePairIterator(f)):139                frag = singlecellmultiomics.fragment.Fragment([R1,R2])140                frag.set_sample(f'TEST{i}')141                self.assertEqual(frag.get_sample(), f'TEST{i}' )142    def test_strand(self):143        """Test if the strand can be obtained (doesn't check validity)"""144        with pysam.AlignmentFile('./data/mini_nla_test.bam') as f:145            for i,(R1,R2) in enumerate(pysamiterators.iterators.MatePairIterator(f)):146                frag = singlecellmultiomics.fragment.Fragment([R1,R2])147                self.assertIn( frag.get_strand(), [None,0,1])148                if frag.is_mapped:149                    self.assertIn( frag.get_strand(), [0,1])150if __name__ == '__main__':...

test_lcadre.py

Source:test_lcadre.py

1#!usr/bin/env/python32from collections import Counter3import logging4import os5import unittest6import lcadre7logger = logging.getLogger("lcadre")8logger.setLevel(logging.CRITICAL)9class MockPair:10    def __init__(self,11                 reference_name=None,12                 reference_start=None,13                 reference_end=None,14                 is_reverse=False,15                 query_alignment_start=0):16        self.reference_name = reference_name17        self.reference_start = reference_start18        self.reference_end = reference_end19        self.is_reverse = is_reverse20        self.query_alignment_start = query_alignment_start21    def __eq__(self, other):22        return self.reference_name == other.reference_name and \23               self.reference_start == other.reference_start and \24               self.reference_end == other.reference_end and \25               (self.is_reverse == other.is_reverse) and \26               self.query_alignment_start == other.query_alignment_start27    def __repr__(self):28        return f"{self.reference_name}:{self.reference_start}-{self.reference_end}:{self.is_reverse}:{self.query_alignment_start}"29class TestLcadre(unittest.TestCase):30    @classmethod31    def setUpClass(self):32        pass33    def setUp(self):34        self.dirname = os.path.dirname(__file__)35        self.collateral_dir = self.dirname + "/collateral/"36        self.input_dir = self.collateral_dir + "input/"37        self.n = 10000000038        self.fn = os.path.join(self.input_dir, "simple.bam")39        self.empty_fn = os.path.join(self.input_dir, "empty.bam")40        self.test = lcadre.Lcadre(self.fn, self.n)41        self.read_blank = MockPair()42        self.read_a = MockPair(reference_name="chr1", reference_start=100, reference_end=200, query_alignment_start=10)43        self.read_b = MockPair(reference_name="chr1", reference_start=100, reference_end=200, query_alignment_start=10,44                               is_reverse=True)45        self.read_c = MockPair(reference_name="chr2", reference_start=100, reference_end=200, query_alignment_start=10)46    def tearDown(self):47        pass48    def testFileTypes(self):49        fts = ["sam", "bam", "cram", "infer"]50        for ft in fts:51            test = lcadre.Lcadre(self.fn, self.n, ft)52            self.assertTrue(test.file_type in lcadre.ALLOWED_FILE_TYPES)53        fail_types = ["fasta", "", None, 1]54        for ft in fail_types:55            with self.assertRaises(ValueError):56                test = lcadre.Lcadre(self.fn, self.n, ft)57        with self.assertRaises(ValueError):58            test = lcadre.Lcadre("test.txt", self.n)59    def testEstimation(self):60        counter = Counter()61        for item in [1, 1, 2, 4, 1, 3, 4, 5, 0]:62            counter[item] += 163        test = lcadre.Lcadre(self.fn, self.n)64        test.counter = counter65        test.perform_estimation()66        self.assertEqual(test.singletons, 4)67        self.assertEqual(test.doubletons, 1)68        self.assertEqual(test.total_read_pairs, 9)69        self.assertEqual(test.total_signatures, 6)70        self.assertEqual(test.chao1, 14)71        self.assertEqual(test.m_star, self.n - 9)72        self.assertEqual(test.f0_hat, 8)73        self.assertEqual(test.s_ind, 14)74        self.assertEqual(test.dup_rate_obs, 0.33333333333333337)75        self.assertEqual(test.dup_rate_extrap, 0.99999986)76    def testProcessPair(self):77        test = lcadre.Lcadre(self.fn, self.n)78        test.process_pair((self.read_blank, self.read_blank))79        self.assertTrue("-1.-1" in test.counter)80        self.assertEqual(test.counter["-1.-1"], 1)81        self.assertEqual(sum(test.counter.values()), 1)82    def testGetPos(self):83        self.assertEqual(self.test.get_pos(self.read_blank), None)84        self.assertEqual(self.test.get_pos(self.read_a), 90)85        self.assertEqual(self.test.get_pos(self.read_b), 210)86    def testGetReadString(self):87        self.assertEqual(self.test.get_read_string(self.read_blank, -1), "-1")88        self.assertEqual(self.test.get_read_string(self.read_a, 90), "chr1.90.+")89        self.assertEqual(self.test.get_read_string(self.read_b, 210), "chr1.210.-")90    def testOrderPair(self):91        read_a, pos_a, read_b, pos_b = self.test.order_pair(self.read_a, 90, self.read_b, 210)92        self.assertEqual(self.read_a, read_a)93        self.assertEqual(pos_a, 90)94        self.assertEqual(self.read_b, read_b)95        self.assertEqual(pos_b, 210)96        read_a, pos_a, read_b, pos_b = self.test.order_pair(self.read_b, 210, self.read_a, 90)97        self.assertEqual(self.read_a, read_a)98        self.assertEqual(pos_a, 90)99        self.assertEqual(self.read_b, read_b)100        self.assertEqual(pos_b, 210)101        read_b, pos_b, read_blank, pos_blank = self.test.order_pair(self.read_b, 210, self.read_blank, -1)102        self.assertEqual(self.read_b, read_b)103        self.assertEqual(pos_b, 210)104        self.assertEqual(self.read_blank, read_blank)105        self.assertEqual(pos_blank, -1)106        read_b, pos_b, read_blank, pos_blank = self.test.order_pair(self.read_blank, -1, self.read_b, 210)107        self.assertEqual(self.read_b, read_b)108        self.assertEqual(pos_b, 210)109        self.assertEqual(self.read_blank, read_blank)110        self.assertEqual(pos_blank, -1)111        read_b, pos_b, read_c, pos_c = self.test.order_pair(self.read_c, 90, self.read_b, 210)112        self.assertEqual(self.read_b, read_b)113        self.assertEqual(pos_b, 210)114        self.assertEqual(self.read_c, read_c)115        self.assertEqual(pos_c, 90)116        read_b, pos_b, read_a, pos_a = self.test.order_pair(self.read_a, None, self.read_b, 210)117        self.assertEqual(self.read_a, read_a)118        self.assertEqual(pos_a, None)119        self.assertEqual(self.read_b, read_b)120        self.assertEqual(pos_b, 210)121    def testRun(self):122        self.test.run()123    def testEmptyBam(self):124        test = lcadre.Lcadre(self.empty_fn, self.n)125        with self.assertRaises(ZeroDivisionError):126            test.run()127if __name__ == '__main__':...

proxy.py

Source:proxy.py

1import asyncio2import collections3Connection = collections.namedtuple("Connection", ["read_fn", "write_fn"])4class Connection():5    """6    A connection that can be used for proxying. Every connection contains two7    methods: a read function and a write function.8    The read function should be a coroutine and takes no parameters. This9    gets called to read data from this connection. Should return None or ""10    when the connection is closed.11    The write function should take one parameter and gets called whenever data12    should be written to this connection.13    """14    def __init__(self, read_fn, write_fn):15        """16        Constructor.17        read_fn: a coroutine function that takes no parameter18        write_fn: a function that takes one parameter19        """20        self.read_fn = read_fn21        self.write_fn = write_fn22class Proxy():23    """24    A proxy between two connections: A and B. Everything read from one25    connection gets sent to the other one, and the other way around.26    """27    def __init__(self, connection_a, connection_b):28        """29        Constructor.30        connection_a and connection_b should both be Connection instances.31        """32        self.connection_a = connection_a33        self.connection_b = connection_b34    @asyncio.coroutine35    def run(self):36        """37        Runs the proxy, returning whenever one side closed the connection.38        """39        read_a = asyncio.async(self.connection_a.read_fn())40        read_b = asyncio.async(self.connection_b.read_fn())41        while read_a or read_b:42            waits = [t for t in (read_a, read_b) if t]43            done, pending = yield from asyncio.wait(waits,44                return_when=asyncio.FIRST_COMPLETED)45            for task in done:46                try:47                    data = task.result()48                except ConnectionResetError:49                    data = None50                if task == read_a:51                    if data:52                        read_a = asyncio.async(self.connection_a.read_fn())53                        self.connection_b.write_fn(data)54                    else:55                        read_a = None56                        read_b.cancel()57                        read_b = None58                elif task == read_b:59                    if data:60                        read_b = asyncio.async(self.connection_b.read_fn())61                        self.connection_a.write_fn(data)62                    else:63                        read_b = None64                        read_a.cancel()...

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.