How to use pop_all method in Slash

Best Python code snippet using slash

test_worker.py

Source:test_worker.py

...78                'url': "%s/foo" % (srv.url,),79                'method': 'POST'80            }81        }82        pop_all(msgs)83        yield worker.retry(req)84        self.assertEqual(pop_all(msgs), [85            ('Retrying request', req),86            ('Retry successful (200)', req),87        ])88        [req] = reqs89        self.assertEqual(req.method, 'POST')90        self.assertEqual((yield zitems(worker.redis, pending_key('test'))), [])91    @inlineCallbacks92    def test_retry_reschedule(self):93        msgs = self.patch_log()94        worker = yield self.mk_worker()95        srv = yield ToyServer.from_test(self)96        yield worker.stop()97        @srv.app.route('/foo')98        def route(req):99            req.setResponseCode(500)100        req1 = {101            'owner_id': '1234',102            'timestamp': 5,103            'attempts': 0,104            'intervals': [10, 20],105            'request': {106                'url': "%s/foo" % (srv.url,),107                'method': 'POST'108            }109        }110        req2 = {111            'owner_id': '1234',112            'timestamp': 10,113            'attempts': 0,114            'intervals': [10, 30],115            'request': {116                'url': "%s/foo" % (srv.url,),117                'method': 'POST'118            }119        }120        pop_all(msgs)121        yield worker.retry(req1)122        self.assertEqual(pop_all(msgs), [123            ('Retrying request', req1),124            ('Retry failed (500)', req1),125            ('Rescheduling retry', req1),126        ])127        yield worker.retry(req2)128        self.assertEqual(pop_all(msgs), [129            ('Retrying request', req2),130            ('Retry failed (500)', req2),131            ('Rescheduling retry', req2),132        ])133        pending = yield zitems(worker.redis, pending_key('test'))134        self.assertEqual(pending, [135            (5 + 20, {136                'owner_id': '1234',137                'timestamp': 5,138                'attempts': 1,139                'intervals': [10, 20],140                'request': {141                    'url': "%s/foo" % (srv.url,),142                    'method': 'POST'143                }144            }),145            (10 + 30, {146                'owner_id': '1234',147                'timestamp': 10,148                'attempts': 1,149                'intervals': [10, 30],150                'request': {151                    'url': "%s/foo" % (srv.url,),152                    'method': 'POST'153                }154            })155        ])156    @inlineCallbacks157    def test_retry_end(self):158        msgs = self.patch_log()159        worker = yield self.mk_worker()160        srv = yield ToyServer.from_test(self)161        yield worker.stop()162        @srv.app.route('/foo')163        def route(req):164            req.setResponseCode(500)165        req1 = {166            'owner_id': '1234',167            'timestamp': 5,168            'attempts': 1,169            'intervals': [10, 20],170            'request': {171                'url': "%s/foo" % (srv.url,),172                'method': 'POST'173            }174        }175        req2 = {176            'owner_id': '1234',177            'timestamp': 10,178            'attempts': 2,179            'intervals': [10, 30, 40],180            'request': {181                'url': "%s/foo" % (srv.url,),182                'method': 'POST'183            }184        }185        pop_all(msgs)186        yield worker.retry(req1)187        self.assertEqual(pop_all(msgs), [188            ('Retrying request', req1),189            ('Retry failed (500)', req1),190            ('No remaining retry intervals, discarding request', req1),191        ])192        yield worker.retry(req2)193        self.assertEqual(pop_all(msgs), [194            ('Retrying request', req2),195            ('Retry failed (500)', req2),196            ('No remaining retry intervals, discarding request', req2),197        ])198        self.assertEqual((yield zitems(worker.redis, pending_key('test'))), [])199    @inlineCallbacks200    def test_retry_timeout_reschedule(self):201        k = pending_key('test')202        msgs = self.patch_log()203        worker = yield self.mk_worker({'timeout': 3})204        srv = yield ToyServer.from_test(self)205        self.patch_reactor_call_later(worker.clock)206        yield worker.stop()207        @srv.app.route('/foo')208        def route(req):209            return Deferred()210        req = {211            'owner_id': '1234',212            'timestamp': 5,213            'attempts': 0,214            'intervals': [10, 20],215            'request': {216                'url': "%s/foo" % (srv.url,),217                'method': 'POST'218            }219        }220        pop_all(msgs)221        d = worker.retry(req)222        worker.clock.advance(2)223        self.assertEqual((yield zitems(worker.redis, k)), [])224        worker.clock.advance(4)225        yield d226        self.assertEqual(pop_all(msgs), [227            ('Retrying request', req),228            ('Retry timed out', req),229            ('Rescheduling retry', req),230        ])231        self.assertEqual((yield zitems(worker.redis, k)), [232            (5 + 20, {233                'owner_id': '1234',234                'timestamp': 5,235                'attempts': 1,236                'intervals': [10, 20],237                'request': {238                    'url': "%s/foo" % (srv.url,),239                    'method': 'POST'240                }241            }),242        ])243    @inlineCallbacks244    def test_retry_timeout_end(self):245        k = pending_key('test')246        msgs = self.patch_log()247        worker = yield self.mk_worker({'timeout': 3})248        srv = yield ToyServer.from_test(self)249        self.patch_reactor_call_later(worker.clock)250        yield worker.stop()251        @srv.app.route('/foo')252        def route(req):253            return Deferred()254        req = {255            'owner_id': '1234',256            'timestamp': 5,257            'attempts': 1,258            'intervals': [10, 20],259            'request': {260                'url': "%s/foo" % (srv.url,),261                'method': 'POST'262            }263        }264        pop_all(msgs)265        d = worker.retry(req)266        worker.clock.advance(2)267        self.assertEqual((yield zitems(worker.redis, k)), [])268        worker.clock.advance(4)269        yield d270        self.assertEqual(pop_all(msgs), [271            ('Retrying request', req),272            ('Retry timed out', req),273            ('No remaining retry intervals, discarding request', req),274        ])275        self.assertEqual((yield zitems(worker.redis, k)), [])276    @inlineCallbacks277    def test_retry_dec_req_count_success(self):278        worker = yield self.mk_worker()279        srv = yield ToyServer.from_test(self)280        @srv.app.route('/')281        def route(req):282            pass283        yield set_req_count(worker.redis, 'test', '1234', 3)284        yield worker.retry({285            'owner_id': '1234',286            'timestamp': 5,287            'attempts': 1,288            'intervals': [10, 20],289            'request': {290                'url': srv.url,291                'method': 'GET'292            }293        })294        self.assertEqual(295            (yield get_req_count(worker.redis, 'test', '1234')), 2)296    @inlineCallbacks297    def test_retry_dec_req_count_no_reattempt(self):298        worker = yield self.mk_worker()299        srv = yield ToyServer.from_test(self)300        @srv.app.route('/')301        def route(req):302            pass303        yield set_req_count(worker.redis, 'test', '1234', 3)304        yield worker.retry({305            'owner_id': '1234',306            'timestamp': 5,307            'attempts': 1,308            'intervals': [10, 20],309            'request': {310                'url': srv.url,311                'method': 'GET'312            }313        })314        self.assertEqual(315            (yield get_req_count(worker.redis, 'test', '1234')), 2)316    @inlineCallbacks317    def test_retry_no_dec_req_count_on_reattempt(self):318        worker = yield self.mk_worker()319        srv = yield ToyServer.from_test(self)320        @srv.app.route('/')321        def route(req):322            req.setResponseCode(500)323        yield set_req_count(worker.redis, 'test', '1234', 3)324        yield worker.retry({325            'owner_id': '1234',326            'timestamp': 5,327            'attempts': 0,328            'intervals': [10, 20],329            'request': {330                'url': srv.url,331                'method': 'GET'332            }333        })334        self.assertEqual(335            (yield get_req_count(worker.redis, 'test', '1234')), 3)336    @inlineCallbacks337    def test_loop(self):338        k = ready_key('test')339        msgs = self.patch_log()340        retries = self.patch_retry()341        worker = yield self.mk_worker({'frequency': 5})342        reqs = [{343            'owner_id': '1234',344            'timestamp': t,345            'attempts': 0,346            'intervals': [10],347            'request': {'foo': t}348        } for t in range(5, 30, 5)]349        yield add_ready(worker.redis, 'test', reqs)350        self.assertEqual(pop_all(msgs), [351            ('Polling for requests to retry',),352            ('Retrieving next request from ready set',),353            ('Ready set is empty, rechecking on next poll',),354        ])355        worker.clock.advance(5)356        req = yield retries.get()357        self.assertEqual(req, reqs[0])358        self.assertEqual((yield lvalues(worker.redis, k)), reqs[1:])359        self.assertEqual(pop_all(msgs), [360            ('Polling for requests to retry',),361            ('Retrieving next request from ready set',),362            ('Scheduling request for retrying', reqs[0]),363            ('Retrieving next request from ready set',),364        ])365        req = yield retries.get()366        self.assertEqual(req, reqs[1])367        self.assertEqual((yield lvalues(worker.redis, k)), reqs[2:])368        self.assertEqual(pop_all(msgs), [369            ('Scheduling request for retrying', reqs[1]),370            ('Retrieving next request from ready set',),371        ])372        req = yield retries.get()373        self.assertEqual(req, reqs[2])374        self.assertEqual((yield lvalues(worker.redis, k)), reqs[3:])375        self.assertEqual(pop_all(msgs), [376            ('Scheduling request for retrying', reqs[2]),377            ('Retrieving next request from ready set',),378        ])379        req = yield retries.get()380        self.assertEqual(req, reqs[3])381        self.assertEqual((yield lvalues(worker.redis, k)), reqs[4:])382        self.assertEqual(pop_all(msgs), [383            ('Scheduling request for retrying', reqs[3]),384            ('Retrieving next request from ready set',),385        ])386        req = yield retries.get()387        self.assertEqual(req, reqs[4])388        self.assertEqual((yield lvalues(worker.redis, k)), [])389        self.assertEqual(pop_all(msgs), [390            ('Scheduling request for retrying', reqs[4]),391            ('Retrieving next request from ready set',),392        ])393        worker.clock.advance(10)394        reqs = [{395            'owner_id': '1234',396            'timestamp': t,397            'attempts': 0,398            'intervals': [10],399            'request': {'foo': t}400        } for t in range(5, 15, 5)]401        yield add_ready(worker.redis, 'test', reqs)402        self.assertEqual(pop_all(msgs), [403            ('Ready set is empty, rechecking on next poll',),404        ])405        worker.clock.advance(5)406        req = yield retries.get()407        self.assertEqual(req, reqs[0])408        self.assertEqual((yield lvalues(worker.redis, k)), reqs[1:])409        self.assertEqual(pop_all(msgs), [410            ('Polling for requests to retry',),411            ('Retrieving next request from ready set',),412            ('Scheduling request for retrying', reqs[0]),413            ('Retrieving next request from ready set',),414        ])415        worker.clock.advance(5)416        req = yield retries.get()417        self.assertEqual(req, reqs[1])418        self.assertEqual((yield lvalues(worker.redis, k)), [])419        self.assertEqual(pop_all(msgs), [420            ('Scheduling request for retrying', reqs[1]),421            ('Retrieving next request from ready set',),422        ])423    @inlineCallbacks424    def test_loop_error(self):425        e = Exception(':/')426        def bad_poll():427            raise e428        errors = self.patch_on_error()429        self.patch_poll(staticmethod(bad_poll))430        worker = yield self.mk_worker({'frequency': 5})431        self.assertEqual(errors, [e])432        worker.clock.advance(5)433        self.assertEqual(errors, [e])...

