x/synergy-service
Code Issues Proposed changes

Synergy should never raise Exception directly

Browse Source 
The basic Exception can be used anywhere in the code but however, no
program nor library should ever raise Exception directly: it's not
specific enough to be helpful. This fix better handles the exceptions by
removing the Exception occurencies and adding a new SynergyError type.

Bug: #1690795
Change-Id: I202e063198ee9aef7397bad9b8398c24d52b5fe1
Sem-Ver: bugfix
This commit is contained in:
Lisa Zangrando 2017-05-15 15:04:39 +02:00
parent b542db3beb
commit a8c06a001c
11 changed files with 137 additions and 562 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
synergy/client/exception.py Normal file
View File

@ -0,0 +1,21 @@
__author__ = "Lisa Zangrando"
__email__ = "lisa.zangrando[AT]pd.infn.it"
__copyright__ = """Copyright (c) 2015 INFN - INDIGO-DataCloud
All Rights Reserved
Licensed under the Apache License, Version 2.0;
you may not use this file except in compliance with the
License. You may obtain a copy of the License at:
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
either express or implied.
See the License for the specific language governing
permissions and limitations under the License."""
class SynergyError(Exception):
pass

430
synergy/client/keystone_v3.py
View File

@ -3,6 +3,7 @@ import os.path
import requests
from datetime import datetime
from synergy.client.exception import SynergyError
__author__ = "Lisa Zangrando"
@ -24,63 +25,6 @@ See the License for the specific language governing
permissions and limitations under the License."""
class Trust(object):
def __init__(self, data):
data = data["trust"]
self.id = data["id"]
self.impersonations = data["impersonation"]
self.roles_links = data["roles_links"]
self.trustor_user_id = data["trustor_user_id"]
self.trustee_user_id = data["trustee_user_id"]
self.links = data["links"]
self.roles = data["roles"]
self.remaining_uses = data["remaining_uses"]
self.expires_at = None
if data["expires_at"] is not None:
self.expires_at = datetime.strptime(data["expires_at"],
"%Y-%m-%dT%H:%M:%S.%fZ")
self.project_id = data["project_id"]
def getId(self):
return self.id
def isImpersonations(self):
return self.impersonations
def getRolesLinks(self):
return self.roles_links
def getTrustorUserId(self):
return self.trustor_user_id
def getTrusteeUserId(self):
return self.trustee_user_id
def getlinks(self):
return self.links
def getProjectId(self):
return self.project_id
def getRoles(self):
return self.roles
def getRemainingUses(self):
return self.remaining_uses
def getExpiration(self):
return self.expires_at
def isExpired(self):
if self.getExpiration() is None:
return False
return self.getExpiration() < datetime.utcnow()
class Token(object):
def __init__(self, token, data):
@ -99,16 +43,8 @@ class Token(object):
if "extras" in data:
self.extras = data["extras"]
def getCatalog(self, service_name=None, interface="public"):
if service_name:
for service in self.catalog:
if service["name"] == service_name:
for endpoint in service["endpoints"]:
if endpoint["interface"] == interface:
return endpoint
return None
else:
return self.catalog
def getCatalog(self):
return self.catalog
def getExpiration(self):
return self.expires_at
@ -173,7 +109,7 @@ class Token(object):
return Token(data["id"], data)
# if the file is empty the ValueError will be thrown
except ValueError as ex:
raise ex
raise SynergyError(ex)
def isotime(self, at=None, subsecond=False):
"""Stringify time in ISO 8601 format."""
@ -193,66 +129,13 @@ class Token(object):
st += ('Z' if tz == 'UTC' else tz)
return st
"""The trustor or grantor of a trust is the person who creates the trust.
The trustor is the one who contributes property to the trust.
The trustee is the person who manages the trust, and is usually appointed
by the trustor. The trustor is also often the trustee in living trusts.
"""
def trust(self, trustee_user, expires_at=None,
project_id=None, roles=None, impersonation=True):
if self.isExpired():
raise Exception("token expired!")
headers = {"Content-Type": "application/json",
"Accept": "application/json",
"User-Agent": "python-novaclient",
"X-Auth-Token": self.getId()}
if roles is None:
roles = self.getRoles()
if project_id is None:
project_id = self.getProject().get("id")
data = {}
data["trust"] = {"impersonation": impersonation,
"project_id": project_id,
"roles": roles,
"trustee_user_id": trustee_user,
"trustor_user_id": self.getUser().get("id")}
if expires_at is not None:
data["trust"]["expires_at"] = self.isotime(expires_at, True)
endpoint = self.getCatalog(service_name="keystone")
if not endpoint:
raise Exception("keystone endpoint not found!")
if "v2.0" in endpoint["url"]:
endpoint["url"] = endpoint["url"].replace("v2.0", "v3")
response = requests.post(url=endpoint["url"] + "/OS-TRUST/trusts",
headers=headers,
data=json.dumps(data))
if response.status_code != requests.codes.ok:
response.raise_for_status()
if not response.text:
raise Exception("trust token failed!")
return Trust(response.json())
class KeystoneClient(object):
def __init__(self, auth_url, username, password,
user_domain_id=None,
def __init__(self, auth_url, username, password, user_domain_id=None,
user_domain_name="default", project_id=None,
project_name=None, project_domain_id=None,
project_domain_name="default", timeout=None,
default_trust_expiration=None, ca_cert=None):
project_domain_name="default", timeout=None, ca_cert=None):
self.auth_url = auth_url
self.username = username
self.password = password
@ -266,11 +149,6 @@ class KeystoneClient(object):
self.token = None
self.ca_cert = ca_cert
if default_trust_expiration:
self.default_trust_expiration = default_trust_expiration
else:
self.default_trust_expiration = 24
def authenticate(self):
if self.token is not None:
if self.token.isExpired():
@ -323,297 +201,25 @@ class KeystoneClient(object):
response.raise_for_status()
if not response.text:
raise Exception("authentication failed!")
# print(response.__dict__)
raise SynergyError("authentication failed!")
token_subject = response.headers["X-Subject-Token"]
token_data = response.json()
self.token = Token(token_subject, token_data)
def getUser(self, id):
try:
response = self.getResource("users/%s" % id, "GET")
except requests.exceptions.HTTPError as ex:
response = ex.response.json()
raise Exception("error on retrieving the user info (id=%r): %s"
% (id, response["error"]["message"]))
def getService(self, name):
for service in self.token.getCatalog():
if service["name"] == name:
return service
if response:
response = response["user"]
raise SynergyError("service %s not found!" % name)
return response
def getEndpoint(self, name, interface="public"):
service = self.getService(name)
def getUsers(self):
try:
response = self.getResource("users", "GET")
except requests.exceptions.HTTPError as ex:
response = ex.response.json()
raise Exception("error on retrieving the users list: %s"
% response["error"]["message"])
for endpoint in service["endpoints"]:
if endpoint["interface"] == interface:
return endpoint
if response:
response = response["users"]
return response
def getUserProjects(self, id):
try:
response = self.getResource("users/%s/projects" % id, "GET")
except requests.exceptions.HTTPError as ex:
response = ex.response.json()
raise Exception("error on retrieving the users's projects "
"(id=%r): %s" % (id, response["error"]["message"]))
if response:
response = response["projects"]
return response
def getUserRoles(self, user_id, project_id):
try:
response = self.getResource("/projects/%s/users/%s/roles"
% (project_id, user_id), "GET")
except requests.exceptions.HTTPError as ex:
response = ex.response.json()
raise Exception("error on retrieving the user's roles (usrId=%r, "
"prjId=%r): %s" % (user_id,
project_id,
response["error"]["message"]))
if response:
response = response["roles"]
return response
def getProject(self, id):
try:
response = self.getResource("/projects/%s" % id, "GET")
except requests.exceptions.HTTPError as ex:
response = ex.response.json()
raise Exception("error on retrieving the project (id=%r): %s"
% (id, response["error"]["message"]))
if response:
response = response["project"]
return response
def getProjects(self):
try:
response = self.getResource("/projects", "GET")
except requests.exceptions.HTTPError as ex:
response = ex.response.json()
raise Exception("error on retrieving the projects list: %s"
% response["error"]["message"])
if response:
response = response["projects"]
return response
def getRole(self, id):
try:
response = self.getResource("/roles/%s" % id, "GET")
except requests.exceptions.HTTPError as ex:
response = ex.response.json()
raise Exception("error on retrieving the role info (id=%r): %s"
% (id, response["error"]["message"]))
if response:
response = response["role"]
return response
def getRoles(self):
try:
response = self.getResource("/roles", "GET")
except requests.exceptions.HTTPError as ex:
response = ex.response.json()
raise Exception("error on retrieving the roles list: %s"
% response["error"]["message"])
if response:
response = response["roles"]
return response
def getToken(self):
self.authenticate()
return self.token
def deleteToken(self, id):
if self.token is None:
return
headers = {"Content-Type": "application/json",
"Accept": "application/json",
"User-Agent": "python-novaclient",
"X-Auth-Project-Id": self.token.getProject()["name"],
"X-Auth-Token": self.token.getId(),
"X-Subject-Token": id}
response = requests.delete(url=self.auth_url + "/auth/tokens",
headers=headers,
timeout=self.timeout,
verify=self.ca_cert)
self.token = None
if response.status_code != requests.codes.ok:
response.raise_for_status()
def validateToken(self, id):
self.authenticate()
headers = {"Content-Type": "application/json",
"Accept": "application/json",
"User-Agent": "python-novaclient",
"X-Auth-Project-Id": self.token.getProject()["name"],
"X-Auth-Token": self.token.getId(),
"X-Subject-Token": id}
response = requests.get(url=self.auth_url + "/auth/tokens",
headers=headers,
timeout=self.timeout,
verify=self.ca_cert)
if response.status_code != requests.codes.ok:
response.raise_for_status()
if not response.text:
raise Exception("token not found!")
token_subject = response.headers["X-Subject-Token"]
token_data = response.json()
return Token(token_subject, token_data)
def getEndpoint(self, id=None, service_id=None):
if id:
try:
response = self.getResource("/endpoints/%s" % id, "GET")
except requests.exceptions.HTTPError as ex:
response = ex.response.json()
raise Exception("error on retrieving the endpoint (id=%r): %s"
% (id, response["error"]["message"]))
if response:
response = response["endpoint"]
return response
elif service_id:
try:
endpoints = self.getEndpoints()
except requests.exceptions.HTTPError as ex:
response = ex.response.json()
raise Exception("error on retrieving the endpoints list"
"(serviceId=%r): %s"
% (service_id, response["error"]["message"]))
if endpoints:
for endpoint in endpoints:
if endpoint["service_id"] == service_id:
return endpoint
return None
def getEndpoints(self):
try:
response = self.getResource("/endpoints", "GET")
except requests.exceptions.HTTPError as ex:
response = ex.response.json()
raise Exception("error on retrieving the endpoints list: %s"
% response["error"]["message"])
if response:
response = response["endpoints"]
return response
def getService(self, id=None, name=None):
if id:
try:
response = self.getResource("/services/%s" % id, "GET")
except requests.exceptions.HTTPError as ex:
response = ex.response.json()
raise Exception("error on retrieving the service info (id=%r)"
": %s" % (id, response["error"]["message"]))
if response:
response = response["service"]
return response
elif name:
services = self.getServices()
if services:
for service in services:
if service["name"] == name:
return service
return None
def getServices(self):
try:
response = self.getResource("/services", "GET")
except requests.exceptions.HTTPError as ex:
response = ex.response.json()
raise Exception("error on retrieving the services list: %s"
% response["error"]["message"])
if response:
response = response["services"]
return response
def getResource(self, resource, method, data=None):
self.authenticate()
url = self.auth_url + "/" + resource
headers = {"Content-Type": "application/json",
"Accept": "application/json",
"User-Agent": "python-novaclient",
"X-Auth-Project-Id": self.token.getProject()["name"],
"X-Auth-Token": self.token.getId()}
if method == "GET":
response = requests.get(url,
headers=headers,
params=data,
timeout=self.timeout,
verify=self.ca_cert)
elif method == "POST":
response = requests.post(url,
headers=headers,
data=json.dumps(data),
timeout=self.timeout,
verify=self.ca_cert)
elif method == "PUT":
response = requests.put(url,
headers=headers,
data=json.dumps(data),
timeout=self.timeout,
verify=self.ca_cert)
elif method == "HEAD":
response = requests.head(url,
headers=headers,
data=json.dumps(data),
timeout=self.timeout,
verify=self.ca_cert)
elif method == "DELETE":
response = requests.delete(url,
headers=headers,
data=json.dumps(data),
timeout=self.timeout,
verify=self.ca_cert)
else:
raise Exception("wrong HTTP method: %s" % method)
if response.status_code != requests.codes.ok:
response.raise_for_status()
if response.text:
return response.json()
else:
return None
raise SynergyError("endpoint for service %s not found!" % name)

17
synergy/client/shell.py
View File

@ -129,16 +129,16 @@ def main():
synergy_url = bypass_url
else:
if not os_username:
raise Exception("'os-username' not defined!")
raise ValueError("'os-username' not defined!")
if not os_password:
raise Exception("'os-password' not defined!")
raise ValueError("'os-password' not defined!")
if not os_project_name:
raise Exception("'os-project-name' not defined!")
raise ValueError("'os-project-name' not defined!")
if not os_auth_url:
raise Exception("'os-auth-url' not defined!")
raise ValueError("'os-auth-url' not defined!")
if not os_user_domain_name:
os_user_domain_name = "default"
@ -159,14 +159,7 @@ def main():
client.authenticate()
synergy_service = client.getService(name="synergy")
if not synergy_service:
print("Synergy service not found into the Keystone catalog!")
sys.exit(1)
synergy_endpoint = client.getEndpoint(
service_id=synergy_service["id"])
synergy_endpoint = client.getEndpoint("synergy")
synergy_url = synergy_endpoint["url"]

12
synergy/common/manager.py
View File

@ -1,4 +1,7 @@
import logging
from serializer import SynergyObject
from synergy.exception import SynergyError
from threading import Condition
from threading import Event
from threading import Thread
@ -22,6 +25,8 @@ either express or implied.
See the License for the specific language governing
permissions and limitations under the License."""
LOG = logging.getLogger(__name__)
class Manager(SynergyObject, Thread):
@ -133,5 +138,8 @@ class Manager(SynergyObject, Thread):
try:
self.task()
self.condition.wait(self.getRate() * 60)
except Exception as ex:
print("task %r: %s" % (self.getName(), ex))
except NotImplementedError:
LOG.debug("task() not implemented by %s"
% self.getName())
except SynergyError as ex:
LOG.error("task %s: %s" % (self.getName(), ex))

