StartDiscoveryTestCase.test_run_poller_lookup_primary() - Code Metrics - Inspection of "Merge pull request #487 from Pulsant/jjm/orion-add..." - StackStorm/st2contrib - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( f1a13c...e53aa3 )

by Tomaz

created 2016-05-17 15:09 UTC

test_run_poller_lookup_primary() A

↳ Parent: StartDiscoveryTestCase

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	7
rs	9.4285

# Licensed to the StackStorm, Inc ('StackStorm') under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and

from mock import Mock, MagicMock

from orion_base_action_test_case import OrionBaseActionTestCase

from start_discovery import StartDiscovery

__all__ = [
    'StartDiscoveryTestCase'
]


class StartDiscoveryTestCase(OrionBaseActionTestCase):
    __test__ = True
    action_cls = StartDiscovery

    def test_run_connect_fail(self):
        action = self.setup_connect_fail()
        self.assertRaises(ValueError,
                          action.run,
                          name="Unit Test Discovery",
                          platform="orion",
                          poller="primary",
                          snmp_communities=["public"],
                          nodes=["router1"],
                          subnets=None,
                          ip_ranges=None,
                          no_icmp_only=True,
                          auto_import=False)

    def test_run_fail_on_no_targets(self):
        action = self.setup_query_blank_results()

        # nodes = 0, subnets = 0, ip_ranges = 0
        self.assertRaises(ValueError,
                          action.run,
                          name="Unit Test Discovery",
                          platform="orion",
                          poller="primary",
                          snmp_communities=["public"],
                          nodes=None,
                          subnets=None,
                          ip_ranges=None,
                          no_icmp_only=True,
                          auto_import=False)

    def test_run_fail_on_mutliple_targets(self):
        action = self.setup_query_blank_results()

        # nodes = 1, subnets = 1, ip_ranges = 1
        nodes = ["192.168.1.1", "192.168.1.10"]
        subnets = ["192.168.1.0/24"]
        ip_ranges = ["192.168.1.1:192.168.1.10"]

        self.assertRaises(ValueError,
                          action.run,
                          name="Unit Test Discovery",
                          platform="orion",
                          poller="primary",
                          snmp_communities=["public"],
                          nodes=nodes,
                          subnets=subnets,
                          ip_ranges=ip_ranges,
                          no_icmp_only=True,
                          auto_import=False)

        # nodes = 1, subnets = 0, ip_ranges = 1
        nodes = ["192.168.1.1", "192.168.1.10"]
        subnets = None
        ip_ranges = ["192.168.1.1:192.168.1.10"]

        self.assertRaises(ValueError,
                          action.run,
                          name="Unit Test Discovery",
                          platform="orion",
                          poller="primary",
                          snmp_communities=["public"],
                          nodes=nodes,
                          subnets=subnets,
                          ip_ranges=ip_ranges,
                          no_icmp_only=True,
                          auto_import=False)

        # nodes = 1, subnets = 1, ip_ranges = 0
        nodes = ["192.168.1.1", "192.168.1.10"]
        subnets = ["192.168.1.0/24"]
        ip_ranges = None

        self.assertRaises(ValueError,
                          action.run,
                          name="Unit Test Discovery",
                          platform="orion",
                          poller="primary",
                          snmp_communities=["public"],
                          nodes=nodes,
                          subnets=subnets,
                          ip_ranges=ip_ranges,
                          no_icmp_only=True,
                          auto_import=False)

        # nodes = 0, subnets = 1, ip_ranges = 1
        nodes = None
        subnets = ["192.168.1.0/24"]
        ip_ranges = ["192.168.1.1:192.168.1.10"]

        self.assertRaises(ValueError,
                          action.run,
                          name="Unit Test Discovery",
                          platform="orion",
                          poller="primary",
                          snmp_communities=["public"],
                          nodes=nodes,
                          subnets=subnets,
                          ip_ranges=ip_ranges,
                          no_icmp_only=True,
                          auto_import=False)

    def test_run_poller_lookup_primary(self):
        expected = 1

        action = self.get_action_instance(config=self.full_config)

        result = action.get_engine_id("primary")
        self.assertEqual(result, expected)

    def test_run_poller_lookup_fail(self):
        action = self.setup_query_blank_results()
        self.assertRaises(ValueError,
                          action.get_engine_id,
                          "foobar")

    def test_run_snmp_communities_fail(self):
        query_data = []
        query_data.append(self.query_no_results)
        query_data.append(self.query_no_results)

        action = self.get_action_instance(config=self.full_config)
        action.connect = MagicMock(return_value=True)
        action.query = MagicMock(side_effect=query_data)

        self.assertRaises(ValueError,
                          action.get_snmp_cred_id,
                          "foobar")

        self.assertRaises(ValueError,
                          action.get_snmp_cred_id,
                          "internal")

    def test_run_discovery_create_completes(self):
        expected = 10

        query_data = []
        query_data.append(self.load_yaml('results_orion_snmp_cred.yaml'))
        query_data.append({'results': [{'ID': 1}]})

        invoke_data = []
        invoke_data.append("CorePluginConfiguration")
        invoke_data.append(10)

        action = self.get_action_instance(config=self.full_config)
        action.connect = MagicMock(return_value=True)
        action.query = MagicMock(side_effect=query_data)
        action.invoke = Mock(side_effect=invoke_data)

        result = action.run(name="Unit Test Discovery",
                            platform="orion",
                            poller="primary",
                            snmp_communities=["public"],
                            nodes=["router1"],
                            subnets=None,
                            ip_ranges=None,
                            no_icmp_only=True,
                            auto_import=False)

        self.assertEqual(result, expected)

    def test_run_discovery_create_completes_on_poller(self):
        expected = 10

        query_data = []
        query_data.append(self.load_yaml('results_orion_snmp_cred.yaml'))
        query_data.append(self.load_yaml('results_orion_engines.yaml'))
        query_data.append({'results': [{'ID': 1}]})

        invoke_data = []
        invoke_data.append("CorePluginConfiguration")
        invoke_data.append(10)

        action = self.get_action_instance(config=self.full_config)
        action.connect = MagicMock(return_value=True)
        action.query = MagicMock(side_effect=query_data)
        action.invoke = Mock(side_effect=invoke_data)

        result = action.run(name="Unit Test Discovery",
                            platform="orion",
                            poller="primary",
                            snmp_communities=["public"],
                            nodes=["router1"],
                            subnets=None,
                            ip_ranges=None,
                            no_icmp_only=True,
                            auto_import=False)

        self.assertEqual(result, expected)


