Merge pull request #28 from xsuite/release/v0.3.5

Release 0.3.5

Author: freddieknets

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "xaux"
-version = "0.3.4"
+version = "0.3.5"
 description = "Support tools for Xsuite packages"
 authors = ["Frederik F. Van der Veken <[email protected]>",
            "Thomas Pugnat <[email protected]>",

tests/test_singleton.py

@@ -5,7 +5,7 @@
 
 import re
 import pytest
-import numpy as np
+from random import randrange
 
 from xaux import singleton
 
@@ -226,10 +226,9 @@ def __init__(self, *args, value3=19870, **kwargs):
 
 
 def _assert_is_singleton(cls, other_cls_instances, value1_init, value2_init=None, value3_init=None):
-    rng = np.random.default_rng()
-    value1_test_vals = [value1_init, *rng.integers(low=-587692, high=6284724, size=10_000)]
-    value2_test_vals = [value2_init, *rng.integers(low=-587692, high=6284724, size=10_000)]
-    value3_test_vals = [value3_init, *rng.integers(low=-587692, high=6284724, size=10_000)]
+    value1_test_vals = [value1_init, *(randrange(-587692, 6284724) for _ in range(10_000))]
+    value2_test_vals = [value2_init, *(randrange(-587692, 6284724) for _ in range(10_000))]
+    value3_test_vals = [value3_init, *(randrange(-587692, 6284724) for _ in range(10_000))]
 
     # These tests are overly overly verbose and expanded, to try to catch all possible corner cases.
     # We are comparing every time again all instances to each other.

tests/test_version.py

@@ -6,5 +6,5 @@
 from xaux import __version__
 
 def test_version():
-    assert __version__ == '0.3.4'
+    assert __version__ == '0.3.5'
 

xaux/fs/eos.py

@@ -161,33 +161,49 @@ def _set_eos_path(self, _eos_instance=None):
 
     # Overwrite Path methods
     # ======================
+    if sys.version_info >= (3, 12):
+        def exists(self, *args, follow_symlinks=True, **kwargs):
+            if self.is_symlink() and follow_symlinks:
+                return self.resolve().exists(*args, **kwargs)
+            return _eos_exists(self.expanduser(), *args, **kwargs)
+
+    else:
+        def exists(self, *args, **kwargs):
+            if self.is_symlink():
+                return self.resolve().exists(*args, **kwargs)
+            return _eos_exists(self.expanduser(), *args, **kwargs)
+
+    if sys.version_info >= (3, 13):
+        def is_file(self, *args, follow_symlinks=True, **kwargs):
+            if self.is_symlink() and follow_symlinks:
+                return self.resolve().is_file(*args, **kwargs)
+            return _eos_is_file(self.expanduser(), *args, **kwargs)
+
+        def is_dir(self, *args, follow_symlinks=True, **kwargs):
+            if self.is_symlink() and follow_symlinks:
+                return self.resolve().is_dir(*args, **kwargs)
+            return _eos_is_dir(self.expanduser(), *args, **kwargs)
+
+    else:
+        def is_file(self, *args, **kwargs):
+            if self.is_symlink():
+                return self.resolve().is_file(*args, **kwargs)
+            return _eos_is_file(self.expanduser(), *args, **kwargs)
+
+        def is_dir(self, *args, **kwargs):
+            if self.is_symlink():
+                return self.resolve().is_dir(*args, **kwargs)
+            return _eos_is_dir(self.expanduser(), *args, **kwargs)
 
-    def exists(self, *args, **kwargs):
-        if self.is_symlink():
-            return self.resolve().exists(*args, **kwargs)
-        return _eos_exists(self.expanduser(), *args, **kwargs)
+    def is_symlink(self, *args, **kwargs):
+        return _eos_is_symlink(self.expanduser(), *args, **kwargs)
 
     def stat(self, *args, **kwargs):
-        # if self.is_symlink():
-        #     return self.resolve().stat(*args, **kwargs)
         return _eos_stat(self.expanduser(), *args, **kwargs)
 
     def lstat(self, *args, **kwargs):
         return _eos_lstat(self.expanduser(), *args, **kwargs)
 
-    def is_file(self, *args, **kwargs):
-        if self.is_symlink():
-            return self.resolve().is_file(*args, **kwargs)
-        return _eos_is_file(self.expanduser(), *args, **kwargs)
-
-    def is_dir(self, *args, **kwargs):
-        if self.is_symlink():
-            return self.resolve().is_dir(*args, **kwargs)
-        return _eos_is_dir(self.expanduser(), *args, **kwargs)
-
-    def is_symlink(self, *args, **kwargs):
-        return _eos_is_symlink(self.expanduser(), *args, **kwargs)
-
     def touch(self, *args, **kwargs):
         return _eos_touch(self.expanduser(), *args, **kwargs)
 

xaux/fs/eos_methods.py

@@ -4,6 +4,7 @@
 # ######################################### #
 
 import os
+import sys
 import stat
 from subprocess import run, PIPE, CalledProcessError
 from pathlib import Path
@@ -109,17 +110,6 @@ def is_egroup_member(egroup, verbose=False):
 # Overwrite Path methods
 # ======================
 
-def _eos_exists(path, *args, **kwargs):
-    _assert_eos_accessible("Cannot stat EOS paths.")
-    try:
-        ftype, _ = _get_type(path, *args, **kwargs)
-    except FileNotFoundError:
-        return False
-    if ftype is not None:
-        return True
-    return Path(path.eos_path).exists(*args, **kwargs)
-
-
 def _get_type(path, *args, **kwargs):
     success, result = _run_eos(['eos', 'stat', path.eos_path], mgm=path.mgm,
                         _false_if_stderr_contains='failed to stat', **kwargs)
@@ -138,6 +128,16 @@ def _get_type(path, *args, **kwargs):
             raise NotImplementedError(f"File type not known. Output:\n{result}")
     return None, None
 
+def _eos_exists(path, *args, **kwargs):
+    _assert_eos_accessible("Cannot stat EOS paths.")
+    try:
+        ftype, _ = _get_type(path, *args, **kwargs)
+    except FileNotFoundError:
+        return False
+    if ftype is not None:
+        return True
+    return Path(path.eos_path).exists()
+
 def _eos_is_file(path, *args, **kwargs):
     _assert_eos_accessible("Cannot stat EOS paths.")
     try:
@@ -146,7 +146,7 @@ def _eos_is_file(path, *args, **kwargs):
         return False
     if ftype is not None:
         return ftype == stat.S_IFREG
-    return Path.is_file(path, *args, **kwargs)
+    return Path.is_file(path)
 
 def _eos_is_dir(path, *args, **kwargs):
     _assert_eos_accessible("Cannot stat EOS paths.")
@@ -156,7 +156,7 @@ def _eos_is_dir(path, *args, **kwargs):
         return False
     if ftype is not None:
         return ftype == stat.S_IFDIR
-    return Path.is_dir(path, *args, **kwargs)
+    return Path.is_dir(path)
 
 def _eos_is_symlink(path, *args, **kwargs):
     _assert_eos_accessible("Cannot stat EOS paths.")
@@ -166,7 +166,7 @@ def _eos_is_symlink(path, *args, **kwargs):
         return False
     if ftype is not None:
         return ftype == stat.S_IFLNK
-    return Path.is_symlink(path, *args, **kwargs)
+    return Path.is_symlink(path)
 
 
 def _parse_fileinfo(fileinfo, ftype=None, st_size=None):
@@ -221,69 +221,104 @@ def _parse_fileinfo(fileinfo, ftype=None, st_size=None):
     #       st_fstype, st_rsize, st_creator, st_type, st_file_attributes, st_reparse_tag
     return make_stat_result(stat_dict)
 
-def _eos_lstat(path, *args, **kwargs):
-    _assert_eos_accessible("Cannot stat EOS paths.")
-    ftype, st_size = _get_type(path, *args, **kwargs)
-    if ftype is None:
-        return Path.lstat(path, *args, **kwargs)
-    elif ftype == stat.S_IFLNK:
-        # Special treatment: do NOT follow link
-        # Temporary solution: no way to retrieve other info currently
-        return make_stat_result({'st_mode': ftype+0o0777, 'st_size': st_size})
-    else:
+if sys.version_info >= (3, 10):
+    def _eos_lstat(path, *args, **kwargs):
+        _assert_eos_accessible("Cannot stat EOS paths.")
+        ftype, st_size = _get_type(path, *args, **kwargs)
+        if ftype is None:
+            return os.stat(path.as_posix(), follow_symlinks=False)
+        elif ftype == stat.S_IFLNK:
+            # Special treatment: do NOT follow link
+            # Temporary solution: no way to retrieve other info currently
+            return make_stat_result({'st_mode': ftype+0o0777, 'st_size': st_size})
+        else:
+            success, result = _run_eos(['eos', 'fileinfo', path.eos_path], mgm=path.mgm,
+                            _false_if_stderr_contains='No such file or directory', **kwargs)
+            if not success:
+                return os.stat(path.as_posix(), follow_symlinks=False)
+            else:
+                if not result:
+                    raise FileNotFoundError
+                return _parse_fileinfo(result, ftype, st_size)
+
+    def _eos_stat(path, *args, follow_symlinks=True, **kwargs):
+        _assert_eos_accessible("Cannot stat EOS paths.")
+        # The command `eos fileinfo` automatically resolves symlinks
+        if not follow_symlinks:
+            return _eos_lstat(path, *args, **kwargs)
         success, result = _run_eos(['eos', 'fileinfo', path.eos_path], mgm=path.mgm,
-                           _false_if_stderr_contains='No such file or directory', **kwargs)
+                            _false_if_stderr_contains='No such file or directory', **kwargs)
         if not success:
-            return Path.lstat(path, *args, **kwargs)
+            return os.stat(path, follow_symlinks=True)
+        if not result:
+            raise FileNotFoundError
+        ftype, _ = _get_type(path, *args, **kwargs)
+        return _parse_fileinfo(result, ftype)
+
+else:
+    def _eos_lstat(path, *args, **kwargs):
+        _assert_eos_accessible("Cannot stat EOS paths.")
+        ftype, st_size = _get_type(path, *args, **kwargs)
+        if ftype is None:
+            return os.lstat(path.as_posix())
+        elif ftype == stat.S_IFLNK:
+            # Special treatment: do NOT follow link
+            # Temporary solution: no way to retrieve other info currently
+            return make_stat_result({'st_mode': ftype+0o0777, 'st_size': st_size})
         else:
-            if not result:
-                raise FileNotFoundError
-            return _parse_fileinfo(result, ftype, st_size)
+            success, result = _run_eos(['eos', 'fileinfo', path.eos_path], mgm=path.mgm,
+                            _false_if_stderr_contains='No such file or directory', **kwargs)
+            if not success:
+                return os.lstat(path.as_posix())
+            else:
+                if not result:
+                    raise FileNotFoundError
+                return _parse_fileinfo(result, ftype, st_size)
 
-def _eos_stat(path, *args, **kwargs):
-    _assert_eos_accessible("Cannot stat EOS paths.")
-    ftype, st_size = _get_type(path, *args, **kwargs)
-    # The command `eos fileinfo` automatically resolves symlinks
-    success, result = _run_eos(['eos', 'fileinfo', path.eos_path], mgm=path.mgm,
-                        _false_if_stderr_contains='No such file or directory', **kwargs)
-    if not success:
-        return Path.stat(path, *args, **kwargs)
-    if not result:
-        raise FileNotFoundError
-    return _parse_fileinfo(result, ftype)
-
-
-def _eos_touch(path, *args, **kwargs):
+    def _eos_stat(path, *args, **kwargs):
+        _assert_eos_accessible("Cannot stat EOS paths.")
+        # The command `eos fileinfo` automatically resolves symlinks
+        success, result = _run_eos(['eos', 'fileinfo', path.eos_path], mgm=path.mgm,
+                            _false_if_stderr_contains='No such file or directory', **kwargs)
+        if not success:
+            return os.stat(path.as_posix())
+        if not result:
+            raise FileNotFoundError
+        ftype, _ = _get_type(path, *args, **kwargs)
+        return _parse_fileinfo(result, ftype)
+
+
+def _eos_touch(path, mode=0o666, exist_ok=True, **kwargs):
     _assert_eos_accessible("Cannot touch EOS paths.")
     success, result = _run_eos(['eos', 'touch', path.eos_path], mgm=path.mgm, **kwargs)
     if success:
         return result