22
synergy/common/serializer.py
View File

@ -1,6 +1,8 @@
from datetime import datetime
from json import JSONEncoder
from synergy.common import utils
from synergy.exception import SynergyError
__author__ = "Lisa Zangrando"
__email__ = "lisa.zangrando[AT]pd.infn.it"
@ -66,26 +68,26 @@ class SynergyObject(object):
@classmethod
def deserialize(cls, entity):
if "synergy_object" not in entity:
raise Exception("it seems not a Synergy object!")
raise SynergyError("it seems not a Synergy object!")
synergy_object = entity["synergy_object"]
if "namespace" not in synergy_object:
raise Exception("synergy_object.namespace not defined!")
raise SynergyError("synergy_object.namespace not defined!")
if "name" not in synergy_object:
raise Exception("synergy_object.name not defined!")
raise SynergyError("synergy_object.name not defined!")
if "version" not in synergy_object:
raise Exception("synergy_object.version mismatch!")
raise SynergyError("synergy_object.version mismatch!")
if synergy_object["version"] != cls.VERSION:
raise Exception("synergy_object.version mis!")
raise SynergyError("synergy_object.version mis!")
if synergy_object["namespace"] != "synergy":
raise Exception("unsupported object objtype='%s.%s"
% (synergy_object["namespace"],
synergy_object["name"]))
raise SynergyError("unsupported object objtype='%s.%s"
% (synergy_object["namespace"],
synergy_object["name"]))
objInstance = None
@ -94,8 +96,8 @@ class SynergyObject(object):
objClass = utils.import_class(objName)
objInstance = objClass()
except Exception as ex:
raise Exception("error on deserializing the object %r: %s"
% (objName, ex))
raise SynergyError("error on deserializing the object %r: %s"
% (objName, ex))
del entity["synergy_object"]

