OrionBaseAction.get_snmp_cred_id() - Code Metrics - Inspection of "Additional actions, aliases, rule and testing for..." - StackStorm/st2contrib - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#487)

unknown

created 2016-05-07 17:45 UTC

OrionBaseAction.get_snmp_cred_id() B

↳ Parent: OrionBaseAction

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	3
dl	0
loc	26
rs	8.8571

# 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
# limitations under the License.

import time

from st2actions.runners.pythonrunner import Action
from orionsdk import SwisClient


class OrionBaseAction(Action):
    def __init__(self, config):
        super(OrionBaseAction, self).__init__(config)

        self.client = None

        if "orion" not in self.config:
            raise ValueError("Orion host details not in the config.yaml")

    def connect(self, platform):
        """
        Connect to an Orion platform from the packs config.yaml.
        """
        try:
            self.client = SwisClient(
                self.config['orion'][platform]['host'],
                self.config['orion'][platform]['user'],
                self.config['orion'][platform]['password'])
        except KeyError:
            raise ValueError("Orion host details not in the config.yaml")

    def query(self, swql, **kargs):
        """
        Run SWQL against the Orion Platform.
        """
        return self.client.query(swql, **kargs)

    def invoke(self, entity, verb, *args):
        """
        Run an Invoke against the Orion Platform.
        """
        return self.client.invoke(entity, verb, *args)

    def create(self, entity, **kargs):
        """
        Run an Create against the Orion Platform.
        """
        return self.client.create(entity, **kargs)

    def read(self, uri):
        """
        Run an Read against the Orion Platform.
        """
        return self.client.read(uri)

    def update(self, uri, **kargs):
        """
        Run an Update against the Orion Platform.
        """
        return self.client.update(uri, **kargs)

    def delete(self, uri):
        """
        Run an Delete of an URI against the Orion Platform.
        """
        return self.client.delete(uri)

    def node_exists(self, caption, ip_address):
        """
        Check if an Node exists (caption and or ip) on the Orion platform.

        Returns: True or False.
        """
        swql = """SELECT NodeID, IPAddress FROM Orion.Nodes
                  WHERE Caption=@caption"""
        kargs = {'caption': caption}
        caption_data = self.query(swql, **kargs)

        if len(caption_data['results']) >= 1:
            self.logger.debug(
                "One (or more) Nodes match '{}' Caption.".format(caption))
            return True

        swql = """SELECT NodeID, IPAddress FROM Orion.Nodes
                  WHERE IPAddress=@ip_address"""
        kargs = {'ip_address': ip_address}
        ip_data = self.query(swql, **kargs)

        if len(ip_data['results']) >= 1:
            self.logger.debug(
                "One (or more) Nodes match '{}' IP.".format(ip_address))
            return True
        else:
            return False

    def get_snmp_community(self, community, std_community):
        """
        Return the correct SNMP comminity to use.
        """
        if community is not None:
            return community
        elif std_community is not None:
            try:
                return self.config['defaults']['snmp'][std_community]
            except KeyError:
                raise ValueError("Invalid standard community")
        elif std_community is None:
            raise ValueError("Need one of community or std_community")

    def get_snmp_cred_id(self, community):
        """
        Look up an SNMP community in the config and then look up
        the Orion ID for the Credential.
        """

        # Check if community is a know standard, otherwise
        # use it as the community.
        try:
            name = self.get_snmp_community(None, community)
        except ValueError:
            name = community

        swql = """SELECT ID FROM Orion.Credential
        WHERE CredentialType=@CredentialType and Name=@name"""

        kargs = {'CredentialType':
                 'SolarWinds.Orion.Core.Models.Credentials.SnmpCredentialsV2',
                 'name': name}
        orion_data = self.query(swql, **kargs)

        if len(orion_data['results']) == 1:
            return orion_data['results'][0]['ID']
        else:
            raise ValueError(
                "Failed to lookup community in Orion.Credential!")

    def get_node_id(self, caption):
        """
        Gets an NodeID from the Orion platform.

        Raises: ValueError on muliple or no matching caption.

        Returns: the NodeID (int)
        """
        swql = "SELECT NodeID FROM Orion.Nodes WHERE Caption=@caption"
        kargs = {'caption': caption}
        data = self.query(swql, **kargs)

        if len(data['results']) == 1:
            try:
                return data['results'][0]['NodeID']
            except IndexError:
                raise ValueError("Invalid Node")
        elif len(data['results']) >= 2:
            self.logger.debug(
                "Muliple Nodes match '{}' Caption: {}".format(
                    caption, data))
            raise ValueError("Muliple Nodes match '{}' Caption".format(
                caption))
        elif len(data['results']) == 0:
            self.logger.debug(
                "No Nodes match '{}' Caption: {}".format(
                    caption, data))
            raise ValueError("No matching Caption for '{}'".format(
                caption))

    def get_node_uri(self, caption):
        """
        Gets an NodeID from the Orion platform.

        Raises: ValueError on muliple or no matching caption.

        Returns:
            string: the Node's URI
        """
        swql = "SELECT Uri FROM Orion.Nodes WHERE Caption=@caption"
        kargs = {'caption': caption}
        data = self.query(swql, **kargs)

        if len(data['results']) == 1:
            try:
                return data['results'][0]['Uri']
            except IndexError:
                raise ValueError("Invalid Node")
        elif len(data['results']) >= 2:
            self.logger.debug(
                "Muliple Nodes match '{}' Caption: {}".format(
                    caption, data))
            raise ValueError("Muliple Nodes match '{}' Caption".format(
                caption))
        elif len(data['results']) == 0:
            self.logger.debug(
                "No Nodes match '{}' Caption: {}".format(
                    caption, data))
            raise ValueError("No matching Caption for '{}'".format(
                caption))

    def get_engine_id(self, poller):
        """
        Takes a poller name (or primary) and returns the EngineID for
        the poller.

        Raises: ValueError on an invaild poller.

        Returns: The EngineID (int)
        """

        if poller == "primary":
            return 1
        else:
            swql = """SELECT EngineID, ServerName, IP, ServerType
            FROM Orion.Engines
            WHERE ServerName=@poller"""
            kargs = {'poller': poller}
            data = self.query(swql, **kargs)

            if len(data['results']) == 1:
                return data['results'][0]['EngineID']
            else:
                self.send_user_error("Invalid poller name")
                raise ValueError("Invalid poller name")

    def get_ncm_node_id(self, caption):
        """
        Queries the Network configuration Manager nodes table on the Orion
        platform for the NodeID of a given node name (aka NodeCaption).

        Raises: IndexError on Invalid number of nodes (e.g. 0 or 2+).

        Returns: A single node id.
        """

        swql = "SELECT NodeID FROM Cirrus.Nodes WHERE NodeCaption=@node"
        kargs = {'node': caption}
        data = self.query(swql, **kargs)

        if len(data['results']) == 1:
            try:
                return data['results'][0]['NodeID']
            except IndexError:
                raise IndexError("Invalid Node")
        elif len(data['results']) >= 2:
            raise IndexError("Muliple Nodes match '{}' NodeCaption".format(
                caption))
        elif len(data['results']) == 0:
            raise IndexError("No matching NodeCaption for '{}'".format(
                caption))

    def get_ncm_transfer_results(self, transfer_id, sleep_delay=10):
        """
        Gets the completed (waits until finished). NCM job transfer status
        from Orion.

        Retruns: The completed status.
        """
        ts = {}
        while True:
            swql = """SELECT TransferID, Action, Status, ErrorMessage,
            DeviceOutput FROM NCM.TransferResults
            WHERE TransferID=@transfer_id"""
            kargs = {'transfer_id': transfer_id}

            transfer_data = self.query(swql, **kargs)
            status = transfer_data['results'][0]['Status']

            if status == 1:
                time.sleep(sleep_delay)
            elif status == 2:
                ts['status'] = "Complete"
                break
            elif status == 3:
                ts['status'] = "Error"
                ts['ErrorMessage'] = transfer_data['results'][0][
                    'ErrorMessage']
                break
            else:
                ts['status'] = "Unknown"
                ts['ErrorMessage'] = "Invalid stauts: {}".format(status)
                break

        return ts

    def status_code_to_text(self, status):
        """
        Takes an Solarwinds Orion status code and translates it to
        human text and also a colour that can be used in Slack.
        """

        if status == 0:
            return ("Unknown", "grey")  # aka slack 'grey'
        elif status == 1:
            return ("Up", "good")  # slack 'good'
        elif status == 2:
            return ("Down", "#7CD197")  # slack 'danger'
        elif status == 3:
            return ("Warning", "warning")  # slack 'warning'
        elif status == 14:
            return ("Critical", "#7CD197")  # slack 'danger'

    def send_user_error(self, message):
        """
        Prints an user error message.
        """
        print(message)


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			# limitations under the License.
15
16			import time
17
18			from st2actions.runners.pythonrunner import Action
19			from orionsdk import SwisClient
20
21
22			class OrionBaseAction(Action):
23			def __init__(self, config):
24			super(OrionBaseAction, self).__init__(config)
25
26			self.client = None
27
28			if "orion" not in self.config:
29			raise ValueError("Orion host details not in the config.yaml")
30
31			def connect(self, platform):
32			"""
33			Connect to an Orion platform from the packs config.yaml.
34			"""
35			try:
36			self.client = SwisClient(
37			self.config['orion'][platform]['host'],
38			self.config['orion'][platform]['user'],
39			self.config['orion'][platform]['password'])
40			except KeyError:
41			raise ValueError("Orion host details not in the config.yaml")
42
43			def query(self, swql, **kargs):
44			"""
45			Run SWQL against the Orion Platform.
46			"""
47			return self.client.query(swql, **kargs)
48
49			def invoke(self, entity, verb, *args):
50			"""
51			Run an Invoke against the Orion Platform.
52			"""
53			return self.client.invoke(entity, verb, *args)
54
55			def create(self, entity, **kargs):
56			"""
57			Run an Create against the Orion Platform.
58			"""
59			return self.client.create(entity, **kargs)
60
61			def read(self, uri):
62			"""
63			Run an Read against the Orion Platform.
64			"""
65			return self.client.read(uri)
66
67			def update(self, uri, **kargs):
68			"""
69			Run an Update against the Orion Platform.
70			"""
71			return self.client.update(uri, **kargs)
72
73			def delete(self, uri):
74			"""
75			Run an Delete of an URI against the Orion Platform.
76			"""
77			return self.client.delete(uri)
78
79			def node_exists(self, caption, ip_address):
80			"""
81			Check if an Node exists (caption and or ip) on the Orion platform.
82
83			Returns: True or False.
84			"""
85			swql = """SELECT NodeID, IPAddress FROM Orion.Nodes
86			WHERE Caption=@caption"""
87			kargs = {'caption': caption}
88			caption_data = self.query(swql, **kargs)
89
90			if len(caption_data['results']) >= 1:
91			self.logger.debug(
92			"One (or more) Nodes match '{}' Caption.".format(caption))
93			return True
94
95			swql = """SELECT NodeID, IPAddress FROM Orion.Nodes
96			WHERE IPAddress=@ip_address"""
97			kargs = {'ip_address': ip_address}
98			ip_data = self.query(swql, **kargs)
99
100			if len(ip_data['results']) >= 1:
101			self.logger.debug(
102			"One (or more) Nodes match '{}' IP.".format(ip_address))
103			return True
104			else:
105			return False
106
107			def get_snmp_community(self, community, std_community):
108			"""
109			Return the correct SNMP comminity to use.
110			"""
111			if community is not None:
112			return community
113			elif std_community is not None:
114			try:
115			return self.config['defaults']['snmp'][std_community]
116			except KeyError:
117			raise ValueError("Invalid standard community")
118			elif std_community is None:
119			raise ValueError("Need one of community or std_community")
120
121			def get_snmp_cred_id(self, community):
122			"""
123			Look up an SNMP community in the config and then look up
124			the Orion ID for the Credential.
125			"""
126
127			# Check if community is a know standard, otherwise
128			# use it as the community.
129			try:
130			name = self.get_snmp_community(None, community)
131			except ValueError:
132			name = community
133
134			swql = """SELECT ID FROM Orion.Credential
135			WHERE CredentialType=@CredentialType and Name=@name"""
136
137			kargs = {'CredentialType':
138			'SolarWinds.Orion.Core.Models.Credentials.SnmpCredentialsV2',
139			'name': name}
140			orion_data = self.query(swql, **kargs)
141
142			if len(orion_data['results']) == 1:
143			return orion_data['results'][0]['ID']
144			else:
145			raise ValueError(
146			"Failed to lookup community in Orion.Credential!")
147
148			def get_node_id(self, caption):
149			"""
150			Gets an NodeID from the Orion platform.
151
152			Raises: ValueError on muliple or no matching caption.
153
154			Returns: the NodeID (int)
155			"""
156			swql = "SELECT NodeID FROM Orion.Nodes WHERE Caption=@caption"
157			kargs = {'caption': caption}
158			data = self.query(swql, **kargs)
159
160			if len(data['results']) == 1:
161			try:
162			return data['results'][0]['NodeID']
163			except IndexError:
164			raise ValueError("Invalid Node")
165			elif len(data['results']) >= 2:
166			self.logger.debug(
167			"Muliple Nodes match '{}' Caption: {}".format(
168			caption, data))
169			raise ValueError("Muliple Nodes match '{}' Caption".format(
170			caption))
171			elif len(data['results']) == 0:
172			self.logger.debug(
173			"No Nodes match '{}' Caption: {}".format(
174			caption, data))
175			raise ValueError("No matching Caption for '{}'".format(
176			caption))
177
178			def get_node_uri(self, caption):
179			"""
180			Gets an NodeID from the Orion platform.
181
182			Raises: ValueError on muliple or no matching caption.
183
184			Returns:
185			string: the Node's URI
186			"""
187			swql = "SELECT Uri FROM Orion.Nodes WHERE Caption=@caption"
188			kargs = {'caption': caption}
189			data = self.query(swql, **kargs)
190
191			if len(data['results']) == 1:
192			try:
193			return data['results'][0]['Uri']
194			except IndexError:
195			raise ValueError("Invalid Node")
196			elif len(data['results']) >= 2:
197			self.logger.debug(
198			"Muliple Nodes match '{}' Caption: {}".format(
199			caption, data))
200			raise ValueError("Muliple Nodes match '{}' Caption".format(
201			caption))
202			elif len(data['results']) == 0:
203			self.logger.debug(
204			"No Nodes match '{}' Caption: {}".format(
205			caption, data))
206			raise ValueError("No matching Caption for '{}'".format(
207			caption))
208
209			def get_engine_id(self, poller):
210			"""
211			Takes a poller name (or primary) and returns the EngineID for
212			the poller.
213
214			Raises: ValueError on an invaild poller.
215
216			Returns: The EngineID (int)
217			"""
218
219			if poller == "primary":
220			return 1
221			else:
222			swql = """SELECT EngineID, ServerName, IP, ServerType
223			FROM Orion.Engines
224			WHERE ServerName=@poller"""
225			kargs = {'poller': poller}
226			data = self.query(swql, **kargs)
227
228			if len(data['results']) == 1:
229			return data['results'][0]['EngineID']
230			else:
231			self.send_user_error("Invalid poller name")
232			raise ValueError("Invalid poller name")
233
234			def get_ncm_node_id(self, caption):
235			"""
236			Queries the Network configuration Manager nodes table on the Orion
237			platform for the NodeID of a given node name (aka NodeCaption).
238
239			Raises: IndexError on Invalid number of nodes (e.g. 0 or 2+).
240
241			Returns: A single node id.
242			"""
243
244			swql = "SELECT NodeID FROM Cirrus.Nodes WHERE NodeCaption=@node"
245			kargs = {'node': caption}
246			data = self.query(swql, **kargs)
247
248			if len(data['results']) == 1:
249			try:
250			return data['results'][0]['NodeID']
251			except IndexError:
252			raise IndexError("Invalid Node")
253			elif len(data['results']) >= 2:
254			raise IndexError("Muliple Nodes match '{}' NodeCaption".format(
255			caption))
256			elif len(data['results']) == 0:
257			raise IndexError("No matching NodeCaption for '{}'".format(
258			caption))
259
260			def get_ncm_transfer_results(self, transfer_id, sleep_delay=10):
261			"""
262			Gets the completed (waits until finished). NCM job transfer status
263			from Orion.
264
265			Retruns: The completed status.
266			"""
267			ts = {}
268			while True:
269			swql = """SELECT TransferID, Action, Status, ErrorMessage,
270			DeviceOutput FROM NCM.TransferResults
271			WHERE TransferID=@transfer_id"""
272			kargs = {'transfer_id': transfer_id}
273
274			transfer_data = self.query(swql, **kargs)
275			status = transfer_data['results'][0]['Status']
276
277			if status == 1:
278			time.sleep(sleep_delay)
279			elif status == 2:
280			ts['status'] = "Complete"
281			break
282			elif status == 3:
283			ts['status'] = "Error"
284			ts['ErrorMessage'] = transfer_data['results'][0][
285			'ErrorMessage']
286			break
287			else:
288			ts['status'] = "Unknown"
289			ts['ErrorMessage'] = "Invalid stauts: {}".format(status)
290			break
291
292			return ts
293
294			def status_code_to_text(self, status):
295			"""
296			Takes an Solarwinds Orion status code and translates it to
297			human text and also a colour that can be used in Slack.
298			"""
299
300			if status == 0:
301			return ("Unknown", "grey") # aka slack 'grey'
302			elif status == 1:
303			return ("Up", "good") # slack 'good'
304			elif status == 2:
305			return ("Down", "#7CD197") # slack 'danger'
306			elif status == 3:
307			return ("Warning", "warning") # slack 'warning'
308			elif status == 14:
309			return ("Critical", "#7CD197") # slack 'danger'
310
311			def send_user_error(self, message):
312			"""
313			Prints an user error message.
314			"""
315			print(message)
316

StackStorm / st2contrib

Pull Request — master (#487)

OrionBaseAction.get_snmp_cred_id() B

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like