ExceptionsHandlingTestCase.test_simple_failure_without_raise() - Code Metrics - jacquev6/ActionTree - Measure and Improve Code Quality continuously with Scrutinizer

test_simple_failure_without_raise() A
last analyzed 2018-06-11 07:43 UTC

↳ Parent: ExceptionsHandlingTestCase

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
c	1
b	0
f	0
dl	0
loc	8
rs	9.4285
cc	1

# coding: utf8

# Copyright 2013-2018 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)
        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, FAILED)

    def test_exit(self):

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

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

        self.assertEqual(len(catcher.exception.exceptions), 1)
        self.assertEqual(catcher.exception.exceptions[0].__class__, SystemExit)

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

    def test_keyboard_interrupt(self):

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

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

        self.assertEqual(len(catcher.exception.exceptions), 1)
        self.assertEqual(catcher.exception.exceptions[0].__class__, KeyboardInterrupt)

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

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

        report = execute(a, do_raise=False)

        self.assertFalse(report.is_success)
        self.assertEqual(report.get_action_status(a).status, 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)
        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, CANCELED)
        self.assertEqual(report.get_action_status(b).status, FAILED)

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

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

        self.assertFalse(report.is_success)
        self.assertEqual(report.get_action_status(a).status, SUCCESSFUL)
        self.assertEqual(report.get_action_status(b).status, 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, cpu_cores=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, CANCELED)
        self.assertEqual(report.get_action_status(b).status, FAILED)
        self.assertEqual(report.get_action_status(c).status, FAILED)
        self.assertEqual(report.get_action_status(d).status, 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, cpu_cores=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, CANCELED)
        self.assertEqual(report.get_action_status(b).status, SUCCESSFUL)
        self.assertEqual(report.get_action_status(c).status, SUCCESSFUL)
        self.assertEqual(report.get_action_status(d).status, CANCELED)
        self.assertEqual(report.get_action_status(e).status, FAILED)
        self.assertEqual(report.get_action_status(f).status, SUCCESSFUL)
        self.assertEqual(report.get_action_status(g).status, 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):  # Not unittested: 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, cpu_cores=1, keep_going=False, do_raise=False)

            self.assertEqual(report.get_action_status(a).status, CANCELED)
            self.assertEqual(report.get_action_status(b).status, FAILED)
            if report.get_action_status(c).status == CANCELED:
                c_was_canceled = True
            else:
                self.assertEqual(report.get_action_status(c).status, 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-2018 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)
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, FAILED)
26
27	View Code Duplication	def test_exit(self):
		0 ignored issues – show Duplication introduced 2017-06-17 15:01 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
28		a = self._action("a", exception=SystemExit())
29
30		with self.assertRaises(CompoundException) as catcher:
31		execute(a)
32		report = catcher.exception.execution_report
33
34		self.assertEqual(len(catcher.exception.exceptions), 1)
35		self.assertEqual(catcher.exception.exceptions[0].__class__, SystemExit)
36
37		self.assertFalse(report.is_success)
38		self.assertEqual(report.get_action_status(a).status, FAILED)
39
40	View Code Duplication	def test_keyboard_interrupt(self):
		0 ignored issues – show Duplication introduced 2017-06-17 15:01 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
41		a = self._action("a", exception=KeyboardInterrupt())
42
43		with self.assertRaises(CompoundException) as catcher:
44		execute(a)
45		report = catcher.exception.execution_report
46
47		self.assertEqual(len(catcher.exception.exceptions), 1)
48		self.assertEqual(catcher.exception.exceptions[0].__class__, KeyboardInterrupt)
49
50		self.assertFalse(report.is_success)
51		self.assertEqual(report.get_action_status(a).status, FAILED)
52
53		def test_simple_failure_without_raise(self):
54		a = self._action("a", exception=Exception("foobar"))
55
56		report = execute(a, do_raise=False)
57
58		self.assertFalse(report.is_success)
59		self.assertEqual(report.get_action_status(a).status, FAILED)
60		self.assertEqual(report.get_action_status(a).exception.args, ("foobar",))
61
62	View Code Duplication	def test_exception_in_dependency(self):
		0 ignored issues – show Duplication introduced 2017-06-12 16:29 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
