# terms of the Eclipse Public License v1.0 which accompanies this distribution,
# and is available at http://www.eclipse.org/legal/epl-v10.html
+import json
+import requests
+
+
__author__ = "Vratko Polak"
__copyright__ = "Copyright(c) 2015, Cisco Systems, Inc."
__license__ = "Eclipse Public License v1.0"
__email__ = "vrpolak@cisco.com"
-import json
-import requests
-
-
#
# Karaf Keyword definitions.
#
def __init__(self, ip, username="", password="", port="8181"):
"""Initialize session using hardcoded text data, remember credentials."""
- self.rest_prefix = "http://" + ip + ":" + port + "/restconf/"
+ self.rest_prefix = "http://" + ip + ":" + port + "/rests/"
self.session = requests.Session()
if username:
self.session.auth = (username, password) # May work with non-string values
else:
- self.session.auth = None # Supports "no authentication mode" as in odl-restconf-noauth
+ self.session.auth = (
+ None # Supports "no authentication mode" as in odl-restconf-noauth
+ )
def robust_method(self, method, uri, **kwargs):
"""Try method once using session credentials. Return last response."""
def __init__(self, ip, username="", password="", port="8181"):
"""Prepare session initialization data using hardcoded text, remember credentials."""
- self.rest_prefix = "http://" + ip + ":" + port + "/restconf/"
+ self.rest_prefix = "http://" + ip + ":" + port + "/rests/"
if username:
self.auth = (username, password) # May work with non-string values
else:
- self.auth = None # Supports "no authentication mode" as in odl-restconf-noauth
+ self.auth = (
+ None # Supports "no authentication mode" as in odl-restconf-noauth
+ )
self.session = None
def robust_method(self, method, uri, **kwargs):
def __init__(self, ip, username, password, scope, port="8181"):
"""Initialize session using hardcoded text data."""
self.auth_url = "http://" + ip + ":" + port + "/oauth2/token"
- self.rest_prefix = "http://" + ip + ":" + port + "/restconf/"
+ self.rest_prefix = "http://" + ip + ":" + port + "/rests/"
self.auth_data = "grant_type=password&username=" + username
self.auth_data += "&password=" + password + "&scope=" + scope
self.auth_header = {"Content-Type": "application/x-www-form-urlencoded"}
if self.session:
self.session.close()
self.session = requests.Session()
- resp = self.session.post(self.auth_url, data=self.auth_data, headers=self.auth_header)
+ resp = self.session.post(
+ self.auth_url, data=self.auth_data, headers=self.auth_header
+ )
resp_obj = json.loads(resp.text)
try:
token = resp_obj["access_token"]
raise RuntimeError("Parse failed: " + resp.text)
self.token = token
# TODO: Use logging so that callers could see token refreshes.
- # print "DEBUG: token:", token
# We keep self.session to use for the following restconf requests.
def oneshot_method(self, method, uri, **kwargs):
def __init__(self, ip, username, password, scope, port="8181"):
"""Prepare session initialization data using hardcoded text."""
self.auth_url = "http://" + ip + ":" + port + "/oauth2/token"
- self.rest_prefix = "http://" + ip + ":" + port + "/restconf/"
+ self.rest_prefix = "http://" + ip + ":" + port + "/rests/"
self.auth_data = "grant_type=password&username=" + username
self.auth_data += "&password=" + password + "&scope=" + scope
self.auth_header = {"Content-Type": "application/x-www-form-urlencoded"}
if self.session:
self.session.close()
self.session = requests.Session()
- resp = self.session.post(self.auth_url, data=self.auth_data, headers=self.auth_header)
+ resp = self.session.post(
+ self.auth_url, data=self.auth_data, headers=self.auth_header
+ )
resp_obj = json.loads(resp.text)
try:
token = resp_obj["access_token"]
raise RuntimeError("Parse failed: " + resp.text)
self.token = token
# TODO: Use logging so that callers could see token refreshes.
- # print "DEBUG: token:", token
# We keep self.session to use for the following restconf requests.
def oneshot_method(self, method, uri, **kwargs):