-    return Path.touch(path, *args, **kwargs)
+    return Path.touch(path, mode=mode, exist_ok=exist_ok)
 
-def _eos_unlink(path, *args, **kwargs):
+def _eos_unlink(path, missing_ok=False, **kwargs):
     _assert_eos_accessible("Cannot unlink EOS paths.")
-    if not path.is_symlink() and path.is_dir():
+    if not path.is_symlink(**kwargs) and path.is_dir(**kwargs):
         raise IsADirectoryError(f"{path} is a directory.")
     success, result = _run_eos(['eos', 'rm', path.eos_path], mgm=path.mgm, **kwargs)
     if success:
         return result
-    return Path.unlink(path, *args, **kwargs)
+    return Path.unlink(path, missing_ok=missing_ok)
 
-def _eos_mkdir(path, *args, **kwargs):
+def _eos_mkdir(path, mode=0o777, parents=False, exist_ok=False, **kwargs):
     _assert_eos_accessible("Cannot rmdir EOS paths.")
     success, result = _run_eos(['eos', 'mkdir', path.eos_path], mgm=path.mgm, **kwargs)
     if success:
         return result
-    return Path.mkdir(path, *args, **kwargs)
+    return Path.mkdir(path, mode=mode, parents=parents, exist_ok=exist_ok)
 
 def _eos_rmdir(path, *args, **kwargs):
     _assert_eos_accessible("Cannot rmdir EOS paths.")
-    if path.is_symlink() or not path.is_dir():
+    if path.is_symlink(*args, **kwargs) or not path.is_dir(*args, **kwargs):
         raise NotADirectoryError(f"{path} is not a directory.")
     success, result = _run_eos(['eos', 'rmdir', path.eos_path], mgm=path.mgm, **kwargs)
     if success:
         return result
-    return Path.rmdir(path, *args, **kwargs)
+    return Path.rmdir(path)
 
 
 # Overwrite FsPath methods

xaux/fs/fs.py

@@ -149,34 +149,49 @@ def resolve(self, *args, **kwargs):
         else:
             return new_path
 
-    def is_file(self, *args, **kwargs):
-        return Path.is_file(self.expanduser(), *args, **kwargs)
+    if sys.version_info >= (3, 12):
+        def exists(self, *args, follow_symlinks=True, **kwargs):
+            if self.is_symlink(*args, **kwargs) and follow_symlinks:
+                return self.resolve(*args, **kwargs).exists(*args, **kwargs)
+            return Path.exists(self.expanduser(), follow_symlinks=follow_symlinks)
+
+    else:
+        def exists(self, *args, **kwargs):
+            if self.is_symlink(*args, **kwargs):
+                return self.resolve(*args, **kwargs).exists(*args, **kwargs)
+            return Path.exists(self.expanduser())
 
-    def is_dir(self, *args, **kwargs):
-        return Path.is_dir(self.expanduser(), *args, **kwargs)
+    if sys.version_info >= (3, 13):
+        def is_file(self, *args, follow_symlinks=True, **kwargs):
+            return Path.is_file(self.expanduser(), follow_symlinks=follow_symlinks)
 
-    def is_symlink(self, *args, **kwargs):
-        return Path.is_symlink(self.expanduser(), *args, **kwargs)
+        def is_dir(self, *args, follow_symlinks=True, **kwargs):
+            return Path.is_dir(self.expanduser(), follow_symlinks=follow_symlinks)
 
-    def exists(self, *args, **kwargs):
-        if self.is_symlink(*args, **kwargs):
-            return self.resolve(*args, **kwargs).exists(*args, **kwargs)
-        return Path.exists(self.expanduser(), *args, **kwargs)
+    else:
+        def is_file(self, *args, **kwargs):
+            return Path.is_file(self.expanduser())
+
+        def is_dir(self, *args, **kwargs):
+            return Path.is_dir(self.expanduser())
+
+    def is_symlink(self, *args, **kwargs):
+        return Path.is_symlink(self.expanduser())
 
     def symlink_to(self, target, target_is_directory=False, **kwargs):
         target = FsPath(target)
         return Path.symlink_to(self.expanduser().resolve(**kwargs), target.expanduser(),
-                               target_is_directory=target.is_dir(**kwargs), **kwargs)
+                               target_is_directory=target.is_dir(**kwargs))
 
-    def unlink(self, *args, **kwargs):
-        if not self.is_symlink(*args, **kwargs) and self.is_dir(*args, **kwargs):
+    def unlink(self, missing_ok=False, **kwargs):
+        if not self.is_symlink(**kwargs) and self.is_dir(**kwargs):
             raise IsADirectoryError(f"{self} is a directory.")
-        Path.unlink(self.expanduser(), *args, **kwargs)
+        Path.unlink(self.expanduser(), missing_ok=missing_ok)
 
     def rmdir(self, *args, **kwargs):
         if not self.is_dir(*args, **kwargs):
             raise NotADirectoryError(f"{self} is not a directory.")
-        Path.rmdir(self.expanduser(), *args, **kwargs)
+        Path.rmdir(self.expanduser())
 
     def __eq__(self, other):
         other = FsPath(other).expanduser().resolve()