TestGroupedOptimizer.test_calling_functions() - Code Metrics - Inspection of "Initial test (#49)" - qbahn/grortir - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( aa0ea9...358d48 )

by Wojtek

created 2016-10-15 17:39 UTC

TestGroupedOptimizer.test_calling_functions() A

↳ Parent: TestGroupedOptimizer

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	1
c	1
b	0
f	0
dl	0
loc	9
rs	9.6666

"""This test is checking that algorithm with grouping is working correctly."""
# pylint: disable=too-many-instance-attributes,too-many-arguments

# pylint: disable=signature-differs,arguments-differ


import unittest.mock as mock
from unittest import TestCase

from grortir.main.model.core.abstract_process import AbstractProcess
from grortir.main.optimizers.grouped_optimizer import GroupedOptimizer
from grortir.main.optimizers.grouping_strategy import GroupingStrategy
from grortir.main.model.core.abstract_stage import AbstractStage
from grortir.main.pso.calls_optimization_strategy import \
    CallsOptimizationStrategy
from grortir.main.pso.pso_algorithm import PsoAlgorithm
from tests.framework.calls.calls_registry import CallsRegistry
from tests.framework.calls.input_validator import InputValidator


class TestGroupedOptimizer(TestCase):
    """Class for testing Optimizer."""

    def setUp(self):
        """Set up environment."""
        self.some_process = AbstractProcess()
        self.first_stage = AbstractStage()
        self.second_stage = AbstractStage()
        self.third_stage_a = mock.Mock()
        self.third_stage_b = mock.Mock()
        self.some_process.add_path([self.first_stage, self.second_stage,

                                    self.third_stage_a])
        self.some_process.add_edge(self.second_stage, self.third_stage_b)


    def test___init__(self):
        """Testing creating object."""
        optimizer = GroupedOptimizer(self.some_process, mock.Mock(),
                                     mock.Mock())
        self.assertIsNotNone(optimizer)
        self.assertEqual(optimizer.process, self.some_process)

    def test_calling_functions(self):
        """Test correct order of calling function."""
        optimizer = GroupedOptimizer(TESTED_PROCESS, GROUPING_STRATEGY,
                                     PSO_ALGORITHM)
        optimizer.optimize_process()
        input_validator = InputValidator(CALLS_REGISTRY)
        is_valid = input_validator.validate_input(GROUPING_STRATEGY,
                                                  TESTED_PROCESS)
        self.assertTrue(is_valid)


CALLS_REGISTRY = CallsRegistry()


class DeterministicStage(AbstractStage):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
    def __init__(self, name, max_get_output_of_stage_count=10,
                 max_is_enough_quality_count=10,
                 max_could_be_optimized_count=10, max_get_quality_count=10,
                 max_get_cost_count=10):
        super().__init__((0, 0, 0, 0))
        self.name = name
        self.get_cost_count = 0
        self.get_quality_count = 0
        self.could_be_optimized_count = 0
        self.is_enough_quality_count = 0
        self.get_output_of_stage_count = 0
        self.get_maximal_acceptable_cost_count = 0

        self.maximum_acceptable_quality = 1
        self.maximal_acceptable_cost = 100
        self.max_get_cost_count = max_get_cost_count
        self.max_get_quality_count = max_get_quality_count
        self.max_could_be_optimized_count = max_could_be_optimized_count
        self.max_is_enough_quality_count = max_is_enough_quality_count
        self.max_get_output_of_stage_count = max_get_output_of_stage_count
        CALLS_REGISTRY.add_call(self.name, '__init__',
                                [name, max_get_output_of_stage_count,
                                 max_is_enough_quality_count,
                                 max_could_be_optimized_count,
                                 max_get_quality_count,
                                 max_get_cost_count], None, 1)

    def get_cost(self):
        self.get_cost_count += 1
        CALLS_REGISTRY.add_call(self.name, "get_cost", [], self.get_cost_count,
                                self.get_cost_count)
        return self.get_cost_count

    def get_quality(self, input_vector, control_params=None):
        self.get_quality_count += 1
        result = 0.1 if self.get_quality_count > int(self.name) else 1000
        CALLS_REGISTRY.add_call(self.name, "get_quality",
                                [input_vector, control_params], result,
                                self.get_quality_count)
        return result

    def could_be_optimized(self):
        self.could_be_optimized_count += 1
        CALLS_REGISTRY.add_call(self.name, "could_be_optimized", [], True,
                                self.could_be_optimized_count)
        return True

    def is_enough_quality(self, value):
        self.is_enough_quality_count += 1
        result = value < 1
        CALLS_REGISTRY.add_call(self.name, "is_enough_quality", [value], result,
                                self.is_enough_quality_count)
        return result

    def get_output_of_stage(self, input_vector, control_params):
        self.get_output_of_stage_count += 1
        output = [self.get_output_of_stage_count] * len(input_vector)
        output[0] = int(self.name)
        CALLS_REGISTRY.add_call(self.name, "get_output_of_stage",
                                [input_vector, control_params], output,
                                self.get_output_of_stage_count)
        return output

    def get_maximal_acceptable_cost(self):
        self.get_maximal_acceptable_cost_count += 1
        CALLS_REGISTRY.add_call(self.name, "get_maximal_acceptable_cost",
                                [], self.maximal_acceptable_cost,
                                self.get_output_of_stage_count)
        return self.maximal_acceptable_cost


