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

Completed

Pull Request — master (#795)

unknown

created 2018-05-29 03:08 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 unittest

import six
from six.moves import queue as Queue

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

binux / pyspider

Pull Request — master (#795)

TestTaskQueue.test_task_queue_in_time_order() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like