https://github.com/python/cpython/commit/22a442181d5f1ac496dac08e24fd852de40... commit: 22a442181d5f1ac496dac08e24fd852de404882f branch: main author: Barney Gale <barney.gale@gmail.com> committer: barneygale <barney.gale@gmail.com> date: 2025-01-11T19:27:47Z summary: GH-128520: Divide pathlib ABCs into three classes (#128523) In the private pathlib ABCs, rename `PurePathBase` to `JoinablePath`, and split `PathBase` into `ReadablePath` and `WritablePath`. This improves the API fit for read-only virtual filesystems. The split of `PathBase` entails a similar split of `CopyWorker` (implements copying) and the test cases in `test_pathlib_abc`. In a later patch, we'll make `WritablePath` inherit directly from `JoinablePath` rather than `ReadablePath`. For a couple of reasons, this isn't quite possible yet. files: M Lib/pathlib/_abc.py M Lib/pathlib/_local.py M Lib/pathlib/_types.py M Lib/test/test_pathlib/test_pathlib.py M Lib/test/test_pathlib/test_pathlib_abc.py diff --git a/Lib/pathlib/_abc.py b/Lib/pathlib/_abc.py index 7de2bb066f8f99..38bc660e0aeb30 100644 --- a/Lib/pathlib/_abc.py +++ b/Lib/pathlib/_abc.py @@ -7,8 +7,8 @@ it's developed alongside pathlib. If it finds success and maturity as a PyPI package, it could become a public part of the standard library. -Two base classes are defined here -- PurePathBase and PathBase -- that -resemble pathlib's PurePath and Path respectively. +Three base classes are defined here -- JoinablePath, ReadablePath and +WritablePath. """ import functools @@ -56,13 +56,13 @@ def concat_path(path, text): return path.with_segments(str(path) + text) -class CopyWorker: +class CopyReader: """ Class that implements copying between path objects. An instance of this - class is available from the PathBase.copy property; it's made callable so - that PathBase.copy() can be treated as a method. + class is available from the ReadablePath.copy property; it's made callable + so that ReadablePath.copy() can be treated as a method. - The target path's CopyWorker drives the process from its _create() method. + The target path's CopyWriter drives the process from its _create() method. Files and directories are exchanged by calling methods on the source and target paths, and metadata is exchanged by calling source.copy._read_metadata() and target.copy._write_metadata(). @@ -77,11 +77,15 @@ def __call__(self, target, follow_symlinks=True, dirs_exist_ok=False, """ Recursively copy this file or directory tree to the given destination. """ - if not isinstance(target, PathBase): + if not isinstance(target, ReadablePath): target = self._path.with_segments(target) - # Delegate to the target path's CopyWorker object. - return target.copy._create(self._path, follow_symlinks, dirs_exist_ok, preserve_metadata) + # Delegate to the target path's CopyWriter object. + try: + create = target.copy._create + except AttributeError: + raise TypeError(f"Target is not writable: {target}") from None + return create(self._path, follow_symlinks, dirs_exist_ok, preserve_metadata) _readable_metakeys = frozenset() @@ -91,6 +95,10 @@ def _read_metadata(self, metakeys, *, follow_symlinks=True): """ raise NotImplementedError + +class CopyWriter(CopyReader): + __slots__ = () + _writable_metakeys = frozenset() def _write_metadata(self, metadata, *, follow_symlinks=True): @@ -182,7 +190,7 @@ def _ensure_distinct_path(self, source): raise err -class PurePathBase: +class JoinablePath: """Base class for pure path objects. This class *does not* provide several magic methods that are defined in @@ -334,7 +342,7 @@ def match(self, path_pattern, *, case_sensitive=None): is matched. The recursive wildcard '**' is *not* supported by this method. """ - if not isinstance(path_pattern, PurePathBase): + if not isinstance(path_pattern, JoinablePath): path_pattern = self.with_segments(path_pattern) if case_sensitive is None: case_sensitive = _is_case_sensitive(self.parser) @@ -359,7 +367,7 @@ def full_match(self, pattern, *, case_sensitive=None): Return True if this path matches the given glob-style pattern. The pattern is matched against the entire path. """ - if not isinstance(pattern, PurePathBase): + if not isinstance(pattern, JoinablePath): pattern = self.with_segments(pattern) if case_sensitive is None: case_sensitive = _is_case_sensitive(self.parser) @@ -369,7 +377,7 @@ def full_match(self, pattern, *, case_sensitive=None): -class PathBase(PurePathBase): +class ReadablePath(JoinablePath): """Base class for concrete path objects. This class provides dummy implementations for many methods that derived @@ -434,25 +442,6 @@ def read_text(self, encoding=None, errors=None, newline=None): with self.open(mode='r', encoding=encoding, errors=errors, newline=newline) as f: return f.read() - def write_bytes(self, data): - """ - Open the file in bytes mode, write to it, and close the file. - """ - # type-check for the buffer interface before truncating the file - view = memoryview(data) - with self.open(mode='wb') as f: - return f.write(view) - - def write_text(self, data, encoding=None, errors=None, newline=None): - """ - Open the file in text mode, write to it, and close the file. - """ - if not isinstance(data, str): - raise TypeError('data must be str, not %s' % - data.__class__.__name__) - with self.open(mode='w', encoding=encoding, errors=errors, newline=newline) as f: - return f.write(data) - def _scandir(self): """Yield os.DirEntry-like objects of the directory contents. @@ -474,7 +463,7 @@ def glob(self, pattern, *, case_sensitive=None, recurse_symlinks=True): """Iterate over this subtree and yield all existing files (of any kind, including directories) matching the given relative pattern. """ - if not isinstance(pattern, PurePathBase): + if not isinstance(pattern, JoinablePath): pattern = self.with_segments(pattern) anchor, parts = _explode_path(pattern) if anchor: @@ -496,7 +485,7 @@ def rglob(self, pattern, *, case_sensitive=None, recurse_symlinks=True): directories) matching the given relative pattern, anywhere in this subtree. """ - if not isinstance(pattern, PurePathBase): + if not isinstance(pattern, JoinablePath): pattern = self.with_segments(pattern) pattern = '**' / pattern return self.glob(pattern, case_sensitive=case_sensitive, recurse_symlinks=recurse_symlinks) @@ -543,6 +532,28 @@ def readlink(self): """ raise NotImplementedError + copy = property(CopyReader, doc=CopyReader.__call__.__doc__) + + def copy_into(self, target_dir, *, follow_symlinks=True, + dirs_exist_ok=False, preserve_metadata=False): + """ + Copy this file or directory tree into the given existing directory. + """ + name = self.name + if not name: + raise ValueError(f"{self!r} has an empty name") + elif isinstance(target_dir, ReadablePath): + target = target_dir / name + else: + target = self.with_segments(target_dir, name) + return self.copy(target, follow_symlinks=follow_symlinks, + dirs_exist_ok=dirs_exist_ok, + preserve_metadata=preserve_metadata) + + +class WritablePath(ReadablePath): + __slots__ = () + def symlink_to(self, target, target_is_directory=False): """ Make this path a symlink pointing to the target path. @@ -556,20 +567,23 @@ def mkdir(self, mode=0o777, parents=False, exist_ok=False): """ raise NotImplementedError - copy = property(CopyWorker, doc=CopyWorker.__call__.__doc__) + def write_bytes(self, data): + """ + Open the file in bytes mode, write to it, and close the file. + """ + # type-check for the buffer interface before truncating the file + view = memoryview(data) + with self.open(mode='wb') as f: + return f.write(view) - def copy_into(self, target_dir, *, follow_symlinks=True, - dirs_exist_ok=False, preserve_metadata=False): + def write_text(self, data, encoding=None, errors=None, newline=None): """ - Copy this file or directory tree into the given existing directory. + Open the file in text mode, write to it, and close the file. """ - name = self.name - if not name: - raise ValueError(f"{self!r} has an empty name") - elif isinstance(target_dir, PathBase): - target = target_dir / name - else: - target = self.with_segments(target_dir, name) - return self.copy(target, follow_symlinks=follow_symlinks, - dirs_exist_ok=dirs_exist_ok, - preserve_metadata=preserve_metadata) + if not isinstance(data, str): + raise TypeError('data must be str, not %s' % + data.__class__.__name__) + with self.open(mode='w', encoding=encoding, errors=errors, newline=newline) as f: + return f.write(data) + + copy = property(CopyWriter, doc=CopyWriter.__call__.__doc__) diff --git a/Lib/pathlib/_local.py b/Lib/pathlib/_local.py index 1da85ddea24376..d6afb31424265c 100644 --- a/Lib/pathlib/_local.py +++ b/Lib/pathlib/_local.py @@ -20,7 +20,7 @@ grp = None from pathlib._os import copyfile -from pathlib._abc import CopyWorker, PurePathBase, PathBase +from pathlib._abc import CopyWriter, JoinablePath, WritablePath __all__ = [ @@ -65,7 +65,7 @@ def __repr__(self): return "<{}.parents>".format(type(self._path).__name__) -class _LocalCopyWorker(CopyWorker): +class _LocalCopyWriter(CopyWriter): """This object implements the Path.copy callable. Don't try to construct it yourself.""" __slots__ = () @@ -158,7 +158,7 @@ def _create_file(self, source, metakeys): try: source = os.fspath(source) except TypeError: - if not isinstance(source, PathBase): + if not isinstance(source, WritablePath): raise super()._create_file(source, metakeys) else: @@ -190,7 +190,7 @@ def _ensure_different_file(self, source): raise err -class PurePath(PurePathBase): +class PurePath(JoinablePath): """Base class for manipulating paths without I/O. PurePath represents a filesystem path and offers operations which @@ -646,7 +646,7 @@ def full_match(self, pattern, *, case_sensitive=None): Return True if this path matches the given glob-style pattern. The pattern is matched against the entire path. """ - if not isinstance(pattern, PurePathBase): + if not isinstance(pattern, PurePath): pattern = self.with_segments(pattern) if case_sensitive is None: case_sensitive = self.parser is posixpath @@ -683,7 +683,7 @@ class PureWindowsPath(PurePath): __slots__ = () -class Path(PathBase, PurePath): +class Path(WritablePath, PurePath): """PurePath subclass that can make system calls. Path represents a filesystem path but unlike PurePath, also offers @@ -830,7 +830,7 @@ def read_text(self, encoding=None, errors=None, newline=None): # Call io.text_encoding() here to ensure any warning is raised at an # appropriate stack level. encoding = io.text_encoding(encoding) - return PathBase.read_text(self, encoding, errors, newline) + return super().read_text(encoding, errors, newline) def write_text(self, data, encoding=None, errors=None, newline=None): """ @@ -839,7 +839,7 @@ def write_text(self, data, encoding=None, errors=None, newline=None): # Call io.text_encoding() here to ensure any warning is raised at an # appropriate stack level. encoding = io.text_encoding(encoding) - return PathBase.write_text(self, data, encoding, errors, newline) + return super().write_text(data, encoding, errors, newline) _remove_leading_dot = operator.itemgetter(slice(2, None)) _remove_trailing_slash = operator.itemgetter(slice(-1)) @@ -1122,7 +1122,7 @@ def replace(self, target): os.replace(self, target) return self.with_segments(target) - copy = property(_LocalCopyWorker, doc=_LocalCopyWorker.__call__.__doc__) + copy = property(_LocalCopyWriter, doc=_LocalCopyWriter.__call__.__doc__) def move(self, target): """ @@ -1134,7 +1134,7 @@ def move(self, target): except TypeError: pass else: - if not isinstance(target, PathBase): + if not isinstance(target, WritablePath): target = self.with_segments(target_str) target.copy._ensure_different_file(self) try: @@ -1155,7 +1155,7 @@ def move_into(self, target_dir): name = self.name if not name: raise ValueError(f"{self!r} has an empty name") - elif isinstance(target_dir, PathBase): + elif isinstance(target_dir, WritablePath): target = target_dir / name else: target = self.with_segments(target_dir, name) diff --git a/Lib/pathlib/_types.py b/Lib/pathlib/_types.py index 72dac2e276fce0..84032bb5b4ff1a 100644 --- a/Lib/pathlib/_types.py +++ b/Lib/pathlib/_types.py @@ -9,7 +9,7 @@ class Parser(Protocol): """Protocol for path parsers, which do low-level path manipulation. Path parsers provide a subset of the os.path API, specifically those - functions needed to provide PurePathBase functionality. Each PurePathBase + functions needed to provide JoinablePath functionality. Each JoinablePath subclass references its path parser via a 'parser' class attribute. """ diff --git a/Lib/test/test_pathlib/test_pathlib.py b/Lib/test/test_pathlib/test_pathlib.py index 6548577f4de12c..ad5a9f9c8de9d6 100644 --- a/Lib/test/test_pathlib/test_pathlib.py +++ b/Lib/test/test_pathlib/test_pathlib.py @@ -75,7 +75,7 @@ def test_is_notimplemented(self): # Tests for the pure classes. # -class PurePathTest(test_pathlib_abc.DummyPurePathTest): +class PurePathTest(test_pathlib_abc.DummyJoinablePathTest): cls = pathlib.PurePath # Make sure any symbolic links in the base test path are resolved. @@ -924,7 +924,7 @@ class cls(pathlib.PurePath): # Tests for the concrete classes. # -class PathTest(test_pathlib_abc.DummyPathTest, PurePathTest): +class PathTest(test_pathlib_abc.DummyWritablePathTest, PurePathTest): """Tests for the FS-accessing functionalities of the Path classes.""" cls = pathlib.Path can_symlink = os_helper.can_symlink() @@ -980,15 +980,15 @@ def tempdir(self): self.addCleanup(os_helper.rmtree, d) return d - def test_matches_pathbase_docstrings(self): - path_names = {name for name in dir(pathlib._abc.PathBase) if name[0] != '_'} + def test_matches_writablepath_docstrings(self): + path_names = {name for name in dir(pathlib._abc.WritablePath) if name[0] != '_'} for attr_name in path_names: if attr_name == 'parser': - # On Windows, Path.parser is ntpath, but PathBase.parser is + # On Windows, Path.parser is ntpath, but WritablePath.parser is # posixpath, and so their docstrings differ. continue our_attr = getattr(self.cls, attr_name) - path_attr = getattr(pathlib._abc.PathBase, attr_name) + path_attr = getattr(pathlib._abc.WritablePath, attr_name) self.assertEqual(our_attr.__doc__, path_attr.__doc__) def test_concrete_class(self): @@ -3019,7 +3019,7 @@ def test_group_windows(self): P('c:/').group() -class PathWalkTest(test_pathlib_abc.DummyPathWalkTest): +class PathWalkTest(test_pathlib_abc.DummyReadablePathWalkTest): cls = pathlib.Path base = PathTest.base can_symlink = PathTest.can_symlink diff --git a/Lib/test/test_pathlib/test_pathlib_abc.py b/Lib/test/test_pathlib/test_pathlib_abc.py index 87aef0c130cf9e..6ba012e0208a53 100644 --- a/Lib/test/test_pathlib/test_pathlib_abc.py +++ b/Lib/test/test_pathlib/test_pathlib_abc.py @@ -4,7 +4,7 @@ import errno import unittest -from pathlib._abc import PurePathBase, PathBase +from pathlib._abc import JoinablePath, ReadablePath, WritablePath from pathlib._types import Parser import posixpath @@ -31,8 +31,8 @@ def needs_windows(fn): # -class PurePathBaseTest(unittest.TestCase): - cls = PurePathBase +class JoinablePathTest(unittest.TestCase): + cls = JoinablePath def test_magic_methods(self): P = self.cls @@ -51,7 +51,7 @@ def test_parser(self): self.assertIs(self.cls.parser, posixpath) -class DummyPurePath(PurePathBase): +class DummyJoinablePath(JoinablePath): __slots__ = ('_segments',) def __init__(self, *segments): @@ -63,7 +63,7 @@ def __str__(self): return '' def __eq__(self, other): - if not isinstance(other, DummyPurePath): + if not isinstance(other, DummyJoinablePath): return NotImplemented return str(self) == str(other) @@ -77,8 +77,8 @@ def with_segments(self, *pathsegments): return type(self)(*pathsegments) -class DummyPurePathTest(unittest.TestCase): - cls = DummyPurePath +class DummyJoinablePathTest(unittest.TestCase): + cls = DummyJoinablePath # Use a base path that's unrelated to any real filesystem path. base = f'/this/path/kills/fascists/{TESTFN}' @@ -916,9 +916,9 @@ def test_with_suffix_invalid(self): # -class DummyPathIO(io.BytesIO): +class DummyWritablePathIO(io.BytesIO): """ - Used by DummyPath to implement `open('w')` + Used by DummyWritablePath to implement `open('w')` """ def __init__(self, files, path): @@ -931,10 +931,10 @@ def close(self): super().close() -class DummyPath(PathBase): +class DummyReadablePath(ReadablePath): """ - Simple implementation of PathBase that keeps files and directories in - memory. + Simple implementation of DummyReadablePath that keeps files and + directories in memory. """ __slots__ = ('_segments') @@ -950,7 +950,7 @@ def __str__(self): return '' def __eq__(self, other): - if not isinstance(other, DummyPath): + if not isinstance(other, DummyReadablePath): return NotImplemented return str(self) == str(other) @@ -990,10 +990,11 @@ def open(self, mode='r', buffering=-1, encoding=None, raise FileNotFoundError(errno.ENOENT, "File not found", path) stream = io.BytesIO(self._files[path]) elif mode == 'w': + # FIXME: move to DummyWritablePath parent, name = posixpath.split(path) if parent not in self._directories: raise FileNotFoundError(errno.ENOENT, "File not found", parent) - stream = DummyPathIO(self._files, path) + stream = DummyWritablePathIO(self._files, path) self._files[path] = b'' self._directories[parent].add(name) else: @@ -1011,6 +1012,10 @@ def iterdir(self): else: raise FileNotFoundError(errno.ENOENT, "File not found", path) + +class DummyWritablePath(DummyReadablePath, WritablePath): + __slots__ = () + def mkdir(self, mode=0o777, parents=False, exist_ok=False): path = str(self) parent = str(self.parent) @@ -1029,24 +1034,11 @@ def mkdir(self, mode=0o777, parents=False, exist_ok=False): self.parent.mkdir(parents=True, exist_ok=True) self.mkdir(mode, parents=False, exist_ok=exist_ok) - def _delete(self): - path = str(self) - if path in self._files: - del self._files[path] - elif path in self._directories: - for name in list(self._directories[path]): - self.joinpath(name)._delete() - del self._directories[path] - else: - raise FileNotFoundError(errno.ENOENT, "File not found", path) - parent = str(self.parent) - self._directories[parent].remove(self.name) - -class DummyPathTest(DummyPurePathTest): - """Tests for PathBase methods that use stat(), open() and iterdir().""" +class DummyReadablePathTest(DummyJoinablePathTest): + """Tests for ReadablePathTest methods that use stat(), open() and iterdir().""" - cls = DummyPath + cls = DummyReadablePath can_symlink = False # (self.base) @@ -1138,213 +1130,6 @@ def test_open_common(self): self.assertIsInstance(f, io.BufferedIOBase) self.assertEqual(f.read().strip(), b"this is file A") - def test_read_write_bytes(self): - p = self.cls(self.base) - (p / 'fileA').write_bytes(b'abcdefg') - self.assertEqual((p / 'fileA').read_bytes(), b'abcdefg') - # Check that trying to write str does not truncate the file. - self.assertRaises(TypeError, (p / 'fileA').write_bytes, 'somestr') - self.assertEqual((p / 'fileA').read_bytes(), b'abcdefg') - - def test_read_write_text(self): - p = self.cls(self.base) - (p / 'fileA').write_text('äbcdefg', encoding='latin-1') - self.assertEqual((p / 'fileA').read_text( - encoding='utf-8', errors='ignore'), 'bcdefg') - # Check that trying to write bytes does not truncate the file. - self.assertRaises(TypeError, (p / 'fileA').write_text, b'somebytes') - self.assertEqual((p / 'fileA').read_text(encoding='latin-1'), 'äbcdefg') - - def test_read_text_with_newlines(self): - p = self.cls(self.base) - # Check that `\n` character change nothing - (p / 'fileA').write_bytes(b'abcde\r\nfghlk\n\rmnopq') - self.assertEqual((p / 'fileA').read_text(newline='\n'), - 'abcde\r\nfghlk\n\rmnopq') - # Check that `\r` character replaces `\n` - (p / 'fileA').write_bytes(b'abcde\r\nfghlk\n\rmnopq') - self.assertEqual((p / 'fileA').read_text(newline='\r'), - 'abcde\r\nfghlk\n\rmnopq') - # Check that `\r\n` character replaces `\n` - (p / 'fileA').write_bytes(b'abcde\r\nfghlk\n\rmnopq') - self.assertEqual((p / 'fileA').read_text(newline='\r\n'), - 'abcde\r\nfghlk\n\rmnopq') - - def test_write_text_with_newlines(self): - p = self.cls(self.base) - # Check that `\n` character change nothing - (p / 'fileA').write_text('abcde\r\nfghlk\n\rmnopq', newline='\n') - self.assertEqual((p / 'fileA').read_bytes(), - b'abcde\r\nfghlk\n\rmnopq') - # Check that `\r` character replaces `\n` - (p / 'fileA').write_text('abcde\r\nfghlk\n\rmnopq', newline='\r') - self.assertEqual((p / 'fileA').read_bytes(), - b'abcde\r\rfghlk\r\rmnopq') - # Check that `\r\n` character replaces `\n` - (p / 'fileA').write_text('abcde\r\nfghlk\n\rmnopq', newline='\r\n') - self.assertEqual((p / 'fileA').read_bytes(), - b'abcde\r\r\nfghlk\r\n\rmnopq') - # Check that no argument passed will change `\n` to `os.linesep` - os_linesep_byte = bytes(os.linesep, encoding='ascii') - (p / 'fileA').write_text('abcde\nfghlk\n\rmnopq') - self.assertEqual((p / 'fileA').read_bytes(), - b'abcde' + os_linesep_byte + b'fghlk' + os_linesep_byte + b'\rmnopq') - - def test_copy_file(self): - base = self.cls(self.base) - source = base / 'fileA' - target = base / 'copyA' - result = source.copy(target) - self.assertEqual(result, target) - self.assertTrue(target.exists()) - self.assertEqual(source.read_text(), target.read_text()) - - def test_copy_file_to_existing_file(self): - base = self.cls(self.base) - source = base / 'fileA' - target = base / 'dirB' / 'fileB' - result = source.copy(target) - self.assertEqual(result, target) - self.assertTrue(target.exists()) - self.assertEqual(source.read_text(), target.read_text()) - - def test_copy_file_to_existing_directory(self): - base = self.cls(self.base) - source = base / 'fileA' - target = base / 'dirA' - self.assertRaises(OSError, source.copy, target) - - def test_copy_file_empty(self): - base = self.cls(self.base) - source = base / 'empty' - target = base / 'copyA' - source.write_bytes(b'') - result = source.copy(target) - self.assertEqual(result, target) - self.assertTrue(target.exists()) - self.assertEqual(target.read_bytes(), b'') - - def test_copy_file_to_itself(self): - base = self.cls(self.base) - source = base / 'empty' - source.write_bytes(b'') - self.assertRaises(OSError, source.copy, source) - self.assertRaises(OSError, source.copy, source, follow_symlinks=False) - - def test_copy_dir_simple(self): - base = self.cls(self.base) - source = base / 'dirC' - target = base / 'copyC' - result = source.copy(target) - self.assertEqual(result, target) - self.assertTrue(target.is_dir()) - self.assertTrue(target.joinpath('dirD').is_dir()) - self.assertTrue(target.joinpath('dirD', 'fileD').is_file()) - self.assertEqual(target.joinpath('dirD', 'fileD').read_text(), - "this is file D\n") - self.assertTrue(target.joinpath('fileC').is_file()) - self.assertTrue(target.joinpath('fileC').read_text(), - "this is file C\n") - - def test_copy_dir_complex(self, follow_symlinks=True): - def ordered_walk(path): - for dirpath, dirnames, filenames in path.walk(follow_symlinks=follow_symlinks): - dirnames.sort() - filenames.sort() - yield dirpath, dirnames, filenames - base = self.cls(self.base) - source = base / 'dirC' - - if self.can_symlink: - # Add some symlinks - source.joinpath('linkC').symlink_to('fileC') - source.joinpath('linkD').symlink_to('dirD', target_is_directory=True) - - # Perform the copy - target = base / 'copyC' - result = source.copy(target, follow_symlinks=follow_symlinks) - self.assertEqual(result, target) - - # Compare the source and target trees - source_walk = ordered_walk(source) - target_walk = ordered_walk(target) - for source_item, target_item in zip(source_walk, target_walk, strict=True): - self.assertEqual(source_item[0].parts[len(source.parts):], - target_item[0].parts[len(target.parts):]) # dirpath - self.assertEqual(source_item[1], target_item[1]) # dirnames - self.assertEqual(source_item[2], target_item[2]) # filenames - # Compare files and symlinks - for filename in source_item[2]: - source_file = source_item[0].joinpath(filename) - target_file = target_item[0].joinpath(filename) - if follow_symlinks or not source_file.is_symlink(): - # Regular file. - self.assertEqual(source_file.read_bytes(), target_file.read_bytes()) - elif source_file.is_dir(): - # Symlink to directory. - self.assertTrue(target_file.is_dir()) - self.assertEqual(source_file.readlink(), target_file.readlink()) - else: - # Symlink to file. - self.assertEqual(source_file.read_bytes(), target_file.read_bytes()) - self.assertEqual(source_file.readlink(), target_file.readlink()) - - def test_copy_dir_complex_follow_symlinks_false(self): - self.test_copy_dir_complex(follow_symlinks=False) - - def test_copy_dir_to_existing_directory(self): - base = self.cls(self.base) - source = base / 'dirC' - target = base / 'copyC' - target.mkdir() - target.joinpath('dirD').mkdir() - self.assertRaises(FileExistsError, source.copy, target) - - def test_copy_dir_to_existing_directory_dirs_exist_ok(self): - base = self.cls(self.base) - source = base / 'dirC' - target = base / 'copyC' - target.mkdir() - target.joinpath('dirD').mkdir() - result = source.copy(target, dirs_exist_ok=True) - self.assertEqual(result, target) - self.assertTrue(target.is_dir()) - self.assertTrue(target.joinpath('dirD').is_dir()) - self.assertTrue(target.joinpath('dirD', 'fileD').is_file()) - self.assertEqual(target.joinpath('dirD', 'fileD').read_text(), - "this is file D\n") - self.assertTrue(target.joinpath('fileC').is_file()) - self.assertTrue(target.joinpath('fileC').read_text(), - "this is file C\n") - - def test_copy_dir_to_itself(self): - base = self.cls(self.base) - source = base / 'dirC' - self.assertRaises(OSError, source.copy, source) - self.assertRaises(OSError, source.copy, source, follow_symlinks=False) - - def test_copy_dir_into_itself(self): - base = self.cls(self.base) - source = base / 'dirC' - target = base / 'dirC' / 'dirD' / 'copyC' - self.assertRaises(OSError, source.copy, target) - self.assertRaises(OSError, source.copy, target, follow_symlinks=False) - self.assertFalse(target.exists()) - - def test_copy_into(self): - base = self.cls(self.base) - source = base / 'fileA' - target_dir = base / 'dirA' - result = source.copy_into(target_dir) - self.assertEqual(result, target_dir / 'fileA') - self.assertTrue(result.exists()) - self.assertEqual(source.read_text(), result.read_text()) - - def test_copy_into_empty_name(self): - source = self.cls('') - target_dir = self.base - self.assertRaises(ValueError, source.copy_into, target_dir) - def test_iterdir(self): P = self.cls p = P(self.base) @@ -1574,9 +1359,220 @@ def test_is_symlink(self): self.assertIs((P / 'linkA\x00').is_file(), False) -class DummyPathWalkTest(unittest.TestCase): - cls = DummyPath - base = DummyPathTest.base +class DummyWritablePathTest(DummyReadablePathTest): + cls = DummyWritablePath + + def test_read_write_bytes(self): + p = self.cls(self.base) + (p / 'fileA').write_bytes(b'abcdefg') + self.assertEqual((p / 'fileA').read_bytes(), b'abcdefg') + # Check that trying to write str does not truncate the file. + self.assertRaises(TypeError, (p / 'fileA').write_bytes, 'somestr') + self.assertEqual((p / 'fileA').read_bytes(), b'abcdefg') + + def test_read_write_text(self): + p = self.cls(self.base) + (p / 'fileA').write_text('äbcdefg', encoding='latin-1') + self.assertEqual((p / 'fileA').read_text( + encoding='utf-8', errors='ignore'), 'bcdefg') + # Check that trying to write bytes does not truncate the file. + self.assertRaises(TypeError, (p / 'fileA').write_text, b'somebytes') + self.assertEqual((p / 'fileA').read_text(encoding='latin-1'), 'äbcdefg') + + def test_read_text_with_newlines(self): + p = self.cls(self.base) + # Check that `\n` character change nothing + (p / 'fileA').write_bytes(b'abcde\r\nfghlk\n\rmnopq') + self.assertEqual((p / 'fileA').read_text(newline='\n'), + 'abcde\r\nfghlk\n\rmnopq') + # Check that `\r` character replaces `\n` + (p / 'fileA').write_bytes(b'abcde\r\nfghlk\n\rmnopq') + self.assertEqual((p / 'fileA').read_text(newline='\r'), + 'abcde\r\nfghlk\n\rmnopq') + # Check that `\r\n` character replaces `\n` + (p / 'fileA').write_bytes(b'abcde\r\nfghlk\n\rmnopq') + self.assertEqual((p / 'fileA').read_text(newline='\r\n'), + 'abcde\r\nfghlk\n\rmnopq') + + def test_write_text_with_newlines(self): + p = self.cls(self.base) + # Check that `\n` character change nothing + (p / 'fileA').write_text('abcde\r\nfghlk\n\rmnopq', newline='\n') + self.assertEqual((p / 'fileA').read_bytes(), + b'abcde\r\nfghlk\n\rmnopq') + # Check that `\r` character replaces `\n` + (p / 'fileA').write_text('abcde\r\nfghlk\n\rmnopq', newline='\r') + self.assertEqual((p / 'fileA').read_bytes(), + b'abcde\r\rfghlk\r\rmnopq') + # Check that `\r\n` character replaces `\n` + (p / 'fileA').write_text('abcde\r\nfghlk\n\rmnopq', newline='\r\n') + self.assertEqual((p / 'fileA').read_bytes(), + b'abcde\r\r\nfghlk\r\n\rmnopq') + # Check that no argument passed will change `\n` to `os.linesep` + os_linesep_byte = bytes(os.linesep, encoding='ascii') + (p / 'fileA').write_text('abcde\nfghlk\n\rmnopq') + self.assertEqual((p / 'fileA').read_bytes(), + b'abcde' + os_linesep_byte + b'fghlk' + os_linesep_byte + b'\rmnopq') + + def test_copy_file(self): + base = self.cls(self.base) + source = base / 'fileA' + target = base / 'copyA' + result = source.copy(target) + self.assertEqual(result, target) + self.assertTrue(target.exists()) + self.assertEqual(source.read_text(), target.read_text()) + + def test_copy_file_to_existing_file(self): + base = self.cls(self.base) + source = base / 'fileA' + target = base / 'dirB' / 'fileB' + result = source.copy(target) + self.assertEqual(result, target) + self.assertTrue(target.exists()) + self.assertEqual(source.read_text(), target.read_text()) + + def test_copy_file_to_existing_directory(self): + base = self.cls(self.base) + source = base / 'fileA' + target = base / 'dirA' + self.assertRaises(OSError, source.copy, target) + + def test_copy_file_empty(self): + base = self.cls(self.base) + source = base / 'empty' + target = base / 'copyA' + source.write_bytes(b'') + result = source.copy(target) + self.assertEqual(result, target) + self.assertTrue(target.exists()) + self.assertEqual(target.read_bytes(), b'') + + def test_copy_file_to_itself(self): + base = self.cls(self.base) + source = base / 'empty' + source.write_bytes(b'') + self.assertRaises(OSError, source.copy, source) + self.assertRaises(OSError, source.copy, source, follow_symlinks=False) + + def test_copy_dir_simple(self): + base = self.cls(self.base) + source = base / 'dirC' + target = base / 'copyC' + result = source.copy(target) + self.assertEqual(result, target) + self.assertTrue(target.is_dir()) + self.assertTrue(target.joinpath('dirD').is_dir()) + self.assertTrue(target.joinpath('dirD', 'fileD').is_file()) + self.assertEqual(target.joinpath('dirD', 'fileD').read_text(), + "this is file D\n") + self.assertTrue(target.joinpath('fileC').is_file()) + self.assertTrue(target.joinpath('fileC').read_text(), + "this is file C\n") + + def test_copy_dir_complex(self, follow_symlinks=True): + def ordered_walk(path): + for dirpath, dirnames, filenames in path.walk(follow_symlinks=follow_symlinks): + dirnames.sort() + filenames.sort() + yield dirpath, dirnames, filenames + base = self.cls(self.base) + source = base / 'dirC' + + if self.can_symlink: + # Add some symlinks + source.joinpath('linkC').symlink_to('fileC') + source.joinpath('linkD').symlink_to('dirD', target_is_directory=True) + + # Perform the copy + target = base / 'copyC' + result = source.copy(target, follow_symlinks=follow_symlinks) + self.assertEqual(result, target) + + # Compare the source and target trees + source_walk = ordered_walk(source) + target_walk = ordered_walk(target) + for source_item, target_item in zip(source_walk, target_walk, strict=True): + self.assertEqual(source_item[0].parts[len(source.parts):], + target_item[0].parts[len(target.parts):]) # dirpath + self.assertEqual(source_item[1], target_item[1]) # dirnames + self.assertEqual(source_item[2], target_item[2]) # filenames + # Compare files and symlinks + for filename in source_item[2]: + source_file = source_item[0].joinpath(filename) + target_file = target_item[0].joinpath(filename) + if follow_symlinks or not source_file.is_symlink(): + # Regular file. + self.assertEqual(source_file.read_bytes(), target_file.read_bytes()) + elif source_file.is_dir(): + # Symlink to directory. + self.assertTrue(target_file.is_dir()) + self.assertEqual(source_file.readlink(), target_file.readlink()) + else: + # Symlink to file. + self.assertEqual(source_file.read_bytes(), target_file.read_bytes()) + self.assertEqual(source_file.readlink(), target_file.readlink()) + + def test_copy_dir_complex_follow_symlinks_false(self): + self.test_copy_dir_complex(follow_symlinks=False) + + def test_copy_dir_to_existing_directory(self): + base = self.cls(self.base) + source = base / 'dirC' + target = base / 'copyC' + target.mkdir() + target.joinpath('dirD').mkdir() + self.assertRaises(FileExistsError, source.copy, target) + + def test_copy_dir_to_existing_directory_dirs_exist_ok(self): + base = self.cls(self.base) + source = base / 'dirC' + target = base / 'copyC' + target.mkdir() + target.joinpath('dirD').mkdir() + result = source.copy(target, dirs_exist_ok=True) + self.assertEqual(result, target) + self.assertTrue(target.is_dir()) + self.assertTrue(target.joinpath('dirD').is_dir()) + self.assertTrue(target.joinpath('dirD', 'fileD').is_file()) + self.assertEqual(target.joinpath('dirD', 'fileD').read_text(), + "this is file D\n") + self.assertTrue(target.joinpath('fileC').is_file()) + self.assertTrue(target.joinpath('fileC').read_text(), + "this is file C\n") + + def test_copy_dir_to_itself(self): + base = self.cls(self.base) + source = base / 'dirC' + self.assertRaises(OSError, source.copy, source) + self.assertRaises(OSError, source.copy, source, follow_symlinks=False) + + def test_copy_dir_into_itself(self): + base = self.cls(self.base) + source = base / 'dirC' + target = base / 'dirC' / 'dirD' / 'copyC' + self.assertRaises(OSError, source.copy, target) + self.assertRaises(OSError, source.copy, target, follow_symlinks=False) + self.assertFalse(target.exists()) + + def test_copy_into(self): + base = self.cls(self.base) + source = base / 'fileA' + target_dir = base / 'dirA' + result = source.copy_into(target_dir) + self.assertEqual(result, target_dir / 'fileA') + self.assertTrue(result.exists()) + self.assertEqual(source.read_text(), result.read_text()) + + def test_copy_into_empty_name(self): + source = self.cls('') + target_dir = self.base + self.assertRaises(ValueError, source.copy_into, target_dir) + + +class DummyReadablePathWalkTest(unittest.TestCase): + cls = DummyReadablePath + base = DummyReadablePathTest.base can_symlink = False def setUp(self):