TimingTestCase - Code Metrics - jacquev6/ActionTree - Measure and Improve Code Quality continuously with Scrutinizer

TimingTestCase A
last analyzed 2018-06-11 07:43 UTC

↳ Parent: Project

Complexity

Total Complexity

Size/Duplication

Total Lines	141
Duplicated Lines	15.6 %

Importance

Changes	11
Bugs	0	Features	1

Metric	Value
dl	22
loc	141
rs	10
c	11
b	0
f	1
wmc	19

10 Methods

Rating	Name	Duplication	Size	Complexity
A	test_failure()	11	11	1
A	test_success()	11	11	1
A	test_leaves_have_same_ready_time()	0	13	1
A	test_cancelation_with_keep_going()	0	11	1
A	test_cancelation_before_ready()	0	20	1
A	test_many_dependencies_with_limited_cpu_cores()	0	12	4
A	test_many_dependencies_with_one_cpu_cores()	0	12	4
A	test_abundant_resource_with_many_cpu_cores()	0	14	1
A	test_scarce_resource_with_many_cpu_cores()	0	16	1
A	test_many_dependencies_with_unlimited_cpu_cores()	0	10	4

How to fix Duplicated Code

# coding: utf8

# Copyright 2013-2018 Vincent Jacques <[email protected]>

from __future__ import division, absolute_import, print_function

import unittest

from ActionTree import *
from . import *


class TimingTestCase(ActionTreeTestCase):
    def test_success(self):

        a = self._action("a")

        report = execute(a)

        self.assertIsInstance(report.get_action_status(a).pending_time, datetime.datetime)
        self.assertEqual(report.get_action_status(a).ready_time, report.get_action_status(a).pending_time)
        self.assertIsNone(report.get_action_status(a).cancel_time)
        self.assertEqual(report.get_action_status(a).start_time, report.get_action_status(a).ready_time)
        self.assertGreater(report.get_action_status(a).success_time, report.get_action_status(a).start_time)
        self.assertIsNone(report.get_action_status(a).failure_time)

    def test_failure(self):

        a = self._action("a", exception=Exception())

        report = execute(a, do_raise=False)

        self.assertIsInstance(report.get_action_status(a).pending_time, datetime.datetime)
        self.assertEqual(report.get_action_status(a).ready_time, report.get_action_status(a).pending_time)
        self.assertIsNone(report.get_action_status(a).cancel_time)
        self.assertEqual(report.get_action_status(a).start_time, report.get_action_status(a).ready_time)
        self.assertIsNone(report.get_action_status(a).success_time)
        self.assertGreater(report.get_action_status(a).failure_time, report.get_action_status(a).start_time)

    def test_cancelation_before_ready(self):
        a = self._action("a")
        b = self._action("b", exception=Exception())
        a.add_dependency(b)

        report = execute(a, do_raise=False)

        self.assertIsInstance(report.get_action_status(b).pending_time, datetime.datetime)
        self.assertEqual(report.get_action_status(b).ready_time, report.get_action_status(b).pending_time)
        self.assertIsNone(report.get_action_status(b).cancel_time)
        self.assertEqual(report.get_action_status(b).start_time, report.get_action_status(b).ready_time)
        self.assertIsNone(report.get_action_status(b).success_time)
        self.assertGreater(report.get_action_status(b).failure_time, report.get_action_status(b).start_time)

        self.assertIsInstance(report.get_action_status(a).pending_time, datetime.datetime)
        self.assertIsNone(report.get_action_status(a).ready_time)
        self.assertEqual(report.get_action_status(a).cancel_time, report.get_action_status(b).failure_time)
        self.assertIsNone(report.get_action_status(a).start_time)
        self.assertIsNone(report.get_action_status(a).success_time)
        self.assertIsNone(report.get_action_status(a).failure_time)

    def test_cancelation_with_keep_going(self):
        a = self._action("a")
        b = self._action("b")
        a.add_dependency(b)
        c = self._action("c", exception=Exception())
        b.add_dependency(c)

        report = execute(a, keep_going=True, do_raise=False)

        self.assertEqual(report.get_action_status(b).cancel_time, report.get_action_status(c).failure_time)
        self.assertEqual(report.get_action_status(a).cancel_time, report.get_action_status(b).cancel_time)

    def test_leaves_have_same_ready_time(self):
        a = self._action("a")
        b = self._action("b")
        c = self._action("c")
        d = self._action("d")
        a.add_dependency(b)
        a.add_dependency(c)
        a.add_dependency(d)

        report = execute(a)

        self.assertEqual(report.get_action_status(c).ready_time, report.get_action_status(b).ready_time)
        self.assertEqual(report.get_action_status(d).ready_time, report.get_action_status(b).ready_time)

    def test_many_dependencies_with_unlimited_cpu_cores(self):
        MANY = 20
        a = self._action("a")
        deps = [self._action(str(i)) for i in range(MANY)]
        for dep in deps:
            a.add_dependency(dep)

        report = execute(a, cpu_cores=UNLIMITED)
        for dep in deps[1:]:
            self.assertEqual(report.get_action_status(dep).start_time, report.get_action_status(deps[0]).start_time)

    def test_many_dependencies_with_one_cpu_cores(self):
        MANY = 20
        a = self._action("a")
        deps = [self._action(str(i)) for i in range(MANY)]
        for dep in deps:
            a.add_dependency(dep)

        report = execute(a, cpu_cores=1)

        # No two actions have started at the same time
        start_times = set(report.get_action_status(dep).start_time for dep in deps)
        self.assertEqual(len(start_times), MANY)

    def test_many_dependencies_with_limited_cpu_cores(self):
        MANY = 20
        a = self._action("a")
        deps = [self._action(str(i)) for i in range(MANY)]
        for dep in deps:
            a.add_dependency(dep)

        report = execute(a, cpu_cores=3)

        # Only the first three actions have started at the same time
        start_times = set(report.get_action_status(dep).start_time for dep in deps)
        self.assertEqual(len(start_times), MANY - 2)

    def test_scarce_resource_with_many_cpu_cores(self):
        r = Resource(1)
        a = self._action("a")
        b = self._action("b")
        b.require_resource(r, 1)
        c = self._action("c")
        c.require_resource(r, 1)
        a.add_dependency(b)
        a.add_dependency(c)

        report = execute(a, cpu_cores=6)

        # @todo Start next action at the same timestamp
        self.assertTrue(
            report.get_action_status(b).start_time > report.get_action_status(c).success_time or
            report.get_action_status(c).start_time > report.get_action_status(b).success_time
        )

    def test_abundant_resource_with_many_cpu_cores(self):
        r = Resource(2)
        a = self._action("a")
        b = self._action("b")
        b.require_resource(r, 1)
        c = self._action("c")
        c.require_resource(r, 1)
        a.add_dependency(b)
        a.add_dependency(c)

        report = execute(a, cpu_cores=6)

        self.assertEqual(report.get_action_status(c).ready_time, report.get_action_status(b).ready_time)
        self.assertEqual(report.get_action_status(c).start_time, report.get_action_status(b).start_time)


