[#455] introduce low level ticket api changes

This ensures the application of a ticket to a session via supply()
filters down to each connection instantiated for that session object.

One test (test_object_read_and_write_tickets) was updated to ensure
call to cleanup() on a session, in order to clear out existing connections
(with their associated tickets) when required.  Due to the flawed approach
previously used in the ticket API, this step had not been strictly
necessary for the test to pass.

Author: d-w-moore

irods/manager/data_object_manager.py

@@ -64,7 +64,7 @@ def should_parallelize_transfer( self,
         # Example: $ export IRODS_VERSION_OVERRIDE="4,2,9" ;  python -m irods.parallel ...
         # ---
         # Delete the following line on resolution of https://github.com/irods/irods/issues/5932 :
-        if getattr(self.sess,'ticket__',None) is not None: return False
+        if self.sess.ticket__: return False
         server_version = ( ast.literal_eval(os.environ.get('IRODS_VERSION_OVERRIDE', '()' )) or server_version_hint or 
                            self.server_version )
         if num_threads == 1 or ( server_version < parallel.MINIMUM_SERVER_VERSION ):
@@ -126,7 +126,7 @@ def get(self, path, local_path = None, num_threads = DEFAULT_NUMBER_OF_THREADS,
             .filter(DataObject.collection_id == parent.id)\
             .add_keyword(kw.ZONE_KW, path.split('/')[1])
 
-        if hasattr(self.sess,'ticket__'):
+        if self.sess.ticket__:
             query = query.filter(Collection.id != 0) # a no-op, but necessary because CAT_SQL_ERR results if the ticket
                                                      # is for a DataObject and we don't explicitly join to Collection
 

irods/pool.py

@@ -3,9 +3,11 @@
 import logging
 import threading
 import os
+import weakref
 
 from irods import DEFAULT_CONNECTION_TIMEOUT
 from irods.connection import Connection
+from irods.ticket import Ticket
 
 logger = logging.getLogger(__name__)
 
@@ -22,13 +24,14 @@ def method_(self,*s,**kw):
 
 class Pool(object):
 
-    def __init__(self, account, application_name='', connection_refresh_time=-1):
+    def __init__(self, account, application_name='', connection_refresh_time=-1, session = None):
         '''
         Pool( account , application_name='' )
         Create an iRODS connection pool; 'account' is an irods.account.iRODSAccount instance and
         'application_name' specifies the application name as it should appear in an 'ips' listing.
         '''
 
+        self.session_ref = weakref.ref(session) if session is not None else lambda:None
         self._thread_local = threading.local()
         self.account = account
         self._lock = threading.RLock()
@@ -77,6 +80,11 @@ def get_connection(self):
 
             self.active.add(conn)
 
+            sess = self.session_ref()
+            if sess and sess.ticket__ and not sess.ticket_applied.get(conn,False):
+                Ticket._lowlevel_api_request(conn, "session", sess.ticket__)
+                sess.ticket_applied[conn] = True
+
             logger.debug("Adding connection with id {} to active set".format(id(conn)))
 
         logger.debug('num active: {}'.format(len(self.active)))
@@ -100,3 +108,4 @@ def release_connection(self, conn, destroy=False):
                 self.idle.remove(conn)
         logger.debug('num active: {}'.format(len(self.active)))
         logger.debug('num idle: {}'.format(len(self.idle)))
+

irods/session.py

@@ -1,5 +1,6 @@
 from __future__ import absolute_import
 import atexit
+import copy
 import os
 import ast
 import json
@@ -145,7 +146,9 @@ def __init__(self, configure = True, auto_cleanup = True, **kwargs):
         self.user_groups = GroupManager(self)
         self.resources = ResourceManager(self)
         self.zones = ZoneManager(self)
-
+        self._auto_cleanup = auto_cleanup
+        self.ticket__ = ''
+        self.ticket_applied = weakref.WeakKeyDictionary() # conn -> ticket applied
         if auto_cleanup:
             _weakly_reference(self)
 
@@ -162,14 +165,33 @@ def __del__(self):
         if self.pool is not None:
             self.cleanup()
 
-    def cleanup(self):
-        for conn in self.pool.active | self.pool.idle:
-            try:
-                conn.disconnect()
-            except NetworkException:
-                pass
-            conn.release(True)
+    def clone(self, **kwargs):
+        other = copy.copy(self)
+        other.pool = None
+        for k,v in vars(other).items():
+            setter = getattr(v,'_set_manager_session',None)
+            if setter:
+                setter(other)
+        other.cleanup(new_host = kwargs.pop('host',''))
+        other.ticket__ = kwargs.pop('ticket',self.ticket__)
+        self.ticket_applied = weakref.WeakKeyDictionary() # conn -> ticket applied
+        if other._auto_cleanup:
+            _weakly_reference(other)
+        return other
+
+    def cleanup(self, new_host = ''):
+        if self.pool:
+            for conn in self.pool.active | self.pool.idle:
+                try:
+                    conn.disconnect()
+                except NetworkException:
+                    pass
+                conn.release(True)
         if self.do_configure: 
+            if new_host:
+                d = self.do_configure.setdefault('_overrides',{})
+                d['irods_host'] = new_host
+                self.__configured = None
             self.__configured = self.configure(**self.do_configure)
 
     def _configure_account(self, **kwargs):

irods/test/ticket_test.py

@@ -163,9 +163,10 @@ def test_ticket_expiry (self):
     def test_object_read_and_write_tickets(self):
         if self.alice is None or self.bob is None:
             self.skipTest("A rodsuser (alice and/or bob) could not be created.")
-        t=None
-        data_objs=[]
-        tmpfiles=[]
+        t = None
+        data_objs = []
+        tmpfiles = []
+        tickets = {}
         try:
             # Create ticket for read access to alice's home collection.
             alice = self.login(self.alice)
@@ -174,14 +175,14 @@ def test_object_read_and_write_tickets(self):
             # Create 'R' and 'W' in alice's home collection.
             data_objs = [helpers.make_object(alice,home.path+"/"+name,content='abcxyz') for name in ('R','W')]
             tickets = {
-                'R': Ticket(alice).issue('read',  home.path + "/R").string,
-                'W': Ticket(alice).issue('write', home.path + "/W").string
+                'R': Ticket(alice).issue('read',  home.path + "/R"),
+                'W': Ticket(alice).issue('write', home.path + "/W")
             }
             # Test only write ticket allows upload.
             with self.login(self.bob) as bob:
                 rw_names={}
                 for name in  ('R','W'):
-                    Ticket( bob, tickets[name] ).supply()
+                    Ticket( bob, tickets[name].string ).supply()
                     with tempfile.NamedTemporaryFile (delete=False) as tmpf:
                         tmpfiles += [tmpf]
                         rw_names[name] = tmpf.name
@@ -197,16 +198,17 @@ def test_object_read_and_write_tickets(self):
                             raise AssertionError("A read ticket allowed a data object write operation to happen without error.")
 
             # Test upload was successful, by getting and confirming contents.
-
             with self.login(self.bob) as bob:  # This check must be in a new session or we get CollectionDoesNotExist. - Possibly a new issue [ ]
                 for name in  ('R','W'):
-                    Ticket( bob, tickets[name] ).supply()
+                    bob.cleanup() # clear out existing connections
+                    Ticket( bob, tickets[name].string ).supply()
                     bob.data_objects.get(home.path+"/"+name,rw_names[ name ],**{kw.FORCE_FLAG_KW:''})
                     with open(rw_names[ name ],'r') as tmpread:
                         self.assertEqual(tmpread.read(),
                                          'abcxyz' if name == 'R' else 'hello')
         finally:
-            if t: t.delete()
+            for t in tickets.values():
+                t.delete()
             for d in data_objs:
                 d.unlink(force=True)
             for file_ in tmpfiles: os.unlink( file_.name )

irods/ticket.py

@@ -57,15 +57,19 @@ def _generate(self, length=15, allow_punctuation = False):
             source_characters += string.punctuation
         return ''.join(random.SystemRandom().choice(source_characters) for _ in range(length))
 
-    def _api_request(self,cmd_string,*args, **opts):
-        message_body = TicketAdminRequest(cmd_string, self.ticket, *args, **opts)
-        message = iRODSMessage("RODS_API_REQ", msg=message_body, int_info=api_number['TICKET_ADMIN_AN'])
-
+    def _api_request(self, cmd_string, *args, **opts):
         with self.session.pool.get_connection() as conn:
-            conn.send(message)
-            response = conn.recv()
+            self._lowlevel_api_request(conn, cmd_string, self.ticket, *args, **opts)
         return self
 
+    @staticmethod
+    def _lowlevel_api_request(conn_, cmd_string, ticket_string, *args, **opts):
+        message_body = TicketAdminRequest(cmd_string, ticket_string, *args, **opts)
+        message = iRODSMessage("RODS_API_REQ", msg=message_body, int_info=api_number['TICKET_ADMIN_AN'])
+        conn_.send(message)
+        response = conn_.recv()
+        return response
+
     def issue(self,permission,target,**opt): return self._api_request("create",permission,target,**opt)
 
     create = issue
@@ -77,9 +81,8 @@ def modify(self,*args,**opt):
         return self._api_request("mod",*arglist,**opt)
 
     def supply(self,**opt):
-        object_ = self._api_request("session",**opt)
         self.session.ticket__ = self._ticket
-        return object_
+        return self
 
     def delete(self,**opt):
         """