TestTaskQueue.test_task_queue_in_time_order() - Code Metrics - Inspection of "optimise scheduler select limit and task queue" - binux/pyspider - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#794)

unknown

created 2018-05-29 02:32 UTC

TestTaskQueue.test_task_queue_in_time_order() B

↳ Parent: TestTaskQueue

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	5
dl	0
loc	33
rs	8.0894
c	1
b	0
f	0

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import Queue
import unittest

import six

from pyspider.scheduler.task_queue import InQueueTask, TaskQueue


class TestTaskQueue(unittest.TestCase):
    """
        TestTaskQueue
    """

    def test_task_queue_in_time_order(self):
        tq = TaskQueue(rate=300, burst=1000)

        queues = dict()
        tasks = dict()

        for i in range(0, 100):
            it = InQueueTask(str(i), priority=i / 10, exetime=0)
            tq.put(it.taskid, it.priority, it.exetime)

            if it.priority not in queues:
                queues[it.priority] = Queue.Queue()

            q = queues[it.priority]  # type:Queue.Queue
            q.put(it)
            tasks[it.taskid] = it

            six.print_('put, ', it)

        for i in range(0, 100):
            task_id = tq.get()
            task = tasks[task_id]
            q = queues[task.priority]  # type: Queue.Queue
            expect_task = q.get()
            self.assertEqual(task_id, expect_task.taskid)
            self.assertEqual(task.priority, 9 - i / 10)
            six.print_('get, ', task)

        self.assertEqual(tq.priority_queue.qsize(), 0)
        self.assertEqual(tq.processing.qsize(), 100)
        for q in six.itervalues(queues):  # type:Queue.Queue
            self.assertEqual(q.qsize(), 0)
        pass

    pass


if __name__ == '__main__':
    unittest.main()


1			#!/usr/bin/env python
2			# -- coding: utf-8 --
3
4			import Queue
5			import unittest
6
7			import six
8
9			from pyspider.scheduler.task_queue import InQueueTask, TaskQueue
10
11
12			class TestTaskQueue(unittest.TestCase):
13			"""
14			TestTaskQueue
15			"""
16
17			def test_task_queue_in_time_order(self):
18			tq = TaskQueue(rate=300, burst=1000)
19
20			queues = dict()
21			tasks = dict()
22
23			for i in range(0, 100):
24			it = InQueueTask(str(i), priority=i / 10, exetime=0)
25			tq.put(it.taskid, it.priority, it.exetime)
26
27			if it.priority not in queues:
28			queues[it.priority] = Queue.Queue()
29
30			q = queues[it.priority] # type:Queue.Queue
31			q.put(it)
32			tasks[it.taskid] = it
33
34			six.print_('put, ', it)
35
36			for i in range(0, 100):
37			task_id = tq.get()
38			task = tasks[task_id]
39			q = queues[task.priority] # type: Queue.Queue
40			expect_task = q.get()
41			self.assertEqual(task_id, expect_task.taskid)
42			self.assertEqual(task.priority, 9 - i / 10)
43			six.print_('get, ', task)
44
45			self.assertEqual(tq.priority_queue.qsize(), 0)
46			self.assertEqual(tq.processing.qsize(), 100)
47			for q in six.itervalues(queues): # type:Queue.Queue
48			self.assertEqual(q.qsize(), 0)
49			pass
50
51			pass
52
53
54			if __name__ == '__main__':
55			unittest.main()
56

binux / pyspider

Pull Request — master (#794)

TestTaskQueue.test_task_queue_in_time_order() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like