path: root/Lib/test/test_pathlib/test_pathlib_abc.py

import io
import os
import errno
import unittest

from pathlib.types import _JoinablePath, _ReadablePath, _WritablePath
import posixpath

from test.support.os_helper import TESTFN


_tests_needing_posix = set()
_tests_needing_windows = set()


def needs_posix(fn):
    """Decorator that marks a test as requiring a POSIX-flavoured path class."""
    _tests_needing_posix.add(fn.__name__)
    return fn

def needs_windows(fn):
    """Decorator that marks a test as requiring a Windows-flavoured path class."""
    _tests_needing_windows.add(fn.__name__)
    return fn


#
# Tests for the pure classes.
#


class DummyJoinablePath(_JoinablePath):
    __slots__ = ('_segments',)

    parser = posixpath

    def __init__(self, *segments):
        self._segments = segments

    def __str__(self):
        if self._segments:
            return self.parser.join(*self._segments)
        return ''

    def __eq__(self, other):
        if not isinstance(other, DummyJoinablePath):
            return NotImplemented
        return str(self) == str(other)

    def __hash__(self):
        return hash(str(self))

    def __repr__(self):
        return "{}({!r})".format(self.__class__.__name__, str(self))

    def with_segments(self, *pathsegments):
        return type(self)(*pathsegments)


class JoinablePathTest(unittest.TestCase):
    cls = DummyJoinablePath

    # Use a base path that's unrelated to any real filesystem path.
    base = f'/this/path/kills/fascists/{TESTFN}'

    def setUp(self):
        name = self.id().split('.')[-1]
        if name in _tests_needing_posix and self.cls.parser is not posixpath:
            self.skipTest('requires POSIX-flavoured path class')
        if name in _tests_needing_windows and self.cls.parser is posixpath:
            self.skipTest('requires Windows-flavoured path class')
        p = self.cls('a')
        self.parser = p.parser
        self.sep = self.parser.sep
        self.altsep = self.parser.altsep

    def _check_str_subclass(self, *args):
        # Issue #21127: it should be possible to construct a PurePath object
        # from a str subclass instance, and it then gets converted to
        # a pure str object.
        class StrSubclass(str):
            pass
        P = self.cls
        p = P(*(StrSubclass(x) for x in args))
        self.assertEqual(p, P(*args))
        for part in p.parts:
            self.assertIs(type(part), str)

    def test_str_subclass_common(self):
        self._check_str_subclass('')
        self._check_str_subclass('.')
        self._check_str_subclass('a')
        self._check_str_subclass('a/b.txt')
        self._check_str_subclass('/a/b.txt')

    @needs_windows
    def test_str_subclass_windows(self):
        self._check_str_subclass('.\\a:b')
        self._check_str_subclass('c:')
        self._check_str_subclass('c:a')
        self._check_str_subclass('c:a\\b.txt')
        self._check_str_subclass('c:\\')
        self._check_str_subclass('c:\\a')
        self._check_str_subclass('c:\\a\\b.txt')
        self._check_str_subclass('\\\\some\\share')
        self._check_str_subclass('\\\\some\\share\\a')
        self._check_str_subclass('\\\\some\\share\\a\\b.txt')

    def _check_str(self, expected, args):
        p = self.cls(*args)
        self.assertEqual(str(p), expected.replace('/', self.sep))

    def test_str_common(self):
        # Canonicalized paths roundtrip.
        for pathstr in ('a', 'a/b', 'a/b/c', '/', '/a/b', '/a/b/c'):
            self._check_str(pathstr, (pathstr,))
        # Other tests for str() are in test_equivalences().

    @needs_windows
    def test_str_windows(self):
        p = self.cls('a/b/c')
        self.assertEqual(str(p), 'a\\b\\c')
        p = self.cls('c:/a/b/c')
        self.assertEqual(str(p), 'c:\\a\\b\\c')
        p = self.cls('//a/b')
        self.assertEqual(str(p), '\\\\a\\b\\')
        p = self.cls('//a/b/c')
        self.assertEqual(str(p), '\\\\a\\b\\c')
        p = self.cls('//a/b/c/d')
        self.assertEqual(str(p), '\\\\a\\b\\c\\d')


#
# Tests for the virtual classes.
#