class ExampleProcess(AbstractProcess):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
    pass


TESTED_PROCESS = ExampleProcess()
stages = {}

for i in range(7):
    stages[i] = DeterministicStage(str(i))

# Our graph:
#   0
#   |
#   1
#   |\
#   2 4
#   | |\
#   3 5 6
# All edges directed to down
# Order of nodes is the same as names
TESTED_PROCESS.add_edge(stages[0], stages[1])

TESTED_PROCESS.add_edge(stages[1], stages[2])

TESTED_PROCESS.add_edge(stages[2], stages[3])


TESTED_PROCESS.add_edge(stages[1], stages[4])

TESTED_PROCESS.add_edge(stages[4], stages[5])

TESTED_PROCESS.add_edge(stages[4], stages[6])


# Groups:
#   (0)0
#      |
#   (0)1
#      |\
#   (0)2 4(1)
#      | | \
#   (1)3 5(1)6(2)

GROUPING_STRATEGY = GroupingStrategy(list(stages.values()))
GROUPING_STRATEGY.define_group((stages[0], stages[1], stages[2]))
GROUPING_STRATEGY.define_group((stages[3], stages[4], stages[5]))
GROUPING_STRATEGY.define_group((stages[6],))

OPTIMIZATION_STARTEGY = CallsOptimizationStrategy()

PSO_ALGORITHM = PsoAlgorithm(TESTED_PROCESS, GROUPING_STRATEGY,
                             OPTIMIZATION_STARTEGY, 2)


1			"""This test is checking that algorithm with grouping is working correctly."""
2			# pylint: disable=too-many-instance-attributes,too-many-arguments
			0 ignored issues – show introduced 2016-10-15 17:37 UTC by Report Bug Copy Issue Report Locally disabling too-many-instance-attributes (R0902) Loading history... introduced 2016-10-15 17:37 UTC by Report Bug Copy Issue Report Locally disabling too-many-arguments (R0913) Loading history...
3			# pylint: disable=signature-differs,arguments-differ
			0 ignored issues – show introduced 2016-10-15 17:37 UTC by Report Bug Copy Issue Report Locally disabling signature-differs (W0222) Loading history... introduced 2016-10-15 17:37 UTC by Report Bug Copy Issue Report Locally disabling arguments-differ (W0221) Loading history...
4
5			import unittest.mock as mock
6			from unittest import TestCase
7
8			from grortir.main.model.core.abstract_process import AbstractProcess
9			from grortir.main.optimizers.grouped_optimizer import GroupedOptimizer
10			from grortir.main.optimizers.grouping_strategy import GroupingStrategy
11			from grortir.main.model.core.abstract_stage import AbstractStage
12			from grortir.main.pso.calls_optimization_strategy import \
13			CallsOptimizationStrategy
14			from grortir.main.pso.pso_algorithm import PsoAlgorithm
15			from tests.framework.calls.calls_registry import CallsRegistry
16			from tests.framework.calls.input_validator import InputValidator
17
18
19			class TestGroupedOptimizer(TestCase):
20			"""Class for testing Optimizer."""
21
22			def setUp(self):
23			"""Set up environment."""
24			self.some_process = AbstractProcess()
25			self.first_stage = AbstractStage()
26			self.second_stage = AbstractStage()
27			self.third_stage_a = mock.Mock()
28			self.third_stage_b = mock.Mock()
29			self.some_process.add_path([self.first_stage, self.second_stage,
			0 ignored issues – show Bug introduced 2016-10-15 09:55 UTC by Report Bug Copy Issue Report The Instance of `AbstractProcess` does not seem to have a member named `add_path`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
30			self.third_stage_a])
31			self.some_process.add_edge(self.second_stage, self.third_stage_b)
			0 ignored issues – show Bug introduced 2016-10-15 09:55 UTC by Report Bug Copy Issue Report The Instance of `AbstractProcess` does not seem to have a member named `add_edge`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
