-# -*- coding: utf-8 -*-\r
-##\r
-# Copyright 2019 Atos - CoE Telco NFV Team\r
-# All Rights Reserved.\r
-#\r
-# Contributors: Oscar Luis Peral, Atos\r
-#\r
-# Licensed under the Apache License, Version 2.0 (the "License"); you may\r
-# not use this file except in compliance with the License. You may obtain\r
-# a copy of the License at\r
-#\r
-# http://www.apache.org/licenses/LICENSE-2.0\r
-#\r
-# Unless required by applicable law or agreed to in writing, software\r
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT\r
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the\r
-# License for the specific language governing permissions and limitations\r
-# under the License.\r
-#\r
-# For those usages not covered by the Apache License, Version 2.0 please\r
-# contact with: <oscarluis.peral@atos.net>\r
-#\r
-# Neither the name of Atos nor the names of its\r
-# contributors may be used to endorse or promote products derived from\r
-# this software without specific prior written permission.\r
-#\r
-# This work has been performed in the context of Arista Telefonica OSM PoC.\r
-##\r
-\r
-from jsonrpclib import Server\r
-import socket\r
-import ssl\r
-\r
-\r
-class AristaSwitch():\r
- """\r
- Used to run switch commands through eAPI and check command output\r
- """\r
-\r
- def __init__(self, name=None, host=None, user=None, passwd=None,\r
- verify_ssl=False, unix_socket=None,\r
- logger=None):\r
-\r
- self.host = host\r
- self.user = user\r
- self.passwd = passwd\r
-\r
- self.unix_socket = unix_socket\r
- self.local_ep = Server(unix_socket) \\r
- if unix_socket is not None else None\r
-\r
- s = "https://{user}:{passwd}@{host}/command-api"\r
- self.url = s.format(user=user, passwd=passwd, host=host)\r
- self.ep = Server(self.url)\r
- self.verify_ssl = verify_ssl\r
- if not self.verify_ssl:\r
- try:\r
- ssl._create_default_https_context = ssl.\\r
- _create_unverified_context\r
- except AttributeError:\r
- # Old python versions do not verify certs by default\r
- pass\r
-\r
- self.log = logger\r
-\r
- def _multilinestr_to_list(self, multilinestr=None):\r
- """\r
- Returns a list, each item been one line of a (multi)line string\r
- Handy for running multiple lines commands through one API call\r
- """\r
- mylist = \\r
- [x.strip() for x in multilinestr.split('\n') if x.strip() != '']\r
- return mylist\r
-\r
- def run(self, cmds=None, timeout=10, local_run=False):\r
- """\r
- Runs commands through eAPI\r
-\r
- If local_run is True eAPI call will be done using local unix socket\r
- If local run is False eAPI call will be done using TCPIP\r
- """\r
- socket.setdefaulttimeout(timeout)\r
-\r
- r = None\r
-\r
- if type(cmds) is str:\r
- run_list = self._multilinestr_to_list(cmds)\r
-\r
- if type(cmds) is list:\r
- run_list = cmds\r
-\r
- if local_run:\r
- ep = self.local_ep\r
- ep_log = "local unix socket {}".format(str(self.unix_socket))\r
- else:\r
- ep = self.ep\r
- ep_log = "tcpip socket {}".format(str(self.host))\r
-\r
- self.log.debug("Calling eAPI at {} with commands {}".\r
- format(ep_log, str(run_list)))\r
-\r
- try:\r
- r = ep.runCmds(1, run_list)\r
- except Exception as e:\r
- self.log.error(str(e))\r
- raise(e)\r
-\r
- return r\r