class DummyWritablePathIO(io.BytesIO):
    """
    Used by DummyWritablePath to implement `__open_wb__()`
    """

    def __init__(self, files, path):
        super().__init__()
        self.files = files
        self.path = path

    def close(self):
        self.files[self.path] = self.getvalue()
        super().close()


class DummyReadablePathInfo:
    __slots__ = ('_is_dir', '_is_file')

    def __init__(self, is_dir, is_file):
        self._is_dir = is_dir
        self._is_file = is_file

    def exists(self, *, follow_symlinks=True):
        return self._is_dir or self._is_file

    def is_dir(self, *, follow_symlinks=True):
        return self._is_dir

    def is_file(self, *, follow_symlinks=True):
        return self._is_file

    def is_symlink(self):
        return False


class DummyReadablePath(_ReadablePath, DummyJoinablePath):
    """
    Simple implementation of DummyReadablePath that keeps files and
    directories in memory.
    """
    __slots__ = ('_info')

    _files = {}
    _directories = {}
    parser = posixpath

    def __init__(self, *segments):
        super().__init__(*segments)
        self._info = None

    @property
    def info(self):
        if self._info is None:
            path_str = str(self)
            self._info = DummyReadablePathInfo(
                is_dir=path_str.rstrip('/') in self._directories,
                is_file=path_str in self._files)
        return self._info

    def __open_rb__(self, buffering=-1):
        path = str(self)
        if path in self._directories:
            raise IsADirectoryError(errno.EISDIR, "Is a directory", path)
        elif path not in self._files:
            raise FileNotFoundError(errno.ENOENT, "File not found", path)
        return io.BytesIO(self._files[path])

    def iterdir(self):
        path = str(self).rstrip('/')
        if path in self._files:
            raise NotADirectoryError(errno.ENOTDIR, "Not a directory", path)
        elif path in self._directories:
            return iter([self / name for name in self._directories[path]])
        else:
            raise FileNotFoundError(errno.ENOENT, "File not found", path)

    def readlink(self):
        raise NotImplementedError


class DummyWritablePath(_WritablePath, DummyJoinablePath):
    __slots__ = ()

    def __open_wb__(self, buffering=-1):
        path = str(self)
        if path in self._directories:
            raise IsADirectoryError(errno.EISDIR, "Is a directory", path)
        parent, name = posixpath.split(path)
        if parent not in self._directories:
            raise FileNotFoundError(errno.ENOENT, "File not found", parent)
        self._files[path] = b''
        self._directories[parent].add(name)
        return DummyWritablePathIO(self._files, path)

    def mkdir(self):
        path = str(self)
        parent = str(self.parent)
        if path in self._directories:
            raise FileExistsError(errno.EEXIST, "File exists", path)
        try:
            if self.name:
                self._directories[parent].add(self.name)
            self._directories[path] = set()
        except KeyError:
            raise FileNotFoundError(errno.ENOENT, "File not found", parent) from None

    def symlink_to(self, target, target_is_directory=False):
        raise NotImplementedError