1		# coding: utf8
2
3		# Copyright 2013-2018 Vincent Jacques <[email protected]>
4
5		from __future__ import division, absolute_import, print_function
6
7		import unittest
8
9		from ActionTree import *
10		from . import *
11
12
13		class TimingTestCase(ActionTreeTestCase):
14	View Code Duplication	def test_success(self):
		0 ignored issues – show Duplication introduced 2017-06-13 12:17 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
15		a = self._action("a")
16
17		report = execute(a)
18
19		self.assertIsInstance(report.get_action_status(a).pending_time, datetime.datetime)
20		self.assertEqual(report.get_action_status(a).ready_time, report.get_action_status(a).pending_time)
21		self.assertIsNone(report.get_action_status(a).cancel_time)
22		self.assertEqual(report.get_action_status(a).start_time, report.get_action_status(a).ready_time)
23		self.assertGreater(report.get_action_status(a).success_time, report.get_action_status(a).start_time)
24		self.assertIsNone(report.get_action_status(a).failure_time)
25
26	View Code Duplication	def test_failure(self):
		0 ignored issues – show Duplication introduced 2017-06-13 12:17 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
27		a = self._action("a", exception=Exception())
28
29		report = execute(a, do_raise=False)
30
31		self.assertIsInstance(report.get_action_status(a).pending_time, datetime.datetime)
32		self.assertEqual(report.get_action_status(a).ready_time, report.get_action_status(a).pending_time)
33		self.assertIsNone(report.get_action_status(a).cancel_time)
34		self.assertEqual(report.get_action_status(a).start_time, report.get_action_status(a).ready_time)
35		self.assertIsNone(report.get_action_status(a).success_time)
36		self.assertGreater(report.get_action_status(a).failure_time, report.get_action_status(a).start_time)
37
38		def test_cancelation_before_ready(self):
39		a = self._action("a")
40		b = self._action("b", exception=Exception())
41		a.add_dependency(b)
42
43		report = execute(a, do_raise=False)
44
45		self.assertIsInstance(report.get_action_status(b).pending_time, datetime.datetime)
46		self.assertEqual(report.get_action_status(b).ready_time, report.get_action_status(b).pending_time)
47		self.assertIsNone(report.get_action_status(b).cancel_time)
48		self.assertEqual(report.get_action_status(b).start_time, report.get_action_status(b).ready_time)
49		self.assertIsNone(report.get_action_status(b).success_time)
50		self.assertGreater(report.get_action_status(b).failure_time, report.get_action_status(b).start_time)
51
52		self.assertIsInstance(report.get_action_status(a).pending_time, datetime.datetime)
53		self.assertIsNone(report.get_action_status(a).ready_time)
54		self.assertEqual(report.get_action_status(a).cancel_time, report.get_action_status(b).failure_time)
55		self.assertIsNone(report.get_action_status(a).start_time)
56		self.assertIsNone(report.get_action_status(a).success_time)
57		self.assertIsNone(report.get_action_status(a).failure_time)
58
59		def test_cancelation_with_keep_going(self):
60		a = self._action("a")
61		b = self._action("b")
62		a.add_dependency(b)
63		c = self._action("c", exception=Exception())
64		b.add_dependency(c)
65
66		report = execute(a, keep_going=True, do_raise=False)
67
68		self.assertEqual(report.get_action_status(b).cancel_time, report.get_action_status(c).failure_time)
69		self.assertEqual(report.get_action_status(a).cancel_time, report.get_action_status(b).cancel_time)
70
71		def test_leaves_have_same_ready_time(self):
72		a = self._action("a")
73		b = self._action("b")
74		c = self._action("c")
75		d = self._action("d")
76		a.add_dependency(b)
77		a.add_dependency(c)
78		a.add_dependency(d)
79
80		report = execute(a)
81
82		self.assertEqual(report.get_action_status(c).ready_time, report.get_action_status(b).ready_time)
83		self.assertEqual(report.get_action_status(d).ready_time, report.get_action_status(b).ready_time)
84
85		def test_many_dependencies_with_unlimited_cpu_cores(self):
86		MANY = 20
87		a = self._action("a")
88		deps = [self._action(str(i)) for i in range(MANY)]
89		for dep in deps:
90		a.add_dependency(dep)
91
92		report = execute(a, cpu_cores=UNLIMITED)
93		for dep in deps[1:]:
94		self.assertEqual(report.get_action_status(dep).start_time, report.get_action_status(deps[0]).start_time)
95
96		def test_many_dependencies_with_one_cpu_cores(self):
97		MANY = 20
98		a = self._action("a")
99		deps = [self._action(str(i)) for i in range(MANY)]
100		for dep in deps:
101		a.add_dependency(dep)
102
103		report = execute(a, cpu_cores=1)
104
105		# No two actions have started at the same time
106		start_times = set(report.get_action_status(dep).start_time for dep in deps)
107		self.assertEqual(len(start_times), MANY)
108
109		def test_many_dependencies_with_limited_cpu_cores(self):
110		MANY = 20
111		a = self._action("a")
112		deps = [self._action(str(i)) for i in range(MANY)]
113		for dep in deps:
114		a.add_dependency(dep)
115
116		report = execute(a, cpu_cores=3)
117
118		# Only the first three actions have started at the same time
119		start_times = set(report.get_action_status(dep).start_time for dep in deps)
120		self.assertEqual(len(start_times), MANY - 2)
121
122		def test_scarce_resource_with_many_cpu_cores(self):
123		r = Resource(1)
124		a = self._action("a")
125		b = self._action("b")
126		b.require_resource(r, 1)
127		c = self._action("c")
128		c.require_resource(r, 1)
129		a.add_dependency(b)
130		a.add_dependency(c)
131
132		report = execute(a, cpu_cores=6)
133
134		# @todo Start next action at the same timestamp
135		self.assertTrue(
136		report.get_action_status(b).start_time > report.get_action_status(c).success_time or
137		report.get_action_status(c).start_time > report.get_action_status(b).success_time
138		)
139
140		def test_abundant_resource_with_many_cpu_cores(self):
141		r = Resource(2)
142		a = self._action("a")
143		b = self._action("b")
144		b.require_resource(r, 1)
145		c = self._action("c")
146		c.require_resource(r, 1)
147		a.add_dependency(b)
148		a.add_dependency(c)
149
150		report = execute(a, cpu_cores=6)
151
152		self.assertEqual(report.get_action_status(c).ready_time, report.get_action_status(b).ready_time)
153		self.assertEqual(report.get_action_status(c).start_time, report.get_action_status(b).start_time)
154

jacquev6 / ActionTree

TimingTestCase A last analyzed 2018-06-11 07:43 UTC

Complexity

Size/Duplication

Importance

10 Methods

How to fix Duplicated Code

Duplicated Code

Duplication Side-by-Side

Filter issues like

TimingTestCase A
last analyzed 2018-06-11 07:43 UTC