63		a = self._action("a")
64		b = self._action("b", exception=Exception("foobar"))
65		a.add_dependency(b)
66
67		with self.assertRaises(CompoundException) as catcher:
68		execute(a)
69		report = catcher.exception.execution_report
70
71		self.assertEqual(len(catcher.exception.exceptions), 1)
72		self.assertEqual(catcher.exception.exceptions[0].args, ("foobar",))
73
74		self.assertFalse(report.is_success)
75		self.assertEqual(report.get_action_status(a).status, CANCELED)
76		self.assertEqual(report.get_action_status(b).status, FAILED)
77
78		def test_exceptions_in_dependency_with_accept_failed_dependencies(self):
79		a = self._action("a", accept_failed_dependencies=True)
80		b = self._action("b", exception=Exception("foobar"))
81		a.add_dependency(b)
82
83		with self.assertRaises(CompoundException) as catcher:
84		execute(a)
85		report = catcher.exception.execution_report
86
87		self.assertFalse(report.is_success)
88		self.assertEqual(report.get_action_status(a).status, SUCCESSFUL)
89		self.assertEqual(report.get_action_status(b).status, FAILED)
90
91		def test_exceptions_in_dependencies_with_keep_going(self):
92		a = self._action("a")
93		b = self._action("b", exception=Exception("eb"))
94		c = self._action("c", exception=Exception("ec"))
95		d = self._action("d")
96		a.add_dependency(b)
97		a.add_dependency(c)
98		a.add_dependency(d)
99
100		with self.assertRaises(CompoundException) as catcher:
101		execute(a, cpu_cores=1, keep_going=True)
102		report = catcher.exception.execution_report
103
104		self.assertEqual(len(catcher.exception.exceptions), 2)
105		self.assertEqual(sorted(ex.args for ex in catcher.exception.exceptions), [("eb",), ("ec",)])
106
107		self.assertEqual(report.get_action_status(a).status, CANCELED)
108		self.assertEqual(report.get_action_status(b).status, FAILED)
109		self.assertEqual(report.get_action_status(c).status, FAILED)
110		self.assertEqual(report.get_action_status(d).status, SUCCESSFUL)
111
112		def test_exceptions_in_long_branch_dependencies_with_keep_going(self):
113		a = self._action("a")
114		b = self._action("b")
115		c = self._action("c")
116		d = self._action("d")
117		e = self._action("e", exception=Exception("foobar"))
118		f = self._action("f")
119		g = self._action("g")
120		a.add_dependency(b)
121		b.add_dependency(c)
122		a.add_dependency(d)
123		d.add_dependency(e)
124		a.add_dependency(f)
125		f.add_dependency(g)
126
127		with self.assertRaises(CompoundException) as catcher:
128		execute(a, cpu_cores=1, keep_going=True)
129		report = catcher.exception.execution_report
130
131		self.assertEqual(len(catcher.exception.exceptions), 1)
132		self.assertEqual(catcher.exception.exceptions[0].args, ("foobar",))
133
134		self.assertEqual(report.get_action_status(a).status, CANCELED)
135		self.assertEqual(report.get_action_status(b).status, SUCCESSFUL)
136		self.assertEqual(report.get_action_status(c).status, SUCCESSFUL)
137		self.assertEqual(report.get_action_status(d).status, CANCELED)
138		self.assertEqual(report.get_action_status(e).status, FAILED)
139		self.assertEqual(report.get_action_status(f).status, SUCCESSFUL)
140		self.assertEqual(report.get_action_status(g).status, SUCCESSFUL)
141
142		def test_exceptions_in_dependencies_without_keep_going(self):
143		c_was_canceled = False
144		c_was_executed = False
145		b_was_added_before_c = False
146		b_was_added_after_c = False
147		# This is a probabilistic test:
148		# - when c is executed before b, c cannot be canceled
149		# - when c is executed after b, c is canceled
150		# We repeat the test until we see both behaviors or we reach a limit
151		for i in xrange(100): # Not unittested: test code
152		a = self._action("a")
153		b = self._action("b", exception=Exception())
154		c = self._action("c")
155		if random.random() < 0.5:
156		a.add_dependency(b)
157		a.add_dependency(c)
158		b_was_added_before_c = True
159		else:
160		a.add_dependency(c)
161		a.add_dependency(b)
162		b_was_added_after_c = True
163
164		report = execute(a, cpu_cores=1, keep_going=False, do_raise=False)
165
166		self.assertEqual(report.get_action_status(a).status, CANCELED)
167		self.assertEqual(report.get_action_status(b).status, FAILED)
168		if report.get_action_status(c).status == CANCELED:
169		c_was_canceled = True
170		else:
171		self.assertEqual(report.get_action_status(c).status, SUCCESSFUL)
172		c_was_executed = True
173		if c_was_canceled and c_was_executed and b_was_added_before_c and b_was_added_after_c:
174		break
175		self.assertTrue(c_was_canceled)
176		self.assertTrue(c_was_executed)
177

jacquev6 / ActionTree

test_simple_failure_without_raise() A last analyzed 2018-06-11 07:43 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

test_simple_failure_without_raise() A
last analyzed 2018-06-11 07:43 UTC