class ReadablePathTest(JoinablePathTest):
    """Tests for ReadablePathTest methods that use stat(), open() and iterdir()."""

    cls = DummyReadablePath
    can_symlink = False

    # (self.base)
    #  |
    #  |-- brokenLink -> non-existing
    #  |-- dirA
    #  |   `-- linkC -> ../dirB
    #  |-- dirB
    #  |   |-- fileB
    #  |   `-- linkD -> ../dirB
    #  |-- dirC
    #  |   |-- dirD
    #  |   |   `-- fileD
    #  |   `-- fileC
    #  |   `-- novel.txt
    #  |-- dirE  # No permissions
    #  |-- fileA
    #  |-- linkA -> fileA
    #  |-- linkB -> dirB
    #  `-- brokenLinkLoop -> brokenLinkLoop
    #

    def setUp(self):
        super().setUp()
        self.createTestHierarchy()

    def createTestHierarchy(self):
        cls = self.cls
        cls._files = {
            f'{self.base}/fileA': b'this is file A\n',
            f'{self.base}/dirB/fileB': b'this is file B\n',
            f'{self.base}/dirC/fileC': b'this is file C\n',
            f'{self.base}/dirC/dirD/fileD': b'this is file D\n',
            f'{self.base}/dirC/novel.txt': b'this is a novel\n',
        }
        cls._directories = {
            f'{self.base}': {'fileA', 'dirA', 'dirB', 'dirC', 'dirE'},
            f'{self.base}/dirA': set(),
            f'{self.base}/dirB': {'fileB'},
            f'{self.base}/dirC': {'fileC', 'dirD', 'novel.txt'},
            f'{self.base}/dirC/dirD': {'fileD'},
            f'{self.base}/dirE': set(),
        }

    def tearDown(self):
        cls = self.cls
        cls._files.clear()
        cls._directories.clear()

    def tempdir(self):
        path = self.cls(self.base).with_name('tmp-dirD')
        path.mkdir()
        return path

    def assertFileNotFound(self, func, *args, **kwargs):
        with self.assertRaises(FileNotFoundError) as cm:
            func(*args, **kwargs)
        self.assertEqual(cm.exception.errno, errno.ENOENT)

    def assertEqualNormCase(self, path_a, path_b):
        normcase = self.parser.normcase
        self.assertEqual(normcase(path_a), normcase(path_b))

    @needs_posix
    def test_glob_posix(self):
        P = self.cls
        p = P(self.base)
        q = p / "FILEa"
        given = set(p.glob("FILEa"))
        expect = {q} if q.info.exists() else set()
        self.assertEqual(given, expect)
        self.assertEqual(set(p.glob("FILEa*")), set())

    @needs_windows
    def test_glob_windows(self):
        P = self.cls
        p = P(self.base)
        self.assertEqual(set(p.glob("FILEa")), { P(self.base, "fileA") })
        self.assertEqual(set(p.glob("*a\\")), { P(self.base, "dirA/") })
        self.assertEqual(set(p.glob("F*a")), { P(self.base, "fileA") })


class WritablePathTest(JoinablePathTest):
    cls = DummyWritablePath


class DummyRWPath(DummyWritablePath, DummyReadablePath):
    __slots__ = ()


class RWPathTest(WritablePathTest, ReadablePathTest):
    cls = DummyRWPath
    can_symlink = False

    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(result.info.exists())
        self.assertEqual(source.read_text(), result.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(result.info.exists())
        self.assertEqual(source.read_text(), result.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(result.info.exists())
        self.assertEqual(result.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(result.info.is_dir())
        self.assertTrue(result.joinpath('dirD').info.is_dir())
        self.assertTrue(result.joinpath('dirD', 'fileD').info.is_file())
        self.assertEqual(result.joinpath('dirD', 'fileD').read_text(),
                         "this is file D\n")
        self.assertTrue(result.joinpath('fileC').info.is_file())
        self.assertTrue(result.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(result)
        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.info.is_symlink():
                    # Regular file.
                    self.assertEqual(source_file.read_bytes(), target_file.read_bytes())
                elif source_file.info.is_dir():
                    # Symlink to directory.
                    self.assertTrue(target_file.info.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_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.info.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.info.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 ReadablePathWalkTest(unittest.TestCase):
    cls = DummyReadablePath
    base = ReadablePathTest.base
    can_symlink = False

    def setUp(self):
        self.walk_path = self.cls(self.base, "TEST1")
        self.sub1_path = self.walk_path / "SUB1"
        self.sub11_path = self.sub1_path / "SUB11"
        self.sub2_path = self.walk_path / "SUB2"
        self.link_path = self.sub2_path / "link"
        self.sub2_tree = (self.sub2_path, [], ["tmp3"])
        self.createTestHierarchy()

    def createTestHierarchy(self):
        cls = self.cls
        cls._files = {
            f'{self.base}/TEST1/tmp1': b'this is tmp1\n',
            f'{self.base}/TEST1/SUB1/tmp2': b'this is tmp2\n',
            f'{self.base}/TEST1/SUB2/tmp3': b'this is tmp3\n',
            f'{self.base}/TEST2/tmp4': b'this is tmp4\n',
        }
        cls._directories = {
            f'{self.base}': {'TEST1', 'TEST2'},
            f'{self.base}/TEST1': {'SUB1', 'SUB2', 'tmp1'},
            f'{self.base}/TEST1/SUB1': {'SUB11', 'tmp2'},
            f'{self.base}/TEST1/SUB1/SUB11': set(),
            f'{self.base}/TEST1/SUB2': {'tmp3'},
            f'{self.base}/TEST2': {'tmp4'},
        }

    def tearDown(self):
        cls = self.cls
        cls._files.clear()
        cls._directories.clear()


if __name__ == "__main__":
    unittest.main()