Skip to content

Commit

Permalink
support for resolved symlinks in copytree
Browse files Browse the repository at this point in the history
  • Loading branch information
brno32 committed Jul 19, 2023
1 parent b81d5f1 commit 38cb37e
Show file tree
Hide file tree
Showing 2 changed files with 62 additions and 5 deletions.
29 changes: 24 additions & 5 deletions src/smbclient/shutil.py
Original file line number Diff line number Diff line change
Expand Up @@ -293,6 +293,16 @@ def copytree(
:param kwargs: Common arguments used to build the SMB Session for any UNC paths.
:return: The dst path.
"""

class LocalDirEntry:
"""Mimics the structure of os.DirEntry which is not exposed https://github.com/python/cpython/issues/71225"""

def __init__(self, path) -> None:
self.path = path

Check warning on line 301 in src/smbclient/shutil.py

View check run for this annotation

Codecov / codecov/patch

src/smbclient/shutil.py#L301

Added line #L301 was not covered by tests

def is_dir(self):
return os.path.isdir(self.path)

Check warning on line 304 in src/smbclient/shutil.py

View check run for this annotation

Codecov / codecov/patch

src/smbclient/shutil.py#L304

Added line #L304 was not covered by tests

if is_remote_path(src):
dir_entries = list(scandir(src, **kwargs))
else:
Expand All @@ -315,20 +325,29 @@ def copytree(
src_path = _join_local_or_remote_path(src, dir_entry.name)
dst_path = _join_local_or_remote_path(dst, dir_entry.name)

is_smb_dir_entry = isinstance(dir_entry, SMBDirEntry)

try:
if dir_entry.is_symlink():
if not isinstance(dir_entry, SMBDirEntry):
raise AssertionError("copytree doesn't yet support symlinks for local to remote operations")

link_target = readlink(src_path, **kwargs)
if is_smb_dir_entry:
link_target = readlink(src_path, **kwargs)
else:
link_target = os.readlink(src_path)

Check warning on line 335 in src/smbclient/shutil.py

View check run for this annotation

Codecov / codecov/patch

src/smbclient/shutil.py#L335

Added line #L335 was not covered by tests
if symlinks:
if not is_smb_dir_entry:
raise AssertionError(

Check warning on line 338 in src/smbclient/shutil.py

View check run for this annotation

Codecov / codecov/patch

src/smbclient/shutil.py#L338

Added line #L338 was not covered by tests
"copytree doesn't yet support unresolved symlinks for local to remote operations"
)
symlink(link_target, dst_path, **kwargs)
copystat(src_path, dst_path, follow_symlinks=False)
continue
else:
# Manually override the dir_entry with a new one that is the link target and copy that below.
try:
dir_entry = SMBDirEntry.from_path(link_target, **kwargs)
if not is_remote_path(link_target):
dir_entry = LocalDirEntry(link_target)

Check warning on line 348 in src/smbclient/shutil.py

View check run for this annotation

Codecov / codecov/patch

src/smbclient/shutil.py#L348

Added line #L348 was not covered by tests
else:
dir_entry = SMBDirEntry.from_path(link_target, **kwargs)
except OSError as err:
if err.errno == errno.ENOENT and ignore_dangling_symlinks:
continue
Expand Down
38 changes: 38 additions & 0 deletions tests/test_smbclient_shutil.py
Original file line number Diff line number Diff line change
Expand Up @@ -1203,6 +1203,44 @@ def test_copytree_with_local_src(smb_share, tmp_path):
assert fd.read() == "file3.txt"


@pytest.mark.skipif(os.name != "nt", reason="Samba target doesn't support symlinks.")
def test_copytree_with_local_src_and_symlinks(smb_share, tmp_path):
src_dirname = str(tmp_path / "source")
alt_src_dirname = str(tmp_path / "source2")
dst_dirname = "%s\\target" % smb_share
alt_dst_dirname = "%s\\target2" % smb_share

os.makedirs(os.path.join(src_dirname, "dir1", "subdir1"))
with open(os.path.join(src_dirname, "file1.txt"), mode="w") as fd:
fd.write("file1.txt")
with open(os.path.join(src_dirname, "dir1", "file2.txt"), mode="w") as fd:
fd.write("file2.txt")
with open(os.path.join(src_dirname, "dir1", "subdir1", "file3.txt"), mode="w") as fd:
fd.write("file3.txt")

path_to_symlink = os.path.join(src_dirname, "dir1", "subdir1", "symlink-to-resolve")
path_to_symlink_target = os.path.join(alt_src_dirname, "symlink-target.txt")
with open(path_to_symlink_target, mode="w") as fd:
fd.write("symlink-target.txt")
os.symlink(path_to_symlink_target, path_to_symlink)

actual = copytree(src_dirname, dst_dirname)
assert actual == dst_dirname

assert sorted(list(listdir(dst_dirname))) == ["dir1", "file1.txt"]
assert sorted(list(listdir("%s\\dir1" % dst_dirname))) == ["file2.txt", "subdir1"]
assert sorted(list(listdir("%s\\dir1\\subdir1" % dst_dirname))) == ["file3.txt"]

with open_file("%s\\file1.txt" % dst_dirname) as fd:
assert fd.read() == "file1.txt"
with open_file("%s\\dir1\\file2.txt" % dst_dirname) as fd:
assert fd.read() == "file2.txt"
with open_file("%s\\dir1\\subdir1\\file3.txt" % dst_dirname) as fd:
assert fd.read() == "file3.txt"
with open_file("%s\\symlink-target.txt" % alt_dst_dirname) as fd:
assert fd.read() == "symlink-target.txt"


@pytest.mark.skipif(
os.name != "nt" and not os.environ.get("SMB_FORCE", False), reason="Samba does not update timestamps"
)
Expand Down

0 comments on commit 38cb37e

Please sign in to comment.