type1.py

Source:type1.py

...208        del self.stack[-n:]209        if n == 1:210            return result[0]211        return result212    def pop_all(self):213        result = self.stack[:]214        del self.stack[:]215        if len(result) == 1:216            return result[0]217        return result218    def endchar(self):219        self.new_path()220        self.flush_stack()221    commands[14] = endchar222    def hsbw(self):223        # horizontal sidebearing and width224        sbx, wx = self.pop_all()225        self.cur = Point(sbx, 0)226    commands[13] = hsbw227    def seac(self):228        # standard encoding accented character229        asb, adx, ady, bchar, achar = self.pop_all()230    commands[32 + 6] = seac231    def sbw(self):232        # sidebearing and width233        sbx, sby, wx, wy = self.pop_all()234        self.cur = Point(sbx, sby)235    commands[32 + 7] = sbw236    def closepath(self):237        self.pop_all()238        self.closed = 1239    commands[9] = closepath240    def rlineto(self):241        dx, dy = self.pop_all()242        self.cur = self.cur + Point(dx, dy)243        self.path.append(tuple(self.cur))244    commands[5] = rlineto245    def hlineto(self):246        dx = self.pop_all()247        self.cur = self.cur + Point(dx, 0)248        self.path.append(tuple(self.cur))249    commands[6] = hlineto250    def vlineto(self):251        dy = self.pop_all()252        self.cur = self.cur + Point(0, dy)253        self.path.append(tuple(self.cur))254    commands[7] = vlineto255    def rmoveto(self):256        dx, dy = self.pop_all()257        self.cur = self.cur + Point(dx, dy)258        if self.in_flex:259            self.flex.append(self.cur)260        else:261            self.new_path()262            self.path.append(tuple(self.cur))263    commands[21] = rmoveto264    def hmoveto(self):265        dx = self.pop_all()266        self.cur = self.cur + Point(dx, 0)267        self.new_path()268        self.path.append(tuple(self.cur))269    commands[22] = hmoveto270    def vmoveto(self):271        dy = self.pop_all()272        self.cur = self.cur + Point(0, dy)273        self.new_path()274        self.path.append(tuple(self.cur))275    commands[4] = vmoveto276    def rrcurveto(self):277        dx1, dy1, dx2, dy2, dx3, dy3 = self.pop_all()278        d1 = self.cur + Point(dx1, dy1)279        d2 = d1 + Point(dx2, dy2)280        d3 = d2 + Point(dx3, dy3)281        self.cur = d3282        self.path.append(tuple(d1) +tuple(d2) + tuple(d3))283    commands[8] = rrcurveto284    def hvcurveto(self):285        dx1, dx2, dy2, dy3 = self.pop_all()286        d1 = self.cur + Point(dx1, 0)287        d2 = d1 + Point(dx2, dy2)288        d3 = d2 + Point(0, dy3)289        self.cur = d3290        self.path.append(tuple(d1) + tuple(d2) + tuple(d3))291    commands[31] = hvcurveto292    def vhcurveto(self):293        dy1, dx2, dy2, dx3 = self.pop_all()294        d1 = self.cur + Point(0, dy1)295        d2 = d1 + Point(dx2, dy2)296        d3 = d2 + Point(dx3, 0)297        self.cur = d3298        self.path.append(tuple(d1) + tuple(d2) + tuple(d3))299    commands[30] = vhcurveto300    def start_flex(self):301        self.in_flex = 1302        self.flex = []303    def end_flex(self):304        size, x, y = self.pop_all()305        d1, d2, d3 = self.flex[1:4]306        self.path.append(tuple(d1) + tuple(d2) + tuple(d3))307        d1, d2, d3 = self.flex[4:7]308        self.path.append(tuple(d1) + tuple(d2) + tuple(d3))309    def div(self):310        num1, num2 = self.pop(2)311        self.stack.append(float(num1) / num2)312    commands[32 + 12] = div313    def callothersubr(self):314        n, sn = self.pop(2)315        if n:316            self.ps_stack = self.pop(n)317        if sn == 3:318            self.ps_stack = [3]319    commands[32 + 16] = callothersubr320    def callsubr(self):321        num = self.pop(1)322        if num == 0:323            self.end_flex()324        elif num == 1:325            self.start_flex()326        elif 2 <= num <= 3:327            return328        else:329            self.execute(self.subrs[num])330    commands[10] = callsubr331    def pop_ps(self):332        value = self.ps_stack[-1]333        del self.ps_stack[-1]334        self.stack.append(value)335    commands[32 + 17] = pop_ps336    def subr_return(self):337        raise SubrReturn338    commands[11] = subr_return339    def setcurrentpoint(self):340        x, y = self.pop_all()341        self.cur = Point(x, y)342    commands[32 + 33] = setcurrentpoint343#344# read_outlines(FILENAME)345#346# Return the outlines of the glyphs in the Type1 font stored in the file347# FILENAME as a tuple (CHAR_STRINGS, INTERPRETER). CHAR_STRINGS is a348# dictionary mapping glyph names to strings containing the outline349# description, INTERPRETER is an instance of CharStringInterpreter350# initialized with the appropriate Subrs.351def read_outlines(filename):352    data = read_type1_file(filename)353    data = streamfilter.StringDecode(data, None)354    subrs, char_strings = parse_type1_file(data)...

