src/osm_ngsa/osm_mon/sdnc_connectors/onos.py

   1 # Copyright 2018 Whitestack, LLC
   2 # *************************************************************
   3
   4 # This file is part of OSM Monitoring module
   5 # All Rights Reserved to Whitestack, LLC
   6
   7 # Licensed under the Apache License, Version 2.0 (the "License"); you may
   8 # not use this file except in compliance with the License. You may obtain
   9 # a copy of the License at
  10
  11 #         http://www.apache.org/licenses/LICENSE-2.0
  12
  13 # Unless required by applicable law or agreed to in writing, software
  14 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  15 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  16 # License for the specific language governing permissions and limitations
  17 # under the License.
  18
  19 # For those usages not covered by the Apache License, Version 2.0 please
  20 # contact: bdiaz@whitestack.com or glavado@whitestack.com
  21 ##
  22 import logging
  23 from typing import Dict
  24
  25 from osm_mon.sdnc_connectors.base_sdnc import SDNCConnector
  26 import requests
  27 from requests.auth import HTTPBasicAuth
  28
  29 log = logging.getLogger(__name__)
  30
  31
  32 class OnosInfraCollector(SDNCConnector):
  33     def __init__(self, sdnc_account: Dict):
  34         self.sdnc_account = sdnc_account
  35
  36     def _obtain_url(self):
  37         url = self.sdnc_account.get("url")
  38         if url:
  39             return url
  40         else:
  41             if not self.sdnc_account.get("ip") or not self.sdnc_account.get("port"):
  42                 raise Exception("You must provide a URL to contact the SDN Controller")
  43             else:
  44                 return "http://{}:{}/onos/v1/devices".format(
  45                     self.sdnc_account["ip"], self.sdnc_account["port"]
  46                 )
  47
  48     def is_sdnc_ok(self) -> bool:
  49         try:
  50             url = self._obtain_url()
  51             user = self.sdnc_account["user"]
  52             password = self.sdnc_account["password"]
  53
  54             requests.get(url, auth=HTTPBasicAuth(user, password))
  55             return True
  56         except Exception:
  57             log.exception("SDNC status is not OK!")
  58             return False