ExceptionsHandlingTestCase.test_exceptions_in_dependencies_without_keep_going() - Code Metrics - Inspection of "Rebuild doc" - jacquev6/ActionTree - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 669a7b...91f4e1 )

by Vincent

created 2017-06-13 17:46 UTC

test_exceptions_in_dependencies_without_keep_going() D

↳ Parent: ExceptionsHandlingTestCase

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	4
Bugs	0	Features	0

Metric	Value
c	4
b	0
f	0
dl	0
loc	35
rs	4
cc	8

# coding: utf8

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

from __future__ import division, absolute_import, print_function

import random

from ActionTree import *
from . import *


class ExceptionsHandlingTestCase(ActionTreeTestCase):
    def test_simple_failure(self):

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

        with self.assertRaises(CompoundException) as catcher:
            execute(a, jobs=1)
        report = catcher.exception.execution_report

        self.assertEqual(len(catcher.exception.exceptions), 1)
        self.assertEqual(catcher.exception.exceptions[0].args, ("foobar",))

        self.assertFalse(report.is_success)
        self.assertEqual(report.get_action_status(a).status, ExecutionReport.ActionStatus.FAILED)

    def test_simple_failure_without_raise(self):
        a = self._action("a", exception=Exception("foobar"))

        report = execute(a, jobs=1, do_raise=False)

        self.assertFalse(report.is_success)
        self.assertEqual(report.get_action_status(a).status, ExecutionReport.ActionStatus.FAILED)
        self.assertEqual(report.get_action_status(a).exception.args, ("foobar",))


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

        with self.assertRaises(CompoundException) as catcher:
            execute(a, jobs=1)
        report = catcher.exception.execution_report

        self.assertEqual(len(catcher.exception.exceptions), 1)
        self.assertEqual(catcher.exception.exceptions[0].args, ("foobar",))

        self.assertFalse(report.is_success)
        self.assertEqual(report.get_action_status(a).status, ExecutionReport.ActionStatus.CANCELED)
        self.assertEqual(report.get_action_status(b).status, ExecutionReport.ActionStatus.FAILED)

    def test_exceptions_in_dependency_with_weak_dependencies(self):

        a = self._action("a", weak_dependencies=True)
        b = self._action("b", exception=Exception("foobar"))
        a.add_dependency(b)

        with self.assertRaises(CompoundException) as catcher:
            execute(a, jobs=1)
        report = catcher.exception.execution_report

        self.assertFalse(report.is_success)
        self.assertEqual(report.get_action_status(a).status, ExecutionReport.ActionStatus.SUCCESSFUL)
        self.assertEqual(report.get_action_status(b).status, ExecutionReport.ActionStatus.FAILED)

    def test_exceptions_in_dependencies_with_keep_going(self):
        a = self._action("a")
        b = self._action("b", exception=Exception("eb"))
        c = self._action("c", exception=Exception("ec"))
        d = self._action("d")
        a.add_dependency(b)
        a.add_dependency(c)
        a.add_dependency(d)

        with self.assertRaises(CompoundException) as catcher:
            execute(a, jobs=1, keep_going=True)
        report = catcher.exception.execution_report

        self.assertEqual(len(catcher.exception.exceptions), 2)
        self.assertEqual(sorted(ex.args for ex in catcher.exception.exceptions), [("eb",), ("ec",)])

        self.assertEqual(report.get_action_status(a).status, ExecutionReport.ActionStatus.CANCELED)
        self.assertEqual(report.get_action_status(b).status, ExecutionReport.ActionStatus.FAILED)
        self.assertEqual(report.get_action_status(c).status, ExecutionReport.ActionStatus.FAILED)
        self.assertEqual(report.get_action_status(d).status, ExecutionReport.ActionStatus.SUCCESSFUL)

    def test_exceptions_in_long_branch_dependencies_with_keep_going(self):
        a = self._action("a")
        b = self._action("b")
        c = self._action("c")
        d = self._action("d")
        e = self._action("e", exception=Exception("foobar"))
        f = self._action("f")
        g = self._action("g")
        a.add_dependency(b)
        b.add_dependency(c)
        a.add_dependency(d)
        d.add_dependency(e)
        a.add_dependency(f)
        f.add_dependency(g)

        with self.assertRaises(CompoundException) as catcher:
            execute(a, jobs=1, keep_going=True)
        report = catcher.exception.execution_report

        self.assertEqual(len(catcher.exception.exceptions), 1)
        self.assertEqual(catcher.exception.exceptions[0].args, ("foobar",))

        self.assertEqual(report.get_action_status(a).status, ExecutionReport.ActionStatus.CANCELED)
        self.assertEqual(report.get_action_status(b).status, ExecutionReport.ActionStatus.SUCCESSFUL)
        self.assertEqual(report.get_action_status(c).status, ExecutionReport.ActionStatus.SUCCESSFUL)
        self.assertEqual(report.get_action_status(d).status, ExecutionReport.ActionStatus.CANCELED)
        self.assertEqual(report.get_action_status(e).status, ExecutionReport.ActionStatus.FAILED)
        self.assertEqual(report.get_action_status(f).status, ExecutionReport.ActionStatus.SUCCESSFUL)
        self.assertEqual(report.get_action_status(g).status, ExecutionReport.ActionStatus.SUCCESSFUL)

    def test_exceptions_in_dependencies_without_keep_going(self):
        c_was_canceled = False
        c_was_executed = False
        b_was_added_before_c = False
        b_was_added_after_c = False
        # This is a probabilistic test:
        # - when c is executed before b, c cannot be canceled
        # - when c is executed after b, c is canceled
        # We repeat the test until we see both behaviors or we reach a limit
        for i in xrange(100):  # pragma no cover (Test code)
            a = self._action("a")
            b = self._action("b", exception=Exception())
            c = self._action("c")
            if random.random() < 0.5:
                a.add_dependency(b)
                a.add_dependency(c)
                b_was_added_before_c = True
            else:
                a.add_dependency(c)
                a.add_dependency(b)
                b_was_added_after_c = True

            report = execute(a, jobs=1, keep_going=False, do_raise=False)

            self.assertEqual(report.get_action_status(a).status, ExecutionReport.ActionStatus.CANCELED)
            self.assertEqual(report.get_action_status(b).status, ExecutionReport.ActionStatus.FAILED)
            if report.get_action_status(c).status == ExecutionReport.ActionStatus.CANCELED:
                c_was_canceled = True
            else:
                self.assertEqual(report.get_action_status(c).status, ExecutionReport.ActionStatus.SUCCESSFUL)
                c_was_executed = True
            if c_was_canceled and c_was_executed and b_was_added_before_c and b_was_added_after_c:
                break
        self.assertTrue(c_was_canceled)
        self.assertTrue(c_was_executed)


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 random
8
9		from ActionTree import *
10		from . import *
11
12
13		class ExceptionsHandlingTestCase(ActionTreeTestCase):
14	View Code Duplication	def test_simple_failure(self):
		0 ignored issues – show Duplication introduced 2017-06-13 17:47 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