buffer_test.py

Source:buffer_test.py

...89        values = [x for x in p.pop_until(0)]90        values_bis = [x for x in p.pop_until(0)]91        self.assertEqual(values, [])92        self.assertEqual(p.pos, 0)93    def test_pop_all(self):94        p = ArrayBuffer(10, 10)95        for i in range(10): p[i] = i96        values = [x for x in p.pop_all()]97        values_bis = [x for x in p.pop_all()]98        self.assertEqual(values, [0, 1, 2, 3, 4, 5, 6, 7, 8, 9])99        self.assertEqual(values_bis, [])100        self.assertEqual(p.pos, 10)101        p = ArrayBuffer(10, 10)102        for i in range(10): p[i] = i103        for _ in range(5): list(p.pop_one())104        values = [x for x in p.pop_all()]105        values_bis = [x for x in p.pop_all()]106        self.assertEqual(values, [5, 6, 7, 8, 9])107        self.assertEqual(values_bis, [])108        self.assertEqual(p.pos, 10)109        p = ArrayBuffer(10, 10)110        for i in range(10): p[i] = i111        for _ in range(10): list(p.pop_one())112        values = [x for x in p.pop_all()]113        values_bis = [x for x in p.pop_all()]114        self.assertEqual(values, [])115        self.assertEqual(values_bis, [])116        self.assertEqual(p.pos, 10)117        p = ArrayBuffer(0, 10)118        values = [x for x in p.pop_all()]119        values_bis = [x for x in p.pop_all()]120        self.assertEqual(values, [])121        self.assertEqual(p.pos, 0)122        p = ArrayBuffer(10, 5, -123)123        values = [x for x in p.pop_all()]124        values_bis = [x for x in p.pop_all()]125        self.assertEqual(values, [-123] * 10)126        self.assertEqual(values_bis, [])127class TestPrefixSumBuffer(unittest.TestCase):128    def test_pop_one(self):129        p = PrefixSumBuffer(10, 5, 123)130        values = [x for x in p.pop_one()]131        values_bis = [x for x in p.pop_one()]132        self.assertEqual(values, [123])133        self.assertEqual(values_bis, [123])134        p = PrefixSumBuffer(10, 5)135        values = [x for x in p.pop_one()]136        self.assertEqual(values, [0])137        p = PrefixSumBuffer(1, 1)138        values = [x for x in p.pop_one()]139        values_bis = [x for x in p.pop_one()]140        self.assertEqual(values, [0])141        self.assertEqual(values_bis, [])142    def test_add_interval(self):143        p = PrefixSumBuffer(10, 10)144        p.add_interval(0, 10, 5)145        values = [x for x in p.pop_all()]146        self.assertEqual(values, [5] * 10)147        p = PrefixSumBuffer(10, 10, 3)148        p.add_interval(5, 7, 10)149        values = [x for x in p.pop_all()]150        self.assertEqual(values, [3, 3, 3, 3, 3, 13, 13, 3, 3, 3])151        p = PrefixSumBuffer(10, 10)152        for i in range(10): p.add_interval(i+1, 10)153        values = [x for x in p.pop_all()]154        self.assertEqual(values, [0, 1, 2, 3, 4, 5, 6, 7, 8, 9])155    def test_pop_until(self):156        p = PrefixSumBuffer(10, 10)157        for i in range(10): p.add_interval(i+1, 10)158        values = [x for x in p.pop_until(5)]159        values_bis = [x for x in p.pop_until(5)]160        self.assertEqual(values, [0, 1, 2, 3, 4])161        self.assertEqual(values_bis, [])162        self.assertEqual(p.pos, 5)163        p = PrefixSumBuffer(10, 10)164        for i in range(10): p.add_interval(i+1, 10)165        for _ in range(5): list(p.pop_one())166        values = [x for x in p.pop_until(10)]167        values_bis = [x for x in p.pop_until(10)]168        self.assertEqual(values, [5, 6, 7, 8, 9])169        self.assertEqual(values_bis, [])170        self.assertEqual(p.pos, 10)171        p = PrefixSumBuffer(10, 10)172        for i in range(10): p.add_interval(i+1, 10)173        for _ in range(5): list(p.pop_one())174        values = [x for x in p.pop_until(5)]175        values_bis = [x for x in p.pop_until(5)]176        self.assertEqual(values, [])177        self.assertEqual(values_bis, [])178        self.assertEqual(p.pos, 5)179        p = PrefixSumBuffer(10, 10)180        for i in range(10): p.add_interval(i+1, 10)181        for _ in range(10): list(p.pop_one())182        values = [x for x in p.pop_until(10)]183        values_bis = [x for x in p.pop_until(10)]184        self.assertEqual(values, [])185        self.assertEqual(values_bis, [])186        self.assertEqual(p.pos, 10)187        p = PrefixSumBuffer(0, 10)188        values = [x for x in p.pop_until(0)]189        values_bis = [x for x in p.pop_until(0)]190        self.assertEqual(values, [])191        self.assertEqual(p.pos, 0)192    def test_pop_all(self):193        p = PrefixSumBuffer(10, 10)194        for i in range(10): p.add_interval(i+1, 10)195        values = [x for x in p.pop_all()]196        values_bis = [x for x in p.pop_all()]197        self.assertEqual(values, [0, 1, 2, 3, 4, 5, 6, 7, 8, 9])198        self.assertEqual(values_bis, [])199        self.assertEqual(p.pos, 10)200        p = PrefixSumBuffer(10, 10)201        for i in range(10): p.add_interval(i+1, 10)202        for _ in range(5): list(p.pop_one())203        values = [x for x in p.pop_all()]204        values_bis = [x for x in p.pop_all()]205        self.assertEqual(values, [5, 6, 7, 8, 9])206        self.assertEqual(values_bis, [])207        self.assertEqual(p.pos, 10)208        p = PrefixSumBuffer(10, 10)209        for i in range(10): p.add_interval(i+1, 10)210        for _ in range(10): list(p.pop_one())211        values = [x for x in p.pop_all()]212        values_bis = [x for x in p.pop_all()]213        self.assertEqual(values, [])214        self.assertEqual(values_bis, [])215        self.assertEqual(p.pos, 10)216        p = PrefixSumBuffer(0, 10)217        values = [x for x in p.pop_all()]218        values_bis = [x for x in p.pop_all()]219        self.assertEqual(values, [])220        self.assertEqual(p.pos, 0)221        p = PrefixSumBuffer(10, 5)222        values = [x for x in p.pop_all()]...

