diff --git a/docs/conf.py b/docs/conf.py index b80c159..f92e0bf 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -17,9 +17,9 @@ # -- Project information ----------------------------------------------------- -project = u"fs_irods" -copyright = u"2023, Helge Hecht" -author = u"Helge Hecht" +project = "fs_irods" +copyright = "2023, Helge Hecht" +author = "Helge Hecht" # The version info for the project you're documenting, acts as replacement for # |version| and |release|, also used in various other places throughout the @@ -61,7 +61,7 @@ # -- Use autoapi.extension to run sphinx-apidoc ------- -autoapi_dirs = ['../fs_irods'] +autoapi_dirs = ["../fs_irods"] # -- Options for HTML output ---------------------------------------------- @@ -78,11 +78,12 @@ # -- Options for Intersphinx -intersphinx_mapping = {'python': ('https://docs.python.org/3', None), - # Commonly used libraries, uncomment when used in package - # 'numpy': ('http://docs.scipy.org/doc/numpy/', None), - # 'scipy': ('http://docs.scipy.org/doc/scipy/reference/', None), - # 'scikit-learn': ('https://scikit-learn.org/stable/', None), - # 'matplotlib': ('https://matplotlib.org/stable/', None), - # 'pandas': ('http://pandas.pydata.org/docs/', None), - } +intersphinx_mapping = { + "python": ("https://docs.python.org/3", None), + # Commonly used libraries, uncomment when used in package + # 'numpy': ('http://docs.scipy.org/doc/numpy/', None), + # 'scipy': ('http://docs.scipy.org/doc/scipy/reference/', None), + # 'scikit-learn': ('https://scikit-learn.org/stable/', None), + # 'matplotlib': ('https://matplotlib.org/stable/', None), + # 'pandas': ('http://pandas.pydata.org/docs/', None), +} diff --git a/fs_irods/__init__.py b/fs_irods/__init__.py index 6799d13..be26405 100644 --- a/fs_irods/__init__.py +++ b/fs_irods/__init__.py @@ -9,4 +9,4 @@ __email__ = "helge.hecht@recetox.muni.cz" __version__ = "0.2.0" -__all__ = ["iRODSFS", "can_create"] \ No newline at end of file +__all__ = ["iRODSFS", "can_create"] diff --git a/fs_irods/iRODSFS.py b/fs_irods/iRODSFS.py index b4fbd3a..fc4f51a 100644 --- a/fs_irods/iRODSFS.py +++ b/fs_irods/iRODSFS.py @@ -1,53 +1,76 @@ import datetime -from io import BufferedRandom import io +import logging import os - +from io import BufferedRandom from multiprocessing import RLock -from typing import Text +from weakref import WeakKeyDictionary from fs.base import FS -from fs.errors import DirectoryExists, ResourceNotFound, RemoveRootError, DirectoryExpected, FileExpected, FileExists, DirectoryNotEmpty, DestinationExists +from fs.errors import DestinationExists +from fs.errors import DirectoryExists +from fs.errors import DirectoryExpected +from fs.errors import DirectoryNotEmpty +from fs.errors import FileExists +from fs.errors import FileExpected +from fs.errors import RemoveRootError +from fs.errors import ResourceNotFound from fs.info import Info from fs.permissions import Permissions from fs.walk import Walker - -from irods.session import iRODSSession +from irods.at_client_exit import register_for_execution_before_prc_cleanup from irods.collection import iRODSCollection -from irods.path import iRODSPath from irods.data_object import iRODSDataObject +from irods.path import iRODSPath +from irods.session import iRODSSession +from fs_irods.utils import can_create +fses = WeakKeyDictionary() +_logger = logging.getLogger(__name__) -from fs_irods.utils import can_create -_utc=datetime.timezone(datetime.timedelta(0)) +# Close out dangling file handles. +def finalize(): + for fs in list(fses): + fs._finalize_files() + + +register_for_execution_before_prc_cleanup(finalize) + +_utc = datetime.timezone(datetime.timedelta(0)) + class iRODSFS(FS): - def __init__(self, session: iRODSSession, root: str|None = None) -> None: + def __init__(self, session: iRODSSession, root: str | None = None) -> None: super().__init__() self._lock = RLock() self._host = session.host self._port = session.port self._zone = session.zone - self._session = session + self._finalizing = False + self.files = WeakKeyDictionary() + fses[self] = None self._root = root if root else self._zone def wrap(self, path: str) -> str: return str(iRODSPath(self._root, path)) - + def parent(self, path: str): return os.path.dirname(path) - - def getinfo(self, path: str, namespaces: list|None = None) -> Info: + + def getinfo(self, path: str, namespaces: list | None = None) -> Info: """Get information about a resource on the filesystem. + Args: path (str): A path to a resource on the filesystem. namespaces (list, optional): Info namespaces to query. If namespaces is None, then all available namespaces are queried. Defaults to None. + Returns: Info: An Info object containing information about the resource. + Raises: ResourceNotFound: If the path does not exist. """ @@ -56,7 +79,7 @@ def getinfo(self, path: str, namespaces: list|None = None) -> Info: with self._lock: raw_info: dict = {"basic": {}, "details": {}, "access": {}} path = self.wrap(path) - data_object: iRODSDataObject|iRODSCollection = None + data_object: iRODSDataObject | iRODSCollection = None if self._session.data_objects.exists(path): data_object = self._session.data_objects.get(path) @@ -65,7 +88,7 @@ def getinfo(self, path: str, namespaces: list|None = None) -> Info: raw_info["details"]["size"] = data_object.size raw_info["details"]["checksum"] = data_object.checksum raw_info["details"]["comments"] = data_object.comments - raw_info["details"]["expiry"] = data_object.expiry # datatype: string + raw_info["details"]["expiry"] = data_object.expiry # datatype: string elif self._session.collections.exists(path): data_object = self._session.collections.get(path) raw_info["basic"]["is_dir"] = True @@ -78,13 +101,16 @@ def getinfo(self, path: str, namespaces: list|None = None) -> Info: raw_info["details"]["created"] = data_object.create_time.replace(tzinfo=_utc).timestamp() return Info(raw_info) - + def listdir(self, path: str) -> list: """List a directory on the filesystem. + Args: path (str): A path to a directory on the filesystem. + Returns: list: A list of resources in the directory. + Raises: ResourceNotFound: If the path does not exist. DirectoryExpected: If the path is not a directory. @@ -94,8 +120,9 @@ def listdir(self, path: str) -> list: coll: iRODSCollection = self._session.collections.get(self.wrap(path)) return [item.path for item in coll.data_objects + coll.subcollections] - def makedir(self, path: str, permissions: Permissions|None = None, recreate: bool = False): + def makedir(self, path: str, permissions: Permissions | None = None, recreate: bool = False): """Make a directory on the filesystem. + Args: path (str): A path to a directory on the filesystem. permissions (Permissions, optional): A Permissions instance, @@ -103,6 +130,7 @@ def makedir(self, path: str, permissions: Permissions|None = None, recreate: boo recreate (bool, optional): If False (the default) raise an error if the directory already exists, if True do not raise an error. Defaults to False. + Raises: DirectoryExists: If the directory already exists and recreate is False. @@ -110,15 +138,72 @@ def makedir(self, path: str, permissions: Permissions|None = None, recreate: boo """ if self.isdir(path) and not recreate: raise DirectoryExists(path) - + if not self.isdir(os.path.dirname(path)): raise ResourceNotFound(path) - + with self._lock: self._session.collections.create(self.wrap(path), recurse=False) - - def openbin(self, path: str, mode:str = "r", buffering: int = -1, **options) -> BufferedRandom: + + def _finalize_files(self): + self._finalizing = True + l = list(self.files) + while l: + f = l.pop() + if not f.closed: + f.close() + + def __del__(self): + if not self._finalizing: + self._finalize_files() + + def open( + self, + path: str, + mode: str = "r", + buffering: int = -1, + encoding: str | None = None, + errors: str | None = None, + newline: str = "", + **options, + ): + """Open a file. + + Stores weak references to open file handles that maintain a hard reference to the iRODSFS object. + In this way, the iRODSFS can only be destructed once these file handles are gone. + + Arguments: + path (str): A path to a file on the filesystem. + mode (str): Mode to open the file object with + (defaults to *r*). + buffering (int): Buffering policy (-1 to use + default buffering, 0 to disable buffering, 1 to select + line buffering, of any positive integer to indicate + a buffer size). + encoding (str): Encoding for text files (defaults to + ``utf-8``) + errors (str, optional): What to do with unicode decode errors + (see `codecs` module for more information). + newline (str): Newline parameter. + **options: keyword arguments for any additional information + required by the filesystem (if any). + + Returns: + io.IOBase: a *file-like* object. + + Raises: + fs.errors.FileExpected: If the path is not a file. + fs.errors.FileExists: If the file exists, and *exclusive mode* + is specified (``x`` in the mode). + fs.errors.ResourceNotFound: If the path does not exist. + """ + fd = super().open(path, mode, buffering, encoding, errors, newline, **options) + self.files[fd] = self + return fd + + def openbin(self, path: str, mode: str = "r", buffering: int = -1, **options) -> BufferedRandom: """Open a binary file-like object on the filesystem. + Args: path (str): A path to a file on the filesystem. mode (str, optional): The mode to open the file in, see @@ -128,8 +213,10 @@ def openbin(self, path: str, mode:str = "r", buffering: int = -1, **options) -> file, see the built-in open() function for details. Defaults to -1. **options: Additional options to pass to the open() function. + Returns: IO: A file-like object representing the file. + Raises: ResourceNotFound: If the path does not exist and mode does not imply creating the file, or if any ancestor of path does not exist. @@ -147,34 +234,35 @@ def openbin(self, path: str, mode:str = "r", buffering: int = -1, **options) -> with self._lock: mode = mode.replace("b", "") file = self._session.data_objects.open( - self.wrap(path), - mode, - create, - allow_redirect=False, - auto_close=False, - **options + self.wrap(path), mode, create, allow_redirect=False, auto_close=False, **options ) - if 'a' in mode: + if "a" in mode: file.seek(0, io.SEEK_END) + + self.files[file] = self return file - + def remove(self, path: str): """Remove a file from the filesystem. + Args: path (str): A path to a file on the filesystem. + Raises: ResourceNotFound: If the path does not exist. FileExpected: If the path is not a file. """ self._check_isfile(path) - + with self._lock: self._session.data_objects.unlink(self.wrap(path)) def _check_isfile(self, path: str): """Check if a path points to a file and raise an FileExpected error if not. + Args: path (str): A path to a file on the filesystem. + Raises: ResourceNotFound: If the path does not exist. FileExpected: If the path is not a file. @@ -182,11 +270,13 @@ def _check_isfile(self, path: str): self._check_exists(path) if not self.isfile(path): raise FileExpected(path) - + def removedir(self, path: str): """Remove a directory from the filesystem. + Args: path (str): A path to a directory on the filesystem. + Raises: ResourceNotFound: If the path does not exist. DirectoryExpected: If the path is not a directory. @@ -215,11 +305,12 @@ def _is_root(self, path: str) -> bool: return path in ["/", "", self._zone] def removetree(self, path: str): - """Recursively remove a directory and all its contents. + """Recursively remove a directory and all its contents. This method is similar to removedir, but will remove the contents of the directory if it is not empty. Args: path (str): A path to a directory on the filesystem. + Raises: ResourceNotFound: If the path does not exist. DirectoryExpected: If the path is not a directory. @@ -241,8 +332,10 @@ def removetree(self, path: str): def _check_isdir(self, path: str): """Check if a path is a directory. + Args: path (str): A path to a resource on the filesystem. + Raises: ResourceNotFound: If the path does not exist. DirectoryExpected: If the path is not a directory. @@ -250,16 +343,16 @@ def _check_isdir(self, path: str): self._check_exists(path) if not self.isdir(path): raise DirectoryExpected(path) - + def setinfo(self, path: str, info: dict) -> None: """Set information about a resource on the filesystem. - + Supports setting file metadata via the 'details' namespace including: - modified: Unix timestamp for modification time - created: Unix timestamp for creation time - comments: Text comments/description - expiry: Str timestamp for expiration/retention date - + Args: path (str): A path to a resource on the filesystem. info (dict): A dictionary containing the information to set. @@ -269,15 +362,15 @@ def setinfo(self, path: str, info: dict) -> None: "comments": , "expiry": }} + Raises: ResourceNotFound: If the path does not exist. FileExpected: If the path is not a file. ValueError: If any field value is invalid. """ - self._check_exists(path) self._check_isfile(path) - + wrapped_path = self.wrap(path) meta_dict = {} @@ -320,22 +413,21 @@ def setinfo(self, path: str, info: dict) -> None: if expiry_timestamp < 0: raise ValueError("'expiry' timestamp must be >= 0") meta_dict["dataExpiry"] = str(expiry_timestamp) - + # If there are no fields to set, return early if not meta_dict: return - + with self._lock: # Use modDataObjMeta to update the metadata - self._session.data_objects.modDataObjMeta( - {"objPath": wrapped_path}, - meta_dict - ) + self._session.data_objects.modDataObjMeta({"objPath": wrapped_path}, meta_dict) - def _check_exists(self, path:str): + def _check_exists(self, path: str): """Check if a resource exists. + Args: path (str): A path to a resource on the filesystem. + Raises: ResourceNotFound: If the path does not exist. """ @@ -343,31 +435,37 @@ def _check_exists(self, path:str): path = self.wrap(path) if not self._session.data_objects.exists(path) and not self._session.collections.exists(path): raise ResourceNotFound(path) - + def isfile(self, path: str) -> bool: """Check if a path is a file. + Args: path (str): A path to a resource on the filesystem. + Returns: bool: True if the path is a file, False otherwise. - """ + """ with self._lock: return self._session.data_objects.exists(self.wrap(path)) - + def isdir(self, path: str) -> bool: """Check if a path is a directory. + Args: path (str): A path to a resource on the filesystem. + Returns: bool: True if the path is a directory, False otherwise. """ with self._lock: return self._session.collections.exists(self.wrap(path)) - def create(self, path:str): + def create(self, path: str): """Create a file on the filesystem. + Args: path (str): A path to a file on the filesystem. + Raises: ResourceNotFound: If any ancestor of path does not exist. FileExists: If the path exists. @@ -405,8 +503,10 @@ def points_into_collection(self, path: str) -> bool: def exists(self, path: str) -> bool: """Check if a resource exists. + Args: path (str): A path to a resource on the filesystem. + Returns: bool: True if the path exists, False otherwise. """ @@ -415,13 +515,14 @@ def exists(self, path: str) -> bool: return self._session.data_objects.exists(path) or self._session.collections.exists(path) def move(self, src_path: str, dst_path: str, overwrite: bool = False, preserve_time: bool = False) -> None: - """Move a file to the specified location + """Move a file to the specified location. Args: src_path (str): Path to the current location of the file dst_path (str): Path to the target location of the file overwrite (bool, optional): Set to True to overwrite an existing destination file. Defaults to False. preserve_time (bool, optional): Set to True to preserve the original modification time. Defaults to False. + Raises: ResourceNotFound: If the path does not exist. FileExpected: If the source path is not a file. @@ -434,15 +535,16 @@ def move(self, src_path: str, dst_path: str, overwrite: bool = False, preserve_t raise DestinationExists(dst_path) with self._lock: self._session.data_objects.move(self.wrap(src_path), self.wrap(dst_path)) - + def copy(self, src_path: str, dst_path: str, overwrite: bool = False, preserve_time: bool = False): - """copy a file from one position to another + """Copy a file from one position to another. Args: src_path (str): Path to source file to copy dst_path (str): Destination overwrite (bool, optional): Whether to overwrite if the destination exists. Defaults to False. preserve_time (bool, optional): Whether to preserve the original modification time. Defaults to False. + Raises: DestinationExists: If ``dst_path`` exists and ``overwrite`` is `False`. ResourceNotFound: If a parent directory of ``dst_path`` does not exist. @@ -460,7 +562,7 @@ def copy(self, src_path: str, dst_path: str, overwrite: bool = False, preserve_t self.remove(dst_path) else: self._check_points_into_collection(dst_path) - + with self._lock: self._session.data_objects.copy(self.wrap(src_path), self.wrap(dst_path)) @@ -470,7 +572,6 @@ def copy(self, src_path: str, dst_path: str, overwrite: bool = False, preserve_t if modified_time is not None: self.setinfo(dst_path, {"details": {"modified": int(modified_time)}}) - def copydir(self, src_path: str, dst_path: str, create: bool = False, preserve_time: bool = False): """Copy the contents of the folder src_path to dst_path. @@ -479,13 +580,13 @@ def copydir(self, src_path: str, dst_path: str, create: bool = False, preserve_t dst_path (str): Where to copy the folder to. create (bool, optional): Create the target directory if it does not exist. Defaults to False. preserve_time (bool, optional): Perserve the modification time. Defaults to False. + Raises: ResourceNotFound: If the ``dst_path`` does not exist, and ``create`` is not `True`. DirectoryExpected: If ``src_path`` is not a directory. """ - self._check_isdir(src_path) - + src_basename = os.path.basename(src_path) dst = os.path.join(dst_path, src_basename) @@ -518,7 +619,7 @@ def copydir(self, src_path: str, dst_path: str, create: bool = False, preserve_t modified_time = src_info.raw.get("details", {}).get("modified") if modified_time is not None: self.setinfo(dst_dir, {"details": {"modified": int(modified_time)}}) - + for file_entry in files: # file_entry may be a string name or an Info-like object file_name = getattr(file_entry, "name", file_entry) @@ -526,8 +627,8 @@ def copydir(self, src_path: str, dst_path: str, create: bool = False, preserve_t src_file = os.path.join(path, file_name) dst_file = os.path.join(target_dir, file_name) self.copy(src_file, dst_file, overwrite=True, preserve_time=preserve_time) - - def upload(self, path: str, file: io.IOBase | str, chunk_size: int|None = None, **options): + + def upload(self, path: str, file: io.IOBase | str, chunk_size: int | None = None, **options): """Set a file to the contents of a binary file object. This method copies bytes from an open binary file to a file on @@ -561,15 +662,10 @@ def upload(self, path: str, file: io.IOBase | str, chunk_size: int|None = None, elif isinstance(file, str): self._check_points_into_collection(path) with self._lock: - self._session.data_objects.put( - file, - self.wrap(path), - allow_redirect=False, - auto_close=False - ) + self._session.data_objects.put(file, self.wrap(path), allow_redirect=False, auto_close=False) else: raise NotImplementedError() - + def download(self, path: str, file: io.IOBase | str, chunk_size=None, **options): """Copy a file from the filesystem to a file-like object. @@ -599,14 +695,9 @@ def download(self, path: str, file: io.IOBase | str, chunk_size=None, **options) """ if isinstance(file, io.IOBase): super().download(path, file, chunk_size=chunk_size, **options) - elif(isinstance(file, str)): + elif isinstance(file, str): with self._lock: self._check_exists(path) - self._session.data_objects.get( - self.wrap(path), - file, - allow_redirect=False, - auto_close=False - ) + self._session.data_objects.get(self.wrap(path), file, allow_redirect=False, auto_close=False) else: raise NotImplementedError() diff --git a/pyproject.toml b/pyproject.toml index 93499fc..21d2064 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,7 +27,7 @@ url = "https://pypi.org/simple/" [tool.poetry.dependencies] python = "^3.9" fs = "^2.4.16" -python-irodsclient = "^3.0.0" +python-irodsclient = "^3.2.0" [tool.poetry.group.dev.dependencies] diff --git a/tests/DelayedSession.py b/tests/DelayedSession.py index 75b4d13..65f1f1d 100644 --- a/tests/DelayedSession.py +++ b/tests/DelayedSession.py @@ -1,8 +1,8 @@ -from irods.session import iRODSSession import time +from irods.session import iRODSSession class DelayedSession(iRODSSession): def __exit__(self, exc_type, exc_value, traceback): time.sleep(1) - return super().__exit__(exc_type, exc_value, traceback) \ No newline at end of file + return super().__exit__(exc_type, exc_value, traceback) diff --git a/tests/iRODSFSBuilder.py b/tests/iRODSFSBuilder.py index a8d50b1..d0f68cd 100644 --- a/tests/iRODSFSBuilder.py +++ b/tests/iRODSFSBuilder.py @@ -1,45 +1,46 @@ from fs_irods import iRODSFS - -from irods.session import iRODSSession from tests.DelayedSession import DelayedSession + class iRODSFSBuilder: def __init__(self): - self._host = 'localhost' + self._host = "localhost" self._port = 1247 - self._user = 'rods' - self._password = 'rods' - self._zone = 'tempZone' - self._session = DelayedSession(host=self._host, port=self._port, user=self._user, password=self._password, zone=self._zone) + self._user = "rods" + self._password = "rods" + self._zone = "tempZone" + self._session = DelayedSession( + host=self._host, port=self._port, user=self._user, password=self._password, zone=self._zone + ) self._root = None def with_host(self, host): self._host = host return self - + def with_port(self, port): self._port = port return self - + def with_user(self, user): self._user = user return self - + def with_password(self, password): self._password = password return self - + def with_zone(self, zone): self._zone = zone return self - + def with_session(self, session): self._session = session return self - + def with_root(self, root): self._root = root return self - + def build(self): - return iRODSFS(self._session, self._root) \ No newline at end of file + return iRODSFS(self._session, self._root) diff --git a/tests/test_fs_extension.py b/tests/test_fs_extension.py index 483a047..2b942ba 100644 --- a/tests/test_fs_extension.py +++ b/tests/test_fs_extension.py @@ -3,16 +3,15 @@ import pytest from fs.test import FSTestCases from fs_irods import iRODSFS - from tests.iRODSFSBuilder import iRODSFSBuilder + @pytest.mark.skip class TestMyFS(FSTestCases, unittest.TestCase): - def make_fs(self): sut = iRODSFSBuilder().build() return sut - + def destroy_fs(self, fs: iRODSFS): # fs.removetree("/") if fs.exists("foo"): diff --git a/tests/test_iRODSFS.py b/tests/test_iRODSFS.py index a32a708..ea545fd 100644 --- a/tests/test_iRODSFS.py +++ b/tests/test_iRODSFS.py @@ -1,21 +1,18 @@ +import os import time -from typing import Generator, List +from typing import List import pytest -import os - -from fs_irods import iRODSFS -from unittest.mock import patch -from tests.DelayedSession import DelayedSession -from tests.iRODSFSBuilder import iRODSFSBuilder - +import six from fs.errors import * from fs.walk import Walker -import six - +from fs_irods import iRODSFS +from tests.iRODSFSBuilder import iRODSFSBuilder from tests.test_utils import WalkResult + def assert_bytes(fs: iRODSFS, path: str, contents: bytes): """Assert a file contains the given bytes. + Arguments: path (str): A path on the filesystem. contents (bytes): Bytes to compare. @@ -47,8 +44,8 @@ def fs(): sut.remove("/tempZone/existing_file.txt") if sut.exists("/tempZone/new_collection"): sut.removetree("/tempZone/new_collection") - - del(sut) + + del sut builder._session.cleanup() @@ -59,81 +56,83 @@ def test_default_state(): assert len(list(sut.scandir("/tempZone"))) == 2 - del(sut) + del sut builder._session.cleanup() -@pytest.mark.parametrize("path, expected", [ - ["/", 1], ["/tempZone", 4] -]) +@pytest.mark.parametrize("path, expected", [["/", 1], ["/tempZone", 4]]) def test_scandir(fs: iRODSFS, path: str, expected: int): actual = list(fs.scandir(path)) assert len(actual) == expected - - -@pytest.mark.parametrize("path, expected", [ - ["/tempZone/home", True], - ["/tempZone/home/rods", True], - ["/tempZone/existing_file.txt", False], - ["/tempZone/i_dont_exist", False] -]) + + +@pytest.mark.parametrize( + "path, expected", + [ + ["/tempZone/home", True], + ["/tempZone/home/rods", True], + ["/tempZone/existing_file.txt", False], + ["/tempZone/i_dont_exist", False], + ], +) def test_isdir(fs: iRODSFS, path: str, expected: bool): assert fs.isdir(path) == expected -@pytest.mark.parametrize("path, expected", [ - ["/tempZone/existing_file.txt", True], - ["/tempZone/existing_collection/existing_file.txt", True], - ["/tempZone/i_dont_exist", False], - ["/tempZone/new_collection", False], - ["/tempZone", False], -]) -def test_isfile(fs: iRODSFS, path:str, expected: bool): +@pytest.mark.parametrize( + "path, expected", + [ + ["/tempZone/existing_file.txt", True], + ["/tempZone/existing_collection/existing_file.txt", True], + ["/tempZone/i_dont_exist", False], + ["/tempZone/new_collection", False], + ["/tempZone", False], + ], +) +def test_isfile(fs: iRODSFS, path: str, expected: bool): assert fs.isfile(path) == expected -@pytest.mark.parametrize("path", [ - "/tempZone/test", "/tempZone/home/rods/test" -]) -def test_makedir(fs: iRODSFS, path:str): +@pytest.mark.parametrize("path", ["/tempZone/test", "/tempZone/home/rods/test"]) +def test_makedir(fs: iRODSFS, path: str): fs.makedir(path) - assert fs.isdir(path) == True + assert fs.isdir(path) is True fs.removedir(path) - assert fs.isdir(path) == False + assert fs.isdir(path) is False -@pytest.mark.parametrize("path, exception", [ - ["/tempZone/home", DirectoryExists], - ["/tempZone/test/subcollection", ResourceNotFound] -]) -def test_makedir_exceptions(fs:iRODSFS, path: str, exception: type): +@pytest.mark.parametrize( + "path, exception", [["/tempZone/home", DirectoryExists], ["/tempZone/test/subcollection", ResourceNotFound]] +) +def test_makedir_exceptions(fs: iRODSFS, path: str, exception: type): with pytest.raises(exception): fs.makedir(path) -@pytest.mark.parametrize("path", [ - "/tempZone/test.txt", "/tempZone/home/rods/test.txt" -]) +@pytest.mark.parametrize("path", ["/tempZone/test.txt", "/tempZone/home/rods/test.txt"]) def test_create_remove(fs: iRODSFS, path): fs.create(path) - assert fs.isfile(path) == True + assert fs.isfile(path) is True fs.remove(path) - assert fs.isfile(path) == False + assert fs.isfile(path) is False -@pytest.mark.parametrize("path, exception", [ - ["/tempZone/missing_collection/file.txt", ResourceNotFound], - ["/tempZone/existing_file.txt", FileExists] -]) +@pytest.mark.parametrize( + "path, exception", + [["/tempZone/missing_collection/file.txt", ResourceNotFound], ["/tempZone/existing_file.txt", FileExists]], +) def test_create_exceptions(fs: iRODSFS, path: str, exception: Exception): with pytest.raises(exception): fs.create(path) -@pytest.mark.parametrize("path, is_dir", [ - ["/tempZone/home", True], - ["/tempZone/existing_file.txt", False], -]) +@pytest.mark.parametrize( + "path, is_dir", + [ + ["/tempZone/home", True], + ["/tempZone/existing_file.txt", False], + ], +) def test_get_info(fs: iRODSFS, path: str, is_dir: bool): info = fs.getinfo(path) assert info.name == os.path.basename(path) @@ -142,59 +141,66 @@ def test_get_info(fs: iRODSFS, path: str, is_dir: bool): assert info.modified is not None assert info.created is not None - assert info.accessed is None - - -@pytest.mark.parametrize("path, expected", [ - ["/tempZone/home", True], - ["/tempZone/home/rods", True], - ["/tempZone/fakedir", False], - ["/tempZone/home/other_user", False], - ["/tempZone/existing_file.txt", True], - ["/tempZone/existing_collection/existing_file.txt", True], - ["/tempZone/existing_collection/bad_file.txt", False] -]) + assert info.accessed is None + + +@pytest.mark.parametrize( + "path, expected", + [ + ["/tempZone/home", True], + ["/tempZone/home/rods", True], + ["/tempZone/fakedir", False], + ["/tempZone/home/other_user", False], + ["/tempZone/existing_file.txt", True], + ["/tempZone/existing_collection/existing_file.txt", True], + ["/tempZone/existing_collection/bad_file.txt", False], + ], +) def test_exists(fs: iRODSFS, path: str, expected: bool): assert fs.exists(path) == expected assert fs.exists(path) == expected -@pytest.mark.parametrize("path", [ - "/tempZone/foo", "/tempZone/home/rods/test" -]) +@pytest.mark.parametrize("path", ["/tempZone/foo", "/tempZone/home/rods/test"]) def test_removedir(fs: iRODSFS, path: str): fs.makedir(path) - assert fs.isdir(path) == True + assert fs.isdir(path) is True fs.removedir(path) - assert fs.isdir(path) == False - - -@pytest.mark.parametrize("path, exception", [ - ["", RemoveRootError], - ["/", RemoveRootError], - ["/tempZone/existing_file.txt", DirectoryExpected], - ["/tempZone/home/something", ResourceNotFound], - ["/tempZone/existing_collection", DirectoryNotEmpty] -]) + assert fs.isdir(path) is False + + +@pytest.mark.parametrize( + "path, exception", + [ + ["", RemoveRootError], + ["/", RemoveRootError], + ["/tempZone/existing_file.txt", DirectoryExpected], + ["/tempZone/home/something", ResourceNotFound], + ["/tempZone/existing_collection", DirectoryNotEmpty], + ], +) def test_removedir_exceptions(fs: iRODSFS, path: str, exception: type): with pytest.raises(exception): fs.removedir(path) -@pytest.mark.parametrize("src_path, dst_path, create, preserve_time", [ - ["/tempZone/existing_collection", "/tempZone/home", False, False], - ["/tempZone/existing_collection", "/tempZone/non_existing_collection", True, False], - ["/tempZone/existing_collection", "/tempZone/home/existing_collection", True, False], - ["/tempZone/existing_collection", "/tempZone/new_collection_preserve", True, True], -]) -def test_copydir(fs:iRODSFS, src_path: str, dst_path: str, create: bool, preserve_time: bool): +@pytest.mark.parametrize( + "src_path, dst_path, create, preserve_time", + [ + ["/tempZone/existing_collection", "/tempZone/home", False, False], + ["/tempZone/existing_collection", "/tempZone/non_existing_collection", True, False], + ["/tempZone/existing_collection", "/tempZone/home/existing_collection", True, False], + ["/tempZone/existing_collection", "/tempZone/new_collection_preserve", True, True], + ], +) +def test_copydir(fs: iRODSFS, src_path: str, dst_path: str, create: bool, preserve_time: bool): # Record original modified times if preserve_time is True original_modified = None if preserve_time: src_file = os.path.join(src_path, "existing_file.txt") original_info = fs.getinfo(src_file, namespaces=["details"]) original_modified = original_info.raw["details"]["modified"] - + fs.copydir(src_path, dst_path, create, preserve_time=preserve_time) result_path = os.path.join(dst_path, os.path.basename(src_path)) @@ -212,24 +218,27 @@ def test_copydir(fs:iRODSFS, src_path: str, dst_path: str, create: bool, preserv src_file = os.path.join(src_path, entry.name) dst_file = os.path.join(result_path, entry.name) assert fs.readbytes(src_file) == fs.readbytes(dst_file) - + # Check preserve_time if enabled if preserve_time: dst_info = fs.getinfo(dst_file, namespaces=["details"]) assert dst_info.raw["details"]["modified"] == original_modified - + # Clean up result_path and parent if it was created fs.removetree(result_path) if create and fs.exists(dst_path): fs.removetree(dst_path) -@pytest.mark.parametrize("src_path, dst_path, create, exception", [ - ["/tempZone/existing_file.txt", "/", False, DirectoryExpected], - ["/tempZone/existing_collection", "/tempZone/fakeFolder", False, ResourceNotFound], - ["/tempZone/fakeFolder", "/tempZone/existing_collection", False, ResourceNotFound] -]) -def test_copydir_exceptions(fs: iRODSFS, src_path: str, dst_path: str, create:bool, exception: Exception): +@pytest.mark.parametrize( + "src_path, dst_path, create, exception", + [ + ["/tempZone/existing_file.txt", "/", False, DirectoryExpected], + ["/tempZone/existing_collection", "/tempZone/fakeFolder", False, ResourceNotFound], + ["/tempZone/fakeFolder", "/tempZone/existing_collection", False, ResourceNotFound], + ], +) +def test_copydir_exceptions(fs: iRODSFS, src_path: str, dst_path: str, create: bool, exception: Exception): with pytest.raises(exception): fs.copydir(src_path, dst_path, create=create) @@ -280,7 +289,7 @@ def test_copydir_overwrite_behavior(fs: iRODSFS): def test_copydir_nested_structure(fs: iRODSFS): src = "/tempZone/testsrc_nested" dst_parent = "/tempZone/nested_dst" - + # clean up any existing if fs.exists(src): fs.removetree(src) @@ -307,52 +316,55 @@ def test_copydir_nested_structure(fs: iRODSFS): fs.removetree(dst_parent) -@pytest.mark.parametrize("path, exception", [ - ["/tempZone/home", FileExpected], - ["/tempZone/some_file.txt", ResourceNotFound] -]) +@pytest.mark.parametrize( + "path, exception", [["/tempZone/home", FileExpected], ["/tempZone/some_file.txt", ResourceNotFound]] +) def test_remove_exceptions(fs: iRODSFS, path: str, exception: type): with pytest.raises(exception): fs.remove(path) - -@pytest.mark.parametrize("path, expected", [ - ["/tempZone", ["/tempZone/existing_file.txt", "/tempZone/existing_collection", "/tempZone/home", "/tempZone/trash"]], - ["", ["/tempZone"]], - ["/tempZone/home", ["/tempZone/home/public", "/tempZone/home/rods"]] -]) + +@pytest.mark.parametrize( + "path, expected", + [ + [ + "/tempZone", + ["/tempZone/existing_file.txt", "/tempZone/existing_collection", "/tempZone/home", "/tempZone/trash"], + ], + ["", ["/tempZone"]], + ["/tempZone/home", ["/tempZone/home/public", "/tempZone/home/rods"]], + ], +) def test_listdir(fs: iRODSFS, path: str, expected: list[str]): actual = fs.listdir(path) assert actual == expected -@pytest.mark.parametrize("path, expected", [ - ["/tempZone/home", False], - ["/tempZone/home/rods", True] -]) +@pytest.mark.parametrize("path, expected", [["/tempZone/home", False], ["/tempZone/home/rods", True]]) def test_isempty(fs: iRODSFS, path: str, expected: bool): assert fs.isempty(path) == expected -@pytest.mark.parametrize("path", [ - "/tempZone/test/subdir" -]) -def test_makedirs(fs:iRODSFS, path: str): +@pytest.mark.parametrize("path", ["/tempZone/test/subdir"]) +def test_makedirs(fs: iRODSFS, path: str): fs.makedirs(path) assert fs.isdir(path) fs.removedir(path) fs.removedir(os.path.dirname(path)) - assert fs.isdir(path) == False - assert fs.isdir(os.path.dirname(path)) == False - - -@pytest.mark.parametrize("path, recreate, exception", [ - ["/tempZone/home", False, DirectoryExists], - ["/tempZone/existing_collection/existing_file.txt/subfolder", False, DirectoryExpected] -]) -def test_makedirs_exception(fs: iRODSFS, path:str, recreate: bool, exception: Exception): + assert fs.isdir(path) is False + assert fs.isdir(os.path.dirname(path)) is False + + +@pytest.mark.parametrize( + "path, recreate, exception", + [ + ["/tempZone/home", False, DirectoryExists], + ["/tempZone/existing_collection/existing_file.txt/subfolder", False, DirectoryExpected], + ], +) +def test_makedirs_exception(fs: iRODSFS, path: str, recreate: bool, exception: Exception): with pytest.raises(exception): - fs.makedirs(path, recreate = recreate) + fs.makedirs(path, recreate=recreate) def test_removetree(fs: iRODSFS): @@ -361,9 +373,9 @@ def test_removetree(fs: iRODSFS): assert fs.isfile("/tempZone/test/subdir/file.txt") fs.removetree("/tempZone/test") - assert fs.exists("/tempZone/test/subdir/file.txt") == False - assert fs.exists("/tempZone/test/subdir") == False - assert fs.exists("/tempZone/test") == False + assert fs.exists("/tempZone/test/subdir/file.txt") is False + assert fs.exists("/tempZone/test/subdir") is False + assert fs.exists("/tempZone/test") is False @pytest.mark.skip @@ -372,42 +384,42 @@ def test_removetree_root(fs: iRODSFS): assert fs.listdir("") == ["/tempZone/trash"] -@pytest.mark.parametrize("path, expected", [ - ["home", "/home"], - ["", "/"], - ["/", "/"], - ["/tempZone/home", "/tempZone/home"], - ["/tempZone", "/tempZone"] -]) -def test_wrap(fs: iRODSFS, path: str, expected:str): +@pytest.mark.parametrize( + "path, expected", + [["home", "/home"], ["", "/"], ["/", "/"], ["/tempZone/home", "/tempZone/home"], ["/tempZone", "/tempZone"]], +) +def test_wrap(fs: iRODSFS, path: str, expected: str): assert fs.wrap(path) == expected def test_openbin(fs: iRODSFS): f = fs.openbin("/tempZone/home/rods/existing_file.txt", mode="w") assert f.writable() - assert f.closed == False + assert f.closed is False f.write("test".encode()) f.close() - assert f.closed == True + assert f.closed is True f = fs.openbin("/tempZone/home/rods/existing_file.txt", mode="r") assert f.readable() assert f.readlines() == [b"test"] f.close() - assert f.closed == True + assert f.closed is True fs.remove("/tempZone/home/rods/existing_file.txt") - - -@pytest.mark.parametrize("path, content, expected", [ - ["/tempZone/empty", b"", 0], - ["/tempZone/one", b"a", 1], - ["/tempZone/onethousand", ("b" * 1000).encode("ascii"), 1000] -]) + + +@pytest.mark.parametrize( + "path, content, expected", + [ + ["/tempZone/empty", b"", 0], + ["/tempZone/one", b"a", 1], + ["/tempZone/onethousand", ("b" * 1000).encode("ascii"), 1000], + ], +) def test_getsize(fs: iRODSFS, path: str, content: bytes, expected: int): fs.writebytes(path, content) - assert fs.getsize(path) == expected + assert fs.getsize(path) == expected fs.remove(path) @@ -416,7 +428,7 @@ def test_getsize_exception(fs: iRODSFS): fs.getsize("doesnotexist") -def test_root_dir(fs:iRODSFS): +def test_root_dir(fs: iRODSFS): with pytest.raises(FileExpected): fs.open("/") with pytest.raises(FileExpected): @@ -458,39 +470,40 @@ def test_getmeta(fs: iRODSFS): def test_move(fs: iRODSFS): fs.writetext("/tempZone/existing_file.txt", "test") fs.move("/tempZone/existing_file.txt", "/tempZone/new_file_location.txt") - + assert fs.isfile("/tempZone/new_file_location.txt") assert fs.readtext("/tempZone/new_file_location.txt") == "test" fs.remove("/tempZone/new_file_location.txt") -@pytest.mark.parametrize("source, dest, overwrite, exception", [ - ["/tempZone/non_existing_file", "/tempZone/new_location", False, ResourceNotFound], - ["/tempZone/home", "/tempZone/somewhere", False, FileExpected], - ["/tempZone/existing_file.txt", "/tempZone//existing_collection/existing_file.txt", False, DestinationExists] -]) -def test_move_exceptions(fs:iRODSFS, source: str, dest: str, overwrite:bool, exception: type): +@pytest.mark.parametrize( + "source, dest, overwrite, exception", + [ + ["/tempZone/non_existing_file", "/tempZone/new_location", False, ResourceNotFound], + ["/tempZone/home", "/tempZone/somewhere", False, FileExpected], + ["/tempZone/existing_file.txt", "/tempZone//existing_collection/existing_file.txt", False, DestinationExists], + ], +) +def test_move_exceptions(fs: iRODSFS, source: str, dest: str, overwrite: bool, exception: type): with pytest.raises(exception): fs.move(source, dest, overwrite=overwrite) -@pytest.mark.parametrize("path, content",[ - ["/tempZone/existing_file.txt", "test"] -]) -def test_writetext_readtext(fs:iRODSFS, path: str, content: str): +@pytest.mark.parametrize("path, content", [["/tempZone/existing_file.txt", "test"]]) +def test_writetext_readtext(fs: iRODSFS, path: str, content: str): fs.writetext(path, content) assert fs.readtext(path) == content def test_upload(fs: iRODSFS): testfile = os.path.join(os.path.curdir, "tests", "test-data", "test.txt") - with open(testfile, mode='rb') as file: + with open(testfile, mode="rb") as file: fs.upload("/tempZone/uploaded_file.txt", file) assert fs.readtext("/tempZone/uploaded_file.txt") == "Hello World!" fs.remove("/tempZone/uploaded_file.txt") -def test_upload_put(fs:iRODSFS): +def test_upload_put(fs: iRODSFS): testfile = os.path.join(os.path.curdir, "tests", "test-data", "test.txt") dst_path = "/tempZone/home/rods/uploaded_file.txt" @@ -499,27 +512,30 @@ def test_upload_put(fs:iRODSFS): fs.remove(dst_path) -def test_download(fs:iRODSFS, tmp_path): +def test_download(fs: iRODSFS, tmp_path): tmp_file = os.path.join(tmp_path, "downloads.txt") - with open(tmp_file, mode='wb') as file: + with open(tmp_file, mode="wb") as file: fs.download("/tempZone/existing_collection/existing_file.txt", file) - with(open(tmp_file)) as file: + with open(tmp_file) as file: assert file.read() == "content" -def test_download_get(fs:iRODSFS, tmp_path): +def test_download_get(fs: iRODSFS, tmp_path): tmp_file = os.path.join(tmp_path, "downloads.txt") fs.download("/tempZone/existing_collection/existing_file.txt", tmp_file) - with(open(tmp_file)) as file: + with open(tmp_file) as file: assert file.read() == "content" -@pytest.mark.parametrize("dst_path, result_path, overwrite", [ - ["/tempZone/existing_file_copy.txt", "/tempZone/existing_file_copy.txt", False], - ["/tempZone/home", "/tempZone/home/existing_file.txt", False], - ["/tempZone/existing_collection", "/tempZone/existing_collection/existing_file.txt", True], - ["/tempZone/existing_collection/existing_file.txt", "/tempZone/existing_collection/existing_file.txt", True], -]) +@pytest.mark.parametrize( + "dst_path, result_path, overwrite", + [ + ["/tempZone/existing_file_copy.txt", "/tempZone/existing_file_copy.txt", False], + ["/tempZone/home", "/tempZone/home/existing_file.txt", False], + ["/tempZone/existing_collection", "/tempZone/existing_collection/existing_file.txt", True], + ["/tempZone/existing_collection/existing_file.txt", "/tempZone/existing_collection/existing_file.txt", True], + ], +) def test_copy(fs: iRODSFS, dst_path: str, result_path: str, overwrite: bool): src_path = "/tempZone/existing_file.txt" fs.copy(src_path, dst_path, overwrite) @@ -530,12 +546,20 @@ def test_copy(fs: iRODSFS, dst_path: str, result_path: str, overwrite: bool): fs.remove(result_path) -@pytest.mark.parametrize("dst_path, result_path, overwrite, preserve_time", [ - ["/tempZone/existing_file_copy.txt", "/tempZone/existing_file_copy.txt", False, True], - ["/tempZone/home", "/tempZone/home/existing_file.txt", False, True], - ["/tempZone/existing_collection", "/tempZone/existing_collection/existing_file.txt", True, True], - ["/tempZone/existing_collection/existing_file.txt", "/tempZone/existing_collection/existing_file.txt", True, True], -]) +@pytest.mark.parametrize( + "dst_path, result_path, overwrite, preserve_time", + [ + ["/tempZone/existing_file_copy.txt", "/tempZone/existing_file_copy.txt", False, True], + ["/tempZone/home", "/tempZone/home/existing_file.txt", False, True], + ["/tempZone/existing_collection", "/tempZone/existing_collection/existing_file.txt", True, True], + [ + "/tempZone/existing_collection/existing_file.txt", + "/tempZone/existing_collection/existing_file.txt", + True, + True, + ], + ], +) def test_copy_preserve_time(fs: iRODSFS, dst_path: str, result_path: str, overwrite: bool, preserve_time: bool): src_path = "/tempZone/existing_file.txt" fs.copy(src_path, dst_path, overwrite, preserve_time=preserve_time) @@ -551,26 +575,32 @@ def test_copy_preserve_time(fs: iRODSFS, dst_path: str, result_path: str, overwr fs.remove(result_path) -@pytest.mark.parametrize("src_path, dst_path, overwrite, exception", [ - ["/tempZone/existing_file.txt", "/tempZone/existing_collection", False, DestinationExists], - ["not_existing.txt", "/tempZone/", False, ResourceNotFound], - ["/tempZone/existing_collection", "/tempZone/", False, FileExpected], - ["/tempZone/existing_file.txt", "/tempZone/fakeFolder/existing_file.txt", False, ResourceNotFound], - ["/tempZone/existing_file.txt", "/tempZone/fakeFolder/test", False, ResourceNotFound] -]) +@pytest.mark.parametrize( + "src_path, dst_path, overwrite, exception", + [ + ["/tempZone/existing_file.txt", "/tempZone/existing_collection", False, DestinationExists], + ["not_existing.txt", "/tempZone/", False, ResourceNotFound], + ["/tempZone/existing_collection", "/tempZone/", False, FileExpected], + ["/tempZone/existing_file.txt", "/tempZone/fakeFolder/existing_file.txt", False, ResourceNotFound], + ["/tempZone/existing_file.txt", "/tempZone/fakeFolder/test", False, ResourceNotFound], + ], +) def test_copy_exceptions(fs: iRODSFS, src_path: str, dst_path: str, overwrite: bool, exception: Exception): with pytest.raises(exception): fs.copy(src_path, dst_path, overwrite) -@pytest.mark.parametrize("path, expected", [ - ["/tempZone/", True], - ["/", True], - ["/tempZone/existing_file.txt", True], - ["existing_file.txt", True], - ["/tempZone/fakeFolder", True], - ["/tempZone/fakeFolder/test", False] -]) +@pytest.mark.parametrize( + "path, expected", + [ + ["/tempZone/", True], + ["/", True], + ["/tempZone/existing_file.txt", True], + ["existing_file.txt", True], + ["/tempZone/fakeFolder", True], + ["/tempZone/fakeFolder/test", False], + ], +) def test_points_into_collection(fs: iRODSFS, path: str, expected: bool): assert fs.points_into_collection(path) == expected @@ -586,61 +616,73 @@ def test_walk(fs: iRODSFS): assert len(actual[0].dirs) == 2 -@pytest.mark.parametrize("field, time_offset", [ - ["modified", -600], # 10 minutes earlier - ["created", -86400], # 1 day earlier -]) +@pytest.mark.parametrize( + "field, time_offset", + [ + ["modified", -600], # 10 minutes earlier + ["created", -86400], # 1 day earlier + ], +) def test_setinfo_time_fields(fs: iRODSFS, field: str, time_offset: int): """Test setting modification and creation times of a file.""" path = "/tempZone/existing_file.txt" - + # Get original info original_info = fs.getinfo(path, namespaces=["details"]) original_time = original_info.raw["details"][field] - + # Set a new time new_time = original_time + time_offset fs.setinfo(path, {"details": {field: new_time}}) - + # Verify the time was updated updated_info = fs.getinfo(path, namespaces=["details"]) assert updated_info.raw["details"][field] == new_time -@pytest.mark.parametrize("path, exception, field, value", [ - ["/tempZone/nonexistent_file.txt", ResourceNotFound, "modified", 1000000000], - ["/tempZone/existing_collection", FileExpected, "modified", 1000000000], -]) +@pytest.mark.parametrize( + "path, exception, field, value", + [ + ["/tempZone/nonexistent_file.txt", ResourceNotFound, "modified", 1000000000], + ["/tempZone/existing_collection", FileExpected, "modified", 1000000000], + ], +) def test_setinfo_exceptions(fs: iRODSFS, path: str, exception: Exception, field: str, value): """Test that setinfo raises appropriate exceptions for invalid inputs.""" with pytest.raises(exception): fs.setinfo(path, {"details": {field: value}}) -@pytest.mark.parametrize("field, value", [ - ["modified", -1], # negative timestamp - ["created", "not-a-timestamp"], # non-numeric timestamp - ["comments", 12345], # non-string comments - ["expiry", -1], # negative expiry -]) +@pytest.mark.parametrize( + "field, value", + [ + ["modified", -1], # negative timestamp + ["created", "not-a-timestamp"], # non-numeric timestamp + ["comments", 12345], # non-string comments + ["expiry", -1], # negative expiry + ], +) def test_setinfo_invalid_values(fs: iRODSFS, field: str, value): """Test that setinfo raises ValueError for invalid field values.""" with pytest.raises(ValueError): fs.setinfo("/tempZone/existing_file.txt", {"details": {field: value}}) -@pytest.mark.parametrize("field, get_value", [ - ["comments", lambda: "This is a test comment"], - ["expiry", lambda: int(time.time()) + (30 * 24 * 60 * 60)], # 30 days from now -]) +@pytest.mark.parametrize( + "field, get_value", + [ + ["comments", lambda: "This is a test comment"], + ["expiry", lambda: int(time.time()) + (30 * 24 * 60 * 60)], # 30 days from now + ], +) def test_setinfo_catalog_fields(fs: iRODSFS, field: str, get_value): """Test setting catalog fields (comments, expiry).""" path = "/tempZone/existing_file.txt" value = get_value() - + # Set field fs.setinfo(path, {"details": {field: value}}) - + # Verify field was set correctly updated_info = fs.getinfo(path, namespaces=["details"]) if field == "expiry": @@ -652,23 +694,25 @@ def test_setinfo_catalog_fields(fs: iRODSFS, field: str, get_value): def test_setinfo_all_fields(fs: iRODSFS): """Test setting multiple catalog fields at once.""" path = "/tempZone/existing_file.txt" - + current_time = int(time.time()) - + # Set all supported fields - fs.setinfo(path, { - "details": { - "modified": current_time - 600, # 10 minutes ago - "created": current_time - 86400, # 1 day ago - "comments": "Test file with all catalog", - "expiry": current_time + (90 * 24 * 60 * 60) # 90 days from now - } - }) - + fs.setinfo( + path, + { + "details": { + "modified": current_time - 600, # 10 minutes ago + "created": current_time - 86400, # 1 day ago + "comments": "Test file with all catalog", + "expiry": current_time + (90 * 24 * 60 * 60), # 90 days from now + } + }, + ) + # Verify all fields were updated updated_info = fs.getinfo(path, namespaces=["details"]) assert updated_info.raw["details"]["modified"] == current_time - 600 assert updated_info.raw["details"]["created"] == current_time - 86400 assert updated_info.raw["details"]["comments"] == "Test file with all catalog" assert int(updated_info.raw["details"]["expiry"]) == current_time + (90 * 24 * 60 * 60) - diff --git a/tests/test_utils.py b/tests/test_utils.py index 4037b4f..c29c065 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -1,25 +1,28 @@ import os -from typing import Any, List +from dataclasses import dataclass +from typing import List import pytest +from fs.info import Info from fs_irods import can_create -from dataclasses import dataclass -from fs.info import Info -@pytest.mark.parametrize("mode, expected", [ - ["r", False], - ["w", True], - ["a", True], - ["r+", False], - ["w+", True], - ["a+", True], - ["rb", False], - ["wb", True], - ["ab", True], - ["r+b", False], - ["w+b", True], - ["a+b", True] -]) +@pytest.mark.parametrize( + "mode, expected", + [ + ["r", False], + ["w", True], + ["a", True], + ["r+", False], + ["w+", True], + ["a+", True], + ["rb", False], + ["wb", True], + ["ab", True], + ["r+b", False], + ["w+b", True], + ["a+b", True], + ], +) def test_can_create(mode: str, expected: bool): assert can_create(mode) == expected @@ -31,7 +34,7 @@ class WalkResult: files: List[Info] -def make_destination(src: str, sub: str, dst: str) -> str: +def make_destination(src: str, sub: str, dst: str) -> str: """Get the destination path for an object 'sub' in 'src' when copying to 'dst'. Args: @@ -48,15 +51,17 @@ def make_destination(src: str, sub: str, dst: str) -> str: return os.path.join(dst, rel) -@pytest.mark.parametrize("src_path, dst_path, file_path, expected", [ - ["/bar", "/foo", "/bar/baz.txt", "/foo/bar/baz.txt"], - ["/bar", "/foo", "/bar/bar1/baz.txt", "/foo/bar/bar1/baz.txt"], - ["/bar", "/foo/foo1", "/bar/bar1/baz.txt", "/foo/foo1/bar/bar1/baz.txt"], - ["/bar/bar1", "/foo/foo1", "/bar/bar1/baz.txt", "/foo/foo1/bar1/baz.txt"], - ["/foo", "/foo/bar", "/foo/bar/baz.txt", "/foo/bar/foo/bar/baz.txt"], - ["/foo", "/bar", "/foo/foo1", "/bar/foo/foo1"] -]) +@pytest.mark.parametrize( + "src_path, dst_path, file_path, expected", + [ + ["/bar", "/foo", "/bar/baz.txt", "/foo/bar/baz.txt"], + ["/bar", "/foo", "/bar/bar1/baz.txt", "/foo/bar/bar1/baz.txt"], + ["/bar", "/foo/foo1", "/bar/bar1/baz.txt", "/foo/foo1/bar/bar1/baz.txt"], + ["/bar/bar1", "/foo/foo1", "/bar/bar1/baz.txt", "/foo/foo1/bar1/baz.txt"], + ["/foo", "/foo/bar", "/foo/bar/baz.txt", "/foo/bar/foo/bar/baz.txt"], + ["/foo", "/bar", "/foo/foo1", "/bar/foo/foo1"], + ], +) def test_get_destination(src_path: str, dst_path: str, file_path: str, expected: str): - actual = make_destination(src_path, file_path, dst_path) - assert actual == expected \ No newline at end of file + assert actual == expected