diff --git a/docs/source/pyotrs.txt b/docs/source/pyotrs.txt
index 34daca566c79803668aa722d849fe6e596f60e1f..b173d7fed3c3191822ada1634c14761d8314b02f 100644
--- a/docs/source/pyotrs.txt
+++ b/docs/source/pyotrs.txt
@@ -55,6 +55,17 @@ PyOTRS Lib (all)
:noindex:
+:mod:`pyotrs.lib.SessionIDStore`
+--------------------------------
+
+.. autoclass:: pyotrs.lib.SessionIDStore
+ :members:
+ :private-members:
+ :undoc-members:
+ :show-inheritance:
+ :noindex:
+
+
:mod:`pyotrs.lib.Ticket`
------------------------
diff --git a/docs/source/pyotrs_public.txt b/docs/source/pyotrs_public.txt
index 9dfc1b7693dee934a6d8f1c11f96d5e340b9d1eb..79190f1c3da6d513badd748d39710ef1cda0cea8 100644
--- a/docs/source/pyotrs_public.txt
+++ b/docs/source/pyotrs_public.txt
@@ -50,6 +50,15 @@ PyOTRS Lib (public)
:noindex:
+:mod:`pyotrs.lib.SessionIDStore`
+--------------------------------
+
+.. autoclass:: pyotrs.lib.SessionIDStore
+ :members:
+ :undoc-members:
+ :show-inheritance:
+ :noindex:
+
:mod:`pyotrs.lib.Ticket`
------------------------
diff --git a/pyotrs/lib.py b/pyotrs/lib.py
index 39ed0d4399979304c060ad9be5debf18f7ac470c..23c5500767b2e04183e13efae8cb61c9f4175c00 100644
--- a/pyotrs/lib.py
+++ b/pyotrs/lib.py
@@ -52,6 +52,14 @@ class ResponseJSONParseError(PyOTRSError):
pass
+class SessionIDStoreNoFileError(PyOTRSError):
+ pass
+
+
+class SessionIDStoreNoTimeoutError(PyOTRSError):
+ pass
+
+
class SessionError(PyOTRSError):
pass
@@ -460,14 +468,139 @@ class Ticket(object):
class SessionIDStore(object):
- """Session ID: persistently store to and retrieve from to file"""
- def __init__(self, file_path=None):
+ """Session ID: persistently store to and retrieve from to file
+
+ Args:
+ file_path (unicode): Path on disc
+ session_timeout (int): OTRS Session Timeout Value (to avoid reusing outdated session id
+ session_id_value (unicode): A Session ID as str
+ session_id_created (int): seconds as epoch when a session_id record was created
+ session_id_expires (int): seconds as epoch when a session_id record expires
+
+ Raises:
+ SessionIDStoreNoFileError
+ SessionIDStoreNoTimeoutError
+
+ .. todo::
+ Is session_timeout correct here?! Or should this just return "expires" value and caller
+ needs to decide whether to use it..?!
+
+ """
+ def __init__(self, file_path=None, session_timeout=None):
+ if not file_path:
+ raise SessionIDStoreNoFileError("Argument file_path is required!")
+
+ if not session_timeout:
+ raise SessionIDStoreNoTimeoutError("Argument session_timeout is required!")
+
self.file_path = file_path
+ self.session_timeout = session_timeout
+ self.session_id_value = None
+ self.session_id_created = None
+ self.session_id_expires = None
def __repr__(self):
- return "<{0}>".format(self.__class__.__name__)
+ return "<{0}: {1}>".format(self.__class__.__name__, self.file_path)
+
+ def read(self):
+ """Retrieve a stored Session ID from file
+
+ Returns:
+ Union(str, None): Retrieved Session ID or None (if none could be read)
+ """
+ if not os.path.isfile(self.file_path):
+ return None
+
+ if not SessionIDStore._validate_file_owner_and_permissions(self.file_path):
+ return None
+
+ with open(self.file_path, "r") as f:
+ content = f.read()
+ try:
+ data = json.loads(content)
+ self.session_id_value = data['session_id']
+
+ self.session_id_created = datetime.datetime.utcfromtimestamp(int(data['created']))
+ self.session_id_expires = (self.session_id_created +
+ datetime.timedelta(minutes=self.session_timeout))
+
+ if self.session_id_expires > datetime.datetime.utcnow():
+ return self.session_id_value # still valid
+ except ValueError:
+ return None
+ except KeyError:
+ return None
+ except Exception as err:
+ raise Exception("Exception Type: {0}: {1}".format(type(err), err))
+
+ def write(self):
+ """Write and store a Session ID to file (rw for user only)
+
+ Returns:
+ bool: True if successful, False otherwise.
+
+ .. todo::
+ (RH) Error Handling and return True/False
+ """
+
+ if os.path.isfile(self.file_path):
+ if not SessionIDStore._validate_file_owner_and_permissions(self.file_path):
+ raise IOError("File exists but is not ok (wrong owner/permissions)!")
+
+ with open(self.file_path, 'w') as f:
+ f.write(json.dumps({'created': str(int(time.time())),
+ 'session_id': self.session_id_value}))
+ os.chmod(self.file_path, 384) # 384 is '0600'
+
+ # TODO 2016-04-23 (RH): check this
+ if not SessionIDStore._validate_file_owner_and_permissions(self.file_path):
+ raise IOError("Race condition: Something happened to file during the run!")
+
+ return True
+
+ def delete(self):
+ """remove session id file (e.g. when it only contains an invalid session id
+
+ Raises:
+ NotImplementedError
+
+ Returns:
+ bool: True if successful, False otherwise.
+
+ .. todo::
+ (RH) implement this _remove_session_id_file
+ """
+ raise NotImplementedError("Not yet done")
+
+ @staticmethod
+ def _validate_file_owner_and_permissions(full_file_path):
+ """validate SessionIDStore file ownership and permissions
+
+ Args:
+ full_file_path (unicode): full path to file on disk
+
+ Returns:
+ bool: True if valid and correct, otherwise False
+
+ """
+ if not os.path.isfile(full_file_path):
+ raise IOError("Does not exist or not a file: {0}".format(full_file_path))
+
+ file_lstat = os.lstat(full_file_path)
+ if not file_lstat.st_uid == os.getuid():
+ return False
+
+ if not file_lstat.st_mode & 0o777 == 384:
+ """ check for unix permission User R+W only (0600)
+ >>> oct(384)
+ '0600' Python 2
+ >>> oct(384)
+ '0o600' Python 3 """
+ return False
+
+ return True
+
-
class Client(object):
"""PyOTRS Client class - includes Session handling
@@ -498,16 +631,18 @@ class Client(object):
raise NoWebServiceName("Missing WebServiceName (e.g. GenericTicketConnectorREST)")
self.webservicename = webservicename
- if not session_id_file:
- self.session_id_file = "/tmp/.session_id.tmp"
- else:
- self.session_id_file = session_id_file
-
if not session_timeout:
self.session_timeout = 28800 # 8 hours is OTRS default
else:
self.session_timeout = session_timeout
+ if not session_id_file:
+ self.session_id_store = SessionIDStore(file_path="/tmp/.session_id.tmp",
+ session_timeout=self.session_timeout)
+ else:
+ self.session_id_store = SessionIDStore(file_path=session_id_file,
+ session_timeout=self.session_timeout)
+
self.https_verify = https_verify
# dummy initialization
@@ -602,24 +737,26 @@ class Client(object):
bool: True if successful, False otherwise.
"""
# try to read session_id from file
- if self._read_session_id_from_file():
+ self.session_id = self.session_id_store.read()
+
+ if self.session_id:
# got one.. check whether it's still valid
try:
if self.session_check_is_valid():
- print("Using valid Session ID from ({0})".format(self.session_id_file))
+ print("Using valid Session ID from ({0})".format(self.session_id_store))
return True
except OTRSAPIError:
- """most likely invalid session_id so pass. Remove clear session_id_file.."""
+ """most likely invalid session_id so pass. Remove clear session_id_store.."""
# got no (valid) session_id from file.. try to create new one
if not self.session_create():
raise SessionCreateError("Failed to create a Session ID!")
# safe new created session_id to file
- if not self._write_session_id_to_file():
+ if not self.session_id_store.write():
raise SessionIDFileError("Failed to save Session ID to file!")
else:
- print("Saved new Session ID to file: {0}".format(self.session_id_file))
+ print("Saved new Session ID to file: {0}".format(self.session_id_store))
return True
def _session_create_json(self, url, payload):
@@ -646,60 +783,6 @@ class Client(object):
return True
- def _read_session_id_from_file(self):
- """Retrieve a stored Session ID from file
-
- Returns:
- bool: True if successful, False otherwise.
- """
- if os.path.isfile(self.session_id_file):
- with open(self.session_id_file, "r") as f:
- content = f.read()
- try:
- data = json.loads(content)
- created = datetime.datetime.utcfromtimestamp(int(data['created']))
-
- expires = created + datetime.timedelta(minutes=self.session_timeout)
- if expires > datetime.datetime.utcnow():
- self.session_id = data['session_id'] # still valid
- return True
- except ValueError:
- return False
- except KeyError:
- return False
- except Exception as err:
- raise Exception("Exception Type: {0}: {1}".format(type(err), err))
-
- return False
-
- def _write_session_id_to_file(self):
- """Write and store a Session ID to file (rw for user only)
-
- Returns:
- bool: True if successful, False otherwise.
-
- .. todo::
- (RH) Error Handling and return True/False
- """
- with os.fdopen(os.open(self.session_id_file, os.O_WRONLY | os.O_CREAT, 0o600), 'w') as f:
- f.write(json.dumps({'created': str(int(time.time())),
- 'session_id': self.session_id}))
- return True
-
- def _remove_session_id_file(self):
- """remove session id file (e.g. when it only contains an invalid session id
-
- Raises:
- NotImplementedError
-
- Returns:
- bool: True if successful, False otherwise.
-
- .. todo::
- (RH) implement this _remove_session_id_file
- """
- raise NotImplementedError("Not yet done")
-
"""
GenericInterface::Operation::Ticket::TicketCreate
Methods (public):
diff --git a/tests/test_client.py b/tests/test_client.py
index 17f44f962620ae8c8dfe265209cb6e1f1d08ee9b..042cc6b47626adfc7e48ae8c74cc8dca73a6c457 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -16,10 +16,11 @@ import datetime
import requests # noqa
from pyotrs import Client
+from pyotrs import Article, Attachment, Client, DynamicField, Ticket # noqa
+
from pyotrs.lib import NoBaseURL, NoWebServiceName, ResponseJSONParseError
from pyotrs.lib import SessionError
-from pyotrs import Article, Attachment, Client, DynamicField, Ticket # noqa
from pyotrs.lib import NoBaseURL, NoWebServiceName, NoCredentials # noqa
from pyotrs.lib import SessionCreateError, SessionIDFileError, OTRSAPIError, OTRSHTTPError # noqa
from pyotrs.lib import TicketError, TicketSearchNumberMultipleResults # noqa
@@ -45,9 +46,24 @@ class ClientTests(unittest.TestCase):
def test_init_no_webservice(self):
self.assertRaisesRegex(NoWebServiceName, 'Missing WebSe.*', Client, 'http://localhost', '')
- def test_init_session_id_file(self):
+ def test_init_session_id_store(self):
client = Client(baseurl="http://localhost/", webservicename="foo", session_id_file=".sid")
- self.assertEqual(client.session_id_file, '.sid')
+ self.assertEqual(client.session_id_store.file_path, '.sid')
+
+ def test_init_session_id_store_timeout_default(self):
+ client = Client(baseurl="http://localhost/",
+ webservicename="foo",
+ session_id_file=".sid")
+ self.assertEqual(client.session_id_store.file_path, '.sid')
+ self.assertEqual(client.session_id_store.session_timeout, 28800)
+
+ def test_init_session_id_store_timeout(self):
+ client = Client(baseurl="http://localhost/",
+ webservicename="foo",
+ session_id_file=".sid",
+ session_timeout=815)
+ self.assertEqual(client.session_id_store.file_path, '.sid')
+ self.assertEqual(client.session_id_store.session_timeout, 815)
def test_session_check_is_valid_no_session_id_error(self):
"""Test"""
@@ -113,7 +129,7 @@ class ClientTests(unittest.TestCase):
self.assertDictEqual(result, {'a': 'response'})
@mock.patch('pyotrs.Client.session_check_is_valid', autospec=True)
- @mock.patch('pyotrs.Client._read_session_id_from_file', autospec=True)
+ @mock.patch('pyotrs.SessionIDStore.read', autospec=True)
def test_session_restore_or_set_up_new_with_file_nok(self, mock_read_s_id, mock_is_valid):
"""Tests session_restore_or_set_up_new when read from file successful but session nok"""
# create object
@@ -122,7 +138,7 @@ class ClientTests(unittest.TestCase):
otrs_api_error = OTRSHTTPError("Failed to access OTRS. Check Hostname...")
- mock_read_s_id.return_value = True
+ mock_read_s_id.return_value = "SomeSessionID1"
mock_is_valid.side_effect = otrs_api_error
self.assertRaisesRegex(OTRSHTTPError,
@@ -133,14 +149,14 @@ class ClientTests(unittest.TestCase):
self.assertEqual(mock_is_valid.call_count, 1)
@mock.patch('pyotrs.Client.session_check_is_valid', autospec=True)
- @mock.patch('pyotrs.Client._read_session_id_from_file', autospec=True)
+ @mock.patch('pyotrs.SessionIDStore.read', autospec=True)
def test_session_restore_or_set_up_new_with_file_ok(self, mock_read_s_id, mock_is_valid):
"""Tests session_restore_or_set_up_new when read from file successful and session ok"""
# create object
obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
obj.session_id = "foobar2"
- mock_read_s_id.return_value = True
+ mock_read_s_id.return_value = "SomeSessionID2"
mock_is_valid.return_value = True
result = obj.session_restore_or_set_up_new()
@@ -151,7 +167,7 @@ class ClientTests(unittest.TestCase):
@mock.patch('pyotrs.Client.session_create', autospec=True)
@mock.patch('pyotrs.Client.session_check_is_valid', autospec=True)
- @mock.patch('pyotrs.Client._read_session_id_from_file', autospec=True)
+ @mock.patch('pyotrs.SessionIDStore.read', autospec=True)
def test_session_restore_or_set_up_new_with_file_pass(self,
mock_read_s_id,
mock_is_valid,
@@ -161,7 +177,7 @@ class ClientTests(unittest.TestCase):
obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
obj.session_id = "foobar"
- mock_read_s_id.return_value = True
+ mock_read_s_id.return_value = "SomeSessionID"
mock_is_valid.side_effect = OTRSHTTPError("Failed to access OTRS. Check Hostname...")
mock_s_create.return_value = True
@@ -174,14 +190,14 @@ class ClientTests(unittest.TestCase):
self.assertEqual(mock_is_valid.call_count, 1)
@mock.patch('pyotrs.Client.session_create', autospec=True)
- @mock.patch('pyotrs.Client._read_session_id_from_file', autospec=True)
+ @mock.patch('pyotrs.SessionIDStore.read', autospec=True)
def test_session_restore_or_set_up_new_nok(self, mock_read_s_id, mock_s_create):
"""Tests session_restore_or_set_up_new no file; create unsuccessful"""
# create object
obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
obj.session_id = "foobar"
- mock_read_s_id.return_value = False
+ mock_read_s_id.return_value = None
mock_s_create.return_value = False
self.assertRaisesRegex(SessionCreateError,
@@ -191,16 +207,16 @@ class ClientTests(unittest.TestCase):
self.assertEqual(mock_read_s_id.call_count, 1)
self.assertEqual(mock_s_create.call_count, 1)
- @mock.patch('pyotrs.Client._write_session_id_to_file', autospec=True)
+ @mock.patch('pyotrs.SessionIDStore.write', autospec=True)
@mock.patch('pyotrs.Client.session_create', autospec=True)
- @mock.patch('pyotrs.Client._read_session_id_from_file', autospec=True)
+ @mock.patch('pyotrs.SessionIDStore.read', autospec=True)
def test_session_restore_or_set_up_new_ok_no_wr(self, mock_read_s_id, mock_s_create, mock_wr):
"""Tests session_restore_or_set_up_new no file; create successful; write (wr) not ok"""
# create object
obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
obj.session_id = "foobar"
- mock_read_s_id.return_value = False
+ mock_read_s_id.return_value = None
mock_s_create.return_value = True
mock_wr.return_value = False
@@ -212,16 +228,16 @@ class ClientTests(unittest.TestCase):
self.assertEqual(mock_s_create.call_count, 1)
self.assertEqual(mock_wr.call_count, 1)
- @mock.patch('pyotrs.Client._write_session_id_to_file', autospec=True)
+ @mock.patch('pyotrs.SessionIDStore.write', autospec=True)
@mock.patch('pyotrs.Client.session_create', autospec=True)
- @mock.patch('pyotrs.Client._read_session_id_from_file', autospec=True)
+ @mock.patch('pyotrs.SessionIDStore.read', autospec=True)
def test_session_restore_or_set_up_new_ok_wr_ok(self, mock_read_s_id, mock_s_create, mock_wr):
"""Tests session_restore_or_set_up_new no file; create successful; write (wr) ok"""
# create object
obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
obj.session_id = "foobar"
- mock_read_s_id.return_value = False
+ mock_read_s_id.return_value = None
mock_s_create.return_value = True
mock_wr.return_value = True
@@ -231,146 +247,6 @@ class ClientTests(unittest.TestCase):
self.assertEqual(mock_s_create.call_count, 1)
self.assertEqual(mock_wr.call_count, 1)
- @mock.patch('pyotrs.lib.os.path.isfile', autospec=True)
- def test__read_session_id_from_file_no_file(self, mock_isfile):
- """Tests _read_session_id_from_file no file"""
- # create object
- obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
- obj.session_id_file = "/tmp/.sid"
-
- mock_isfile.return_value = False
-
- self.assertFalse(obj._read_session_id_from_file(), False)
- self.assertEqual(mock_isfile.call_count, 1)
- mock_isfile.assert_called_with("/tmp/.sid")
-
- @mock.patch('pyotrs.lib.os.path.isfile', autospec=True)
- def test__read_session_id_from_file_w_file_valid_json(self, mock_isfile):
- """Tests _read_session_id_from_file with a file with valid json but in past"""
- # create object
- obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
- obj.session_id_file = "/tmp/.sid"
-
- mock_isfile.return_value = True
-
- with mock.patch('pyotrs.lib.open',
- mock.mock_open(read_data='{"created": "1", '
- '"session_id": "foo2"}')) as m:
- obj._read_session_id_from_file()
-
- m.assert_called_once_with('/tmp/.sid', 'r')
- self.assertNotEqual(obj.session_id, 'foo2')
-
- @mock.patch('pyotrs.lib.os.path.isfile', autospec=True)
- def test__read_session_id_from_file_valid_json_in_no_created(self, mock_isfile):
- """Tests _read_session_id_from_file with a file with valid json but no created"""
- # create object TODO: not working
- obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
- obj.session_id_file = "/tmp/.sid"
-
- mock_isfile.return_value = True
-
- with mock.patch('pyotrs.lib.open',
- mock.mock_open(read_data='{"not_epoch_time": "2147483646", '
- '"no_s_id": "fake"}')) as m:
- result = obj._read_session_id_from_file()
- print(result)
- """
- self.assertRaisesRegex(Exception,
- 'Some Exception.*',
- obj._read_session_id_from_file)
- """
- m.assert_called_once_with('/tmp/.sid', 'r')
-
- @mock.patch('pyotrs.lib.os.path.isfile', autospec=True)
- def test__read_session_id_from_file_valid_json_sid_in_p(self, mock_isfile):
- """Tests _read_session_id_from_file with a file with valid json and sid in past"""
- # create object
- obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
- obj.session_id_file = "/tmp/.sid"
-
- mock_isfile.return_value = True
-
- with mock.patch('pyotrs.lib.open',
- mock.mock_open(read_data='{"created": "1", '
- '"session_id": "foo3"}')) as m:
- result = obj._read_session_id_from_file()
-
- m.assert_called_once_with('/tmp/.sid', 'r')
- self.assertEqual(result, False)
- self.assertNotEqual(obj.session_id, 'foo3')
-
- @mock.patch('pyotrs.lib.os.path.isfile', autospec=True)
- def test__read_session_id_from_file_valid_json_sid_in_f(self, mock_isfile):
- """Tests _read_session_id_from_file with a file with valid json and sid in future"""
- # create object
- obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
- obj.session_id_file = "/tmp/.sid"
-
- mock_isfile.return_value = True
-
- with mock.patch('pyotrs.lib.open',
- mock.mock_open(read_data='{"created": "2147483646", '
- '"session_id": "foo4"}')) as m:
- result = obj._read_session_id_from_file()
-
- m.assert_called_once_with('/tmp/.sid', 'r')
- self.assertEqual(result, True)
- self.assertEqual(obj.session_id, 'foo4')
-
- @mock.patch('pyotrs.lib.os.path.isfile', autospec=True)
- def test__read_session_id_from_file_valid_json_in_f_no_sid(self, mock_isfile):
- """Tests _read_session_id_from_file with a file with valid json and in future but no sid"""
- # create object TODO: not working
- obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
- obj.session_id_file = "/tmp/.sid"
-
- mock_isfile.return_value = True
-
- with mock.patch('pyotrs.lib.open',
- mock.mock_open(read_data='{"created": "2147483646", '
- '"no_s_id": "fake"}')) as m:
- result = obj._read_session_id_from_file()
-
- self.assertEqual(result, False)
- m.assert_called_once_with('/tmp/.sid', 'r')
-
- @mock.patch('pyotrs.lib.json.loads', autospec=True)
- @mock.patch('pyotrs.lib.os.path.isfile', autospec=True)
- def test__read_session_id_from_file_w_file_val_exception(self, mock_isfile, mock_json_loads):
- """Tests _read_session_id_from_file with a file with an ValueError exception"""
- # create object
- obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
- obj.session_id_file = "/tmp/.sid"
-
- mock_isfile.return_value = True
- mock_json_loads.side_effect = ValueError("Some value exception")
-
- with mock.patch('pyotrs.lib.open',
- mock.mock_open(read_data='invalid_json')) as m:
- result = obj._read_session_id_from_file()
- self.assertEqual(result, False)
- m.assert_called_once_with('/tmp/.sid', 'r')
-
- @mock.patch('pyotrs.lib.json.loads', autospec=True)
- @mock.patch('pyotrs.lib.os.path.isfile', autospec=True)
- def test__read_session_id_from_file_w_file_unc_exception(self, mock_isfile, mock_json_loads):
- """Tests _read_session_id_from_file with a file with an uncaught exception"""
- # create object
- obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
- obj.session_id_file = "/tmp/.sid"
-
- mock_isfile.return_value = True
- mock_json_loads.side_effect = Exception("Some exception")
-
- with mock.patch('pyotrs.lib.open',
- mock.mock_open(read_data='invalid_json')) as m:
- self.assertRaisesRegex(Exception,
- 'Exception Type',
- obj._read_session_id_from_file)
-
- m.assert_called_once_with('/tmp/.sid', 'r')
-
@mock.patch('pyotrs.Client._ticket_get_json', autospec=True)
def test_ticket_get_by_id_fail(self, mock_ticket_get_json):
"""Tests ticket_get_by_id fail"""
@@ -445,20 +321,6 @@ class ClientTests(unittest.TestCase):
self.assertEqual(mock_ticket_search.call_count, 1)
- def test__write_session_id_to_file(self):
- """Tests _write_session_id_to_file not implemented"""
- # create object
- Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
- # TODO
-
- def test__remove_session_id_file(self):
- """Tests_remove_session_id_file not implemented"""
- # create object
- obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
- self.assertRaisesRegex(NotImplementedError,
- 'Not yet done',
- obj._remove_session_id_file)
-
def test_ticket_create_no_ticket(self):
"""Test ticket_create - no ticket specified"""
obj = Client(baseurl="http://fqdn",
diff --git a/tests/test_session_id_store.py b/tests/test_session_id_store.py
index d476186349cc165ee3670fd73036239250eac793..4d4115091485edbc94813f3218c0c4062e44af34 100644
--- a/tests/test_session_id_store.py
+++ b/tests/test_session_id_store.py
@@ -12,33 +12,346 @@ from __future__ import unicode_literals # support both Python2 and 3
import unittest2 as unittest
import mock
-from pyotrs import SessionIDStore
+from pyotrs.lib import SessionIDStore
+from pyotrs.lib import SessionIDStoreNoFileError, SessionIDStoreNoTimeoutError
class SessionIDStoreTests(unittest.TestCase):
- def test_init(self):
- sis = SessionIDStore()
+ def test_init_no_file(self):
+ self.assertRaisesRegex(SessionIDStoreNoFileError,
+ 'Argument file_path is required!',
+ SessionIDStore)
+
+ def test_init_no_timeout(self):
+ self.assertRaisesRegex(SessionIDStoreNoTimeoutError,
+ 'Argument session_timeout is required!',
+ SessionIDStore,
+ file_path='/tmp/.foo.bar')
+
+ def test_init_ok(self):
+ sis = SessionIDStore(file_path="/tmp/.session_id_store", session_timeout=600)
self.assertIsInstance(sis, SessionIDStore)
+ @mock.patch('pyotrs.lib.os.path.isfile', autospec=True)
+ def test_read_no_file(self, mock_isfile):
+ """Tests _read_session_id_from_file no file"""
+ # create object
+ sis = SessionIDStore(file_path="/tmp/.session_id_store", session_timeout=600)
+
+ mock_isfile.return_value = False
+ result = sis.read()
+
+ self.assertIsNone(result)
+ self.assertEqual(mock_isfile.call_count, 1)
+ mock_isfile.assert_called_with("/tmp/.session_id_store")
+
+ @mock.patch('pyotrs.SessionIDStore._validate_file_owner_and_permissions')
+ @mock.patch('pyotrs.lib.os.path.isfile', autospec=True)
+ def test_read_w_invalid_file(self, mock_isfile, mock_validation):
+ """Tests read with an invalid file"""
+ # create object
+ sis = SessionIDStore(file_path="/tmp/.session_id_store", session_timeout=600)
+
+ mock_isfile.return_value = True
+ mock_validation.return_value = False
+
+ result = sis.read()
+
+ self.assertIsNone(result)
+ self.assertEqual(mock_isfile.call_count, 1)
+ self.assertEqual(mock_validation.call_count, 1)
+
+ @mock.patch('pyotrs.SessionIDStore._validate_file_owner_and_permissions')
+ @mock.patch('pyotrs.lib.os.path.isfile', autospec=True)
+ def test_read_w_file_valid_json(self, mock_isfile, mock_validation):
+ """Tests read with a file with valid json but in past"""
+ # create object
+ sis = SessionIDStore(file_path="/tmp/.session_id_store", session_timeout=600)
+
+ mock_isfile.return_value = True
+ mock_validation.return_value = True
+ r_data = '{"created": "1", "session_id": "foo2"}'
+
+ with mock.patch('pyotrs.lib.open',
+ mock.mock_open(read_data=r_data)) as m:
+ result = sis.read()
+
+ m.assert_called_once_with('/tmp/.session_id_store', 'r')
+ self.assertNotEqual(result, 'foo2')
+
+ @mock.patch('pyotrs.SessionIDStore._validate_file_owner_and_permissions')
+ @mock.patch('pyotrs.lib.os.path.isfile', autospec=True)
+ def test_read_valid_json_sid_in_p(self, mock_isfile, mock_validation):
+ """Tests _read_session_id_from_file with a file with valid json and sid in past"""
+ # create object
+ sis = SessionIDStore(file_path="/tmp/.session_id_store", session_timeout=600)
+
+ mock_isfile.return_value = True
+ mock_validation.return_value = True
+ r_data = '{"created": "1", "session_id": "foo3"}'
+
+ with mock.patch('pyotrs.lib.open',
+ mock.mock_open(read_data=r_data)) as m:
+ result = sis.read()
+
+ m.assert_called_once_with('/tmp/.session_id_store', 'r')
+ self.assertIsNone(result)
+
+ @mock.patch('pyotrs.SessionIDStore._validate_file_owner_and_permissions')
+ @mock.patch('pyotrs.lib.os.path.isfile', autospec=True)
+ def test_read_valid_json_sid_in_f(self, mock_isfile, mock_validation):
+ """Tests _read_session_id_from_file with a file with valid json and sid in future"""
+ # create object
+ sis = SessionIDStore(file_path="/tmp/.session_id_store", session_timeout=600)
+
+ mock_isfile.return_value = True
+ mock_validation.return_value = True
+ r_data = '{"created": "2147483646", "session_id": "foo4"}'
+
+ with mock.patch('pyotrs.lib.open',
+ mock.mock_open(read_data=r_data)) as m:
+ result = sis.read()
+
+ m.assert_called_once_with('/tmp/.session_id_store', 'r')
+ self.assertEqual(result, 'foo4')
+
+ @mock.patch('pyotrs.SessionIDStore._validate_file_owner_and_permissions')
+ @mock.patch('pyotrs.lib.os.path.isfile', autospec=True)
+ def test_read_valid_json_in_f_no_sid(self, mock_isfile, mock_validation):
+ """Tests _read_session_id_from_file with a file with valid json and in future but no sid"""
+ # create object TODO: not working
+ sis = SessionIDStore(file_path="/tmp/.session_id_store", session_timeout=600)
+
+ mock_isfile.return_value = True
+ mock_validation.return_value = True
+ r_data = '{"created": "2147483646", "no_s_id": "fake"}'
+
+ with mock.patch('pyotrs.lib.open',
+ mock.mock_open(read_data=r_data)) as m:
+ result = sis.read()
+
+ self.assertIsNone(result)
+ m.assert_called_once_with('/tmp/.session_id_store', 'r')
+
+ @mock.patch('pyotrs.SessionIDStore._validate_file_owner_and_permissions')
+ @mock.patch('pyotrs.lib.json.loads', autospec=True)
+ @mock.patch('pyotrs.lib.os.path.isfile', autospec=True)
+ def test_read_w_file_val_exception(self, mock_isfile, mock_json_loads, mock_validation):
+ """Tests _read_session_id_from_file with a file with an ValueError exception"""
+ # create object
+ sis = SessionIDStore(file_path="/tmp/.session_id_store", session_timeout=600)
+
+ mock_isfile.return_value = True
+ mock_validation.return_value = True
+ mock_json_loads.side_effect = ValueError("Some value exception")
+ r_data = 'invalid_json'
+
+ with mock.patch('pyotrs.lib.open',
+ mock.mock_open(read_data=r_data)) as m:
+ result = sis.read()
+ self.assertIsNone(result)
+ m.assert_called_once_with('/tmp/.session_id_store', 'r')
+
+ @mock.patch('pyotrs.SessionIDStore._validate_file_owner_and_permissions')
+ @mock.patch('pyotrs.lib.json.loads', autospec=True)
+ @mock.patch('pyotrs.lib.os.path.isfile', autospec=True)
+ def test_read_w_file_unc_exception(self, mock_isfile, mock_json_loads, mock_validation):
+ """Tests _read_session_id_from_file with a file with an uncaught exception"""
+ # create object
+ sis = SessionIDStore(file_path="/tmp/.session_id_store", session_timeout=600)
+
+ mock_isfile.return_value = True
+ mock_validation.return_value = True
+ mock_json_loads.side_effect = Exception("Some exception")
+ r_data = 'invalid_json'
+
+ with mock.patch('pyotrs.lib.open',
+ mock.mock_open(read_data=r_data)) as m:
+ self.assertRaisesRegex(Exception,
+ 'Exception Type',
+ sis.read)
+
+ m.assert_called_once_with('/tmp/.session_id_store', 'r')
+
+ # -- write
+ @mock.patch('pyotrs.lib.os.chmod', autospec=True)
+ @mock.patch('pyotrs.SessionIDStore._validate_file_owner_and_permissions')
+ @mock.patch('pyotrs.lib.os.path.isfile', autospec=True)
+ def test_write_mock_open_file_exists_nok(self, mock_isfile, mock_validation, mock_chmod):
+ """Tests write mock_open file exists; validation not ok"""
+ # create object
+ test_path = "/tmp/mocked_wr1"
+ sis = SessionIDStore(file_path=test_path, session_timeout=600)
+ sis.session_id_value = "some_session_id_value1"
+
+ mock_isfile.return_value = True
+ mock_validation.return_value = False
+ mock_chmod.return_value = True
+
+ self.assertRaisesRegex(IOError,
+ 'File exists but is not ok.*',
+ sis.write)
+
+ @mock.patch('pyotrs.lib.os.chmod', autospec=True)
+ @mock.patch('pyotrs.SessionIDStore._validate_file_owner_and_permissions')
+ @mock.patch('pyotrs.lib.os.path.isfile', autospec=True)
+ def test_write_mock_open_file_exists_ok(self, mock_isfile, mock_validation, mock_chmod):
+ """Tests write mock_open file exists; validation ok"""
+ # create object
+ test_path = "/tmp/mocked_wr2"
+ sis = SessionIDStore(file_path=test_path, session_timeout=600)
+ sis.session_id_value = "some_session_id_value1"
+
+ mock_isfile.return_value = True
+ mock_validation.return_value = True
+
+ with mock.patch('pyotrs.lib.open', mock.mock_open()) as m:
+ result = sis.write()
+
+ self.assertTrue(result)
+ self.assertEqual(mock_isfile.call_count, 1)
+ self.assertEqual(mock_validation.call_count, 2)
+ self.assertEqual(mock_chmod.call_count, 1)
+ m.assert_called_once_with(test_path, 'w')
+
+ @mock.patch('pyotrs.lib.os.chmod', autospec=True)
+ @mock.patch('pyotrs.SessionIDStore._validate_file_owner_and_permissions')
+ @mock.patch('pyotrs.lib.os.path.isfile', autospec=True)
+ def test_write_mock_open_file_not_exist_v_nok(self, mock_isfile, mock_validation, mock_chmod):
+ """Tests write mock_open file does not exist; validation also not ok -> Race Condition?!"""
+ # create object
+ test_path = "/tmp/mocked_wr2"
+ sis = SessionIDStore(file_path=test_path, session_timeout=600)
+ sis.session_id_value = "some_session_id_value1"
+
+ mock_isfile.return_value = False
+ mock_validation.return_value = False
+
+ with mock.patch('pyotrs.lib.open', mock.mock_open()) as m:
+ self.assertRaisesRegex(IOError,
+ 'Race condition.*',
+ sis.write)
+
+ self.assertEqual(mock_isfile.call_count, 1)
+ self.assertEqual(mock_validation.call_count, 1)
+ self.assertEqual(mock_chmod.call_count, 1)
+ m.assert_called_once_with(test_path, 'w')
+
+ def test_delete(self):
+ """Tests remove - not implemented"""
+ # create object
+ sis = SessionIDStore(file_path="/tmp/.session_id_store", session_timeout=600)
+ self.assertRaisesRegex(NotImplementedError,
+ 'Not yet done',
+ sis.delete)
+
+ @mock.patch('pyotrs.lib.os.path.isfile', autospec=True)
+ def test__validate_non_existing_file(self, mock_isfile):
+ """Tests _validate file - existing file - wrong permissions"""
+ # create object
+ test_path = "/tmp/mocked1"
+ mock_isfile.return_value = False
+
+ self.assertRaisesRegex(IOError,
+ 'Does not exist or not a '
+ 'file: /tmp/mocked1',
+ SessionIDStore._validate_file_owner_and_permissions,
+ full_file_path=test_path)
+
+ self.assertEqual(mock_isfile.call_count, 1)
+
+ @mock.patch('pyotrs.lib.os.lstat', autospec=True)
+ @mock.patch('pyotrs.lib.os.getuid', autospec=True)
+ @mock.patch('pyotrs.lib.os.path.isfile', autospec=True)
+ def test__validate_existing_file_wrong_owner(self, mock_isfile, mock_getuid, mock_lstat):
+ """Tests _validate file - existing file - wrong owner"""
+ # create object
+ test_path = "/tmp/mocked2"
+
+ mock_isfile.return_value = True
+ mock_getuid.return_value = 4711
+ mock_lstat.return_value.st_uid = 815
+ mock_lstat.return_value.st_mode = 33188 # 644
+
+ result = SessionIDStore._validate_file_owner_and_permissions(full_file_path=test_path)
+
+ self.assertFalse(result)
+ self.assertEqual(mock_isfile.call_count, 1)
+ self.assertEqual(mock_getuid.call_count, 1)
+ self.assertEqual(mock_lstat.call_count, 1)
+
+ @mock.patch('pyotrs.lib.os.lstat', autospec=True)
+ @mock.patch('pyotrs.lib.os.getuid', autospec=True)
+ @mock.patch('pyotrs.lib.os.path.isfile', autospec=True)
+ def test__validate_existing_file_wrong_permissions(self, mock_isfile, mock_getuid, mock_lstat):
+ """Tests _validate file - existing file - wrong permissions"""
+ # create object
+ test_path = "/tmp/mocked3"
+
+ mock_isfile.return_value = True
+ mock_getuid.return_value = 4712
+ mock_lstat.return_value.st_uid = 4712
+ mock_lstat.return_value.st_mode = 33188 # 644
+
+ result = SessionIDStore._validate_file_owner_and_permissions(full_file_path=test_path)
+
+ self.assertFalse(result)
+ self.assertEqual(mock_isfile.call_count, 1)
+ self.assertEqual(mock_getuid.call_count, 1)
+ self.assertEqual(mock_lstat.call_count, 1)
+
+ @mock.patch('pyotrs.lib.os.lstat', autospec=True)
+ @mock.patch('pyotrs.lib.os.getuid', autospec=True)
+ @mock.patch('pyotrs.lib.os.path.isfile', autospec=True)
+ def test__validate_existing_file_ok(self, mock_isfile, mock_getuid, mock_lstat):
+ """Tests _validate file - existing file - ok"""
+ # create object
+ test_path = "/tmp/mocked4"
+
+ mock_isfile.return_value = True
+ mock_getuid.return_value = 4713
+ mock_lstat.return_value.st_uid = 4713
+ mock_lstat.return_value.st_mode = 33152 # 0600
+
+ result = SessionIDStore._validate_file_owner_and_permissions(full_file_path=test_path)
+
+ self.assertTrue(result)
+ self.assertEqual(mock_isfile.call_count, 1)
+ self.assertEqual(mock_getuid.call_count, 1)
+ self.assertEqual(mock_lstat.call_count, 1)
+
+ # def test__validate_existing_file(self):
+ # """Tests _validate file - existing file - wrong permissions"""
+ # # create object
+ # test_path = "/home/robbie/.bash_history" # TODO (RH): not a valid test!
+ # result = SessionIDStore._validate_file_owner_and_permissions(full_file_path=test_path)
+ # self.assertTrue(result)
+
+ # # Non mocked: !!!
+
+ # def test__validate_file_owner_and_permissions_non_existing_file(self):
+ # """Tests _validate file - non existing file"""
+ # # create object
+ # test_path = "/osirhgiosehgoisrehgioshrgiohserdiob"
+ # self.assertRaisesRegex(IOError,
+ # 'Does not exist or not a '
+ # 'file: /osirhgiosehgoisrehgioshrgiohserdiob',
+ # SessionIDStore._validate_file_owner_and_permissions,
+ # full_file_path=test_path)
+
+ # def test__validate_file_owner_and_permissions_existing_file_wrong_permissions(self):
+ # """Tests _validate file - existing file - wrong permissions"""
+ # # create object
+ # test_path = "/etc/passwd" # TODO 2016-04-23 (RH): is this a valid test?!
+ # result = SessionIDStore._validate_file_owner_and_permissions(full_file_path=test_path)
+ # self.assertFalse(result)
#
- # def test_session_check_is_valid_no_session_id_error(self):
- # """Test"""
- # client = Client(baseurl="http://localhost/", webservicename="foo")
- # self.assertRaisesRegex(SessionError,
- # 'No value set for session_id!',
- # client.session_check_is_valid)
- #
- # @mock.patch('pyotrs.Client._ticket_get_json', autospec=True)
- # def test_session_check_is_valid_session_id(self, mock_ticket_get_json):
- # """Test session_check_is_valid with a given session id"""
- # obj = Client(baseurl="http://localhost/", webservicename="foo")
- # obj.session_id = "some_other_value"
- # mock_ticket_get_json.return_value = True
- #
- # result = obj.session_check_is_valid(session_id="some_value")
- #
- # self.assertEqual(obj.session_id, "some_value")
- # self.assertEqual(result, True)
+ # def test__validate_file_owner_and_permissions_existing_file(self):
+ # """Tests _validate file - existing file - wrong permissions"""
+ # # create object
+ # test_path = "/home/robbie/.bash_history" # TODO (RH): not a valid test!
+ # result = SessionIDStore._validate_file_owner_and_permissions(full_file_path=test_path)
+ # self.assertTrue(result)
def main():