How to use queue method in wpt

Best JavaScript code snippet using wpt

test_func.py

Source:test_func.py

...42        """43        return str(uuid.uuid4())44    def test_enqueue_response_status(self):45        job_id = self._get_job_id()46        response = self.queue.enqueue(47            payload=self._test_payload_1,48            interval=10000,  # 10s (10000ms)49            job_id=job_id,50            queue_id=self._test_queue_id,51            queue_type=self._test_queue_type,52        )53        self.assertEqual(response['status'], 'queued')54    def test_enqueue_job_queue_existence(self):55        job_id = self._get_job_id()56        response = self.queue.enqueue(57            payload=self._test_payload_1,58            interval=10000,  # 10s (10000ms)59            job_id=job_id,60            queue_id=self._test_queue_id,61            queue_type=self._test_queue_type,62        )63        # check if the job queue exists64        queue_name = '%s:%s:%s' % (65            self.queue._key_prefix,66            self._test_queue_type,67            self._test_queue_id68        )69        self.assertTrue(self.queue._r.exists(queue_name))70    def test_enqueue_job_existence_in_job_queue(self):71        job_id = self._get_job_id()72        response = self.queue.enqueue(73            payload=self._test_payload_1,74            interval=10000,  # 10s (10000ms)75            job_id=job_id,76            queue_id=self._test_queue_id,77            queue_type=self._test_queue_type,78        )79        # check if the queue contains the job we just pushed (by peeking)80        queue_name = '%s:%s:%s' % (81            self.queue._key_prefix,82            self._test_queue_type,83            self._test_queue_id84        )85        latest_job_id = self.queue._r.lrange(queue_name, -1, -1)86        self.assertEqual(latest_job_id, [job_id])87    def test_enqueue_job_queue_length(self):88        job_id = self._get_job_id()89        response = self.queue.enqueue(90            payload=self._test_payload_1,91            interval=10000,  # 10s (10000ms)92            job_id=job_id,93            queue_id=self._test_queue_id,94            queue_type=self._test_queue_type,95        )96        # check if the queue length is one97        queue_name = '%s:%s:%s' % (98            self.queue._key_prefix,99            self._test_queue_type,100            self._test_queue_id101        )102        queue_length = self.queue._r.llen(queue_name)103        self.assertEqual(queue_length, 1)104    def test_enqueue_payload_dump(self):105        job_id = self._get_job_id()106        response = self.queue.enqueue(107            payload=self._test_payload_1,108            interval=10000,  # 10s (10000ms)109            job_id=job_id,110            queue_id=self._test_queue_id,111            queue_type=self._test_queue_type,112        )113        # check if the payload is saved in the appropriate structure114        payload_map_name = '%s:payload' % (self.queue._key_prefix)115        # check if the payload map exists116        self.assertTrue(self.queue._r.exists(payload_map_name))117    def test_enqueue_payload_encode_decode(self):118        job_id = self._get_job_id()119        response = self.queue.enqueue(120            payload=self._test_payload_1,121            interval=10000,  # 10s (10000ms)122            job_id=job_id,123            queue_id=self._test_queue_id,124            queue_type=self._test_queue_type,125        )126        payload_map_name = '%s:payload' % (self.queue._key_prefix)127        payload_map_key = '%s:%s:%s' % (128            self._test_queue_type, self._test_queue_id, job_id)129        raw_payload = self.queue._r.hget(payload_map_name, payload_map_key)130        # decode the payload from msgpack to dictionary131        payload = msgpack.unpackb(raw_payload[1:-1])132        self.assertEqual(payload, self._test_payload_1)133    def test_enqueue_interval_map_existence(self):134        job_id = self._get_job_id()135        response = self.queue.enqueue(136            payload=self._test_payload_1,137            interval=10000,  # 10s (10000ms)138            job_id=job_id,139            queue_id=self._test_queue_id,140            queue_type=self._test_queue_type,141        )142        # check if interval is saved in the appropriate structure143        interval_map_name = '%s:interval' % (self.queue._key_prefix)144        # check if interval map exists145        self.assertTrue(self.queue._r.exists(interval_map_name))146    def test_enqueue_interval_value(self):147        job_id = self._get_job_id()148        response = self.queue.enqueue(149            payload=self._test_payload_1,150            interval=10000,  # 10s (10000ms)151            job_id=job_id,152            queue_id=self._test_queue_id,153            queue_type=self._test_queue_type,154        )155        # check if interval is saved in the appropriate structure156        interval_map_name = '%s:interval' % (self.queue._key_prefix)157        interval_map_key = '%s:%s' % (158            self._test_queue_type, self._test_queue_id)159        interval = self.queue._r.hget(160            interval_map_name, interval_map_key)161        self.assertEqual(interval, '10000')  # 10s (10000ms)162    def test_enqueue_requeue_limit_map_existence(self):163        job_id = self._get_job_id()164        response = self.queue.enqueue(165            payload=self._test_payload_1,166            interval=10000,  # 10s (10000ms)167            job_id=job_id,168            queue_id=self._test_queue_id,169            queue_type=self._test_queue_type170            # without a requeue limit parameter171        )172        # check if requeue limit is saved in the appropriate structure173        requeue_limit_map_name = '%s:%s:%s:requeues_remaining' % (174            self.queue._key_prefix,175            self._test_queue_type,176            self._test_queue_id,177        )178        # check if requeue limit  map exists179        self.assertTrue(self.queue._r.exists(requeue_limit_map_name))180        job_id = self._get_job_id()181        response = self.queue.enqueue(182            payload=self._test_payload_1,183            interval=10000,  # 10s (10000ms)184            job_id=job_id,185            queue_id=self._test_queue_id,186            queue_type=self._test_queue_type,187            requeue_limit=self._test_requeue_limit_5188        )189        # check if requeue limit is saved in the appropriate structure190        requeue_limit_map_name = '%s:%s:%s:requeues_remaining' % (191            self.queue._key_prefix,192            self._test_queue_type,193            self._test_queue_id,194        )195        # check if requeue limit  map exists196        self.assertTrue(self.queue._r.exists(requeue_limit_map_name))197    def test_enqueue_requeue_limit_value(self):198        # without requeue limit (but reading from the config)199        job_id = self._get_job_id()200        response = self.queue.enqueue(201            payload=self._test_payload_1,202            interval=10000,  # 10s (10000ms)203            job_id=job_id,204            queue_id=self._test_queue_id,205            queue_type=self._test_queue_type206            # without requeue limit.207        )208        # check if requeue limit is saved in the appropriate structure209        requeue_limit_map_name = '%s:%s:%s:requeues_remaining' % (210            self.queue._key_prefix,211            self._test_queue_type,212            self._test_queue_id,213        )214        requeues_remaining = self.queue._r.hget(215            requeue_limit_map_name, job_id)216        self.assertEqual(requeues_remaining, '-1')  # from the config file.217        # with requeue limit in the enqueue function.218        job_id = self._get_job_id()219        response = self.queue.enqueue(220            payload=self._test_payload_1,221            interval=10000,  # 10s (10000ms)222            job_id=job_id,223            queue_id=self._test_queue_id,224            queue_type=self._test_queue_type,225            requeue_limit=self._test_requeue_limit_5226        )227        # check if requeue limit is saved in the appropriate structure228        requeue_limit_map_name = '%s:%s:%s:requeues_remaining' % (229            self.queue._key_prefix,230            self._test_queue_type,231            self._test_queue_id,232        )233        requeues_remaining = self.queue._r.hget(234            requeue_limit_map_name, job_id)235        self.assertEqual(requeues_remaining, '5') # 5 retries remaining.236    def test_enqueue_ready_set(self):237        job_id = self._get_job_id()238        response = self.queue.enqueue(239            payload=self._test_payload_1,240            interval=10000,  # 10s (10000ms)241            job_id=job_id,242            queue_id=self._test_queue_id,243            queue_type=self._test_queue_type,244        )245        sorted_set_name = '%s:%s' % (246            self.queue._key_prefix, self._test_queue_type)247        self.assertTrue(self.queue._r.exists(sorted_set_name))248    def test_enqueue_ready_set_contents(self):249        job_id = self._get_job_id()250        start_time = str(generate_epoch())251        response = self.queue.enqueue(252            payload=self._test_payload_1,253            interval=10000,  # 10s (10000ms)254            job_id=job_id,255            queue_id=self._test_queue_id,256            queue_type=self._test_queue_type,257        )258        end_time = str(generate_epoch())259        sorted_set_name = '%s:%s' % (260            self.queue._key_prefix, self._test_queue_type)261        queue_id_list = self.queue._r.zrangebyscore(262            sorted_set_name,263            start_time,264            end_time)265        # check if exactly one item in the list266        self.assertEqual(len(queue_id_list), 1)267        # check the value to match the queue_id268        self.assertEqual(queue_id_list[0], self._test_queue_id)269    def test_enqueue_queue_type_ready_set(self):270        job_id = self._get_job_id()271        response = self.queue.enqueue(272            payload=self._test_payload_1,273            interval=10000,  # 10s (10000ms)274            job_id=job_id,275            queue_id=self._test_queue_id,276            queue_type=self._test_queue_type,277        )278        # check the queue type ready set.279        queue_type_ready_set = self.queue._r.smembers(280            '%s:ready:queue_type' % self.queue._key_prefix)281        self.assertEqual(len(queue_type_ready_set), 1)282        self.assertEqual(queue_type_ready_set.pop(), self._test_queue_type)283    def test_enqueue_queue_type_active_set(self):284        job_id = self._get_job_id()285        response = self.queue.enqueue(286            payload=self._test_payload_1,287            interval=10000,  # 10s (10000ms)288            job_id=job_id,289            queue_id=self._test_queue_id,290            queue_type=self._test_queue_type,291        )292        queue_type_ready_set = self.queue._r.smembers(293            '%s:active:queue_type' % self.queue._key_prefix)294        self.assertEqual(len(queue_type_ready_set), 0)295    def test_enqueue_metrics_global_enqueue_counter(self):296        job_id = self._get_job_id()297        response = self.queue.enqueue(298            payload=self._test_payload_1,299            interval=10000,  # 10s (10000ms)300            job_id=job_id,301            queue_id=self._test_queue_id,302            queue_type=self._test_queue_type,303        )304        timestamp = int(generate_epoch())305        # epoch for the minute.306        timestamp_minute = str(int(math.floor(timestamp / 60000.0) * 60000))307        counter_value = self.queue._r.get('%s:enqueue_counter:%s' % (308            self.queue._key_prefix, timestamp_minute))309        self.assertEqual(counter_value, '1')310    def test_enqueue_metrics_per_queue_enqueue_counter(self):311        job_id = self._get_job_id()312        response = self.queue.enqueue(313            payload=self._test_payload_1,314            interval=10000,  # 10s (10000ms)315            job_id=job_id,316            queue_id=self._test_queue_id,317            queue_type=self._test_queue_type,318        )319        response = self.queue.dequeue(320            queue_type=self._test_queue_type321        )322        timestamp = int(generate_epoch())323        # epoch for the minute.324        timestamp_minute = str(int(math.floor(timestamp / 60000.0) * 60000))325        counter_value = self.queue._r.get('%s:%s:%s:enqueue_counter:%s' % (326            self.queue._key_prefix,327            self._test_queue_type,328            self._test_queue_id,329            timestamp_minute))330        self.assertEqual(counter_value, '1')331    def test_enqueue_second_job_status(self):332        # job 1333        job_id = self._get_job_id()334        response = self.queue.enqueue(335            payload=self._test_payload_1,336            interval=10000,  # 10s (10000ms)337            job_id=job_id,338            queue_id=self._test_queue_id,339            queue_type=self._test_queue_type,340        )341        # job 2342        job_id = self._get_job_id()343        response = self.queue.enqueue(344            payload=self._test_payload_2,345            interval=20000,  # 20s (20000ms)346            job_id=job_id,347            queue_id=self._test_queue_id,348            queue_type=self._test_queue_type,349        )350        self.assertEqual(response['status'], 'queued')351    def test_enqueue_second_job_queue_existence(self):352        # job 1353        job_id = self._get_job_id()354        response = self.queue.enqueue(355            payload=self._test_payload_1,356            interval=10000,  # 10s (10000ms)357            job_id=job_id,358            queue_id=self._test_queue_id,359            queue_type=self._test_queue_type,360        )361        # job 2362        job_id = self._get_job_id()363        response = self.queue.enqueue(364            payload=self._test_payload_2,365            interval=20000,  # 20s (20000ms)366            job_id=job_id,367            queue_id=self._test_queue_id,368            queue_type=self._test_queue_type,369        )370        queue_name = '%s:%s:%s' % (371            self.queue._key_prefix, self._test_queue_type, self._test_queue_id)372        self.assertTrue(self.queue._r.exists(queue_name))373    def test_enqueue_second_job_existence_in_job_queue(self):374        # job 1375        job_id = self._get_job_id()376        response = self.queue.enqueue(377            payload=self._test_payload_1,378            interval=10000,  # 10s (10000ms)379            job_id=job_id,380            queue_id=self._test_queue_id,381            queue_type=self._test_queue_type,382        )383        # job 2384        job_id = self._get_job_id()385        response = self.queue.enqueue(386            payload=self._test_payload_2,387            interval=20000,  # 20s (20000ms)388            job_id=job_id,389            queue_id=self._test_queue_id,390            queue_type=self._test_queue_type,391        )392        queue_name = '%s:%s:%s' % (393            self.queue._key_prefix, self._test_queue_type, self._test_queue_id)394        latest_job_id = self.queue._r.lrange(queue_name, -1, -1)395        self.assertEqual(latest_job_id, [job_id])396    def test_enqueue_second_job_queue_length(self):397        # job 1398        job_id = self._get_job_id()399        response = self.queue.enqueue(400            payload=self._test_payload_1,401            interval=10000,  # 10s (10000ms)402            job_id=job_id,403            queue_id=self._test_queue_id,404            queue_type=self._test_queue_type,405        )406        # job 2407        job_id = self._get_job_id()408        response = self.queue.enqueue(409            payload=self._test_payload_2,410            interval=20000,  # 20s (20000ms)411            job_id=job_id,412            queue_id=self._test_queue_id,413            queue_type=self._test_queue_type,414        )415        queue_name = '%s:%s:%s' % (416            self.queue._key_prefix, self._test_queue_type, self._test_queue_id)417        # check if the queue length is two418        queue_length = self.queue._r.llen(queue_name)419        self.assertEqual(queue_length, 2)420    def test_enqueue_second_job_payload_dump(self):421        # job 1422        job_id = self._get_job_id()423        response = self.queue.enqueue(424            payload=self._test_payload_1,425            interval=10000,  # 10s (10000ms)426            job_id=job_id,427            queue_id=self._test_queue_id,428            queue_type=self._test_queue_type,429        )430        # job 2431        job_id = self._get_job_id()432        response = self.queue.enqueue(433            payload=self._test_payload_2,434            interval=20000,  # 20s (20000ms)435            job_id=job_id,436            queue_id=self._test_queue_id,437            queue_type=self._test_queue_type,438        )439        payload_map_name = '%s:payload' % (self.queue._key_prefix)440        # check if the payload map exists441        self.assertTrue(self.queue._r.exists(payload_map_name))442    def test_enqueue_second_job_payload_encode_decode(self):443        # job 1444        job_id = self._get_job_id()445        response = self.queue.enqueue(446            payload=self._test_payload_1,447            interval=10000,  # 10s (10000ms)448            job_id=job_id,449            queue_id=self._test_queue_id,450            queue_type=self._test_queue_type,451        )452        # job 2453        job_id = self._get_job_id()454        response = self.queue.enqueue(455            payload=self._test_payload_2,456            interval=20000,  # 20s (20000ms)457            job_id=job_id,458            queue_id=self._test_queue_id,459            queue_type=self._test_queue_type,460        )461        payload_map_name = '%s:payload' % (self.queue._key_prefix)462        payload_map_key = '%s:%s:%s' % (463            self._test_queue_type, self._test_queue_id, job_id)464        raw_payload = self.queue._r.hget(payload_map_name, payload_map_key)465        # decode the payload from msgpack to dictionary466        payload = msgpack.unpackb(raw_payload[1:-1])467        self.assertEqual(payload, self._test_payload_2)468    def test_enqueue_second_job_interval_map_existence(self):469        # job 1470        job_id = self._get_job_id()471        response = self.queue.enqueue(472            payload=self._test_payload_1,473            interval=10000,  # 10s (10000ms)474            job_id=job_id,475            queue_id=self._test_queue_id,476            queue_type=self._test_queue_type,477        )478        # job 2479        job_id = self._get_job_id()480        response = self.queue.enqueue(481            payload=self._test_payload_2,482            interval=20000,  # 20s (20000ms)483            job_id=job_id,484            queue_id=self._test_queue_id,485            queue_type=self._test_queue_type,486        )487        interval_map_name = '%s:interval' % (self.queue._key_prefix)488        # check if interval map exists489        self.assertTrue(self.queue._r.exists(interval_map_name))490    def test_enqueue_second_job_interval_value(self):491        # job 1492        job_id = self._get_job_id()493        response = self.queue.enqueue(494            payload=self._test_payload_1,495            interval=10000,  # 10s (10000ms)496            job_id=job_id,497            queue_id=self._test_queue_id,498            queue_type=self._test_queue_type,499        )500        # job 2501        job_id = self._get_job_id()502        response = self.queue.enqueue(503            payload=self._test_payload_2,504            interval=20000,  # 20s (20000ms)505            job_id=job_id,506            queue_id=self._test_queue_id,507            queue_type=self._test_queue_type,508        )509        interval_map_name = '%s:interval' % (self.queue._key_prefix)510        interval_map_key = '%s:%s' % (511            self._test_queue_type, self._test_queue_id)512        interval = self.queue._r.hget(interval_map_name, interval_map_key)513        self.assertEqual(interval, '20000')  # 20s (20000ms)514    def test_enqueue_second_job_ready_set(self):515        # job 1516        job_id = self._get_job_id()517        response = self.queue.enqueue(518            payload=self._test_payload_1,519            interval=10000,  # 10s (10000ms)520            job_id=job_id,521            queue_id=self._test_queue_id,522            queue_type=self._test_queue_type,523        )524        # job 2525        job_id = self._get_job_id()526        response = self.queue.enqueue(527            payload=self._test_payload_2,528            interval=20000,  # 20s (20000ms)529            job_id=job_id,530            queue_id=self._test_queue_id,531            queue_type=self._test_queue_type,532        )533        sorted_set_name = '%s:%s' % (534            self.queue._key_prefix, self._test_queue_type)535        self.assertTrue(self.queue._r.exists(sorted_set_name))536    def test_enqueue_second_job_ready_set_contents(self):537        # job 1538        job_id = self._get_job_id()539        response = self.queue.enqueue(540            payload=self._test_payload_1,541            interval=10000,  # 10s (10000ms)542            job_id=job_id,543            queue_id=self._test_queue_id,544            queue_type=self._test_queue_type,545        )546        # sleeping for 500ms to ensure that the547        # time difference between two enqueues is548        # measurable for the test cases.549        time.sleep(0.5)550        # job 2551        job_id = self._get_job_id()552        start_time = str(generate_epoch())553        response = self.queue.enqueue(554            payload=self._test_payload_2,555            interval=20000,  # 20s (20000ms)556            job_id=job_id,557            queue_id=self._test_queue_id,558            queue_type=self._test_queue_type,559        )560        sorted_set_name = '%s:%s' % (561            self.queue._key_prefix, self._test_queue_type)562        end_time = str(generate_epoch())563        queue_id_list = self.queue._r.zrangebyscore(564            sorted_set_name,565            start_time,566            end_time)567        self.assertEqual(len(queue_id_list), 0)568    def test_enqueue_second_job_queue_type_ready_set(self):569        # job 1570        job_id = self._get_job_id()571        response = self.queue.enqueue(572            payload=self._test_payload_1,573            interval=10000,  # 10s (10000ms)574            job_id=job_id,575            queue_id=self._test_queue_id,576            queue_type=self._test_queue_type,577        )578        # job 2579        job_id = self._get_job_id()580        start_time = str(generate_epoch())581        response = self.queue.enqueue(582            payload=self._test_payload_2,583            interval=20000,  # 20s (20000ms)584            job_id=job_id,585            queue_id=self._test_queue_id,586            queue_type=self._test_queue_type,587        )588        # check the queue type ready set.589        queue_type_ready_set = self.queue._r.smembers(590            '%s:ready:queue_type' % self.queue._key_prefix)591        self.assertEqual(len(queue_type_ready_set), 1)592        self.assertEqual(queue_type_ready_set.pop(), self._test_queue_type)593    def test_enqueue_second_job_queue_type_active_set(self):594        # job 1595        job_id = self._get_job_id()596        response = self.queue.enqueue(597            payload=self._test_payload_1,598            interval=10000,  # 10s (10000ms)599            job_id=job_id,600            queue_id=self._test_queue_id,601            queue_type=self._test_queue_type,602        )603        # job 2604        job_id = self._get_job_id()605        start_time = str(generate_epoch())606        response = self.queue.enqueue(607            payload=self._test_payload_2,608            interval=20000,  # 20s (20000ms)609            job_id=job_id,610            queue_id=self._test_queue_id,611            queue_type=self._test_queue_type,612        )613        queue_type_ready_set = self.queue._r.smembers(614            '%s:active:queue_type' % self.queue._key_prefix)615        self.assertEqual(len(queue_type_ready_set), 0)616    def test_dequeue_response_status_failure(self):617        response = self.queue.dequeue(618            queue_type=self._test_queue_type619        )620        self.assertEqual(response['status'], 'failure')621    def test_dequeue_response_status_success_without_requeue_limit(self):622        job_id = self._get_job_id()623        response = self.queue.enqueue(624            payload=self._test_payload_1,625            interval=10000,  # 10s (10000ms)626            job_id=job_id,627            queue_id=self._test_queue_id,628            queue_type=self._test_queue_type,629            # without requeue limit630        )631        # dequeue from the queue_type632        response = self.queue.dequeue(633            queue_type=self._test_queue_type634        )635        # check all the responses636        self.assertEqual(response['status'], 'success')637        self.assertEqual(response['queue_id'], self._test_queue_id)638        self.assertEqual(response['job_id'], job_id)639        self.assertEqual(response['payload'], self._test_payload_1)640        self.assertEqual(response['requeues_remaining'], -1)  # from the config641    def test_dequeue_response_status_success_with_requeue_limit(self):642        # with requeue limit passed explicitly643        job_id = self._get_job_id()644        response = self.queue.enqueue(645            payload=self._test_payload_1,646            interval=10000,  # 10s (10000ms)647            job_id=job_id,648            queue_id=self._test_queue_id,649            queue_type=self._test_queue_type,650            requeue_limit=self._test_requeue_limit_5651        )652        # dequeue from the queue_type653        response = self.queue.dequeue(654            queue_type=self._test_queue_type655        )656        # check all the responses657        self.assertEqual(response['status'], 'success')658        self.assertEqual(response['queue_id'], self._test_queue_id)659        self.assertEqual(response['job_id'], job_id)660        self.assertEqual(response['payload'], self._test_payload_1)661        self.assertEqual(662            response['requeues_remaining'], self._test_requeue_limit_5)663    def test_dequeue_job_queue_existence(self):664        job_id = self._get_job_id()665        response = self.queue.enqueue(666            payload=self._test_payload_1,667            interval=10000,  # 10s (10000ms)668            job_id=job_id,669            queue_id=self._test_queue_id,670            queue_type=self._test_queue_type,671        )672        # dequeue from the queue_type673        response = self.queue.dequeue(674            queue_type=self._test_queue_type675        )676        queue_name = '%s:%s:%s' % (677            self.queue._key_prefix, self._test_queue_type, self._test_queue_id)678        self.assertFalse(self.queue._r.exists(queue_name))679    def test_dequeue_time_keeper_existence(self):680        job_id = self._get_job_id()681        response = self.queue.enqueue(682            payload=self._test_payload_1,683            interval=10000,  # 10s (10000ms)684            job_id=job_id,685            queue_id=self._test_queue_id,686            queue_type=self._test_queue_type,687        )688        # dequeue from the queue_type689        response = self.queue.dequeue(690            queue_type=self._test_queue_type691        )692        # time keeper key should exist693        time_keeper_key_name = '%s:%s:%s:time' % (694            self.queue._key_prefix,695            self._test_queue_type, self._test_queue_id696        )697        self.assertTrue(self.queue._r.exists(time_keeper_key_name))698    def test_dequeue_ready_sorted_set_existence(self):699        job_id = self._get_job_id()700        response = self.queue.enqueue(701            payload=self._test_payload_1,702            interval=10000,  # 10s (10000ms)703            job_id=job_id,704            queue_id=self._test_queue_id,705            queue_type=self._test_queue_type,706        )707        # dequeue from the queue_type708        response = self.queue.dequeue(709            queue_type=self._test_queue_type710        )711        # the sorted set should not exists712        sorted_set_name = '%s:%s' % (713            self.queue._key_prefix, self._test_queue_type)714        self.assertFalse(self.queue._r.exists(sorted_set_name))715    def test_dequeue_active_sorted_set(self):716        job_id = self._get_job_id()717        start_time = str(generate_epoch())718        response = self.queue.enqueue(719            payload=self._test_payload_1,720            interval=10000,  # 10s (10000ms)721            job_id=job_id,722            queue_id=self._test_queue_id,723            queue_type=self._test_queue_type,724        )725        # dequeue from the queue_type726        response = self.queue.dequeue(727            queue_type=self._test_queue_type728        )729        # the job should exist in the active set with the timestamp730        # it was picked up with.731        active_sorted_set_name = '%s:%s:active' % (732            self.queue._key_prefix,733            self._test_queue_type734        )735        end_time = str(generate_epoch())736        job_expire_timestamp = str(737            int(end_time) + self.queue._job_expire_interval)738        job_id_list = self.queue._r.zrangebyscore(739            active_sorted_set_name,740            start_time,741            job_expire_timestamp)742        # check if there is exactly one job in the743        # active sorted set744        self.assertEqual(len(job_id_list), 1)745    def test_dequeue_time_keeper_expiry(self):746        job_id = self._get_job_id()747        response = self.queue.enqueue(748            payload=self._test_payload_1,749            interval=1000,  # 1s (1000ms)750            job_id=job_id,751            queue_id=self._test_queue_id,752            queue_type=self._test_queue_type,753        )754        # dequeue from the queue_type755        response = self.queue.dequeue(756            queue_type=self._test_queue_type757        )758        # wait for the interval duration and check that the759        # time keeper should have expired760        time.sleep(self.queue._job_expire_interval / 1000.00)  # in seconds761        time_keeper_key_name = '%s:%s:%s:time' % (762            self.queue._key_prefix,763            self._test_queue_type, self._test_queue_id764        )765        self.assertFalse(self.queue._r.exists(time_keeper_key_name))766    def test_dequeue_ready_queue_type_set(self):767        job_id = self._get_job_id()768        response = self.queue.enqueue(769            payload=self._test_payload_1,770            interval=10000,  # 10s (10000ms)771            job_id=job_id,772            queue_id=self._test_queue_id,773            queue_type=self._test_queue_type,774        )775        # dequeue from the queue_type776        response = self.queue.dequeue(777            queue_type=self._test_queue_type778        )779        # the ready queue type set should have 0 items780        queue_type_ready_set = self.queue._r.smembers(781            '%s:ready:queue_type' % self.queue._key_prefix)782        self.assertEqual(len(queue_type_ready_set), 0)783    def test_dequeue_active_queue_type_set(self):784        job_id = self._get_job_id()785        response = self.queue.enqueue(786            payload=self._test_payload_1,787            interval=10000,  # 10s (10000ms)788            job_id=job_id,789            queue_id=self._test_queue_id,790            queue_type=self._test_queue_type,791        )792        # dequeue from the queue_type793        response = self.queue.dequeue(794            queue_type=self._test_queue_type795        )796        # the active queue type set should have one item797        queue_type_ready_set = self.queue._r.smembers(798            '%s:active:queue_type' % self.queue._key_prefix)799        self.assertEqual(len(queue_type_ready_set), 1)800        self.assertEqual(queue_type_ready_set.pop(), self._test_queue_type)801    def test_dequeue_metrics_global_dequeue_counter(self):802        job_id = self._get_job_id()803        response = self.queue.enqueue(804            payload=self._test_payload_1,805            interval=10000,  # 10s (10000ms)806            job_id=job_id,807            queue_id=self._test_queue_id,808            queue_type=self._test_queue_type,809        )810        response = self.queue.dequeue(811            queue_type=self._test_queue_type812        )813        timestamp = int(generate_epoch())814        # epoch for the minute.815        timestamp_minute = str(int(math.floor(timestamp / 60000.0) * 60000))816        counter_value = self.queue._r.get('%s:dequeue_counter:%s' % (817            self.queue._key_prefix, timestamp_minute))818        self.assertEqual(counter_value, '1')819    def test_dequeue_metrics_per_queue_dequeue_counter(self):820        job_id = self._get_job_id()821        response = self.queue.enqueue(822            payload=self._test_payload_1,823            interval=10000,  # 10s (10000ms)824            job_id=job_id,825            queue_id=self._test_queue_id,826            queue_type=self._test_queue_type,827        )828        response = self.queue.dequeue(829            queue_type=self._test_queue_type830        )831        timestamp = int(generate_epoch())832        # epoch for the minute.833        timestamp_minute = str(int(math.floor(timestamp / 60000.0) * 60000))834        counter_value = self.queue._r.get('%s:%s:%s:dequeue_counter:%s' % (835            self.queue._key_prefix,836            self._test_queue_type,837            self._test_queue_id,838            timestamp_minute))839        self.assertEqual(counter_value, '1')840    def test_finish_on_empty_queue(self):841        job_id = self._get_job_id()842        response = self.queue.finish(843            job_id=job_id,844            queue_id='doesnotexist',845            queue_type=self._test_queue_type846        )847        self.assertEqual(response['status'], 'failure')848    def test_finish_response_status(self):849        job_id = self._get_job_id()850        response = self.queue.enqueue(851            payload=self._test_payload_1,852            interval=10000,  # 10s (10000ms)853            job_id=job_id,854            queue_id=self._test_queue_id,855            queue_type=self._test_queue_type,856        )857        # dequeue from the queue_type858        response = self.queue.dequeue(859            queue_type=self._test_queue_type860        )861        response = self.queue.finish(862            job_id=job_id,863            queue_id=response['queue_id'],864            queue_type=self._test_queue_type865        )866        self.assertEqual(response['status'], 'success')867    def test_finish_ready_sorted_set_existence(self):868        job_id = self._get_job_id()869        response = self.queue.enqueue(870            payload=self._test_payload_1,871            interval=10000,  # 10s (10000ms)872            job_id=job_id,873            queue_id=self._test_queue_id,874            queue_type=self._test_queue_type,875        )876        # dequeue from the queue_type877        response = self.queue.dequeue(878            queue_type=self._test_queue_type879        )880        response = self.queue.finish(881            job_id=job_id,882            queue_id=response['queue_id'],883            queue_type=self._test_queue_type884        )885        self.assertFalse(886            self.queue._r.exists('%s:%s' % (887                self.queue._key_prefix, self._test_queue_type)))888    def test_finish_active_sorted_set_existence(self):889        job_id = self._get_job_id()890        response = self.queue.enqueue(891            payload=self._test_payload_1,892            interval=10000,  # 10s (10000ms)893            job_id=job_id,894            queue_id=self._test_queue_id,895            queue_type=self._test_queue_type,896        )897        # dequeue from the queue_type898        response = self.queue.dequeue(899            queue_type=self._test_queue_type900        )901        response = self.queue.finish(902            job_id=job_id,903            queue_id=response['queue_id'],904            queue_type=self._test_queue_type905        )906        self.assertFalse(907            self.queue._r.exists('%s:%s:active' % (908                self.queue._key_prefix, self._test_queue_type)))909    def test_finish_payload_existence(self):910        job_id = self._get_job_id()911        response = self.queue.enqueue(912            payload=self._test_payload_1,913            interval=10000,  # 10s (10000ms)914            job_id=job_id,915            queue_id=self._test_queue_id,916            queue_type=self._test_queue_type,917        )918        # dequeue from the queue_type919        response = self.queue.dequeue(920            queue_type=self._test_queue_type921        )922        response = self.queue.finish(923            job_id=job_id,924            queue_id=response['queue_id'],925            queue_type=self._test_queue_type926        )927        self.assertFalse(928            self.queue._r.exists('%s:payload' % self.queue._key_prefix))929    def test_finish_interval_existence(self):930        job_id = self._get_job_id()931        response = self.queue.enqueue(932            payload=self._test_payload_1,933            interval=10000,  # 10s (10000ms)934            job_id=job_id,935            queue_id=self._test_queue_id,936            queue_type=self._test_queue_type,937        )938        # dequeue from the queue_type939        response = self.queue.dequeue(940            queue_type=self._test_queue_type941        )942        response = self.queue.finish(943            job_id=job_id,944            queue_id=response['queue_id'],945            queue_type=self._test_queue_type946        )947        self.assertFalse(948            self.queue._r.exists('%s:interval' % self.queue._key_prefix))949    def test_finish_requeue_limit_existence(self):950        job_id = self._get_job_id()951        response = self.queue.enqueue(952            payload=self._test_payload_1,953            interval=10000,  # 10s (10000ms)954            job_id=job_id,955            queue_id=self._test_queue_id,956            queue_type=self._test_queue_type,957            requeue_limit=self._test_requeue_limit_0958        )959        # dequeue from the queue_type960        response = self.queue.dequeue(961            queue_type=self._test_queue_type962        )963        # mark the job as finished964        response = self.queue.finish(965            job_id=job_id,966            queue_id=response['queue_id'],967            queue_type=self._test_queue_type968        )969        self.assertFalse(970            self.queue._r.exists('%s:%s:%s:requeues_remaining' % (971                self.queue._key_prefix, self._test_queue_type, self._test_queue_id972            ))973        )974    def test_finish_job_queue_existence(self):975        job_id = self._get_job_id()976        response = self.queue.enqueue(977            payload=self._test_payload_1,978            interval=10000,  # 10s (10000ms)979            job_id=job_id,980            queue_id=self._test_queue_id,981            queue_type=self._test_queue_type,982        )983        # dequeue from the queue_type984        response = self.queue.dequeue(985            queue_type=self._test_queue_type986        )987        response = self.queue.finish(988            job_id=job_id,989            queue_id=response['queue_id'],990            queue_type=self._test_queue_type991        )992        self.assertFalse(993            self.queue._r.exists('%s:%s:%s' % (994                self.queue._key_prefix, self._test_queue_type, self._test_queue_id)))995    def test_finish_time_keeper_expire(self):996        job_id = self._get_job_id()997        response = self.queue.enqueue(998            payload=self._test_payload_1,999            interval=10000,  # 10s (10000ms)1000            job_id=job_id,1001            queue_id=self._test_queue_id,1002            queue_type=self._test_queue_type,1003        )1004        # dequeue from the queue_type1005        response = self.queue.dequeue(1006            queue_type=self._test_queue_type1007        )1008        response = self.queue.finish(1009            job_id=job_id,1010            queue_id=response['queue_id'],1011            queue_type=self._test_queue_type1012        )1013        # convert to seconds.1014        time.sleep(self.queue._job_expire_interval / 1000.00)1015        time_keeper_key_name = '%s:%s:%s:time' % (1016            self.queue._key_prefix,1017            self._test_queue_type, self._test_queue_id)1018        self.assertFalse(self.queue._r.exists(time_keeper_key_name))1019    def test_finish_queue_type_ready_set_existence(self):1020        job_id = self._get_job_id()1021        response = self.queue.enqueue(1022            payload=self._test_payload_1,1023            interval=10000,  # 10s (10000ms)1024            job_id=job_id,1025            queue_id=self._test_queue_id,1026            queue_type=self._test_queue_type,1027        )1028        # dequeue from the queue_type1029        response = self.queue.dequeue(1030            queue_type=self._test_queue_type1031        )1032        response = self.queue.finish(1033            job_id=job_id,1034            queue_id=response['queue_id'],1035            queue_type=self._test_queue_type1036        )1037        queue_type_ready_set = self.queue._r.smembers(1038            '%s:ready:queue_type' % self.queue._key_prefix)1039        self.assertEqual(len(queue_type_ready_set), 0)1040    def test_finish_queue_type_active_set_existence(self):1041        job_id = self._get_job_id()1042        response = self.queue.enqueue(1043            payload=self._test_payload_1,1044            interval=10000,  # 10s (10000ms)1045            job_id=job_id,1046            queue_id=self._test_queue_id,1047            queue_type=self._test_queue_type,1048        )1049        # dequeue from the queue_type1050        response = self.queue.dequeue(1051            queue_type=self._test_queue_type1052        )1053        response = self.queue.finish(1054            job_id=job_id,1055            queue_id=response['queue_id'],1056            queue_type=self._test_queue_type1057        )1058        queue_type_active_set = self.queue._r.smembers(1059            '%s:active:queue_type' % self.queue._key_prefix)1060        self.assertEqual(len(queue_type_active_set), 0)1061    def test_requeue_active_sorted_set(self):1062        job_id = self._get_job_id()1063        response = self.queue.enqueue(1064            payload=self._test_payload_1,1065            interval=10000,  # 10s (10000ms)1066            job_id=job_id,1067            queue_id=self._test_queue_id,1068            queue_type=self._test_queue_type,1069        )1070        # dequeue from the queue_type1071        response = self.queue.dequeue(1072            queue_type=self._test_queue_type1073        )1074        # wait until the job expires1075        time.sleep(self.queue._job_expire_interval / 1000.00)1076        # requeue the job1077        self.queue.requeue()1078        self.assertFalse(1079            self.queue._r.exists('%s:%s:active' % (1080                self.queue._key_prefix, self._test_queue_type)))1081    def test_requeue_queue_type_ready_set(self):1082        job_id = self._get_job_id()1083        response = self.queue.enqueue(1084            payload=self._test_payload_1,1085            interval=10000,  # 10s (10000ms)1086            job_id=job_id,1087            queue_id=self._test_queue_id,1088            queue_type=self._test_queue_type,1089        )1090        # dequeue from the queue_type1091        response = self.queue.dequeue(1092            queue_type=self._test_queue_type1093        )1094        # wait until the job expires1095        time.sleep(self.queue._job_expire_interval / 1000.00)1096        # requeue the job1097        self.queue.requeue()1098        queue_type_ready_set = self.queue._r.smembers(1099            '%s:ready:queue_type' % self.queue._key_prefix)1100        self.assertEqual(len(queue_type_ready_set), 1)1101        self.assertEqual(queue_type_ready_set.pop(), self._test_queue_type)1102    def test_requeue_queue_type_active_set(self):1103        job_id = self._get_job_id()1104        response = self.queue.enqueue(1105            payload=self._test_payload_1,1106            interval=10000,  # 10s (10000ms)1107            job_id=job_id,1108            queue_id=self._test_queue_id,1109            queue_type=self._test_queue_type,1110        )1111        # dequeue from the queue_type1112        response = self.queue.dequeue(1113            queue_type=self._test_queue_type1114        )1115        # wait until the job expires1116        time.sleep(self.queue._job_expire_interval / 1000.00)1117        # requeue the job1118        self.queue.requeue()1119        queue_type_active_set = self.queue._r.smembers(1120            '%s:active:queue_type' % self.queue._key_prefix)1121        self.assertEqual(len(queue_type_active_set), 0)1122    def test_requeue_requeue_limit_5(self):1123        # with requeue limit as 51124        job_id = self._get_job_id()1125        response = self.queue.enqueue(1126            payload=self._test_payload_1,1127            interval=10000,  # 10s (10000ms)1128            job_id=job_id,1129            queue_id=self._test_queue_id,1130            queue_type=self._test_queue_type,1131            requeue_limit=self._test_requeue_limit_51132        )1133        # dequeue from the queue_type1134        response = self.queue.dequeue(1135            queue_type=self._test_queue_type1136        )1137        self.assertEqual(1138            response['requeues_remaining'], self._test_requeue_limit_5)1139        # wait until the job expires1140        time.sleep(self.queue._job_expire_interval / 1000.00)1141        # requeue the job1142        self.queue.requeue()1143        # dequeue from the queue_type1144        response = self.queue.dequeue(1145            queue_type=self._test_queue_type1146        )1147        self.assertEqual(1148            response['requeues_remaining'], self._test_requeue_limit_5 - 1)1149        # wait until the job expires1150        time.sleep(self.queue._job_expire_interval / 1000.00)1151        # requeue the job1152        self.queue.requeue()1153        # dequeue from the queue_type1154        response = self.queue.dequeue(1155            queue_type=self._test_queue_type1156        )1157        self.assertEqual(1158            response['requeues_remaining'], self._test_requeue_limit_5 - 2)1159    def test_requeue_requeue_limit_0(self):1160        # with requeue limit as 01161        job_id = self._get_job_id()1162        response = self.queue.enqueue(1163            payload=self._test_payload_1,1164            interval=10000,  # 10s (10000ms)1165            job_id=job_id,1166            queue_id=self._test_queue_id,1167            queue_type=self._test_queue_type,1168            requeue_limit=self._test_requeue_limit_01169        )1170        # dequeue from the queue_type1171        response = self.queue.dequeue(1172            queue_type=self._test_queue_type1173        )1174        self.assertEqual(1175            response['requeues_remaining'], self._test_requeue_limit_0)1176        # wait until the job expires1177        time.sleep(self.queue._job_expire_interval / 1000.00)1178        # requeue the job1179        self.queue.requeue()1180        # dequeue from the queue_type1181        response = self.queue.dequeue(1182            queue_type=self._test_queue_type1183        )1184        self.assertEqual(response['status'], 'failure')1185    def test_requeue_requeue_limit_neg_1(self):1186        # with requeue limit as -1 (requeue infinitely)1187        job_id = self._get_job_id()1188        response = self.queue.enqueue(1189            payload=self._test_payload_1,1190            interval=10000,  # 10s (10000ms)1191            job_id=job_id,1192            queue_id=self._test_queue_id,1193            queue_type=self._test_queue_type,1194            requeue_limit=self._test_requeue_limit_neg_11195        )1196        # dequeue from the queue_type1197        response = self.queue.dequeue(1198            queue_type=self._test_queue_type1199        )1200        self.assertEqual(1201            response['requeues_remaining'], self._test_requeue_limit_neg_1)1202        # wait until the job expires1203        time.sleep(self.queue._job_expire_interval / 1000.00)1204        # requeue the job1205        self.queue.requeue()1206        # dequeue from the queue_type1207        response = self.queue.dequeue(1208            queue_type=self._test_queue_type1209        )1210        self.assertEqual(1211            response['requeues_remaining'], self._test_requeue_limit_neg_1)1212        # wait until the job expires1213        time.sleep(self.queue._job_expire_interval / 1000.00)1214        # requeue the job1215        self.queue.requeue()1216        # dequeue from the queue_type1217        response = self.queue.dequeue(1218            queue_type=self._test_queue_type1219        )1220        self.assertEqual(1221            response['requeues_remaining'], self._test_requeue_limit_neg_1)1222        # wait until the job expires1223        time.sleep(self.queue._job_expire_interval / 1000.00)1224        # requeue the job1225        self.queue.requeue()1226        # dequeue from the queue_type1227        response = self.queue.dequeue(1228            queue_type=self._test_queue_type1229        )1230        self.assertEqual(1231            response['requeues_remaining'], self._test_requeue_limit_neg_1)1232        # wait until the job expires1233        time.sleep(self.queue._job_expire_interval / 1000.00)1234        # requeue the job1235        self.queue.requeue()1236        self.assertEqual(1237            response['requeues_remaining'], self._test_requeue_limit_neg_1)1238        # wait until the job expires1239        time.sleep(self.queue._job_expire_interval / 1000.00)1240    def test_interval_non_existent_queue(self):1241        response = self.queue.interval(1242            interval=1000,1243            queue_id=self._test_queue_id,1244            queue_type=self._test_queue_type1245        )1246        self.assertEqual(response['status'], 'failure')1247        interval_map_name = '%s:interval' % (self.queue._key_prefix)1248        # check if interval map exists1249        self.assertFalse(self.queue._r.exists(interval_map_name))1250    def test_interval_existent_queue(self):1251        job_id = self._get_job_id()1252        response = self.queue.enqueue(1253            payload=self._test_payload_1,1254            interval=10000,  # 10s (10000ms)1255            job_id=job_id,1256            queue_id=self._test_queue_id,1257            queue_type=self._test_queue_type,1258        )1259        # check if interval is saved in the appropriate structure1260        interval_map_name = '%s:interval' % (self.queue._key_prefix)1261        # check if interval map exists1262        self.assertTrue(self.queue._r.exists(interval_map_name))1263        # check the value1264        interval_map_key = '%s:%s' % (1265            self._test_queue_type, self._test_queue_id)1266        interval = self.queue._r.hget(interval_map_name, interval_map_key)1267        self.assertEqual(interval, '10000')1268        # set the interval to 5s (5000ms)1269        response = self.queue.interval(1270            interval=5000,1271            queue_id=self._test_queue_id,1272            queue_type=self._test_queue_type1273        )1274        self.assertEqual(response['status'], 'success')1275        # check if interval is saved in the appropriate structure1276        interval_map_name = '%s:interval' % (self.queue._key_prefix)1277        # check if interval map exists1278        self.assertTrue(self.queue._r.exists(interval_map_name))1279        # check the value1280        # check the value1281        interval_map_key = '%s:%s' % (1282            self._test_queue_type, self._test_queue_id)1283        interval = self.queue._r.hget(interval_map_name, interval_map_key)1284        self.assertEqual(interval, '5000')1285    def test_metrics_response_status(self):1286        response = self.queue.metrics()1287        self.assertEqual(response['status'], 'success')1288        response = self.queue.metrics(self._test_queue_type)1289        self.assertEqual(response['status'], 'success')1290        response = self.queue.metrics(1291            self._test_queue_type, self._test_queue_id)1292        self.assertEqual(response['status'], 'success')1293    def test_metrics_response_queue_types(self):1294        response = self.queue.metrics()1295        self.assertEqual(response['queue_types'], [])1296        self.assertEqual(len(response['enqueue_counts'].values()), 10)1297        self.assertEqual(sum(response['enqueue_counts'].values()), 0)1298        self.assertEqual(len(response['dequeue_counts'].values()), 10)1299        self.assertEqual(sum(response['dequeue_counts'].values()), 0)1300        job_id = self._get_job_id()1301        response = self.queue.enqueue(1302            payload=self._test_payload_1,1303            interval=10000,  # 10s (10000ms)1304            job_id=job_id,1305            queue_id=self._test_queue_id,1306            queue_type=self._test_queue_type,1307        )1308        response = self.queue.metrics()1309        self.assertEqual(response['queue_types'], [self._test_queue_type])1310        self.assertEqual(len(response['enqueue_counts'].values()), 10)1311        self.assertEqual(sum(response['enqueue_counts'].values()), 1)1312        self.assertEqual(len(response['dequeue_counts'].values()), 10)1313        self.assertEqual(sum(response['dequeue_counts'].values()), 0)1314        response = self.queue.dequeue(queue_type=self._test_queue_type)1315        response = self.queue.metrics()1316        self.assertEqual(len(response['dequeue_counts'].values()), 10)1317        self.assertEqual(sum(response['dequeue_counts'].values()), 1)1318    def test_metrics_response_queue_ids(self):1319        response = self.queue.metrics(queue_type=self._test_queue_type)1320        self.assertEqual(response['queue_ids'], [])1321        job_id = self._get_job_id()1322        response = self.queue.enqueue(1323            payload=self._test_payload_1,1324            interval=10000,  # 10s (10000ms)1325            job_id=job_id,1326            queue_id=self._test_queue_id,1327            queue_type=self._test_queue_type,1328        )1329        response = self.queue.metrics(queue_type=self._test_queue_type)1330        self.assertEqual(response['queue_ids'], [self._test_queue_id])1331        response = self.queue.dequeue(1332            queue_type=self._test_queue_type1333        )1334        response = self.queue.metrics(queue_type=self._test_queue_type)1335        self.assertEqual(response['queue_ids'], [self._test_queue_id])1336    def test_metrics_response_enqueue_counts_list(self):1337        response = self.queue.metrics(1338            queue_type=self._test_queue_type, queue_id=self._test_queue_id)1339        self.assertEqual(len(response['enqueue_counts'].values()), 10)1340        self.assertEqual(sum(response['enqueue_counts'].values()), 0)1341        job_id = self._get_job_id()1342        response = self.queue.enqueue(1343            payload=self._test_payload_1,1344            interval=10000,  # 10s (10000ms)1345            job_id=job_id,1346            queue_id=self._test_queue_id,1347            queue_type=self._test_queue_type,1348        )1349        response = self.queue.metrics(1350            queue_type=self._test_queue_type, queue_id=self._test_queue_id)1351        self.assertEqual(len(response['enqueue_counts'].values()), 10)1352        self.assertEqual(sum(response['enqueue_counts'].values()), 1)1353    def test_metrics_response_dequeue_counts_list(self):1354        response = self.queue.metrics(1355            queue_type=self._test_queue_type, queue_id=self._test_queue_id)1356        self.assertEqual(len(response['dequeue_counts'].values()), 10)1357        self.assertEqual(sum(response['dequeue_counts'].values()), 0)1358        response = self.queue.dequeue(queue_type=self._test_queue_type)1359        response = self.queue.metrics(1360            queue_type=self._test_queue_type, queue_id=self._test_queue_id)1361        self.assertEqual(len(response['dequeue_counts'].values()), 10)1362        self.assertEqual(sum(response['dequeue_counts'].values()), 0)1363        job_id = self._get_job_id()1364        response = self.queue.enqueue(1365            payload=self._test_payload_1,1366            interval=10000,  # 10s (10000ms)1367            job_id=job_id,1368            queue_id=self._test_queue_id,1369            queue_type=self._test_queue_type,1370        )1371        response = self.queue.dequeue(queue_type=self._test_queue_type)1372        response = self.queue.metrics(1373            queue_type=self._test_queue_type, queue_id=self._test_queue_id)1374        self.assertEqual(len(response['dequeue_counts'].values()), 10)1375        self.assertEqual(sum(response['dequeue_counts'].values()), 1)1376    def test_metrics_response_queue_length(self):1377        response = self.queue.metrics(1378            queue_type=self._test_queue_type, queue_id=self._test_queue_id)1379        self.assertEqual(response['queue_length'], 0)1380        job_id = self._get_job_id()1381        response = self.queue.enqueue(1382            payload=self._test_payload_1,1383            interval=10000,  # 10s (10000ms)1384            job_id=job_id,1385            queue_id=self._test_queue_id,1386            queue_type=self._test_queue_type,1387        )1388        response = self.queue.metrics(1389            queue_type=self._test_queue_type, queue_id=self._test_queue_id)1390        self.assertEqual(response['queue_length'], 1)1391        response = self.queue.dequeue(queue_type=self._test_queue_type)1392        response = self.queue.metrics(1393            queue_type=self._test_queue_type, queue_id=self._test_queue_id)1394        self.assertEqual(response['queue_length'], 0)1395    def test_metrics_enqueue_sliding_window(self):1396        response = self.queue.metrics(1397            queue_type=self._test_queue_type, queue_id=self._test_queue_id)1398        global_response = self.queue.metrics()1399        self.assertEqual(len(response['enqueue_counts'].values()), 10)1400        self.assertEqual(sum(response['enqueue_counts'].values()), 0)1401        self.assertEqual(len(global_response['enqueue_counts'].values()), 10)1402        self.assertEqual(sum(global_response['enqueue_counts'].values()), 0)1403        # enqueue a job1404        job_id = self._get_job_id()1405        response = self.queue.enqueue(1406            payload=self._test_payload_1,1407            interval=10000,  # 10s (10000ms)1408            job_id=job_id,1409            queue_id=self._test_queue_id,1410            queue_type=self._test_queue_type,1411        )1412        timestamp = int(generate_epoch())1413        # epoch for the minute.1414        timestamp_minute = str(int(math.floor(timestamp / 60000.0) * 60000))1415        response = self.queue.metrics(1416            queue_type=self._test_queue_type, queue_id=self._test_queue_id)1417        global_response = self.queue.metrics()1418        self.assertEqual(response['enqueue_counts'][timestamp_minute], 1)1419        self.assertEqual(1420            global_response['enqueue_counts'][timestamp_minute], 1)1421        # enqueue another job1422        job_id = self._get_job_id()1423        response = self.queue.enqueue(1424            payload=self._test_payload_1,1425            interval=10000,  # 10s (10000ms)1426            job_id=job_id,1427            queue_id=self._test_queue_id,1428            queue_type=self._test_queue_type,1429        )1430        response = self.queue.metrics(1431            queue_type=self._test_queue_type, queue_id=self._test_queue_id)1432        global_response = self.queue.metrics()1433        self.assertEqual(response['enqueue_counts'][timestamp_minute], 2)1434        self.assertEqual(1435            global_response['enqueue_counts'][timestamp_minute], 2)1436        # wait for one minute1437        time.sleep(65)  # 65 seconds1438        # check the last minute value.1439        response = self.queue.metrics(1440            queue_type=self._test_queue_type, queue_id=self._test_queue_id)1441        global_response = self.queue.metrics()1442        self.assertEqual(response['enqueue_counts'][timestamp_minute], 2)1443        self.assertEqual(1444            global_response['enqueue_counts'][timestamp_minute], 2)1445        # save the old value before overwriting1446        old_1_timestamp_minute = timestamp_minute1447        # check the current minute value1448        timestamp = int(generate_epoch())1449        # epoch for the minute.1450        timestamp_minute = str(int(math.floor(timestamp / 60000.0) * 60000))1451        response = self.queue.metrics(1452            queue_type=self._test_queue_type, queue_id=self._test_queue_id)1453        global_response = self.queue.metrics()1454        self.assertEqual(response['enqueue_counts'][timestamp_minute], 0)1455        self.assertEqual(1456            global_response['enqueue_counts'][timestamp_minute], 0)1457        # enqueue a job in the current minute1458        job_id = self._get_job_id()1459        response = self.queue.enqueue(1460            payload=self._test_payload_1,1461            interval=10000,  # 10s (10000ms)1462            job_id=job_id,1463            queue_id=self._test_queue_id,1464            queue_type=self._test_queue_type,1465        )1466        response = self.queue.metrics(1467            queue_type=self._test_queue_type, queue_id=self._test_queue_id)1468        global_response = self.queue.metrics()1469        self.assertEqual(response['enqueue_counts'][timestamp_minute], 1)1470        self.assertEqual(response['enqueue_counts'][old_1_timestamp_minute], 2)1471        self.assertEqual(1472            global_response['enqueue_counts'][timestamp_minute], 1)1473        self.assertEqual(1474            global_response['enqueue_counts'][old_1_timestamp_minute], 2)1475        time.sleep(65)  # sleep for another 65s1476        # save the old timestamp1477        old_2_timestamp_minute = timestamp_minute1478        # check the current minute value1479        timestamp = int(generate_epoch())1480        # epoch for the minute.1481        timestamp_minute = str(int(math.floor(timestamp / 60000.0) * 60000))1482        response = self.queue.metrics(1483            queue_type=self._test_queue_type, queue_id=self._test_queue_id)1484        global_response = self.queue.metrics()1485        self.assertEqual(response['enqueue_counts'][timestamp_minute], 0)1486        self.assertEqual(1487            global_response['enqueue_counts'][timestamp_minute], 0)1488        # enqueue a job in the current minute1489        job_id = self._get_job_id()1490        response = self.queue.enqueue(1491            payload=self._test_payload_1,1492            interval=10000,  # 10s (10000ms)1493            job_id=job_id,1494            queue_id=self._test_queue_id,1495            queue_type=self._test_queue_type,1496        )1497        response = self.queue.metrics(1498            queue_type=self._test_queue_type, queue_id=self._test_queue_id)1499        global_response = self.queue.metrics()1500        self.assertEqual(response['enqueue_counts'][timestamp_minute], 1)1501        self.assertEqual(response['enqueue_counts'][old_1_timestamp_minute], 2)1502        self.assertEqual(response['enqueue_counts'][old_2_timestamp_minute], 1)1503        self.assertEqual(1504            global_response['enqueue_counts'][timestamp_minute], 1)1505        self.assertEqual(1506            global_response['enqueue_counts'][old_1_timestamp_minute], 2)1507        self.assertEqual(1508            global_response['enqueue_counts'][old_2_timestamp_minute], 1)1509    def test_metrics_dequeue_sliding_window(self):1510        response = self.queue.metrics(1511            queue_type=self._test_queue_type, queue_id=self._test_queue_id)1512        global_response = self.queue.metrics()1513        self.assertEqual(len(response['dequeue_counts'].values()), 10)1514        self.assertEqual(sum(response['dequeue_counts'].values()), 0)1515        self.assertEqual(len(global_response['dequeue_counts'].values()), 10)1516        self.assertEqual(sum(global_response['dequeue_counts'].values()), 0)1517        # enqueue a job1518        job_id = self._get_job_id()1519        response = self.queue.enqueue(1520            payload=self._test_payload_1,1521            interval=100,  # 100ms1522            job_id=job_id,1523            queue_id=self._test_queue_id,1524            queue_type=self._test_queue_type,1525        )1526        response = self.queue.dequeue(1527            queue_type=self._test_queue_type1528        )1529        timestamp = int(generate_epoch())1530        # epoch for the minute.1531        timestamp_minute = str(int(math.floor(timestamp / 60000.0) * 60000))1532        response = self.queue.metrics(1533            queue_type=self._test_queue_type, queue_id=self._test_queue_id)1534        global_response = self.queue.metrics()1535        self.assertEqual(response['dequeue_counts'][timestamp_minute], 1)1536        self.assertEqual(1537            global_response['dequeue_counts'][timestamp_minute], 1)1538        # enqueue another job1539        job_id = self._get_job_id()1540        response = self.queue.enqueue(1541            payload=self._test_payload_1,1542            interval=100,  # 100ms1543            job_id=job_id,1544            queue_id=self._test_queue_id,1545            queue_type=self._test_queue_type,1546        )1547        time.sleep(0.1)  # 100ms1548        response = self.queue.dequeue(1549            queue_type=self._test_queue_type1550        )1551        response = self.queue.metrics(1552            queue_type=self._test_queue_type, queue_id=self._test_queue_id)1553        global_response = self.queue.metrics()1554        self.assertEqual(response['dequeue_counts'][timestamp_minute], 2)1555        self.assertEqual(1556            global_response['dequeue_counts'][timestamp_minute], 2)1557        # wait for one minute1558        time.sleep(65)  # 65 seconds1559        # check the last minute value.1560        response = self.queue.metrics(1561            queue_type=self._test_queue_type, queue_id=self._test_queue_id)1562        global_response = self.queue.metrics()1563        self.assertEqual(response['dequeue_counts'][timestamp_minute], 2)1564        self.assertEqual(1565            global_response['dequeue_counts'][timestamp_minute], 2)1566        # save the old value before overwriting1567        old_1_timestamp_minute = timestamp_minute1568        # check the current minute value1569        timestamp = int(generate_epoch())1570        # epoch for the minute.1571        timestamp_minute = str(int(math.floor(timestamp / 60000.0) * 60000))1572        response = self.queue.metrics(1573            queue_type=self._test_queue_type, queue_id=self._test_queue_id)1574        global_response = self.queue.metrics()1575        self.assertEqual(response['dequeue_counts'][timestamp_minute], 0)1576        self.assertEqual(1577            global_response['dequeue_counts'][timestamp_minute], 0)1578        # enqueue a job in the current minute1579        job_id = self._get_job_id()1580        response = self.queue.enqueue(1581            payload=self._test_payload_1,1582            interval=100,  # 100ms1583            job_id=job_id,1584            queue_id=self._test_queue_id,1585            queue_type=self._test_queue_type,1586        )1587        time.sleep(0.1)  # 100ms1588        response = self.queue.dequeue(1589            queue_type=self._test_queue_type1590        )1591        response = self.queue.metrics(1592            queue_type=self._test_queue_type, queue_id=self._test_queue_id)1593        global_response = self.queue.metrics()1594        self.assertEqual(response['dequeue_counts'][timestamp_minute], 1)1595        self.assertEqual(response['dequeue_counts'][old_1_timestamp_minute], 2)1596        self.assertEqual(1597            global_response['dequeue_counts'][timestamp_minute], 1)1598        self.assertEqual(1599            global_response['dequeue_counts'][old_1_timestamp_minute], 2)1600        time.sleep(65)  # sleep for another 65s1601        # save the old timestamp1602        old_2_timestamp_minute = timestamp_minute1603        # check the current minute value1604        timestamp = int(generate_epoch())1605        # epoch for the minute.1606        timestamp_minute = str(int(math.floor(timestamp / 60000.0) * 60000))1607        response = self.queue.metrics(1608            queue_type=self._test_queue_type, queue_id=self._test_queue_id)1609        global_response = self.queue.metrics()1610        self.assertEqual(response['dequeue_counts'][timestamp_minute], 0)1611        self.assertEqual(1612            global_response['dequeue_counts'][timestamp_minute], 0)1613        # enqueue a job in the current minute1614        job_id = self._get_job_id()1615        response = self.queue.enqueue(1616            payload=self._test_payload_1,1617            interval=100,  # 100ms1618            job_id=job_id,1619            queue_id=self._test_queue_id,1620            queue_type=self._test_queue_type,1621        )1622        time.sleep(0.1)  # 100ms1623        response = self.queue.dequeue(1624            queue_type=self._test_queue_type1625        )1626        response = self.queue.metrics(1627            queue_type=self._test_queue_type, queue_id=self._test_queue_id)1628        global_response = self.queue.metrics()1629        self.assertEqual(response['dequeue_counts'][timestamp_minute], 1)1630        self.assertEqual(response['dequeue_counts'][old_1_timestamp_minute], 2)1631        self.assertEqual(response['dequeue_counts'][old_2_timestamp_minute], 1)1632        self.assertEqual(1633            global_response['dequeue_counts'][timestamp_minute], 1)1634        self.assertEqual(1635            global_response['dequeue_counts'][old_1_timestamp_minute], 2)1636        self.assertEqual(1637            global_response['dequeue_counts'][old_2_timestamp_minute], 1)1638    def test_sharq_rate_limiting(self):1639        job_id_1 = self._get_job_id()1640        response = self.queue.enqueue(1641            payload=self._test_payload_1,1642            interval=2000,  # 2s (2000ms)1643            job_id=job_id_1,1644            queue_id=self._test_queue_id,1645            queue_type=self._test_queue_type1646        )1647        job_id_2 = self._get_job_id()1648        response = self.queue.enqueue(1649            payload=self._test_payload_2,1650            interval=2000,  # 2s (2000ms)1651            job_id=job_id_2,1652            queue_id=self._test_queue_id,1653            queue_type=self._test_queue_type1654        )1655        # try to do  back-to-back dequeues.1656        # only the first one should return the job,1657        # the second one should fail and should only1658        # succeed after waiting for the time1659        # interval specified.1660        response = self.queue.dequeue(1661            queue_type=self._test_queue_type1662        )1663        # check all the responses1664        self.assertEqual(response['status'], 'success')1665        self.assertEqual(response['queue_id'], self._test_queue_id)1666        self.assertEqual(response['job_id'], job_id_1)1667        self.assertEqual(response['payload'], self._test_payload_1)1668        response = self.queue.dequeue(1669            queue_type=self._test_queue_type1670        )1671        self.assertEqual(response['status'], 'failure')1672        time.sleep(2)  # 2s1673        # dequeue again1674        response = self.queue.dequeue(1675            queue_type=self._test_queue_type1676        )1677        # check all the responses1678        self.assertEqual(response['status'], 'success')1679        self.assertEqual(response['queue_id'], self._test_queue_id)1680        self.assertEqual(response['job_id'], job_id_2)1681        self.assertEqual(response['payload'], self._test_payload_2)1682    def test_clear_queue_without_purge(self):1683        job_id = self._get_job_id()1684        queue_response = self.queue.enqueue(1685            payload=self._test_payload_1,1686            interval=10000,  # 10s (10000ms)1687            job_id=job_id,1688            queue_id=self._test_queue_id,1689            queue_type=self._test_queue_type,1690        )1691        queue_clear_response = self.queue.clear_queue(1692            queue_type=self._test_queue_type,1693            queue_id=self._test_queue_id)1694        # check the responses1695        self.assertEqual(queue_clear_response['status'], 'success')1696        self.assertEqual(queue_clear_response['message'], 'successfully removed all queued calls')1697        #check in redis1698        job_queue_list = '%s:%s:%s' % (self.queue._key_prefix, 1699                            self._test_queue_type, self._test_queue_id)1700        primary_set = '%s:%s'% (self.queue._key_prefix, self._test_queue_type)1701        primary_sorted_key = self.queue._r.zrange(primary_set, 0, -1)1702        self.assertNotIn(self._test_queue_id, primary_sorted_key)1703        self.assertFalse(self.queue._r.exists(job_queue_list))          1704    1705    def test_clear_queue_with_purge(self):1706        job_id = self._get_job_id()1707        queue_response = self.queue.enqueue(1708            payload=self._test_payload_1,1709            interval=10000,  # 10s (10000ms)1710            job_id=job_id,1711            queue_id=self._test_queue_id,1712            queue_type=self._test_queue_type,1713        )1714        queue_clear_response = self.queue.clear_queue(1715            queue_type=self._test_queue_type,1716            queue_id=self._test_queue_id,1717            purge_all=True)1718        #check the responses1719        self.assertEqual(queue_clear_response['status'], 'success')1720        self.assertEqual(queue_clear_response['message'], 1721                    'successfully removed all queued calls and purged related resources') 1722        #check in the redis if resource is removed 1723        job_queue_list = '%s:%s:%s' % (self.queue._key_prefix, 1724                            self._test_queue_type, self._test_queue_id)1725        primary_set = '%s:%s'% (self.queue._key_prefix, self._test_queue_type)1726        payload_hashset = '%s:payload' % (self.queue._key_prefix)1727        job_payload_key = '%s:%s:%s' % (self._test_queue_type, self._test_queue_id, job_id)1728        interval_set = '%s:interval' % (self.queue._key_prefix)1729        job_interval_key = '%s:%s' % (self._test_queue_type, self._test_queue_id)1730        primary_sorted_key = self.queue._r.zrange(primary_set, 0, -1)1731        self.assertNotIn(self._test_queue_id, primary_sorted_key)1732        self.assertFalse(self.queue._r.hexists(payload_hashset, job_payload_key))1733        self.assertFalse(self.queue._r.hexists(interval_set, job_interval_key))1734        self.assertFalse(self.queue._r.exists(job_queue_list))1735    1736    def test_clear_queue_with_non_existing_queue_id(self):1737        queue_clear_response = self.queue.clear_queue(1738            queue_type=self._test2_queue_type,1739            queue_id=self._test2_queue_id)1740        #check the responses1741        self.assertEqual(queue_clear_response['status'], 'failure')1742        self.assertEqual(queue_clear_response['message'], 1743                    'No queued calls found')1744    def test_clear_queue_with_non_existing_queue_id_with_purge(self):1745        queue_clear_response = self.queue.clear_queue(1746            queue_type=self._test2_queue_type,1747            queue_id=self._test2_queue_id,1748            purge_all=True)1749        #check the responses1750        self.assertEqual(queue_clear_response['status'], 'failure')1751        self.assertEqual(queue_clear_response['message'], 1752                    'No queued calls found')1753    def tearDown(self):1754        # flush all the keys in the test db after each test1755        self.queue._r.flushdb()1756def main():1757    unittest.main()1758if __name__ == '__main__':1759    main()

test_queue.py

Source:test_queue.py

...287            queue_type=self.valid_queue_type288        )289    def test_enqueue_all_ok(self):290        # with a queue_type291        response = self.queue.enqueue(292            payload=self.valid_payload,293            interval=self.valid_interval,294            job_id=self.valid_job_id,295            queue_id=self.valid_queue_id,296            queue_type=self.valid_queue_type297        )298        self.assertEqual(response['status'], 'queued')299        # the result should contain only status300        response.pop('status')301        self.assertEqual(response, {})302        # without a queue_type (queue_type will be 'default')303        response = self.queue.enqueue(304            payload=self.valid_payload,305            interval=self.valid_interval,306            job_id=self.valid_job_id,307            queue_id=self.valid_queue_id308        )309        self.assertEqual(response['status'], 'queued')310        # the result should contain only status311        response.pop('status')312        self.assertEqual(response, {})313        # with requeue_limit 1314        response = self.queue.enqueue(315            payload=self.valid_payload,316            interval=self.valid_interval,317            job_id=self.valid_job_id,318            queue_id=self.valid_queue_id,319            queue_type=self.valid_queue_type,320            requeue_limit=self.valid_requeue_limit_1321        )322        self.assertEqual(response['status'], 'queued')323        # the result should contain only status324        response.pop('status')325        self.assertEqual(response, {})326        # with requeue_limit 2327        response = self.queue.enqueue(328            payload=self.valid_payload,329            interval=self.valid_interval,330            job_id=self.valid_job_id,331            queue_id=self.valid_queue_id,332            queue_type=self.valid_queue_type,333            requeue_limit=self.valid_requeue_limit_2334        )335        self.assertEqual(response['status'], 'queued')336        # the result should contain only status337        response.pop('status')338        self.assertEqual(response, {})339        # with requeue_limit 3340        response = self.queue.enqueue(341            payload=self.valid_payload,342            interval=self.valid_interval,343            job_id=self.valid_job_id,344            queue_id=self.valid_queue_id,345            queue_type=self.valid_queue_type,346            requeue_limit=self.valid_requeue_limit_3347        )348        self.assertEqual(response['status'], 'queued')349        # the result should contain only status350        response.pop('status')351        self.assertEqual(response, {})352        # requeue_limit missing353        response = self.queue.enqueue(354            payload=self.valid_payload,355            interval=self.valid_interval,356            job_id=self.valid_job_id,357            queue_id=self.valid_queue_id,358            queue_type=self.valid_queue_type359        )360        self.assertEqual(response['status'], 'queued')361        # the result should contain only status362        response.pop('status')363        self.assertEqual(response, {})364    def test_dequeue_queue_type_invalid(self):365        # type 1366        self.assertRaisesRegexp(367            BadArgumentException,368            '`queue_type` has an invalid value.',369            self.queue.dequeue,370            queue_type=self.invalid_queue_type_1371        )372        # type 2373        self.assertRaisesRegexp(374            BadArgumentException,375            '`queue_type` has an invalid value.',376            self.queue.dequeue,377            queue_type=self.invalid_queue_type_2378        )379        # type 3380        self.assertRaisesRegexp(381            BadArgumentException,382            '`queue_type` has an invalid value.',383            self.queue.dequeue,384            queue_type=self.invalid_queue_type_3385        )386    def test_dequeue_all_ok(self):387        # first enqueue a job388        self.queue.enqueue(389            payload=self.valid_payload,390            interval=self.valid_interval,391            job_id=self.valid_job_id,392            queue_id=self.valid_queue_id,393            queue_type=self.valid_queue_type394        )395        # with a queue_type396        response = self.queue.dequeue(397            queue_type=self.valid_queue_type398        )399        self.assertEqual(response['status'], 'success')400        response.pop('status')401        # check if it has a key called 'payload'402        self.assertIn('payload', response)403        response.pop('payload')404        # check if it has a key called 'queue_id'405        self.assertIn('queue_id', response)406        response.pop('queue_id')407        # check if it has a key called 'job_id'408        self.assertIn('job_id', response)409        response.pop('job_id')410        # check if it has a key called 'requeues_remaining'411        self.assertIn('requeues_remaining', response)412        response.pop('requeues_remaining')413        # make sure nothing else in response414        # except the above key / value pairs415        self.assertEqual(response, {})416        # enqueue another job417        self.queue.enqueue(418            payload=self.valid_payload,419            interval=self.valid_interval,420            job_id=self.valid_job_id,421            queue_id=self.valid_queue_id422        )423        # without a queue_type424        response = self.queue.dequeue()425        self.assertEqual(response['status'], 'success')426        response.pop('status')427        # check if it has a key called 'payload'428        self.assertIn('payload', response)429        response.pop('payload')430        # check if it has a key called 'queue_id'431        self.assertIn('queue_id', response)432        response.pop('queue_id')433        # check if it has a key called 'job_id'434        self.assertIn('job_id', response)435        response.pop('job_id')436        # check if it has a key called 'requeues_remaining'437        self.assertIn('requeues_remaining', response)438        response.pop('requeues_remaining')...

queuesController.js

Source:queuesController.js

1app.controller('queuesController',2['$scope',3'$q',4'$rootScope',5'$stateParams', 6'$location',7'projectService',8'queueService',9'sharedDataService',10'tableService',11function($scope,12$q,  13$rootScope,14$stateParams,15$location,16projectService,17queueService,18sharedDataService,19tableService){      20  21  var id;22  $rootScope.showAppPanel=true;23  $rootScope.isFullScreen=false;24  $rootScope.page='queues';25  $scope.firstVisit=true;26  //Queues Specific27  $scope.showCreateQueueBox=false;28  $scope.queueList=[];29  $scope.queueListLoading=true;30  $scope.activatedQueue=[];31  $scope.selectedQueue={};32  $scope.queueMessagesList=[];33  $scope.newQueueType="pull";//Default34  $scope.creatingQueue=false;35  $scope.openMsgAdvanceOptions=false;36  $scope.editableQueue=[];  37  $scope.queueSizes=[];38  $scope.queueDataTypes=["Text","JSON"];39  $scope.queueActiveTab="Text";40  $scope.init= function() {            41    id = $stateParams.appId;42    if($rootScope.currentProject && $rootScope.currentProject.appId === id){43      //if the same project is already in the rootScope, then dont load it.44      initCB(); 45      getAllQueues();46      $rootScope.pageHeaderDisplay=$rootScope.currentProject.name;                    47    }else{48      loadProject(id);              49    }50    //Flush Acl data in sharedDataService51    sharedDataService.flushAclArray();52  };  53  $scope.initCreateQueue=function(){54    $("#md-createqueuemodel").modal();55  };56  $scope.createQueue=function(){57    if($scope.newQueueName){58      $scope.queueModalError=null;59      $scope.creatingQueue=true;60      queueService.createQueue($scope.newQueueName,$scope.newQueueType)61      .then(function(queueObj){62        $scope.queueList.push(queueObj);63        $scope.creatingQueue=false; 64        $scope.newQueueName=null;65        $scope.firstVisit=false;66        $scope.queueSizes[$scope.queueList.length-1]=0;67        $("#md-createqueuemodel").modal("hide");                                   68      }, function(error){ 69        $scope.creatingQueue=false; 70        $scope.queueModalError=error;       71      });72    }    73  };74  $scope.openQueueDetails=function(queue){75    var index=$scope.queueList.indexOf(queue);76    if(index!=$scope.previousIndex){77      if($scope.previousIndex==0 || $scope.previousIndex>0){78        $scope.activatedQueue.splice($scope.previousIndex,1);79        var prevQ=$scope.queueList[$scope.previousIndex];80        $scope.selectedQueue[prevQ.id]=false;81      }82      $scope.previousIndex=index; 83      $scope.activatedQueue[index]=queue;84      $scope.selectedQueue[queue.id]=true;85      $scope.messagesLoading=true;86      $scope.messagesError=null;         87      queueService.getAllMessages(queue)88      .then(function(list){ 89        $scope.queueMessagesList=list; 90        $scope.messagesLoading=false;91      }, function(error){ 92        $scope.messagesLoading=false;        93        $scope.messagesError=error;        94      });95    }96    97  };98  $scope.editQueueACL=function(queue){99    var index=$scope.queueList.indexOf(queue);    100    $scope.editableQueue.push(queue);   101    102    //Sharing Data through a service         103    sharedDataService.pushAclObject(queue.ACL);    104    $("#md-queueaclviewer").modal();  105  };106  $scope.saveQueueACL=function(updatedQueueACL){107    $("#md-queueaclviewer").modal("hide"); 108    $scope.editableQueue[$scope.editableQueue.length-1].ACL=updatedQueueACL;  109    110    var currentQ=$scope.editableQueue[$scope.editableQueue.length-1];111    var index=$scope.queueList.indexOf(currentQ);112   113    updateQWrapper(currentQ,index)114    .then(function(resp){               115      $scope.queueList[resp.rowIndex].ACL=updatedQueueACL;116      $scope.editableQueue.splice(resp.rowIndex,1);117      sharedDataService.spliceAclObjectByIndex(resp.rowIndex); 118    }, function(errorResp){                         119      sharedDataService.spliceAclObjectByIndex(resp.rowIndex);120      $scope.editableQueue.splice(resp.rowIndex,1);     121      errorNotify(error);    122    });    123  };124  function updateQWrapper(queue,rowIndex){125    var q=$q.defer();126    queueService.updateQueue(queue)127    .then(function(resp){128      var respObj={129        resp:resp,130        rowIndex:rowIndex131      };              132      q.resolve(respObj);133      134    }, function(error){   135      var respObj={136        error:error,137        rowIndex:rowIndex138      };                         139      q.reject(respObj);     140    });   141    return  q.promise; 142  }143  $scope.initDeleteQueue=function(queue){144    $scope.deletableQueue=queue;145    $scope.confirmQueueName=null;146    $("#md-deletequeue").modal();147  };148  $scope.deleteQueue=function(){149    if($scope.confirmQueueName==$scope.deletableQueue.name){150      var index=$scope.queueList.indexOf($scope.deletableQueue);     151      $scope.queueModalError=null; 152      $scope.confirmSpinner=true;153      queueService.deleteQueue($scope.deletableQueue)154      .then(function(resp){155        $scope.queueList.splice(index,1);156        $scope.activatedQueue.splice(index,1);157        $("#md-deletequeue").modal("hide");158        $scope.confirmSpinner=false; 159        $scope.confirmQueueName=null;160        $scope.deletableQueue=null;161        $scope.queueSizes.splice(index,1);162        if($scope.queueList.length==0){163          $scope.firstVisit=true;164        }165      }, function(error){       166        $scope.queueModalError=error; 167        $scope.confirmSpinner=false;         168      });169    }else{170      $scope.queueModalError="Queue Name doesn't match";171    }172  };173  $scope.initAddNewMessage=function(){174    $scope.queueActiveTab="Text";    175    $scope.newMessage={176      msg:null,177      timeout:null,178      delay:null,179      expires:null,180    };181    $("#md-addnewmsg").modal();182  };183  $scope.addNewMessage=function(){184    var validate=validateQ($scope.newMessage);185    if(!validate){ 186      if($scope.newMessage.msg){187        if($scope.newMessage.expires){188          $scope.newMessage.expires=new Date($scope.newMessage.expires);189        }        190        if(_isJsonString($scope.newMessage.msg)){191          $scope.newMessage.msg=JSON.parse($scope.newMessage.msg);192        }193        $scope.queueModalError=null; 194        $scope.addMsgSpinner=true;195        queueService.insertMessageIntoQueue($scope.activatedQueue[$scope.previousIndex],$scope.newMessage.msg,$scope.newMessage.timeout,$scope.newMessage.delay,$scope.newMessage.expires)196        .then(function(resp){197          $("#md-addnewmsg").modal("hide");198          if(!$scope.queueMessagesList){199            $scope.queueMessagesList=[];200          }201          $scope.queueMessagesList.push(resp);202          $scope.queueModalError=false; 203          $scope.newMessage=null;        204          $scope.addMsgSpinner=false;205          $scope.queueSizes[$scope.previousIndex]=++$scope.queueSizes[$scope.previousIndex];         206        }, function(error){  207          $scope.queueModalError=error; 208          $scope.addMsgSpinner=false;       209        });210      }else{211        $scope.queueModalError="Message shoudn't be empty";212      }213    }else{214      $scope.queueModalError=validate;215    }216  };  217  $scope.initEditMessage=function(msgObj){218    $scope.queueActiveTab="Text";    219    $scope.requestedIndex=$scope.queueMessagesList.indexOf(msgObj);220    $scope.requestedMessage=angular.copy(msgObj);    221    if($scope.requestedMessage.expires){222      var date=new Date($scope.requestedMessage.expires).getDate();223      var month=new Date($scope.requestedMessage.expires).getMonth()+1;224      var year=new Date($scope.requestedMessage.expires).getFullYear();225      226      var hours=new Date($scope.requestedMessage.expires).getHours();227      var minutes=new Date($scope.requestedMessage.expires).getMinutes();228      var seconds=new Date($scope.requestedMessage.expires).getSeconds();      229      $scope.requestedMessage.expires=year+"-"+month+"-"+date+" "+hours+":"+minutes+":"+seconds;      230    }231    if(_isJsonString($scope.requestedMessage.message)){232      $scope.requestedMessage.message=JSON.parse($scope.requestedMessage.message);      233    }234    if(Object.prototype.toString.call($scope.requestedMessage.message)=="[object Object]" || Object.prototype.toString.call($scope.requestedMessage.message)=="[object Array]"){235      $scope.queueActiveTab="JSON";      236      $scope.requestedMessage.message=JSON.stringify($scope.requestedMessage.message,null,2);237    }     238    $scope.requestedSplDelay=$scope.requestedMessage.delay;239    240    $("#md-editmsg").modal();241  };242  $scope.editMessage=function(){ 243    $scope.requestedMessage.delay=angular.copy($scope.requestedSplDelay);244    $scope.requestedSplDelay=null;245    var validate=validateQ($scope.requestedMessage); 246    if(!validate){247     248      if($scope.requestedMessage.message){249        if(_isJsonString($scope.requestedMessage.message)){250          $scope.requestedMessage.message=JSON.parse($scope.requestedMessage.message);251        }252        if($scope.requestedMessage.expires){253          $scope.requestedMessage.expires=new Date($scope.requestedMessage.expires);254        }255        $scope.queueModalError=null;              256        $scope.editMsgSpinner=true;257        258        queueService.editMessage($scope.activatedQueue[$scope.previousIndex],$scope.requestedMessage)259        .then(function(resp){260          $scope.queueMessagesList[$scope.requestedIndex]=resp;261          $("#md-editmsg").modal("hide");     262          $scope.queueModalError=null;              263          $scope.editMsgSpinner=false;264          $scope.requestedMessage=null;265          $scope.requestedIndex=null;266        }, function(error){  267          $scope.queueModalError=error; 268          $scope.editMsgSpinner=false;       269        });270      }else{271        $scope.queueModalError="Message shoudn't be empty";272      }273    }else{274      $scope.queueModalError=validate;275    }276  };277  $scope.initDeleteMessage=function(msgObj){278    $scope.requestedMessage=msgObj;279    $("#md-deletemsg").modal();280  };281  $scope.deleteMessage=function(){282    var index=$scope.queueMessagesList.indexOf($scope.requestedMessage);283    $scope.queueModalError=null;              284    $scope.deleteMsgSpinner=true;285    queueService.deleteMsgById($scope.activatedQueue[$scope.previousIndex],$scope.requestedMessage.id)286    .then(function(resp){287      $("#md-deletemsg").modal("hide");     288      $scope.queueMessagesList.splice(index,1);289      $scope.queueModalError=null;              290      $scope.deleteMsgSpinner=false;291      $scope.requestedMessage=null;292      $scope.queueSizes[$scope.previousIndex]=--$scope.queueSizes[$scope.previousIndex];293    }, function(error){  294      $scope.queueModalError=error; 295      $scope.deleteMsgSpinner=false;       296    });297  };298  299  $scope.toggleMsgAdvanceOptions=function(){300    if($scope.openMsgAdvanceOptions==true){301      $scope.openMsgAdvanceOptions=false;302    }else{303      $scope.openMsgAdvanceOptions=true;304    }305  };  306  function getAllQueues(){307    $scope.queueListLoading=true;308    queueService.getAllQueues()309    .then(function(list){310      if(list && list.length>0){311        $scope.queueList=list;312      }313      314      $scope.queueListLoading=false;315      if($scope.queueList && $scope.queueList.length>0){316        $scope.firstVisit=false;317        $scope.openQueueDetails($scope.queueList[0]);318        for(var i=0;i<$scope.queueList.length;++i){319          $scope.queueSizes[i]=$scope.queueList[i].size;320        } 321      }322     323    }, function(error){  324      $scope.queueListLoading=false;    325      errorNotify(error);        326    });327  }328  //Private Functions329  function loadProject(id){330    331    if($rootScope.currentProject){332      initCB();333      getAllQueues();334      getProjectTables();      335    }else{336      projectService.getProject(id)337      .then(function(currentProject){338        if(currentProject){339          $rootScope.currentProject=currentProject;340          initCB();341          getAllQueues();342          getProjectTables();   343          $rootScope.pageHeaderDisplay=$rootScope.currentProject.name;         344        }                              345      }, function(error){          346      });347    }348    349  }350  function getProjectTables(){351    var promises=[];  352    if(!$rootScope.currentProject.tables || $rootScope.currentProject.tables.length==0){    353      promises.push(tableService.getProjectTables());     354    }  355    $q.all(promises).then(function(list){356      if(list.length==1){      357        $rootScope.currentProject.tables=list[0];           358      }359    }, function(err){      360    });361  }362  function validateQ(q){363    if(q.timeout){364      q.timeout=parseInt(q.timeout);365      if(!q.timeout){366        return "timeout should be a number";367      }368      if(typeof q.timeout!="number"){369        return "timeout should be a number";370      }          371    }372    if(q.delay){373      q.delay=parseInt(q.delay);374      if(!q.delay){375        return "delay should be a number";376      }377      if(typeof q.delay!="number"){378        return "delay should be a number";379      }       380    }381    if(q.expires && !q.expires instanceof Date){382      return "expires should be a date object";383    }384    return null;385  }386  function initCB(){387    CB.CloudApp.init(SERVER_URL,$rootScope.currentProject.appId, $rootScope.currentProject.keys.master);388  }           389    ...

queueService.js

Source:queueService.js

1app.factory('queueService', ['$q','$http','$rootScope',function ($q,$http,$rootScope) {2    var global = {};  3    global.createQueue = function(name,type){4      var q=$q.defer();5      var queue = new CB.CloudQueue(name);6      queue.create({7          success : function(queueObject){8            q.resolve(queueObject); 9            if(!__isDevelopment){10              /****Tracking*********/              11               mixpanel.track('Create Queue', {"App id": $rootScope.currentProject.appId,"Queue Name": name});12              /****End of Tracking*****/13            }14          }, error : function(error){15            q.reject(error);16          }17      });18      return  q.promise;19    };20    global.getAllQueues = function(){21      var q=$q.defer();22      CB.CloudQueue.getAll({23        success : function(list){24          q.resolve(list);25        }, error : function(error){26          q.reject(error);27        }28      });29      return  q.promise;30    };31    global.updateQueue = function(queue){32      var q=$q.defer();33      34      queue.update({35        success : function(updatedQueue){36          q.resolve(updatedQueue);            37        }, error : function(error){38          q.reject(error);39        }40      });41      return  q.promise;42    };43    global.deleteQueue = function(queue){44      var q=$q.defer();45      queue.delete({46        success : function(resp){47          q.resolve(resp);48        }, error : function(error){49          q.reject(error);50        }51      });52      return  q.promise;53    };54    global.getQueueInfo = function(queue){55      var q=$q.defer();56      queue.get({57        success : function(queueInfo){58          q.resolve(queueInfo);59        }, error : function(error){60          q.reject(error);61        }62      });63      return  q.promise;64    };65    global.getAllMessages = function(queue){66      var q=$q.defer();67      queue.getAllMessages({68        success : function(list){69          q.resolve(list);70        }, error : function(error){71          q.reject(error);72        }73      });74      return  q.promise;75    };76    global.getMessageById = function(queue,messageId){77      var q=$q.defer();78      queue.getMessageById(messageId,{79        success : function(message){80          q.resolve(message);81        }, error : function(error){82          q.reject(error);83        }84      });85      return  q.promise;86    };87    global.insertMessageIntoQueue = function(queue,message,timeout,delay,expire){88      var q=$q.defer();89      var queueMessage = new CB.QueueMessage();90      queueMessage.message = message;91      if(timeout>0){92        queueMessage.timeout = timeout;93      }     94      if(delay>0){95        queueMessage.delay = delay;96      }     97      if(expire){98        queueMessage.expires = expire;99      }100      queue.addMessage(queueMessage, {101        success : function(queueMessage){102          q.resolve(queueMessage);  103        }, error : function(error){104          q.reject(error);105        }106      });107      return  q.promise;108    };109    global.editMessage = function(queue,queueMessage){110      var q=$q.defer();      111      queue.updateMessage(queueMessage, {112        success : function(updatedQueueMsg){113          q.resolve(updatedQueueMsg);  114        }, error : function(error){115          q.reject(error);116        }117      });118      return  q.promise;119    };120    global.deleteMsgById = function(queue,messageId){121      var q=$q.defer();122      queue.deleteMessage(messageId, {123        success : function(queueMessage){124          q.resolve(queueMessage); 125        }, error : function(error){126          q.reject(error);127        }128      });129      return  q.promise;130    };131    global.getDataFromQueue = function(queue){132      var q=$q.defer();133      queue.pull({134        success : function(queueMessage){ 135          q.resolve(queueMessage);          136        }, error : function(error){ 137          q.reject(error);        138        }139      });140      return  q.promise;141    };142    return global;...

usedSharedRessources.py

Source:usedSharedRessources.py

1usedSharedRessources = {2    'marvinGui': [3        'processDict',4        'arduinoDict',5        'servoDict',6        'cartDict',7        'environmentDict',8        'sharedDataUpdateQueue',9        'skeletonGuiUpdateQueue',10        'cartGuiUpdateQueue',11        'mapGuiUpdateQueue',12        'mainGuiUpdateQueue',13        'ikUpdateQueue',14        'skeletonRequestQueue',15        'cartRequestQueue',16        'speakRequestQueue',17        'playGestureQueue',18        'navManagerRequestQueue',19        'imageProcessingRequestQueue',20    ],21    'skeletonControl': [22        'processDict',23        #'arduinoDict',24        'servoDict',25        #'cartDict'26        'sharedDataUpdateQueue',27        'skeletonGuiUpdateQueue',28        #'cartGuiUpdateQueue',29        'ikUpdateQueue',30        'skeletonRequestQueue',31        #'cartRequestQueue',32        'speakRequestQueue',33        #'SpeakRespondQueue',34        'imageProcessingRequestQueue',35    ],36    'cartControl': [37        'processDict',38        #'arduinoDict',39        'servoDict',40        #'cartDict'41        'sharedDataUpdateQueue',42        #'skeletonGuiUpdateQueue',43        'cartGuiUpdateQueue',44        #'ikUpdateQueue',45        'skeletonRequestQueue',46        'cartRequestQueue',47        #'speakRequestQueue',48        # 'SpeakRespondQueue',49        'imageProcessingRequestQueue',50    ],51    'randomMoves': [52        'processDict',53        # 'arduinoDict',54        'servoDict',55        # 'cartDict',56        'sharedDataUpdateQueue',57        # 'skeletonGuiUpdateQueue',58        # 'cartGuiUpdateQueue',59        # 'ikUpdateQueue',60        'skeletonRequestQueue',61        # 'cartRequestQueue',62        'speakRequestQueue',63        # 'SpeakRespondQueue',64        # 'imageProcessingRequestQueue',65        ],66    'playGesture': [67        'processDict',68        # 'arduinoDict',69        'servoDict',70        'cartDict',71        'sharedDataUpdateQueue',72        # 'skeletonGuiUpdateQueue',73        # 'cartGuiUpdateQueue',74        # 'ikUpdateQueue',75        'skeletonRequestQueue',76        'cartRequestQueue',77        'speakRequestQueue',78        'playGestureQueue',79        # 'navManagerRequestQueue',80        # 'imageProcessingRequestQueue'81    ],82    'speechControl': [83        'processDict',84        # 'arduinoDict',85        'servoDict',86        # 'cartDict',87        'sharedDataUpdateQueue',88        # 'skeletonGuiUpdateQueue',89        # 'cartGuiUpdateQueue',90        # 'ikUpdateQueue',91        'skeletonRequestQueue',92        # 'cartRequestQueue',93        'speakRequestQueue',94        'playGestureQueue',95        # 'navManagerRequestQueue',96        # 'imageProcessingRequestQueue'97    ],98    'imageProcessing': [99        'processDict',100        # 'arduinoDict',101        # 'servoDict',102        'cartDict',103        'environmentDict',104        'sharedDataUpdateQueue',105        # 'skeletonGuiUpdateQueue',106        # 'cartGuiUpdateQueue',107        # 'ikUpdateQueue',108        'skeletonRequestQueue',109        'cartRequestQueue',110        #'speakRequestQueue',111        # 'SpeakRespondQueue',112        'imageProcessingRequestQueue'113    ],114    'navManager': [115        'processDict',116        # 'arduinoDict',117        'servoDict',118        'cartDict',119        'environmentDict',120        'sharedDataUpdateQueue',121        # 'skeletonGuiUpdateQueue',122        # 'cartGuiUpdateQueue',123        # 'ikUpdateQueue',124        'skeletonRequestQueue',125        'cartRequestQueue',126        # 'speakRequestQueue',127        # 'SpeakRespondQueue',128        'navManagerRequestQueue',129        'imageProcessingRequestQueue'130    ],131    'stickFigure': [132        'processDict',133        # 'arduinoDict',134        'servoDict',135        # 'cartDict',136        #'sharedDataUpdateQueue',137        # 'skeletonGuiUpdateQueue',138        # 'cartGuiUpdateQueue',139        'ikUpdateQueue',140        #'skeletonRequestQueue',141        # 'cartRequestQueue',142        #'speakRequestQueue',143        #'playGestureQueue',144        #'navManagerRequestQueue',145        # 'imageProcessingRequestQueue'146    ],...

Using AI Code Generation

1var wpt = require('webpagetest');2var test = wpt('www.webpagetest.org');3    if (err) {4        console.log(err);5    } else {6        console.log(data);7        var testId = data.data.testId;8        console.log(testId);9        test.getTestStatus(testId, function(err, data) {10            if (err) {11                console.log(err);12            } else {13                console.log(data);14            }15        });16    }17});

Using AI Code Generation

1var wpt = require('webpagetest');2var wpt = new WebPageTest('www.webpagetest.org');3var options = {4    videoParams: {5    }6};7    if (err) return console.error(err);8    console.log('Test Results: ' + data.data.runs);9    console.log('Test Results: ' + data.data.runs[1].firstView.videoFrames);10    console.log('Test Results: ' + data.data.runs[1].firstView.videoFiles);11});12    at Request._callback (/Users/xxx/Desktop/node-wpt/node_modules/webpagetest/lib/webpagetest.js:192:41)13    at Request.self.callback (/Users/xxx/Desktop/node-wpt/node_modules/webpagetest/node_modules/request/request.js:186:22)14    at emitTwo (events.js:87:13)15    at Request.emit (events.js:172:7)16    at Request.<anonymous> (/Users/xxx/Desktop/node-wpt/node_modules/webpagetest/node_modules/request/request.js:1161:10)17    at emitOne (events.js:77:13)18    at Request.emit (events.js:169:7)19    at IncomingMessage.<anonymous> (/Users/xxx/Desktop/node-wpt/node_modules/webpagetest/node_modules/request/request.js:1083:12)20    at emitNone (events.js:72:20)21    at IncomingMessage.emit (events.js:166:7)

Using AI Code Generation

1var wpt = require('webpagetest');2var options = {3};4var wptDriver = new wpt(options);5var runs = 1;6var location = 'Dulles_MotoG4:Chrome.56';7var firstViewOnly = true;8var connectivity = 'Cable';9var label = 'test';10var pollResults = 5;11var timeout = 60;12var private = false;13var video = true;14var login = '';15var password = '';16var basicAuthUser = '';17var basicAuthPassword = '';18var script = '';19var notifyEmail = '';20var requests = '';21var block = '';22var bandwidthDown = '';23var bandwidthUp = '';24var latency = '';25var packetLossRate = '';26var keepOriginalUserAgent = '';27var disableOptimizations = '';28var clearCerts = '';29var disableJavaScript = '';30var disableFlash = '';31var disableCSS = '';32var disableVideo = '';33var disableImages = '';34var disableTCPFastOpen = '';35var disableHTTPKeepAlive = '';36var disableWebSecurity = '';37var tcpdump = '';38var timeline = '';39var trace = '';40var netlog = '';41var chromeTrace = '';42var spof = '';43var spofUrl = '';44var spofAction = '';45var spofWait = '';46var spofWaitResponse = '';47var spofWaitAction = '';48var spofWaitActionResponse = '';49var spofWaitScript = '';50var spofWaitScriptResponse = '';51var spofWaitScriptAction = '';52var spofWaitScriptActionResponse = '';53var spofWaitScriptActionResponse2 = '';54var spofWaitScriptActionResponse3 = '';55var spofWaitScriptActionResponse4 = '';56var spofWaitScriptActionResponse5 = '';57var spofWaitScriptActionResponse6 = '';58var spofWaitScriptActionResponse7 = '';59var spofWaitScriptActionResponse8 = '';60var spofWaitScriptActionResponse9 = '';61var spofWaitScriptActionResponse10 = '';62var spofWaitScriptActionResponse11 = '';63var spofWaitScriptActionResponse12 = '';64var spofWaitScriptActionResponse13 = '';

Using AI Code Generation

1var wpt = require('webpagetest');2var wpt = new WebPageTest('www.webpagetest.org', 'A.3b3a7fcf8e9d9e7b1d2f2d7c8c1d1b7c');3    if (err) {4        return console.error(err);5    }6    console.log('Test submitted');7    console.log('Test ID: ' + data.data.testId);8    console.log('Test status: ' + data.data.statusText);9    console.log('View test at: ' + data.data.summary);10});11    if (err) {12        return console.error(err);13    }14    console.log('Test submitted');15    console.log('Test ID: ' + data.data.testId);16    console.log('Test status: ' + data.data.statusText);17    console.log('View test at: ' + data.data.summary);18});19    if (err) {20        return console.error(err);21    }22    console.log('Test submitted');23    console.log('Test ID: ' + data.data.testId);24    console.log('Test status: ' + data.data.statusText);25    console.log('View test at: ' + data.data.summary);26});27    if (err) {28        return console.error(err);29    }30    console.log('Test submitted');31    console.log('Test ID: ' + data.data.testId);32    console.log('Test status: ' + data.data.statusText);33    console.log('View test at: ' + data.data.summary);34});35    if (err) {36        return console.error(err);37    }38    console.log('Test submitted');39    console.log('Test ID: ' + data.data.testId);40    console.log('Test status: ' + data.data.statusText);

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.