32
33			def test___init__(self):
34			"""Testing creating object."""
35			optimizer = GroupedOptimizer(self.some_process, mock.Mock(),
36			mock.Mock())
37			self.assertIsNotNone(optimizer)
38			self.assertEqual(optimizer.process, self.some_process)
39
40			def test_calling_functions(self):
41			"""Test correct order of calling function."""
42			optimizer = GroupedOptimizer(TESTED_PROCESS, GROUPING_STRATEGY,
43			PSO_ALGORITHM)
44			optimizer.optimize_process()
45			input_validator = InputValidator(CALLS_REGISTRY)
46			is_valid = input_validator.validate_input(GROUPING_STRATEGY,
47			TESTED_PROCESS)
48			self.assertTrue(is_valid)
49
50
51			CALLS_REGISTRY = CallsRegistry()
52
53
54			class DeterministicStage(AbstractStage):
			0 ignored issues – show Coding Style introduced 2016-10-15 09:55 UTC by Report Bug Copy Issue Report This class should have a docstring. The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods: class SomeClass: def some_method(self): """Do x and return foo.""" If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions. Loading history...
55			def __init__(self, name, max_get_output_of_stage_count=10,
56			max_is_enough_quality_count=10,
57			max_could_be_optimized_count=10, max_get_quality_count=10,
58			max_get_cost_count=10):
59			super().__init__((0, 0, 0, 0))
60			self.name = name
61			self.get_cost_count = 0
62			self.get_quality_count = 0
63			self.could_be_optimized_count = 0
64			self.is_enough_quality_count = 0
65			self.get_output_of_stage_count = 0
66			self.get_maximal_acceptable_cost_count = 0
			0 ignored issues – show Coding Style Naming introduced 2016-10-15 17:37 UTC by Report Bug Copy Issue Report The name `get_maximal_acceptable_cost_count` does not conform to the attribute naming conventions (`[a-z_][a-z0-9_]{2,30}$`). This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
67			self.maximum_acceptable_quality = 1
68			self.maximal_acceptable_cost = 100
69			self.max_get_cost_count = max_get_cost_count
70			self.max_get_quality_count = max_get_quality_count
71			self.max_could_be_optimized_count = max_could_be_optimized_count
72			self.max_is_enough_quality_count = max_is_enough_quality_count
73			self.max_get_output_of_stage_count = max_get_output_of_stage_count
74			CALLS_REGISTRY.add_call(self.name, '__init__',
75			[name, max_get_output_of_stage_count,
76			max_is_enough_quality_count,
77			max_could_be_optimized_count,
78			max_get_quality_count,
79			max_get_cost_count], None, 1)
80
81			def get_cost(self):
82			self.get_cost_count += 1
83			CALLS_REGISTRY.add_call(self.name, "get_cost", [], self.get_cost_count,
84			self.get_cost_count)
85			return self.get_cost_count
86
87			def get_quality(self, input_vector, control_params=None):
88			self.get_quality_count += 1
89			result = 0.1 if self.get_quality_count > int(self.name) else 1000
90			CALLS_REGISTRY.add_call(self.name, "get_quality",
91			[input_vector, control_params], result,
92			self.get_quality_count)
93			return result
94
95			def could_be_optimized(self):
96			self.could_be_optimized_count += 1
97			CALLS_REGISTRY.add_call(self.name, "could_be_optimized", [], True,
98			self.could_be_optimized_count)
99			return True
100
101			def is_enough_quality(self, value):
102			self.is_enough_quality_count += 1
103			result = value < 1
104			CALLS_REGISTRY.add_call(self.name, "is_enough_quality", [value], result,
105			self.is_enough_quality_count)
106			return result
107
108			def get_output_of_stage(self, input_vector, control_params):
109			self.get_output_of_stage_count += 1
110			output = [self.get_output_of_stage_count] * len(input_vector)
111			output[0] = int(self.name)
112			CALLS_REGISTRY.add_call(self.name, "get_output_of_stage",
113			[input_vector, control_params], output,
114			self.get_output_of_stage_count)
115			return output
116
117			def get_maximal_acceptable_cost(self):
118			self.get_maximal_acceptable_cost_count += 1
119			CALLS_REGISTRY.add_call(self.name, "get_maximal_acceptable_cost",
120			[], self.maximal_acceptable_cost,
121			self.get_output_of_stage_count)
122			return self.maximal_acceptable_cost
123
124
125			class ExampleProcess(AbstractProcess):
			0 ignored issues – show Coding Style introduced 2016-10-15 09:55 UTC by Report Bug Copy Issue Report This class should have a docstring. The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods: class SomeClass: def some_method(self): """Do x and return foo.""" If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions. Loading history...
