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

Completed

Push — master ( 663472...7037a7 )

by Roy

created 2018-06-10 01:07 UTC

TestTimeQueue.test_time_queue() B

↳ Parent: TestTimeQueue

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	6
c	1
b	0
f	0
dl	0
loc	63
rs	7.7105

How to fix Long Method

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

import time
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, taskid=', it.taskid, 'priority=', it.priority, 'exetime=', it.exetime)
        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, taskid=', task.taskid, 'priority=', task.priority, 'exetime=', task.exetime)

        self.assertEqual(tq.size(), 100)
        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


class TestTimeQueue(unittest.TestCase):
    def test_time_queue(self):

        six.print_('Test time queue order by time only')

        tq = TaskQueue(rate=300, burst=1000)

        fifo_queue = Queue.Queue()

        interval = 5.0 / 1000

        for i in range(0, 20):
            it = InQueueTask(str(i), priority=int(i // 10), exetime=time.time() + (i + 1) * interval)
            tq.put(it.taskid, it.priority, it.exetime)
            fifo_queue.put(it)
            six.print_('put, taskid=', it.taskid, 'priority=', it.priority, 'exetime=', it.exetime)

        self.assertEqual(tq.priority_queue.qsize(), 0)
        self.assertEqual(tq.processing.qsize(), 0)
        self.assertEqual(tq.time_queue.qsize(), 20)

        for i in range(0, 20):
            t1 = fifo_queue.get()
            t2 = tq.time_queue.get()
            self.assertEqual(t1.taskid, t2.taskid)
            six.print_('get, taskid=', t2.taskid, 'priority=', t2.priority, 'exetime=', t2.exetime)
        self.assertEqual(tq.priority_queue.qsize(), 0)
        self.assertEqual(tq.processing.qsize(), 0)
        self.assertEqual(tq.time_queue.qsize(), 0)

        queues = dict()
        tasks = dict()
        for i in range(0, 20):
            priority = int(i // 10)
            it = InQueueTask(str(i), priority=priority, exetime=time.time() + (i + 1) * interval)
            tq.put(it.taskid, it.priority, it.exetime)
            tasks[it.taskid] = it

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

        self.assertEqual(tq.priority_queue.qsize(), 0)
        self.assertEqual(tq.processing.qsize(), 0)
        self.assertEqual(tq.time_queue.qsize(), 20)

        time.sleep(20 * interval)
        tq.check_update()
        self.assertEqual(tq.priority_queue.qsize(), 20)
        self.assertEqual(tq.processing.qsize(), 0)
        self.assertEqual(tq.time_queue.qsize(), 0)
        for i in range(0, 20):
            taskid = tq.get()
            t1 = tasks[taskid]
            t2 = queues[t1.priority].get()
            self.assertEqual(t1.taskid, t2.taskid)

        self.assertEqual(tq.priority_queue.qsize(), 0)
        self.assertEqual(tq.processing.qsize(), 20)
        self.assertEqual(tq.time_queue.qsize(), 0)

        pass

    pass


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


1			#!/usr/bin/env python
2			# -- coding: utf-8 --
3
4			import time
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			six.print_('put, taskid=', it.taskid, 'priority=', it.priority, 'exetime=', it.exetime)
35			for i in range(0, 100):
36			task_id = tq.get()
37			task = tasks[task_id]
38			q = queues[task.priority] # type: Queue.Queue
39			expect_task = q.get()
40			self.assertEqual(task_id, expect_task.taskid)
41			self.assertEqual(task.priority, int(9 - i // 10))
42			six.print_('get, taskid=', task.taskid, 'priority=', task.priority, 'exetime=', task.exetime)
43
44			self.assertEqual(tq.size(), 100)
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			class TestTimeQueue(unittest.TestCase):
55			def test_time_queue(self):
56
57			six.print_('Test time queue order by time only')
58
59			tq = TaskQueue(rate=300, burst=1000)
60
61			fifo_queue = Queue.Queue()
62
63			interval = 5.0 / 1000
64
65			for i in range(0, 20):
66			it = InQueueTask(str(i), priority=int(i // 10), exetime=time.time() + (i + 1) * interval)
67			tq.put(it.taskid, it.priority, it.exetime)
68			fifo_queue.put(it)
69			six.print_('put, taskid=', it.taskid, 'priority=', it.priority, 'exetime=', it.exetime)
70
71			self.assertEqual(tq.priority_queue.qsize(), 0)
72			self.assertEqual(tq.processing.qsize(), 0)
73			self.assertEqual(tq.time_queue.qsize(), 20)
74
75			for i in range(0, 20):
76			t1 = fifo_queue.get()
77			t2 = tq.time_queue.get()
78			self.assertEqual(t1.taskid, t2.taskid)
79			six.print_('get, taskid=', t2.taskid, 'priority=', t2.priority, 'exetime=', t2.exetime)
80			self.assertEqual(tq.priority_queue.qsize(), 0)
81			self.assertEqual(tq.processing.qsize(), 0)
82			self.assertEqual(tq.time_queue.qsize(), 0)
83
84			queues = dict()
85			tasks = dict()
86			for i in range(0, 20):
87			priority = int(i // 10)
88			it = InQueueTask(str(i), priority=priority, exetime=time.time() + (i + 1) * interval)
89			tq.put(it.taskid, it.priority, it.exetime)
90			tasks[it.taskid] = it
91
92			if priority not in queues:
93			queues[priority] = Queue.Queue()
94			q = queues[priority]
95			q.put(it)
96			pass
97
98			self.assertEqual(tq.priority_queue.qsize(), 0)
99			self.assertEqual(tq.processing.qsize(), 0)
100			self.assertEqual(tq.time_queue.qsize(), 20)
101
102			time.sleep(20 * interval)
103			tq.check_update()
104			self.assertEqual(tq.priority_queue.qsize(), 20)
105			self.assertEqual(tq.processing.qsize(), 0)
106			self.assertEqual(tq.time_queue.qsize(), 0)
107			for i in range(0, 20):
108			taskid = tq.get()
109			t1 = tasks[taskid]
110			t2 = queues[t1.priority].get()
111			self.assertEqual(t1.taskid, t2.taskid)
112
113			self.assertEqual(tq.priority_queue.qsize(), 0)
114			self.assertEqual(tq.processing.qsize(), 20)
115			self.assertEqual(tq.time_queue.qsize(), 0)
116
117			pass
118
119			pass
120
121
122			if __name__ == '__main__':
123			unittest.main()
124

binux / pyspider

Push — master ( 663472...7037a7 )

TestTimeQueue.test_time_queue() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like