Commit 2d74a2ad authored by Alex Chauvin's avatar Alex Chauvin

migrate exception named SSD to SDS

parent d55011e8
......@@ -8,19 +8,29 @@
Exceptions for the SOLIDServer modules
"""
__all__ = ["SSDError",
import logging
__all__ = ["SDSAuthError",
"SDSEmptyError",
"SDSInitError",
"SDSRequestError",
"SDSServiceError",
# deprecated versions
"SSDAuthError",
"SSDError",
"SSDInitError",
"SSDServiceError",
"SSDRequestError",
"SDSEmptyError",
"SSDAuthError"]
"SSDServiceError",
"SDSError", ]
# ---- DEPRECATED TO BE SUPPRESSED --------------------------------------
class SSDError(Exception):
""" generic class for any exception in SOLIDServer communication """
def __init__(self, message=""):
super(SSDError, self).__init__()
self.message = message
logging.critical("*deprecated class*, migrate to SDS version")
def __str__(self):
return "{}".format(self.message)
......@@ -61,9 +71,57 @@ class SSDAuthError(SSDError):
def __init__(self, message=""):
super(SSDAuthError, self).__init__("authent: {}".format(message))
# --------------------------------------------------------------------------
class SDSError(Exception):
""" generic class for any exception in SOLIDServer communication """
def __init__(self, message=""):
super(SDSError, self).__init__()
self.message = message
def __str__(self):
return "{}".format(self.message)
class SDSInitError(SDSError):
""" raised when action on non initialized SSD connection """
def __init__(self, message=""):
super(SDSInitError, self).__init__("[init] {}".format(message))
class SDSServiceError(SDSError):
""" raised on unknown service """
def __init__(self, service_name, message=""):
super(SDSServiceError, self).__init__(message)
self.service = service_name
def __str__(self):
return "{} on service {}".format(self.message, self.service)
class SDSRequestError(SDSError):
""" raised when urllib request is failing """
def __init__(self, method, url, headers, message=""):
super(SDSRequestError, self).__init__(message)
self.method = method
self.url = url
self.headers = headers
def __str__(self):
return "{} with {} {}".format(self.message, self.method, self.url)
class SDSAuthError(SDSError):
""" raised when auth on request is wrong """
def __init__(self, message=""):
super(SDSAuthError, self).__init__("authent: {}".format(message))
class SDSEmptyError(SSDError):
class SDSEmptyError(SDSError):
""" raised when empty answer """
def __init__(self, message=""):
......
# -*- Mode: Python; python-indent-offset: 4 -*-
# -*-coding:Utf-8 -*
#
# Time-stamp: <2019-06-24 22:16:52 alex>
# Time-stamp: <2019-07-06 19:04:08 alex>
#
# disable naming convention issue
# pylint: disable=C0103
......@@ -30,13 +30,13 @@ requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
if sys.version_info[0] == 2:
# pylint: disable=F0401
from mapper import SERVICE_MAPPER, METHOD_MAPPER
from Exception import SSDInitError, SSDError
from Exception import SSDServiceError, SSDRequestError
from Exception import SDSInitError, SDSError
from Exception import SDSServiceError, SDSRequestError
# pylint: enable=F0401
else:
from .mapper import SERVICE_MAPPER, METHOD_MAPPER
from .Exception import SSDInitError, SSDError
from .Exception import SSDServiceError, SSDRequestError
from .Exception import SDSInitError, SDSError
from .Exception import SDSServiceError, SDSRequestError
__all__ = ["SOLIDserverRest"]
......@@ -46,13 +46,13 @@ __all__ = ["SOLIDserverRest"]
# effectively few variables in this class, just disabling the warning
# pylint: disable=R0902
class SOLIDserverRest:
""" main SSD class """
""" main SDS class """
CNX_NATIVE = 1
CNX_APIKEY = 2
CNX_BASIC = 3
def __init__(self, host, debug=False):
""" initialize connection with SSD host,
""" initialize connection with SDS host,
this function is not active,
just set host and parameters
"""
......@@ -94,15 +94,15 @@ class SOLIDserverRest:
def __del__(self):
self.clean()
def use_native_ssd(self, user, password):
""" propose to use a native EfficientIP SSD connection with Username
def use_native_sds(self, user, password):
""" propose to use a native EfficientIP SDS connection with Username
and password encoded in the headers of each requests
"""
logging.debug("useNativeSSD %s %s", user, password)
logging.debug("useNativeSDS %s %s", user, password)
# check if SSD connection is established
# check if SDS connection is established
if self.host is None:
raise SSDInitError()
raise SDSInitError()
self.user = user
self.password = password
......@@ -115,14 +115,14 @@ class SOLIDserverRest:
'content-type': 'application/json'
}
def use_basicauth_ssd(self, user, password):
def use_basicauth_sds(self, user, password):
""" propose to use the basic auth implementation on the SDS
"""
logging.debug("useBasicAuthSSD %s %s", user, password)
logging.debug("useBasicAuthSDS %s %s", user, password)
# check if SSD connection is established
# check if SDS connection is established
if self.host is None:
raise SSDInitError()
raise SDSInitError()
self.user = user
self.password = password
......@@ -142,10 +142,10 @@ class SOLIDserverRest:
file_content)
except IOError:
logging.error("cannot load CA file")
raise SSDInitError("cannot load CA file {}".format(file_path))
raise SDSInitError("cannot load CA file {}".format(file_path))
except crypto.Error as error:
logging.error(error)
raise SSDInitError("invalid CA file {}".format(file_path))
raise SDSInitError("invalid CA file {}".format(file_path))
self.session.verify = file_path
self.ssl_verify = True
......@@ -156,7 +156,7 @@ class SOLIDserverRest:
self.ssl_verify = value
else:
logging.error("bad type when calling set_ssl_verify")
raise SSDError("requested bool on set_ssl_verify")
raise SDSError("requested bool on set_ssl_verify")
def query(self, service,
params=None,
......@@ -184,7 +184,7 @@ class SOLIDserverRest:
if method is None:
msg = "no method available for request {}".format(service)
logging.error("no method available for request %s", service)
raise SSDServiceError(service,
raise SDSServiceError(service,
message=msg)
logging.debug("method %s selected for service %s", method, service)
......@@ -199,7 +199,7 @@ class SOLIDserverRest:
svc_mapped = SERVICE_MAPPER.get(service)
if svc_mapped is None:
logging.error("unknown service %s", service)
raise SSDServiceError(service)
raise SDSServiceError(service)
self.last_url = "{}{}".format(svc_mapped, params).strip()
url = "{}{}".format(self.prefix_url, self.last_url)
......@@ -220,19 +220,19 @@ class SOLIDserverRest:
timeout=timeout,
auth=self.auth)
except requests.exceptions.SSLError:
raise SSDRequestError(method,
raise SDSRequestError(method,
url,
self.headers,
message="SSL certificate error")
except BaseException as error:
raise SSDRequestError(method, url, self.headers, message=error)
raise SDSRequestError(method, url, self.headers, message=error)
def get_headers(self):
""" returns the headers attached to this connection """
return self.headers
def get_status(self):
""" returns status of the SSD connection """
""" returns status of the SDS connection """
_r = {
'host': self.host,
'python_version': self.python_version
......@@ -240,7 +240,7 @@ class SOLIDserverRest:
return _r
def clean(self):
""" clean all status of the SSD connection """
""" clean all status of the SDS connection """
self.auth = None
self.cnx_type = None
self.debug = None
......@@ -259,3 +259,15 @@ class SOLIDserverRest:
_s = "SOLIDserverRest: API={}, user={}"
return(_s.format(self.prefix_url,
self.user))
# deprecated method to be suppressed
def use_native_ssd(self, user, password):
"""deprecated version of use_native_sds"""
logging.critical("deprecated method use_native_ssd")
self.use_native_sds(user, password)
def use_basicauth_ssd(self, user, password):
"""deprecated version of use_basicauth_ssd"""
logging.critical("deprecated method use_basicauth_ssd")
self.use_basicauth_sds(user, password)
# -*- Mode: Python; python-indent-offset: 4 -*-
#
# Time-stamp: <2019-06-23 18:37:03 alex>
# Time-stamp: <2019-07-06 18:57:14 alex>
#
# only for python v3
......@@ -13,7 +13,7 @@ import ipaddress
from json.decoder import JSONDecodeError
from SOLIDserverRest.Exception import SSDInitError, SSDAuthError
from SOLIDserverRest.Exception import SDSInitError, SDSAuthError
from SOLIDserverRest.Exception import SDSEmptyError
from SOLIDserverRest import SOLIDserverRest
......@@ -22,8 +22,11 @@ from .class_params import ClassParams
__all__ = ["SDS"]
# more than 7 arguments to class
# pylint: disable=R0902
class SDS(ClassParams):
""" class to get connected to a SDS server """
# ---------------------------
def __init__(self, ip_address=None, user=None, pwd=None):
"""init the SDS object:
......@@ -45,6 +48,7 @@ class SDS(ClassParams):
self.check_certificate = False
self.sds = None
self.timeout = 1
# ---------------------------
def set_server_ip(self, ip_address):
......@@ -52,7 +56,7 @@ class SDS(ClassParams):
try:
ipaddress.IPv4Address(ip_address)
except ipaddress.AddressValueError as error:
raise SSDInitError(message="IPv4 address of server error: {}".
raise SDSInitError(message="IPv4 address of server error: {}".
format(error))
self.sds_ip = ip_address
......@@ -62,12 +66,12 @@ class SDS(ClassParams):
"""add user and login to credentials of this session"""
if user is None or pwd is None:
msg = "missing user or password in credentials"
raise SSDInitError(message=msg)
raise SDSInitError(message=msg)
self.user = user
self.pwd = pwd
# ---------------------------
def connect(self, method="basicauth", cert_file_path=None):
def connect(self, method="basicauth", cert_file_path=None, timeout=1):
"""connects to SOLIDserver and check everything is OK by
querying the version of the admin node in the member list
......@@ -77,17 +81,17 @@ class SDS(ClassParams):
"""
if self.user is None or self.pwd is None:
msg = "missing user or password in credentials for connect"
raise SSDInitError(message=msg)
raise SDSInitError(message=msg)
if self.sds_ip is None:
raise SSDInitError(message="missing ip for server for connect")
raise SDSInitError(message="missing ip for server for connect")
self.sds = SOLIDserverRest(self.sds_ip)
if method == "basicauth":
self.sds.use_basicauth_ssd(self.user, self.pwd)
self.sds.use_basicauth_sds(self.user, self.pwd)
self.auth_method = "basic auth"
elif method == "native":
self.sds.use_native_ssd(self.user, self.pwd)
self.sds.use_native_sds(self.user, self.pwd)
self.auth_method = "native"
# certificate & SSL check
......@@ -97,10 +101,13 @@ class SDS(ClassParams):
else:
self.sds.set_ssl_verify(False)
if timeout != self.timeout:
self.timeout = timeout
self.version = self.get_version()
if self.version is None:
raise SSDInitError(message="version of SOLIDserver not found")
raise SDSInitError(message="version of SOLIDserver not found")
# ---------------------------
def get_version(self):
......@@ -115,8 +122,7 @@ class SDS(ClassParams):
params={
'WHERE': 'member_is_me=1',
},
option=False,
timeout=2)
option=False)
if j is None:
logging.error("error in getting answer on version")
......@@ -133,14 +139,18 @@ class SDS(ClassParams):
def query(self, method, params=None, option=False, timeout=1):
"""execute a query towards the SDS"""
_timeout = self.timeout
if timeout not in (self.timeout, 1):
_timeout = timeout
try:
answer_req = self.sds.query(method,
params=params,
option=option,
timeout=timeout)
timeout=_timeout)
if answer_req.status_code == 401:
raise SSDAuthError(message="authentication error")
raise SDSAuthError(message="authentication error")
if answer_req.status_code == 204:
raise SDSEmptyError(message="204 returned")
......@@ -152,8 +162,8 @@ class SDS(ClassParams):
logging.error("no json in return")
return None
except SSDAuthError as error:
raise SSDAuthError("{}".format(error))
except SDSAuthError as error:
raise SDSAuthError("{}".format(error))
return None
......
#
# -*- Mode: Python; python-indent-offset: 4 -*-
#
# Time-stamp: <2019-06-23 18:40:26 alex>
# Time-stamp: <2019-07-06 18:47:28 alex>
#
"""
......@@ -10,7 +10,7 @@ SOLIDserver space management
# import logging
from SOLIDserverRest.Exception import SSDInitError, SSDError
from SOLIDserverRest.Exception import SDSInitError, SDSError
from SOLIDserverRest.Exception import SDSEmptyError
from .class_params import ClassParams
......@@ -43,26 +43,44 @@ class Space(ClassParams):
super(Space, self).__init__()
def refresh(self):
"""refresh content of the object from the SDS"""
def create(self):
"""creates the space in the SDS"""
if self.sds is None:
raise SSDInitError(message="not connected")
raise SDSInitError(message="not connected")
space_id = self._get_siteid_by_name(self.name)
if space_id is not None:
raise SDSError(message="already existant space")
def _get_siteid_by_name(self, name):
"""get the space ID from its name, return None if non existant"""
try:
rjson = self.sds.query("ip_site_list",
params={
"WHERE": "site_name='{}'".
format(self.name),
format(name),
})
except SDSEmptyError:
raise SDSEmptyError(message="non existant space")
return None
if rjson[0]['errno'] != '0':
raise SSDError("errno raised")
raise SDSError("errno raised")
return rjson[0]['site_id']
def refresh(self):
"""refresh content of the object from the SDS"""
if self.sds is None:
raise SDSInitError(message="not connected")
space_id = self._get_siteid_by_name(self.name)
if space_id is None:
raise SDSEmptyError(message="non existant space")
rjson = self.sds.query("ip_site_info",
params={
"site_id": rjson[0]['site_id'],
"site_id": space_id,
})
rjson = rjson[0]
......@@ -79,7 +97,7 @@ class Space(ClassParams):
'row_enabled',
'multistatus']:
if label not in rjson:
raise SSDError("parameter {} not found in space".format(label))
raise SDSError("parameter {} not found in space".format(label))
self.space_params[label] = rjson[label]
if 'site_class_parameters' in rjson:
......
......@@ -49,7 +49,7 @@ def test_no_server():
testR.use_native_ssd('soliduser', 'solidpass')
print('Test = NO-OK')
except SSDError:
except SDSError:
print('Test = OK')
None
......@@ -76,7 +76,7 @@ def test_few_services():
print('Answer: {}'.format(answerR))
print('Answer: {}'.format(answerR.status_code))
print('Answer: {}'.format(answerR.content))
except SSDError:
except SDSError:
None
#ip_site_update
......@@ -91,7 +91,7 @@ def test_few_services():
print('Answer: {}'.format(answerR))
print('Answer: {}'.format(answerR.status_code))
print('Answer: {}'.format(answerR.content))
except SSDError:
except SDSError:
None
def test_auto_dico():
......@@ -124,7 +124,7 @@ def test_auto_dico():
requierement = value["mandatory_addition_params"]
print('Requierement parameters: {}'.format(value["mandatory_addition_params"]))
except SSDError:
except SDSError:
None
#Parameters preparation
......@@ -141,7 +141,7 @@ def test_auto_dico():
print('Answer: {}'.format(answerR.status_code))
print('Answer:')
print(json.dumps(json.loads(answerR.content), indent=4, sort_keys=True, encoding="utf-8"))
except SSDError:
except SDSError:
None
display_test = display_test + '*'
......@@ -154,4 +154,4 @@ def test_auto_dico():
readme_writting()
print('******************************')
print('* END documentation writting *')
print('******************************')
\ No newline at end of file
print('******************************')
......@@ -15,7 +15,7 @@ print('your_obj = SOLIDServerRest("host", "user", "password")')
con = SOLIDserverRest('192.168.56.254')
try:
con.useNativeSSD(user="api", password="admin")
except SSDInitError:
except SDSInitError:
exit()
print("-------------------")
......
# -*- Mode: Python; python-indent-offset: 4 -*-
#
# Time-stamp: <2019-06-23 16:09:13 alex>
# Time-stamp: <2019-07-06 18:50:34 alex>
#
"""test file for the eip advance suite package, require an SDS to be available
......@@ -10,7 +10,7 @@ this file is used to tdd"""
import logging
import sys
from SOLIDserverRest.Exception import SSDInitError, SSDRequestError, SSDAuthError, SSDError
from SOLIDserverRest.Exception import SDSInitError, SDSRequestError, SDSAuthError, SDSError
if sys.version_info[0] == 3:
try:
......@@ -48,18 +48,18 @@ def test_create_sds_bad_ip():
sds = sdsadv.SDS()
try:
sds.set_server_ip('192.168.56.254.test')
except SSDInitError:
except SDSInitError:
return
assert None, "should have raised an error on bad IP address"
def _test_create_sds_bad_not_responsive_ip():
def test_create_sds_bad_not_responsive_ip():
"""create a connection to a SOLIDserver with a bad ip address format"""
sds = sdsadv.SDS()
sds.set_server_ip('192.168.56.1')
sds.set_credentials(user=USER, pwd=PWD)
try:
sds.connect()
except SSDError:
sds.connect(timeout=0.01)
except SDSError:
None
def test_create_sds_withuser():
......@@ -75,7 +75,7 @@ def test_create_sds_set_user():
try:
sds.set_credentials()
assert None, "should have raised an error on no credentials"
except SSDInitError:
except SDSInitError:
None
sds.set_credentials(user=USER, pwd=PWD)
......@@ -88,7 +88,7 @@ def test_create_sds_set_baduser():
sds.set_credentials(user=USER, pwd="error")
try:
sds.connect()
except SSDAuthError as e:
except SDSAuthError as e:
return
assert None, "should have raised an error on bad credentials"
......@@ -99,7 +99,7 @@ def test_create_sds_connect_wo_user():
sds.set_server_ip(SERVER)
try:
sds.connect()
except SSDInitError as e:
except SDSInitError as e:
return
assert None, "should have raised an error on no credentials"
......@@ -110,7 +110,7 @@ def test_create_sds_connect_wo_ip():
sds.set_credentials(user=USER, pwd=PWD)
try:
sds.connect()
except SSDInitError as e:
except SDSInitError as e:
return
assert None, "should have raised an error on no ip"
......@@ -134,10 +134,10 @@ def test_basic_auth_with_cert():
sds.set_credentials(user=USER, pwd=PWD)
try:
sds.connect(method="basicauth", cert_file_path="ca.crt")
except SSDRequestError:
except SDSRequestError:
logging.debug(e)
assert None, "certifiate validation error"
except SSDInitError as e:
except SDSInitError as e:
logging.debug(e)
assert None, "connection error, probable certificate issue"
......
# -*- Mode: Python; python-indent-offset: 4 -*-
#
# Time-stamp: <2019-06-23 15:10:48 alex>
# Time-stamp: <2019-07-06 18:53:30 alex>
#
"""test file for the eip advance suite package, base class object"""
......@@ -8,7 +8,7 @@
import logging
import sys
from SOLIDserverRest.Exception import SSDInitError, SSDRequestError, SSDAuthError, SSDError
from SOLIDserverRest.Exception import SDSInitError, SDSRequestError, SDSAuthError, SDSError
try:
from tests.data_sample import *
......
# -*- Mode: Python; python-indent-offset: 4 -*-
#
# Time-stamp: <2019-06-23 18:26:01 alex>
# Time-stamp: <2019-07-06 18:56:38 alex>
#
"""test file for the eip advance suite package, require an SDS to be available
......@@ -10,7 +10,9 @@ this file is used to tdd"""
import logging
import sys
from SOLIDserverRest.Exception import SSDInitError, SSDRequestError, SSDAuthError, SSDError
from SOLIDserverRest.Exception import SDSInitError, SDSRequestError
from SOLIDserverRest.Exception import SDSAuthError, SDSError
from SOLIDserverRest.Exception import SDSEmptyError
try:
from tests.data_sample import *
......@@ -20,7 +22,7 @@ except:
from .context import sdsadv
def test_set_space_empty():
"""create a space object and populate it with the content of the SDS"""
"""create a space object"""
space = sdsadv.Space()
obj_string = str(space)
logging.debug(obj_string)
......@@ -30,7 +32,7 @@ def test_refresh_space_not_connected():
space = sdsadv.Space()
try:
space.refresh()
except SSDInitError:
except SDSInitError:
return
assert None, "refresh on non connected space should raise an exc"
......@@ -42,10 +44,10 @@ def test_refresh_space_local():
sds.set_credentials(user=USER, pwd=PWD)
try:
sds.connect(method="basicauth", cert_file_path="ca.crt")
except SSDRequestError:
except SDSRequestError:
logging.debug(e)
assert None, "certifiate validation error"
except SSDInitError as e:
except SDSInitError as e:
logging.debug(e)
assert None, "connection error, probable certificate issue"
......@@ -59,10 +61,10 @@ def test_refresh_space_with_classparams():
sds.set_credentials(user=USER, pwd=PWD)
try:
sds.connect(method="basicauth", cert_file_path="ca.crt")
except SSDRequestError:
except SDSRequestError:
logging.debug(e)
assert None, "certifiate validation error"
except SSDInitError as e:
except SDSInitError as e:
logging.debug(e)
assert None, "connection error, probable certificate issue"
......@@ -76,17 +78,17 @@ def test_refresh_space_not_found():
sds.set_credentials(user=USER, pwd=PWD)
try:
sds.connect(method="basicauth", cert_file_path="ca.crt")
except SSDRequestError:
except SDSRequestError:
logging.debug(e)
assert None, "certifiate validation error"
except SSDInitError as e:
except SDSInitError as e:
logging.debug(e)
assert None, "connection error, probable certificate issue"
space = sdsadv.Space(sds=sds, name="not_known")
try:
space.refresh()
except SSDError:
except SDSEmptyError:
return
assert None, "should not be able to refresh unknown space"
......
......@@ -41,8 +41,8 @@ def test_setup_woinit():
con = SOLIDserverRest(SERVER)
con.clean()
try:
con.use_native_ssd(user=USER, password=PWD)
except SSDInitError:
con.use_native_sds(user=USER, password=PWD)
except SDSInitError:
return
con.clean()
......@@ -50,9 +50,9 @@ def test_setup_woinit():
def test_setup_simple():
""" simple setup of SSD native connection """
""" simple setup of SDS native connection """
con = SOLIDserverRest(SERVER)
con.use_native_ssd(user=USER, password=PWD)
con.use_native_sds(user=USER, password=PWD)
s = con.get_status()
if 'python_version' not in s:
......@@ -66,9 +66,9 @@ def test_setup_simple():
def test_native_simple_call():
""" simple call with SSD native connection """
""" simple call with SDS native connection """
con = SOLIDserverRest(SERVER)