15		a = self._action("a", exception=Exception("foobar"))
16
17		with self.assertRaises(CompoundException) as catcher:
18		execute(a, jobs=1)
19		report = catcher.exception.execution_report
20
21		self.assertEqual(len(catcher.exception.exceptions), 1)
22		self.assertEqual(catcher.exception.exceptions[0].args, ("foobar",))
23
24		self.assertFalse(report.is_success)
25		self.assertEqual(report.get_action_status(a).status, ExecutionReport.ActionStatus.FAILED)
26
27		def test_simple_failure_without_raise(self):
28		a = self._action("a", exception=Exception("foobar"))
29
30		report = execute(a, jobs=1, do_raise=False)
31
32		self.assertFalse(report.is_success)
33		self.assertEqual(report.get_action_status(a).status, ExecutionReport.ActionStatus.FAILED)
34	View Code Duplication	self.assertEqual(report.get_action_status(a).exception.args, ("foobar",))
		0 ignored issues – show Duplication introduced 2017-06-13 17:47 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
35
36		def test_exception_in_dependency(self):
37		a = self._action("a")
38		b = self._action("b", exception=Exception("foobar"))
39		a.add_dependency(b)
40
41		with self.assertRaises(CompoundException) as catcher:
42		execute(a, jobs=1)
43		report = catcher.exception.execution_report
44
45		self.assertEqual(len(catcher.exception.exceptions), 1)
46		self.assertEqual(catcher.exception.exceptions[0].args, ("foobar",))
47
48		self.assertFalse(report.is_success)
49		self.assertEqual(report.get_action_status(a).status, ExecutionReport.ActionStatus.CANCELED)
50		self.assertEqual(report.get_action_status(b).status, ExecutionReport.ActionStatus.FAILED)
51
52	View Code Duplication	def test_exceptions_in_dependency_with_weak_dependencies(self):
		0 ignored issues – show Duplication introduced 2017-06-13 17:47 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