1			# Licensed to the StackStorm, Inc ('StackStorm') under one or more
2			# contributor license agreements. See the NOTICE file distributed with
3			# this work for additional information regarding copyright ownership.
4			# The ASF licenses this file to You under the Apache License, Version 2.0
5			# (the "License"); you may not use this file except in compliance with
6			# the License. You may obtain a copy of the License at
7			#
8			# http://www.apache.org/licenses/LICENSE-2.0
9			#
10			# Unless required by applicable law or agreed to in writing, software
11			# distributed under the License is distributed on an "AS IS" BASIS,
12			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13			# See the License for the specific language governing permissions and
14
15			from mock import Mock, MagicMock
16
17			from orion_base_action_test_case import OrionBaseActionTestCase
18
19			from start_discovery import StartDiscovery
20
21			__all__ = [
22			'StartDiscoveryTestCase'
23			]
24
25
26			class StartDiscoveryTestCase(OrionBaseActionTestCase):
27			__test__ = True
28			action_cls = StartDiscovery
29
30			def test_run_connect_fail(self):
31			action = self.setup_connect_fail()
32			self.assertRaises(ValueError,
33			action.run,
34			name="Unit Test Discovery",
35			platform="orion",
36			poller="primary",
37			snmp_communities=["public"],
38			nodes=["router1"],
39			subnets=None,
40			ip_ranges=None,
41			no_icmp_only=True,
42			auto_import=False)
43
44			def test_run_fail_on_no_targets(self):
45			action = self.setup_query_blank_results()
46
47			# nodes = 0, subnets = 0, ip_ranges = 0
48			self.assertRaises(ValueError,
49			action.run,
50			name="Unit Test Discovery",
51			platform="orion",
52			poller="primary",
53			snmp_communities=["public"],
54			nodes=None,
55			subnets=None,
56			ip_ranges=None,
57			no_icmp_only=True,
58			auto_import=False)
59
60			def test_run_fail_on_mutliple_targets(self):
61			action = self.setup_query_blank_results()
62
63			# nodes = 1, subnets = 1, ip_ranges = 1
64			nodes = ["192.168.1.1", "192.168.1.10"]
65			subnets = ["192.168.1.0/24"]
66			ip_ranges = ["192.168.1.1:192.168.1.10"]
67
68			self.assertRaises(ValueError,
69			action.run,
70			name="Unit Test Discovery",
71			platform="orion",
72			poller="primary",
73			snmp_communities=["public"],
74			nodes=nodes,
75			subnets=subnets,
76			ip_ranges=ip_ranges,
77			no_icmp_only=True,
78			auto_import=False)
79
80			# nodes = 1, subnets = 0, ip_ranges = 1
81			nodes = ["192.168.1.1", "192.168.1.10"]
82			subnets = None
83			ip_ranges = ["192.168.1.1:192.168.1.10"]
84
85			self.assertRaises(ValueError,
86			action.run,
87			name="Unit Test Discovery",
88			platform="orion",
89			poller="primary",
90			snmp_communities=["public"],
91			nodes=nodes,
92			subnets=subnets,
93			ip_ranges=ip_ranges,
94			no_icmp_only=True,
95			auto_import=False)
96
97			# nodes = 1, subnets = 1, ip_ranges = 0
98			nodes = ["192.168.1.1", "192.168.1.10"]
99			subnets = ["192.168.1.0/24"]
100			ip_ranges = None
101
102			self.assertRaises(ValueError,
103			action.run,
104			name="Unit Test Discovery",
105			platform="orion",
106			poller="primary",
107			snmp_communities=["public"],
108			nodes=nodes,
109			subnets=subnets,
110			ip_ranges=ip_ranges,
111			no_icmp_only=True,
112			auto_import=False)
113
114			# nodes = 0, subnets = 1, ip_ranges = 1
115			nodes = None
116			subnets = ["192.168.1.0/24"]
117			ip_ranges = ["192.168.1.1:192.168.1.10"]
118
119			self.assertRaises(ValueError,
120			action.run,
121			name="Unit Test Discovery",
122			platform="orion",
123			poller="primary",
124			snmp_communities=["public"],
125			nodes=nodes,
126			subnets=subnets,
127			ip_ranges=ip_ranges,
128			no_icmp_only=True,
129			auto_import=False)
130
131			def test_run_poller_lookup_primary(self):
132			expected = 1
133
134			action = self.get_action_instance(config=self.full_config)
135
136			result = action.get_engine_id("primary")
137			self.assertEqual(result, expected)
138
139			def test_run_poller_lookup_fail(self):
140			action = self.setup_query_blank_results()
141			self.assertRaises(ValueError,
142			action.get_engine_id,
143			"foobar")
144
145			def test_run_snmp_communities_fail(self):
146			query_data = []
147			query_data.append(self.query_no_results)
148			query_data.append(self.query_no_results)
149
150			action = self.get_action_instance(config=self.full_config)
151			action.connect = MagicMock(return_value=True)
152			action.query = MagicMock(side_effect=query_data)
153
154			self.assertRaises(ValueError,
155			action.get_snmp_cred_id,
156			"foobar")
157
158			self.assertRaises(ValueError,
159			action.get_snmp_cred_id,
160			"internal")
161
162			def test_run_discovery_create_completes(self):
163			expected = 10
164
165			query_data = []
166			query_data.append(self.load_yaml('results_orion_snmp_cred.yaml'))
167			query_data.append({'results': [{'ID': 1}]})
168
169			invoke_data = []
170			invoke_data.append("CorePluginConfiguration")
171			invoke_data.append(10)
172
173			action = self.get_action_instance(config=self.full_config)
174			action.connect = MagicMock(return_value=True)
175			action.query = MagicMock(side_effect=query_data)
176			action.invoke = Mock(side_effect=invoke_data)
177
178			result = action.run(name="Unit Test Discovery",
179			platform="orion",
180			poller="primary",
181			snmp_communities=["public"],
182			nodes=["router1"],
183			subnets=None,
184			ip_ranges=None,
185			no_icmp_only=True,
186			auto_import=False)
187
188			self.assertEqual(result, expected)
189
190			def test_run_discovery_create_completes_on_poller(self):
191			expected = 10
192
193			query_data = []
194			query_data.append(self.load_yaml('results_orion_snmp_cred.yaml'))
195			query_data.append(self.load_yaml('results_orion_engines.yaml'))
196			query_data.append({'results': [{'ID': 1}]})
197
198			invoke_data = []
199			invoke_data.append("CorePluginConfiguration")
200			invoke_data.append(10)
201
202			action = self.get_action_instance(config=self.full_config)
203			action.connect = MagicMock(return_value=True)
204			action.query = MagicMock(side_effect=query_data)
205			action.invoke = Mock(side_effect=invoke_data)
206
207			result = action.run(name="Unit Test Discovery",
208			platform="orion",
209			poller="primary",
210			snmp_communities=["public"],
211			nodes=["router1"],
212			subnets=None,
213			ip_ranges=None,
214			no_icmp_only=True,
215			auto_import=False)
216
217			self.assertEqual(result, expected)
218

StackStorm / st2contrib

Push — master ( f1a13c...e53aa3 )

test_run_poller_lookup_primary() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like