test_readings.py

Source:test_readings.py

...123        # Prepare124        reading = Reading()125        self.queue.push(reading)126        # Execute127        res = self.queue.pop_all()128        # Assert129        self.assertEqual(res, [reading])130    def test_pop_all_several(self):131        # Prepare132        reading1 = Reading()133        self.queue.push(reading1)134        reading2 = Reading()135        self.queue.push(reading2)136        reading3 = Reading()137        self.queue.push(reading3)138        # Execute139        res = self.queue.pop_all()140        # Assert141        self.assertEqual(res, [reading1, reading2, reading3])142    def test_pop_all_empty_origin(self):143        # Prepare144        reading1 = Reading()145        self.queue.push(reading1)146        reading2 = Reading()147        self.queue.push(reading2)148        reading3 = Reading()149        self.queue.push(reading3)150        # Execute151        res = self.queue.pop_all()152        # Assert153        self.assertEqual(self.queue.queue, list())154class TestManageReadings(unittest.TestCase):155    @patch('threading.Thread')156    @patch('rfid_controller.scripts.lib.sources.robot.Robot')157    @patch('rfid_controller.scripts.lib.sources.helpers.StoreReadings')158    def setUp(self, mock1, mock2, mock3):159        self.reading_queue = MagicMock()160        self.manage_readings = ManageReadings(self.reading_queue, "111-111-1-C1", "C1")161        self.manage_readings.db_reading = MagicMock()162        self.count_readings = MagicMock()163        self.robot = MagicMock()164        self.store_readings = MagicMock()165    def test_count_twist_store_call_count(self):...

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.