Completed
Push — master ( a4f2fd...4fbdfd )
by Vincent
01:14
created

TimingTestCase   A

Complexity

Total Complexity 8

Size/Duplication

Total Lines 79
Duplicated Lines 26.58 %

Importance

Changes 8
Bugs 0 Features 0
Metric Value
wmc 8
c 8
b 0
f 0
dl 21
loc 79
rs 10

5 Methods

Rating   Name   Duplication   Size   Complexity  
A test_failure() 10 10 1
A test_success() 11 11 1
A test_leaves_have_same_ready_time() 0 13 1
B test_cancelation_with_keep_going() 0 22 4
A test_cancelation_before_ready() 0 18 1

How to fix   Duplicated Code   

Duplicated Code

Duplicate code is one of the most pungent code smells. A rule that is often used is to re-structure code once it is duplicated in three or more places.

Common duplication problems, and corresponding solutions are:

1
# coding: utf8
2
3
# Copyright 2013-2017 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 by
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.assertGreater(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.assertGreater(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 by
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).ready_time, datetime.datetime)
32
        self.assertIsNone(report.get_action_status(a).cancel_time)
33
        self.assertGreater(report.get_action_status(a).start_time, report.get_action_status(a).ready_time)
34
        self.assertIsNone(report.get_action_status(a).success_time)
35
        self.assertGreater(report.get_action_status(a).failure_time, report.get_action_status(a).start_time)
36
37
    def test_cancelation_before_ready(self):
38
        a = self._action("a")
39
        b = self._action("b", exception=Exception())
40
        a.add_dependency(b)
41
42
        report = execute(a, do_raise=False)
43
44
        self.assertIsInstance(report.get_action_status(b).ready_time, datetime.datetime)
45
        self.assertIsNone(report.get_action_status(b).cancel_time)
46
        self.assertGreater(report.get_action_status(b).start_time, report.get_action_status(b).ready_time)
47
        self.assertIsNone(report.get_action_status(b).success_time)
48
        self.assertGreater(report.get_action_status(b).failure_time, report.get_action_status(b).start_time)
49
50
        self.assertIsNone(report.get_action_status(a).ready_time)
51
        self.assertGreater(report.get_action_status(a).cancel_time, report.get_action_status(b).failure_time)
52
        self.assertIsNone(report.get_action_status(a).start_time)
53
        self.assertIsNone(report.get_action_status(a).success_time)
54
        self.assertIsNone(report.get_action_status(a).failure_time)
55
56
    def test_cancelation_with_keep_going(self):
57
        for i in range(10):
58
            a0 = self._action("a0")
59
            a = self._action("a")
60
            a0.add_dependency(a)
61
            b = self._action("b", exception=Exception())
62
            a.add_dependency(b)
63
            DEPS = 10
64
            deps = []
65
            for i in range(DEPS):
66
                c = self._action("c")
67
                a.add_dependency(c)
68
                deps.append(c)
69
70
            report = execute(a0, keep_going=True, do_raise=False)
71
72
            # a is not canceled before all its dependencies are done
73
            self.assertGreater(report.get_action_status(a).cancel_time, report.get_action_status(b).failure_time)
74
            for dep in deps:
75
                self.assertGreater(report.get_action_status(a).cancel_time, report.get_action_status(dep).success_time)
76
            # a0 is canceled at the same time as a
77
            self.assertEqual(report.get_action_status(a0).cancel_time, report.get_action_status(a).cancel_time)
78
79
    def test_leaves_have_same_ready_time(self):
80
        a = self._action("a")
81
        b = self._action("b")
82
        c = self._action("c")
83
        d = self._action("d")
84
        a.add_dependency(b)
85
        a.add_dependency(c)
86
        a.add_dependency(d)
87
88
        report = execute(a)
89
90
        self.assertEqual(report.get_action_status(c).ready_time, report.get_action_status(b).ready_time)
91
        self.assertEqual(report.get_action_status(d).ready_time, report.get_action_status(b).ready_time)
92