finish and clean up implementation of swap, ver bump for release

augpathlib/__init__.py

@@ -51,4 +51,4 @@ class RepoPath(RepoHelper):
     'RemotePath',
 ]
 
-__version__ = '0.0.18'
+__version__ = '0.0.19'

augpathlib/core.py

@@ -20,6 +20,7 @@
 #from Xlib.display import Display
 #from Xlib import Xatom
 import augpathlib as aug
+from augpathlib import swap
 from augpathlib import exceptions as exc
 from augpathlib.meta import PathMeta
 from augpathlib.utils import log, default_cypher, StatResult, etag
@@ -472,39 +473,9 @@ def swap_carefree(self, target):
             temp.rename(self)
 
     if sys.platform != 'linux': # just look at us not using pathlib's infra ...
-        def _swap(self, target):
-            raise NotImplementedError('Windows has not atomic swap operation!')
-
+        _swap = swap.swap_not_implemented
     else:
-        # linux renameat2
-        # osx renamex_np
-        # osx exchangedata
-        # win MoveFileTransacted very few systems support this
-        import ctypes as _ctypes
-        _SYS_renameat2 = 316  # from /usr/include/asm/unistd_64.h
-        _RENAME_EXCHANGE = (1 << 1)  # /usr/src/linux/include/uapi/linux/fs.h
-        _libc = _ctypes.CDLL(None)
-        _rnat2_syscall = _libc.syscall
-        _rnat2_syscall.restypes = _ctypes.c_int  # returns and int
-        _rnat2_syscall.argtypes = (_ctypes.c_int,  # syscall number
-                                   _ctypes.c_int,  # old dir fd
-                                   _ctypes.POINTER(_ctypes.c_char),  # oldpat
-                                   _ctypes.c_int, # new dir fd
-                                   _ctypes.POINTER(_ctypes.c_char),  # newpath
-                                   _ctypes.c_uint)  # flags
-
-        def _swap(self, target):
-            raise NotImplementedError('nearly ready')
-            old_fd = os.open(self, 0)
-            new_fd = os.open(target, 0)
-            old_path = None  # TODO FIXME
-            new_path = None  # TODO FIXME
-            self._rnat2_syscall(self._SYS_renameat2,
-                                old_fd, old_path,
-                                new_fd, new_path,
-                                self._RENAME_EXCHANGE)
-            os.close(old_fd)
-            os.close(new_fd)
+        _swap = swap.swap_linux
 
     def swap(self, target):
         """ atomic swap of two paths

augpathlib/swap.py

@@ -0,0 +1,39 @@
+import os
+import sys
+import ctypes
+
+# linux renameat2
+# osx renamex_np
+# osx exchangedata
+# win MoveFileTransacted very few systems support this
+
+def swap_not_implemented(self, target):
+    raise NotImplementedError('Windows has not atomic swap operation!')
+
+# setup for calling renameat2
+SYS_renameat2 = 316  # from /usr/include/asm/unistd_64.h
+RENAME_EXCHANGE = (1 << 1)  # /usr/src/linux/include/uapi/linux/fs.h
+libc = ctypes.CDLL(None)
+rnat2_syscall = libc.syscall
+rnat2_syscall.restypes = ctypes.c_int       # returns and int
+rnat2_syscall.argtypes = (ctypes.c_int,     # syscall number
+                          ctypes.c_int,     # old dir fd
+                          ctypes.c_char_p,  # oldpath
+                          ctypes.c_int,     # new dir fd
+                          ctypes.c_char_p,  # newpath
+                          ctypes.c_uint)    # flags
+
+def swap_linux(self, target):
+    """ use renameat2 to perform an atomic swap operation """
+    old_fd = os.open(self, 0)
+    new_fd = os.open(target, 0)
+    old_path = os.fspath(self).encode()
+    new_path = os.fspath(target).encode()
+    value = rnat2_syscall(SYS_renameat2,
+                          old_fd, old_path,
+                          new_fd, new_path,
+                          RENAME_EXCHANGE)
+    os.close(old_fd)
+    os.close(new_fd)
+    if value != 0:
+        raise OSError(value, 'renameat2 failed', self, None, target)

test/test_path.py

@@ -1,10 +1,13 @@
+import sys
 import unittest
 from pathlib import PurePosixPath
 import pytest
+from augpathlib import swap
 from augpathlib import exceptions as exc
 from augpathlib import AugmentedPath, LocalPath
 from augpathlib import SymlinkCache, PrimaryCache
 from augpathlib import PathMeta
+from augpathlib import PathMeta
 from augpathlib.meta import _PathMetaAsSymlink, _PathMetaAsXattrs
 from .common import (log,
                      onerror,
@@ -19,10 +22,17 @@
 SymlinkCache._local_class = AugmentedPath  # have to set a default
 
 class Helper:
-    def setUp(self):
+    @classmethod
+    def setUpClass(cls):
         if not test_base.exists():
             test_base.mkdir()
 
+    @classmethod
+    def tearDownClass(cls):
+        if test_base.exists():
+            test_base.rmtree(onerror=onerror)
+
+    def setUp(self):
         self.test_link = AugmentedPath(test_base, 'evil-symlink')  # FIXME random needed ...
         if self.test_link.is_symlink():
             self.test_link.unlink()
@@ -159,8 +169,27 @@ def setUp(self):
         self.f1.touch()
         self.f2.touch()
 
+    def _doit(self, thunk):
+        assert (self.source_d / self.f1.name).exists()
+        assert (self.target_d / self.f2.name).exists()
+        assert self.f1.exists()
+        assert self.f2.exists()
+        thunk()
+        assert (self.source_d / self.f2.name).exists()
+        assert (self.target_d / self.f1.name).exists()
+        assert not self.f1.exists()
+        assert not self.f2.exists()
+
+    @pytest.mark.skipif(sys.platform != 'linux', reason='not implemented')
+    def test_swap_linux(self):
+        self._doit(lambda :swap.swap_linux(self.source_d, self.target_d))
+
+    @pytest.mark.skipif(sys.platform != 'linux', reason='not implemented')
+    def test_swap(self):
+        self._doit(lambda :self.source_d.swap(self.target_d))
+
     def test_swap_carefree(self):
-        self.source_d.swap_carefree(self.target_d)
+        self._doit(lambda :self.source_d.swap_carefree(self.target_d))
 
 
 class TestACachePath(unittest.TestCase):