126			pass
127
128
129			TESTED_PROCESS = ExampleProcess()
130			stages = {}
			0 ignored issues – show Coding Style Naming introduced 2016-10-15 09:55 UTC by Report Bug Copy Issue Report The name `stages` does not conform to the constant naming conventions (`(([A-Z_][A-Z0-9_])\|(__.__))$`). This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
131			for i in range(7):
132			stages[i] = DeterministicStage(str(i))
133
134			# Our graph:
135			# 0
136			# \|
137			# 1
138			# \|\
139			# 2 4
140			# \| \|\
141			# 3 5 6
142			# All edges directed to down
143			# Order of nodes is the same as names
144			TESTED_PROCESS.add_edge(stages[0], stages[1])
			0 ignored issues – show Bug introduced 2016-10-15 09:55 UTC by Report Bug Copy Issue Report The Instance of `ExampleProcess` does not seem to have a member named `add_edge`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
145			TESTED_PROCESS.add_edge(stages[1], stages[2])
			0 ignored issues – show Bug introduced 2016-10-15 09:55 UTC by Report Bug Copy Issue Report The Instance of `ExampleProcess` does not seem to have a member named `add_edge`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
146			TESTED_PROCESS.add_edge(stages[2], stages[3])
			0 ignored issues – show Bug introduced 2016-10-15 09:55 UTC by Report Bug Copy Issue Report The Instance of `ExampleProcess` does not seem to have a member named `add_edge`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
147
148			TESTED_PROCESS.add_edge(stages[1], stages[4])
			0 ignored issues – show Bug introduced 2016-10-15 09:55 UTC by Report Bug Copy Issue Report The Instance of `ExampleProcess` does not seem to have a member named `add_edge`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
149			TESTED_PROCESS.add_edge(stages[4], stages[5])
			0 ignored issues – show Bug introduced 2016-10-15 09:55 UTC by Report Bug Copy Issue Report The Instance of `ExampleProcess` does not seem to have a member named `add_edge`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
150			TESTED_PROCESS.add_edge(stages[4], stages[6])
			0 ignored issues – show Bug introduced 2016-10-15 09:55 UTC by Report Bug Copy Issue Report The Instance of `ExampleProcess` does not seem to have a member named `add_edge`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
151
152			# Groups:
153			# (0)0
154			# \|
155			# (0)1
156			# \|\
157			# (0)2 4(1)
158			# \| \| \
159			# (1)3 5(1)6(2)
160
161			GROUPING_STRATEGY = GroupingStrategy(list(stages.values()))
162			GROUPING_STRATEGY.define_group((stages[0], stages[1], stages[2]))
163			GROUPING_STRATEGY.define_group((stages[3], stages[4], stages[5]))
164			GROUPING_STRATEGY.define_group((stages[6],))
165
166			OPTIMIZATION_STARTEGY = CallsOptimizationStrategy()
167
168			PSO_ALGORITHM = PsoAlgorithm(TESTED_PROCESS, GROUPING_STRATEGY,
169			OPTIMIZATION_STARTEGY, 2)
170

qbahn / grortir

Push — master ( aa0ea9...358d48 )

TestGroupedOptimizer.test_calling_functions() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like