53		a = self._action("a", weak_dependencies=True)
54		b = self._action("b", exception=Exception("foobar"))
55		a.add_dependency(b)
56
57		with self.assertRaises(CompoundException) as catcher:
58		execute(a, jobs=1)
59		report = catcher.exception.execution_report
60
61		self.assertFalse(report.is_success)
62		self.assertEqual(report.get_action_status(a).status, ExecutionReport.ActionStatus.SUCCESSFUL)
63		self.assertEqual(report.get_action_status(b).status, ExecutionReport.ActionStatus.FAILED)
64
65		def test_exceptions_in_dependencies_with_keep_going(self):
66		a = self._action("a")
67		b = self._action("b", exception=Exception("eb"))
68		c = self._action("c", exception=Exception("ec"))
69		d = self._action("d")
70		a.add_dependency(b)
71		a.add_dependency(c)
72		a.add_dependency(d)
73
74		with self.assertRaises(CompoundException) as catcher:
75		execute(a, jobs=1, keep_going=True)
76		report = catcher.exception.execution_report
77
78		self.assertEqual(len(catcher.exception.exceptions), 2)
79		self.assertEqual(sorted(ex.args for ex in catcher.exception.exceptions), [("eb",), ("ec",)])
80
81		self.assertEqual(report.get_action_status(a).status, ExecutionReport.ActionStatus.CANCELED)
82		self.assertEqual(report.get_action_status(b).status, ExecutionReport.ActionStatus.FAILED)
83		self.assertEqual(report.get_action_status(c).status, ExecutionReport.ActionStatus.FAILED)
84		self.assertEqual(report.get_action_status(d).status, ExecutionReport.ActionStatus.SUCCESSFUL)
85
86		def test_exceptions_in_long_branch_dependencies_with_keep_going(self):
87		a = self._action("a")
88		b = self._action("b")
89		c = self._action("c")
90		d = self._action("d")
91		e = self._action("e", exception=Exception("foobar"))
92		f = self._action("f")
93		g = self._action("g")
94		a.add_dependency(b)
95		b.add_dependency(c)
96		a.add_dependency(d)
97		d.add_dependency(e)
98		a.add_dependency(f)
99		f.add_dependency(g)
100
101		with self.assertRaises(CompoundException) as catcher:
102		execute(a, jobs=1, keep_going=True)
103		report = catcher.exception.execution_report
104
105		self.assertEqual(len(catcher.exception.exceptions), 1)
106		self.assertEqual(catcher.exception.exceptions[0].args, ("foobar",))
107
108		self.assertEqual(report.get_action_status(a).status, ExecutionReport.ActionStatus.CANCELED)
109		self.assertEqual(report.get_action_status(b).status, ExecutionReport.ActionStatus.SUCCESSFUL)
110		self.assertEqual(report.get_action_status(c).status, ExecutionReport.ActionStatus.SUCCESSFUL)
111		self.assertEqual(report.get_action_status(d).status, ExecutionReport.ActionStatus.CANCELED)
112		self.assertEqual(report.get_action_status(e).status, ExecutionReport.ActionStatus.FAILED)
113		self.assertEqual(report.get_action_status(f).status, ExecutionReport.ActionStatus.SUCCESSFUL)
114		self.assertEqual(report.get_action_status(g).status, ExecutionReport.ActionStatus.SUCCESSFUL)
115
116		def test_exceptions_in_dependencies_without_keep_going(self):
117		c_was_canceled = False
118		c_was_executed = False
119		b_was_added_before_c = False
120		b_was_added_after_c = False
121		# This is a probabilistic test:
122		# - when c is executed before b, c cannot be canceled
123		# - when c is executed after b, c is canceled
124		# We repeat the test until we see both behaviors or we reach a limit
125		for i in xrange(100): # pragma no cover (Test code)
126		a = self._action("a")
127		b = self._action("b", exception=Exception())
128		c = self._action("c")
129		if random.random() < 0.5:
130		a.add_dependency(b)
131		a.add_dependency(c)
132		b_was_added_before_c = True
133		else:
134		a.add_dependency(c)
135		a.add_dependency(b)
136		b_was_added_after_c = True
137
138		report = execute(a, jobs=1, keep_going=False, do_raise=False)
139
140		self.assertEqual(report.get_action_status(a).status, ExecutionReport.ActionStatus.CANCELED)
141		self.assertEqual(report.get_action_status(b).status, ExecutionReport.ActionStatus.FAILED)
142		if report.get_action_status(c).status == ExecutionReport.ActionStatus.CANCELED:
143		c_was_canceled = True
144		else:
145		self.assertEqual(report.get_action_status(c).status, ExecutionReport.ActionStatus.SUCCESSFUL)
146		c_was_executed = True
147		if c_was_canceled and c_was_executed and b_was_added_before_c and b_was_added_after_c:
148		break
149		self.assertTrue(c_was_canceled)
150		self.assertTrue(c_was_executed)
151

jacquev6 / ActionTree

Push — master ( 669a7b...91f4e1 )

test_exceptions_in_dependencies_without_keep_going() D

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like