6
synergy/common/utils.py
View File

@ -2,6 +2,8 @@ import sys
import traceback
from datetime import datetime
from synergy.exception import SynergyError
__author__ = "Lisa Zangrando"
__email__ = "lisa.zangrando[AT]pd.infn.it"
@ -30,7 +32,7 @@ def import_class(import_str):
try:
return getattr(sys.modules[mod_str], class_str)
except AttributeError:
raise ImportError(
raise SynergyError(
'Class %s cannot be found (%s)' %
(class_str, traceback.format_exception(*sys.exc_info())))
@ -52,7 +54,7 @@ def objectHookHandler(json_dict):
objClass = import_class(synergy_object["name"])
objInstance = objClass()
return objInstance.deserialize(json_dict)
except Exception as ex:
except SynergyError as ex:
raise ex
else:
return json_dict

7
synergy/common/wsgi.py
View File

@ -9,7 +9,7 @@ import time
from eventlet import greenio as eventlet_greenio
from eventlet import wsgi as eventlet_wsgi
from synergy.exception import SynergyError
from sys import exc_info
from traceback import format_tb
@ -216,9 +216,8 @@ class Server(object):
family = info[0]
bind_addr = info[-1]
except Exception as ex:
LOG.error("Unable to listen on %s:%s: %s"
% (self.host_name, self.host_port, ex))
raise ex
raise SynergyError("Unable to listen on %s:%s: %s"
% (self.host_name, self.host_port, ex))
retry_until = time.time() + self.retry_until_window
exception = None

4
synergy/examples/timer_manager.py
View File

@ -2,6 +2,8 @@ import logging
import time
from synergy.common.manager import Manager
from synergy.exception import SynergyError
__author__ = "Lisa Zangrando"
__email__ = "lisa.zangrando[AT]pd.infn.it"
@ -38,7 +40,7 @@ class TimerManager(Manager):
if command == "GET_TIME":
return {"localtime": time.asctime(time.localtime(time.time()))}
else:
raise Exception("command %r not supported" % command)
raise SynergyError("command %r not supported" % command)
def destroy(self):
LOG.info("%s destroy invoked!" % (self.name))

21
synergy/exception.py Normal file
View File

@ -0,0 +1,21 @@
__author__ = "Lisa Zangrando"
__email__ = "lisa.zangrando[AT]pd.infn.it"
__copyright__ = """Copyright (c) 2015 INFN - INDIGO-DataCloud
All Rights Reserved
Licensed under the Apache License, Version 2.0;
you may not use this file except in compliance with the
License. You may obtain a copy of the License at:
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
either express or implied.
See the License for the specific language governing
permissions and limitations under the License."""
class SynergyError(Exception):
pass

61
synergy/service.py
View File

@ -16,6 +16,7 @@ from synergy.common.manager import Manager
from synergy.common.serializer import SynergyEncoder
from synergy.common.service import Service
from synergy.common.wsgi import Server
from synergy.exception import SynergyError
__author__ = "Lisa Zangrando"
@ -92,7 +93,7 @@ class Synergy(Service):
self.wsgi_server = None
for entry in iter_entry_points(MANAGER_ENTRY_POINT):
LOG.info("loading manager %r", entry.name)
LOG.info("loading manager %s", entry.name)
try:
CONF.register_opts(config.manager_opts, group=entry.name)
@ -111,28 +112,32 @@ class Synergy(Service):
self.managers[manager_obj.getName()] = manager_obj
CONF.register_opts(manager_obj.getOptions(), group=entry.name)
except Exception as ex:
except cfg.Error as ex:
LOG.error("Exception has occured", exc_info=1)
LOG.error("manager %r instantiation error: %s"
LOG.error("manager %s instantiation error: %s"
% (entry.name, ex))
raise Exception("manager %r instantiation error: %s"
% (entry.name, ex))
raise SynergyError("manager %s instantiation error: %s"
% (entry.name, ex))
for name, manager in self.managers.items():
manager.managers = self.managers
try:
LOG.info("initializing the %r manager" % (manager.getName()))
LOG.info("initializing the %s manager" % (manager.getName()))
manager.setup()
LOG.info("manager %r initialized!" % (manager.getName()))
except Exception as ex:
LOG.error("Exception has occured", exc_info=1)
LOG.info("manager %s initialized!" % (manager.getName()))
except NotImplementedError:
message = "manager %s instantiation error: setup() not " \
"implemented!" % name
LOG.error("manager %r instantiation error: %s" % (name, ex))
LOG.error(message)
raise SynergyError(message)
except SynergyError as ex:
LOG.error("manager %s instantiation error: %s" % (name, ex))
self.managers[manager.getName()].setStatus("ERROR")
raise ex
@ -184,7 +189,7 @@ class Synergy(Service):
if len(manager_list) == 1 and len(result) == 0:
start_response("404 NOT FOUND", [("Content-Type", "text/plain")])
return ["manager %r not found!" % manager_list[0]]
return ["manager %s not found!" % manager_list[0]]
start_response("200 OK", [("Content-Type", "text/html")])
return ["%s" % json.dumps(result, cls=SynergyEncoder)]
@ -210,7 +215,7 @@ class Synergy(Service):
if manager_name not in self.managers:
start_response("404 NOT FOUND", [("Content-Type", "text/plain")])
return ["manager %r not found!" % manager_name]
return ["manager %s not found!" % manager_name]
if "command" not in parameters:
start_response("400 BAD REQUEST", [("Content-Type", "text/plain")])
@ -232,7 +237,15 @@ class Synergy(Service):
start_response("200 OK", [("Content-Type", "text/html")])
return ["%s" % json.dumps(result, cls=SynergyEncoder)]
except Exception as ex:
except NotImplementedError:
message = "execute() not implemented!"
LOG.error(message)
start_response("500 INTERNAL SERVER ERROR",
[("Content-Type", "text/plain")])
return ["error: %s" % message]
except SynergyError as ex:
LOG.debug("execute command: error=%s" % ex)
start_response("500 INTERNAL SERVER ERROR",
[("Content-Type", "text/plain")])
@ -272,11 +285,11 @@ class Synergy(Service):
result.append(m)
if manager.getStatus() == "ACTIVE":
LOG.info("starting the %r manager" % (manager_name))
LOG.info("starting the %s manager" % (manager_name))
manager.resume()
LOG.info("%r manager started! (rate=%s min)"
LOG.info("%s manager started! (rate=%s min)"
% (manager_name, manager.getRate()))
m.setStatus("RUNNING")
@ -330,11 +343,11 @@ class Synergy(Service):
result.append(m)
if manager.getStatus() == "RUNNING":
LOG.info("stopping the %r manager" % (manager_name))
LOG.info("stopping the %s manager" % (manager_name))
manager.pause()
LOG.info("%r manager stopped!" % (manager_name))
LOG.info("%s manager stopped!" % (manager_name))
m.setStatus("ACTIVE")
m.set("message", "stopped successfully")
@ -358,12 +371,12 @@ class Synergy(Service):
for name, manager in self.managers.items():
if manager.getStatus() != "ERROR":
try:
LOG.info("starting the %r manager" % (name))
LOG.info("starting the %s manager" % (name))
manager.start()
LOG.info("%r manager started! (rate=%s min, status=%s)"
LOG.info("%s manager started! (rate=%s min, status=%s)"
% (name, manager.getRate(), manager.getStatus()))
except Exception as ex:
except SynergyError as ex:
LOG.error("error occurred during the manager start %s"
% (ex))
manager.setStatus("ERROR")
@ -407,7 +420,13 @@ class Synergy(Service):
manager.destroy()
# manager.join()
# LOG.info("%s manager destroyed" % (name))
except Exception as ex:
except NotImplementedError:
message = "method destroy() not implemented by the " \
"%s manager" % manager.getName()
LOG.error(message)
except SynergyError as ex:
LOG.error("Exception has occured", exc_info=1)
manager.setStatus("ERROR")

98
synergy/tests/unit/test_client_keystonev3_trust.py
View File

@ -1,98 +0,0 @@
# coding: utf-8
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""
Test the Trust class.
"""
import mock
from datetime import datetime
from synergy.client.keystone_v3 import Trust
from synergy.tests import base
class TestTrust(base.TestCase):
def setUp(self):
super(TestTrust, self).setUp()
def test_trust_no_expires_at(self):
data = {
"trust": {
"id": 1,
"impersonation": False,
"roles_links": "some links",
"trustor_user_id": 0,
"trustee_user_id": 1,
"links": "some links",
"roles": "roll roll roll",
"remaining_uses": 10,
"expires_at": None,
"project_id": 46}}
trust = Trust(data)
self.assertEqual(1, trust.getId())
self.assertEqual(False, trust.isImpersonations())
self.assertEqual("some links", trust.getRolesLinks())
self.assertEqual(0, trust.getTrustorUserId())
self.assertEqual(1, trust.getTrusteeUserId())
self.assertEqual("some links", trust.getlinks())
self.assertEqual(46, trust.getProjectId())
self.assertEqual("roll roll roll", trust.getRoles())
self.assertEqual(10, trust.getRemainingUses())
self.assertIsNone(trust.getExpiration())
self.assertEqual(False, trust.isExpired())
def test_trust_not_expired(self):
mock_utcnow = datetime(2000, 1, 1)
data = {
"trust": {
"id": 1,
"impersonation": False,
"roles_links": "some links",
"trustor_user_id": 0,
"trustee_user_id": 1,
"links": "some links",
"roles": "roll roll roll",
"remaining_uses": 10,
"expires_at": "1900-01-01T00:00:00.000Z",
"project_id": 46}}
trust = Trust(data)
self.assertEqual(datetime(1900, 1, 1, 0, 0, 0), trust.getExpiration())
with mock.patch('datetime.datetime') as m:
m.utcnow.return_value = mock_utcnow
self.assertEqual(True, trust.isExpired())
def test_trust_expired(self):
mock_utcnow = datetime(2099, 1, 1)
data = {
"trust": {
"id": 1,
"impersonation": False,
"roles_links": "some links",
"trustor_user_id": 0,
"trustee_user_id": 1,
"links": "some links",
"roles": "roll roll roll",
"remaining_uses": 10,
"expires_at": "2099-01-01T00:00:00.000Z",
"project_id": 46}}
trust = Trust(data)
self.assertEqual(datetime(2099, 1, 1, 0, 0, 0), trust.getExpiration())
with mock.patch('datetime.datetime') as m:
m.utcnow.return_value = mock_utcnow
self.assertEqual(False, trust.isExpired())