custom irods_env_path support; pass along kwargs

cubi_tk/irods_common.py

@@ -35,12 +35,20 @@ class TransferJob:
 
 
 class iRODSCommon:
-    """Implementation of common iRODS utility functions."""
+    """
+    Implementation of common iRODS utility functions.
 
-    def __init__(self, ask: bool = False):
-        #: Path to iRODS environment file
-        # TODO: custom env path
-        self.irods_env_path = Path.home().joinpath(".irods", "irods_environment.json")
+    Attributes:
+    ask -- Confirm with user before certain actions.
+    irods_env_path -- Path to irods_environment.json
+    """
+
+    def __init__(self, ask: bool = False, irods_env_path: Path = None):
+        # Path to iRODS environment file
+        if irods_env_path is None:
+            self.irods_env_path = Path.home().joinpath(".irods", "irods_environment.json")
+        else:
+            self.irods_env_path = irods_env_path
         self.ask = ask
 
         # check for outdated .irodsA file
@@ -136,8 +144,8 @@ class iRODSTransfer(iRODSCommon):
     jobs -- iterable of TransferJob objects
     """
 
-    def __init__(self, jobs: Iterable[TransferJob]):
-        super().__init__()
+    def __init__(self, jobs: Iterable[TransferJob], **kwargs):
+        super().__init__(**kwargs)
         with self._get_irods_sessions(1) as s:
             self.session = s[0]  # TODO: use more sessions
         self.jobs = jobs

tests/test_irods_common.py

@@ -17,6 +17,8 @@ def fake_filesystem(fs):
 @patch("cubi_tk.irods_common.iRODSSession")
 def test_common_init(mocksession):
     assert iRODSCommon().irods_env_path is not None
+    icommon = iRODSCommon(irods_env_path="a/b/c.json")
+    assert icommon.irods_env_path == "a/b/c.json"
     assert type(iRODSCommon().ask) is bool
 
 
@@ -102,6 +104,11 @@ def test_irods_transfer_init(jobs, itransfer):
     assert itransfer.total_bytes == sum([job.bytes for job in jobs])
     assert itransfer.destinations == [job.path_dest for job in jobs]
 
+    with patch("cubi_tk.irods_common.iRODSSession") as mocksession:
+        itransferc = iRODSTransfer(jobs=jobs, irods_env_path="a/b/c", ask=True)
+        assert itransferc.irods_env_path == "a/b/c"
+        assert itransferc.ask == True
+
 
 def test_irods_transfer_put(fs, itransfer, jobs):
     for job in jobs: