Fix separator handling

Author: msg555

dirdiff/differ.py

@@ -16,6 +16,7 @@
 )
 from dirdiff.filelib import DirectoryManager, FileManager, PathManager
 from dirdiff.filelib_tar import TarDirectoryManager, TarFileLoader
+from dirdiff.osshim import posix_join
 from dirdiff.output import DiffOutput, StatInfo
 
 LOGGER = logging.getLogger(__name__)
@@ -100,7 +101,7 @@ def stats_differ(self, x: StatInfo, y: StatInfo) -> bool:
 
 def _open_dir(path_dir: DifferPathLike) -> DirectoryManager:
     if isinstance(path_dir, tarfile.TarFile):
-        return TarDirectoryManager(TarFileLoader(path_dir), os.path.sep)  # type: ignore
+        return TarDirectoryManager(TarFileLoader(path_dir), "/")  # type: ignore
     return DirectoryManager(path_dir)
 
 
@@ -216,7 +217,7 @@ def _diff_dirs(
 
         for dir_entry in merged_entries:
             dir_entry_type = _dir_entry_type(dir_entry)
-            cpath = os.path.join(archive_path, dir_entry.name)
+            cpath = posix_join(archive_path, dir_entry.name)
 
             lower_type = lower_map.pop(dir_entry.name, None)
             if dir_entry_type == DirEntryType.DIRECTORY:
@@ -250,7 +251,7 @@ def _diff_dirs(
         for name in lower_map:
             self._flush_pending()
             try:
-                self.output.delete_marker(os.path.join(archive_path, name))
+                self.output.delete_marker(posix_join(archive_path, name))
             except IOErrors:
                 self._output_error(archive_path, "creating delete marker")
 
@@ -395,7 +396,7 @@ def _insert_dir(
         except IOErrors:
             self._output_error(archive_path, "creating dir")
         for dir_entry in dir_entries:
-            cpath = os.path.join(archive_path, dir_entry.name)
+            cpath = posix_join(archive_path, dir_entry.name)
             if dir_entry.is_dir(follow_symlinks=False):
                 self._insert_dir(cpath, obj.child_dir(dir_entry.name))
             elif dir_entry.is_file(follow_symlinks=False):

dirdiff/filelib_tar.py

@@ -1,14 +1,13 @@
 import collections
 import contextlib
 import functools
-import os
 import stat
 import tarfile
 from tarfile import TarFile, TarInfo
 from typing import Iterator
 
 from dirdiff.filelib import StatInfo
-from dirdiff.osshim import makedev
+from dirdiff.osshim import makedev, posix_join, posix_norm, posix_split
 
 _MODE_MAPPING = {
     tarfile.REGTYPE: stat.S_IFREG,
@@ -21,7 +20,7 @@
 
 
 def _norm_name(name: str) -> str:
-    return os.path.normpath(os.path.join(os.path.sep, name))
+    return posix_norm(posix_join("/", name))
 
 
 class TarFileLoader:
@@ -33,15 +32,15 @@ def __init__(self, tf: TarFile) -> None:
         self.info = {}
         for ti in tf.getmembers():
             path = _norm_name(ti.name)
-            parent_path, filename = os.path.split(path)
+            parent_path, filename = posix_split(path)
             self.info[path] = ti
 
             while filename:
                 self.children[parent_path].append((filename, ti))
                 if parent_path in self.info:
                     break
                 self.info[parent_path] = self.MISSING_DIRECTORY_TARINFO
-                parent_path, filename = os.path.split(parent_path)
+                parent_path, filename = posix_split(parent_path)
 
 
 class TarManagerBase:
@@ -120,10 +119,10 @@ def exists_in_archive(self) -> bool:
         return self.ti is not self.loader.MISSING_DIRECTORY_TARINFO
 
     def child_dir(self, name: str) -> "TarDirectoryManager":
-        return TarDirectoryManager(self.loader, os.path.join(self.name, name))
+        return TarDirectoryManager(self.loader, posix_join(self.name, name))
 
     def child_file(self, name: str) -> TarFileManager:
-        return TarFileManager(self.loader, os.path.join(self.name, name))
+        return TarFileManager(self.loader, posix_join(self.name, name))
 
     def child_path(self, name: str) -> TarPathManager:
-        return TarPathManager(self.loader, os.path.join(self.name, name))
+        return TarPathManager(self.loader, posix_join(self.name, name))

dirdiff/osshim.py

@@ -1,4 +1,17 @@
 # pylint: disable=redefined-outer-name,unused-import
+import os
+from typing import List, Tuple
+
+try:
+    from posixpath import basename as posix_basename
+    from posixpath import join as posix_join
+    from posixpath import normpath as posix_norm
+    from posixpath import split as posix_split
+except ImportError:
+    from os.path import basename as posix_basename
+    from os.path import join as posix_join
+    from os.path import normpath as posix_norm
+    from os.path import split as posix_split
 
 try:
     from os import makedev

dirdiff/output_aufs.py

@@ -1,8 +1,8 @@
 import io
 import logging
-import os
 import stat
 
+from dirdiff.osshim import posix_basename, posix_join, posix_split
 from dirdiff.output import DiffOutputForwarding, OutputBackend, StatInfo
 
 LOGGER = logging.getLogger(__name__)
@@ -18,12 +18,12 @@ def __init__(self, backend: OutputBackend) -> None:
 
     @classmethod
     def is_whiteout_path(cls, path: str) -> bool:
-        return os.path.basename(path).startswith(cls.WHITEOUT_PREFIX)
+        return posix_basename(path).startswith(cls.WHITEOUT_PREFIX)
 
     def delete_marker(self, path: str) -> None:
-        head, tail = os.path.split(path)
+        head, tail = posix_split(path)
         self.backend.write_file(
-            os.path.join(head, self.WHITEOUT_PREFIX + tail),
+            posix_join(head, self.WHITEOUT_PREFIX + tail),
             StatInfo(
                 mode=stat.S_IFREG | 0o444,
                 uid=self.whiteout_uid,

dirdiff/output_tar.py

@@ -1,30 +1,36 @@
 import logging
-import os
 import stat
 import tarfile
 from tarfile import TarFile, TarInfo
 
 from dirdiff.exceptions import DirDiffOutputException
-from dirdiff.osshim import major, minor
+from dirdiff.osshim import major, minor, posix_join
 from dirdiff.output import OutputBackend, StatInfo
 
 LOGGER = logging.getLogger(__name__)
 
 
 class OutputBackendTarfile(OutputBackend):
+    """
+    Tarfile output backend.
+
+    Note that all paths are expected to be '/' separated when using this module.
+    Other separators will be considered parts of file names.
+    """
+
     def __init__(self, tf: TarFile, *, archive_root=".") -> None:
         self.tf = tf
         self.archive_root = archive_root
 
     def _get_tar_info(self, name: str, st: StatInfo) -> TarInfo:
-        if name in (".", os.path.sep):
+        if name in (".", "/"):
             arch_name = self.archive_root
-        elif name.startswith(f".{os.path.sep}"):
-            arch_name = os.path.join(self.archive_root, name[2:])
-        elif name.startswith(os.path.sep):
-            arch_name = os.path.join(self.archive_root, name[1:])
+        elif name.startswith("./"):
+            arch_name = posix_join(self.archive_root, name[2:])
+        elif name.startswith("/"):
+            arch_name = posix_join(self.archive_root, name[1:])
         else:
-            arch_name = os.path.join(self.archive_root, name)
+            arch_name = posix_join(self.archive_root, name)
         ti = TarInfo(arch_name)
         ti.mode = stat.S_IMODE(st.mode)
         ti.mtime = st.mtime

tests/cases/backslash/diff.tgz

@@ -9,7 +9,7 @@
 def tar_summarize(tf: tarfile.TarFile):
     result = {}
     while ti := tf.next():
-        assert ti.name == "." or ti.name.startswith(f".{os.path.sep}")
+        assert ti.name == "." or ti.name.startswith("./")
         assert ti.name not in result
         parent = os.path.dirname(ti.name)
 
@@ -65,4 +65,10 @@ def run_test(name: str) -> None:
 
 
 def test_generic():
+    """Test several different misc things"""
+    run_test("generic")
+
+
+def test_backslash():
+    """Test that files with backslashes work okay for tar archives."""
     run_test("generic")

tests/cases/backslash/lower.tgz

@@ -1,13 +1,12 @@
 import io
-import os
 import stat
 import tarfile
 
 import pytest
 
 from dirdiff.filelib import StatInfo
 from dirdiff.filelib_tar import TarDirectoryManager, TarFileLoader
-from dirdiff.osshim import major, makedev, minor
+from dirdiff.osshim import major, makedev, minor, posix_join
 from dirdiff.output_tar import OutputBackendTarfile
 
 DEFAULT_UID = 1234
@@ -38,14 +37,14 @@ def test_file_write_dir():
     data.seek(0)
     with tarfile.open(fileobj=data, mode="r") as tf:
         assert len(tf.getmembers()) == 1
-        ti = tf.getmember(os.path.join(backend.archive_root, file_name))
+        ti = tf.getmember(posix_join(backend.archive_root, file_name))
         assert st.mode == mode
         assert ti.isdir()
         assert ti.uid == DEFAULT_UID
         assert ti.gid == DEFAULT_GID
 
         loader = TarFileLoader(tf)
-        with TarDirectoryManager(loader, os.path.sep) as dm:
+        with TarDirectoryManager(loader, "/") as dm:
             assert not dm.exists_in_archive()
             assert [(entry.name, entry.is_dir()) for entry in dm] == [(file_name, True)]
             with dm.child_dir(file_name) as sdm:
@@ -67,7 +66,7 @@ def test_file_write_reg():
     data.seek(0)
     with tarfile.open(fileobj=data, mode="r") as tf:
         assert len(tf.getmembers()) == 1
-        ti = tf.getmember(os.path.join(backend.archive_root, file_name))
+        ti = tf.getmember(posix_join(backend.archive_root, file_name))
         assert st.mode == mode
         assert ti.isreg()
         assert ti.uid == DEFAULT_UID
@@ -76,7 +75,7 @@ def test_file_write_reg():
             assert fdata.read() == file_data
 
         loader = TarFileLoader(tf)
-        with TarDirectoryManager(loader, os.path.sep) as dm:
+        with TarDirectoryManager(loader, "/") as dm:
             assert not dm.exists_in_archive()
             assert [(entry.name, entry.is_file()) for entry in dm] == [
                 (file_name, True)
@@ -107,14 +106,14 @@ def test_file_write_link():
     data.seek(0)
     with tarfile.open(fileobj=data, mode="r") as tf:
         assert len(tf.getmembers()) == 2
-        ti = tf.getmember(os.path.join(backend.archive_root, sym_file_name))
+        ti = tf.getmember(posix_join(backend.archive_root, sym_file_name))
         assert ti.issym()
         assert ti.linkname == file_name
         assert ti.uid == DEFAULT_UID
         assert ti.gid == DEFAULT_GID
 
         loader = TarFileLoader(tf)
-        with TarDirectoryManager(loader, os.path.sep) as dm:
+        with TarDirectoryManager(loader, "/") as dm:
             assert not dm.exists_in_archive()
             assert sorted(
                 (entry.name, entry.is_file(follow_symlinks=False)) for entry in dm
@@ -146,15 +145,15 @@ def test_file_write_device(ftype, attest):
     data.seek(0)
     with tarfile.open(fileobj=data, mode="r") as tf:
         assert len(tf.getmembers()) == 1
-        ti = tf.getmember(os.path.join(backend.archive_root, file_name))
+        ti = tf.getmember(posix_join(backend.archive_root, file_name))
         assert attest(ti)
         assert ti.devmajor == dev_major
         assert ti.devminor == dev_minor
         assert ti.uid == DEFAULT_UID
         assert ti.gid == DEFAULT_GID
 
         loader = TarFileLoader(tf)
-        with TarDirectoryManager(loader, os.path.sep) as dm:
+        with TarDirectoryManager(loader, "/") as dm:
             assert not dm.exists_in_archive()
             assert [(entry.name, entry.is_file()) for entry in dm] == [
                 (file_name, False)