OrionBaseAction.get_ncm_node_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-16 14:05 UTC

OrionBaseAction.get_ncm_node_id() B

↳ Parent: OrionBaseAction

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
cc	5
c	2
b	0
f	0
dl	0
loc	25
rs	8.0894

# 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

from lib.node import OrionNode
from lib.utils import send_user_error, is_ip


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.
        """
        if platform is None:
            try:
                platform = self.config['defaults']['platform']
            except IndexError:
                send_user_error("No default Orion platform.")
                raise ValueError("No default Orion platform.")

        self.logger.debug("Connecting to Orion platform: {}".format(platform))

        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")

        return platform

    def get_node(self, node):
        """
        Get an OrionNode object
        """

        orion_node = OrionNode()

        if is_ip(node):
            query_for_where = "IPAddress"
        else:
            query_for_where = "Caption"

        swql = """SELECT NodeID, Uri, IPAddress, Caption
        FROM Orion.Nodes
        WHERE {}=@query_on""".format(query_for_where)
        kargs = {'query_on': node}
        data = self.query(swql, **kargs)

        if len(data['results']) == 1:
            try:
                orion_node.npm_id = data['results'][0]['NodeID']
                orion_node.uri = data['results'][0]['Uri']
                orion_node.ip_address = data['results'][0]['IPAddress']
                orion_node.caption = data['results'][0]['Caption']
            except IndexError:
                pass
        elif len(data['results']) >= 2:
            self.logger.debug(
                "Muliple Nodes match '{}' Caption: {}".format(
                    node, data))
            raise ValueError("Muliple Nodes match '{}' Caption".format(
                node))

        if orion_node.npm:
            swql = """SELECT NodeID
            FROM Cirrus.Nodes
            WHERE CoreNodeID=@CoreNodeID"""
            kargs = {'CoreNodeID': orion_node.npm_id}
            data = self.query(swql, **kargs)

            if len(data['results']) == 1:
                try:
                    orion_node.ncm_id = data['results'][0]['NodeID']
                except IndexError:
                    pass

        return orion_node

    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 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_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:
                send_user_error("Invalid poller name")
                raise ValueError("Invalid poller name")

    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, NodeID, Action, RequestedConfigType,
            RequestedScript, RequestedReboot, ConfigID, TransferProtocol,
            Status, ErrorMessage, DeviceOutput, DateTime, UserName
            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"
                break
            else:
                ts['status'] = "Unknown"
                break

        ts['RequestedScript'] = transfer_data['results'][0]['RequestedScript']
        ts['RequestedReboot'] = transfer_data['results'][0]['RequestedReboot']
        ts['ErrorMessage'] = transfer_data['results'][0]['ErrorMessage']
        ts['DeviceOutput'] = transfer_data['results'][0]['DeviceOutput']
        ts['UserName'] = transfer_data['results'][0]['UserName']

        return ts


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			from lib.node import OrionNode
22			from lib.utils import send_user_error, is_ip
23
24
25			class OrionBaseAction(Action):
26			def __init__(self, config):
27			super(OrionBaseAction, self).__init__(config)
28
29			self.client = None
30
31			if "orion" not in self.config:
32			raise ValueError("Orion host details not in the config.yaml")
33
34			def connect(self, platform):
35			"""
36			Connect to an Orion platform from the packs config.yaml.
37			"""
38			if platform is None:
39			try:
40			platform = self.config['defaults']['platform']
41			except IndexError:
42			send_user_error("No default Orion platform.")
43			raise ValueError("No default Orion platform.")
44
45			self.logger.debug("Connecting to Orion platform: {}".format(platform))
46
47			try:
48			self.client = SwisClient(
49			self.config['orion'][platform]['host'],
50			self.config['orion'][platform]['user'],
51			self.config['orion'][platform]['password'])
52			except KeyError:
53			raise ValueError("Orion host details not in the config.yaml")
54
55			return platform
56
57			def get_node(self, node):
58			"""
59			Get an OrionNode object
60			"""
61
62			orion_node = OrionNode()
63
64			if is_ip(node):
65			query_for_where = "IPAddress"
66			else:
67			query_for_where = "Caption"
68
69			swql = """SELECT NodeID, Uri, IPAddress, Caption
70			FROM Orion.Nodes
71			WHERE {}=@query_on""".format(query_for_where)
72			kargs = {'query_on': node}
73			data = self.query(swql, **kargs)
74
75			if len(data['results']) == 1:
76			try:
77			orion_node.npm_id = data['results'][0]['NodeID']
78			orion_node.uri = data['results'][0]['Uri']
79			orion_node.ip_address = data['results'][0]['IPAddress']
80			orion_node.caption = data['results'][0]['Caption']
81			except IndexError:
82			pass
83			elif len(data['results']) >= 2:
84			self.logger.debug(
85			"Muliple Nodes match '{}' Caption: {}".format(
86			node, data))
87			raise ValueError("Muliple Nodes match '{}' Caption".format(
88			node))
89
90			if orion_node.npm:
91			swql = """SELECT NodeID
92			FROM Cirrus.Nodes
93			WHERE CoreNodeID=@CoreNodeID"""
94			kargs = {'CoreNodeID': orion_node.npm_id}
95			data = self.query(swql, **kargs)
96
97			if len(data['results']) == 1:
98			try:
99			orion_node.ncm_id = data['results'][0]['NodeID']
100			except IndexError:
101			pass
102
103			return orion_node
104
105			def query(self, swql, **kargs):
106			"""
107			Run SWQL against the Orion Platform.
108			"""
109			return self.client.query(swql, **kargs)
110
111			def invoke(self, entity, verb, *args):
112			"""
113			Run an Invoke against the Orion Platform.
114			"""
115			return self.client.invoke(entity, verb, *args)
116
117			def create(self, entity, **kargs):
118			"""
119			Run an Create against the Orion Platform.
120			"""
121			return self.client.create(entity, **kargs)
122
123			def read(self, uri):
124			"""
125			Run an Read against the Orion Platform.
126			"""
127			return self.client.read(uri)
128
129			def update(self, uri, **kargs):
130			"""
131			Run an Update against the Orion Platform.
132			"""
133			return self.client.update(uri, **kargs)
134
135			def delete(self, uri):
136			"""
137			Run an Delete of an URI against the Orion Platform.
138			"""
139			return self.client.delete(uri)
140
141			def get_snmp_community(self, community, std_community):
142			"""
143			Return the correct SNMP comminity to use.
144			"""
145			if community is not None:
146			return community
147			elif std_community is not None:
148			try:
149			return self.config['defaults']['snmp'][std_community]
150			except KeyError:
151			raise ValueError("Invalid standard community")
152			elif std_community is None:
153			raise ValueError("Need one of community or std_community")
154
155			def get_snmp_cred_id(self, community):
156			"""
157			Look up an SNMP community in the config and then look up
158			the Orion ID for the Credential.
159			"""
160
161			# Check if community is a know standard, otherwise
162			# use it as the community.
163			try:
164			name = self.get_snmp_community(None, community)
165			except ValueError:
166			name = community
167
168			swql = """SELECT ID FROM Orion.Credential
169			WHERE CredentialType=@CredentialType and Name=@name"""
170
171			kargs = {'CredentialType':
172			'SolarWinds.Orion.Core.Models.Credentials.SnmpCredentialsV2',
173			'name': name}
174			orion_data = self.query(swql, **kargs)
175
176			if len(orion_data['results']) == 1:
177			return orion_data['results'][0]['ID']
178			else:
179			raise ValueError(
180			"Failed to lookup community in Orion.Credential!")
181
182			def get_engine_id(self, poller):
183			"""
184			Takes a poller name (or primary) and returns the EngineID for
185			the poller.
186
187			Raises: ValueError on an invaild poller.
188
189			Returns: The EngineID (int)
190			"""
191
192			if poller == "primary":
193			return 1
194			else:
195			swql = """SELECT EngineID, ServerName, IP, ServerType
196			FROM Orion.Engines
197			WHERE ServerName=@poller"""
198			kargs = {'poller': poller}
199			data = self.query(swql, **kargs)
200
201			if len(data['results']) == 1:
202			return data['results'][0]['EngineID']
203			else:
204			send_user_error("Invalid poller name")
205			raise ValueError("Invalid poller name")
206
207			def get_ncm_transfer_results(self, transfer_id, sleep_delay=10):
208			"""
209			Gets the completed (waits until finished). NCM job transfer status
210			from Orion.
211
212			Retruns: The completed status.
213			"""
214			ts = {}
215			while True:
216			swql = """SELECT TransferID, NodeID, Action, RequestedConfigType,
217			RequestedScript, RequestedReboot, ConfigID, TransferProtocol,
218			Status, ErrorMessage, DeviceOutput, DateTime, UserName
219			FROM NCM.TransferResults
220			WHERE TransferID=@transfer_id"""
221			kargs = {'transfer_id': transfer_id}
222
223			transfer_data = self.query(swql, **kargs)
224			status = transfer_data['results'][0]['Status']
225
226			if status == 1:
227			time.sleep(sleep_delay)
228			elif status == 2:
229			ts['status'] = "Complete"
230			break
231			elif status == 3:
232			ts['status'] = "Error"
233			break
234			else:
235			ts['status'] = "Unknown"
236			break
237
238			ts['RequestedScript'] = transfer_data['results'][0]['RequestedScript']
239			ts['RequestedReboot'] = transfer_data['results'][0]['RequestedReboot']
240			ts['ErrorMessage'] = transfer_data['results'][0]['ErrorMessage']
241			ts['DeviceOutput'] = transfer_data['results'][0]['DeviceOutput']
242			ts['UserName'] = transfer_data['results'][0]['UserName']
243
244			return ts
245

StackStorm / st2contrib

Pull Request — master (#487)

OrionBaseAction.get_ncm_node_id() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like