1 # Copyright 2018 Whitestack, LLC
2 # *************************************************************
4 # This file is part of OSM Monitoring module
5 # All Rights Reserved to Whitestack, LLC
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
11 # http://www.apache.org/licenses/LICENSE-2.0
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
19 # For those usages not covered by the Apache License, Version 2.0 please
20 # contact: bdiaz@whitestack.com or glavado@whitestack.com
23 from typing
import Dict
25 from osm_mon
.sdnc_connectors
.base_sdnc
import SDNCConnector
27 from requests
.auth
import HTTPBasicAuth
29 log
= logging
.getLogger(__name__
)
32 class OnosInfraCollector(SDNCConnector
):
33 def __init__(self
, sdnc_account
: Dict
):
34 self
.sdnc_account
= sdnc_account
36 def _obtain_url(self
):
37 url
= self
.sdnc_account
.get("url")
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")
44 return "http://{}:{}/onos/v1/devices".format(
45 self
.sdnc_account
["ip"], self
.sdnc_account
["port"]
48 def is_sdnc_ok(self
) -> bool:
50 url
= self
._obtain
_url
()
51 user
= self
.sdnc_account
["user"]
52 password
= self
.sdnc_account
["password"]
54 requests
.get(url
, auth
=HTTPBasicAuth(user
, password
))
57 log
.exception("SDNC status is not OK!")