mirror of
https://github.com/Stone-Red-Code/dvc.git
synced 2026-11-10 07:18:27 +01:00
Ruslan Kuprieiev
625b20833e
* dvcignore: move file/dir/parent checks into DvcIgnoreFilter * dvc: get rid of CleanTree CleanTree is a very awkward wrapper that spreads the context across the codebase and results in unexpected behaviour when using it. This PR starts moving dvcignore-related logic into the trees themselves (it makes a lot of sense, kinda like `state`) so they could deal with it however they like. There are at least two temporary ugly parts about this PR: 1) dvcignore is used by individual trees and not packed into tree/base.py yet; 2) `dvcignore_root` argument. This one is caused by the dvcignore trying to collect everything topdown starting from the certain root dir. What it should do instead is for a certain path that is being checked look up the tree through the parents until it finds repo root (.dvc dir) and then stop. That would handle subrepos as well. At the same time we need to leverage existing dvcignore trie structure to cache those results.
280 lines
8.1 KiB
Python
280 lines
8.1 KiB
Python
import filecmp
|
|
import os
|
|
from unittest import TestCase
|
|
|
|
import pytest
|
|
from mock import patch
|
|
|
|
import dvc
|
|
from dvc.path_info import PathInfo
|
|
from dvc.system import System
|
|
from dvc.tree.local import LocalRemoteTree
|
|
from dvc.utils import relpath
|
|
from dvc.utils.fs import (
|
|
BasePathNotInCheckedPathException,
|
|
contains_symlink_up_to,
|
|
copy_fobj_to_file,
|
|
copyfile,
|
|
get_inode,
|
|
get_mtime_and_size,
|
|
makedirs,
|
|
move,
|
|
path_isin,
|
|
remove,
|
|
walk_files,
|
|
)
|
|
from tests.basic_env import TestDir
|
|
|
|
|
|
class TestMtimeAndSize(TestDir):
|
|
def test(self):
|
|
tree = LocalRemoteTree(
|
|
None, {"url": self.root_dir}, use_dvcignore=True
|
|
)
|
|
file_time, file_size = get_mtime_and_size(self.DATA, tree)
|
|
dir_time, dir_size = get_mtime_and_size(self.DATA_DIR, tree)
|
|
|
|
actual_file_size = os.path.getsize(self.DATA)
|
|
actual_dir_size = os.path.getsize(self.DATA) + os.path.getsize(
|
|
self.DATA_SUB
|
|
)
|
|
|
|
self.assertIs(type(file_time), str)
|
|
self.assertIs(type(file_size), str)
|
|
self.assertEqual(file_size, str(actual_file_size))
|
|
self.assertIs(type(dir_time), str)
|
|
self.assertIs(type(dir_size), str)
|
|
self.assertEqual(dir_size, str(actual_dir_size))
|
|
|
|
|
|
class TestContainsLink(TestCase):
|
|
def test_should_raise_exception_on_base_path_not_in_path(self):
|
|
with self.assertRaises(BasePathNotInCheckedPathException):
|
|
contains_symlink_up_to(os.path.join("foo", "path"), "bar")
|
|
|
|
@patch.object(System, "is_symlink", return_value=True)
|
|
def test_should_return_true_on_symlink_in_path(self, _):
|
|
base_path = "foo"
|
|
path = os.path.join(base_path, "bar")
|
|
self.assertTrue(contains_symlink_up_to(path, base_path))
|
|
|
|
@patch.object(System, "is_symlink", return_value=False)
|
|
def test_should_return_false_on_path_eq_to_base_path(self, _):
|
|
path = "path"
|
|
self.assertFalse(contains_symlink_up_to(path, path))
|
|
|
|
@patch.object(System, "is_symlink", return_value=False)
|
|
@patch.object(os.path, "dirname", side_effect=lambda arg: arg)
|
|
def test_should_return_false_on_no_more_dirs_below_path(
|
|
self, dirname_patch, _
|
|
):
|
|
self.assertFalse(
|
|
contains_symlink_up_to(os.path.join("foo", "path"), "foo")
|
|
)
|
|
dirname_patch.assert_called_once()
|
|
|
|
@patch.object(System, "is_symlink", return_value=True)
|
|
def test_should_return_false_when_base_path_is_symlink(self, _):
|
|
base_path = "foo"
|
|
target_path = os.path.join(base_path, "bar")
|
|
|
|
def base_path_is_symlink(path):
|
|
if path == base_path:
|
|
return True
|
|
return False
|
|
|
|
with patch.object(
|
|
System, "is_symlink", side_effect=base_path_is_symlink
|
|
):
|
|
self.assertFalse(contains_symlink_up_to(target_path, base_path))
|
|
|
|
def test_path_object_and_str_are_valid_arg_types(self):
|
|
base_path = "foo"
|
|
target_path = os.path.join(base_path, "bar")
|
|
self.assertFalse(contains_symlink_up_to(target_path, base_path))
|
|
self.assertFalse(
|
|
contains_symlink_up_to(PathInfo(target_path), PathInfo(base_path))
|
|
)
|
|
|
|
|
|
def test_should_call_recursive_on_no_condition_matched(mocker):
|
|
mocker.patch.object(System, "is_symlink", return_value=False)
|
|
|
|
contains_symlink_spy = mocker.spy(dvc.utils.fs, "contains_symlink_up_to")
|
|
|
|
# call from full path to match contains_symlink_spy patch path
|
|
assert not dvc.utils.fs.contains_symlink_up_to(
|
|
os.path.join("foo", "path"), "foo"
|
|
)
|
|
assert contains_symlink_spy.mock.call_count == 2
|
|
|
|
|
|
@pytest.mark.skipif(os.name != "nt", reason="Windows specific")
|
|
def test_relpath_windows_different_drives():
|
|
path1 = os.path.join("A:", os.sep, "some", "path")
|
|
path2 = os.path.join("B:", os.sep, "other", "path")
|
|
assert relpath(path1, path2) == path1
|
|
|
|
info1, info2 = PathInfo(path1), PathInfo(path2)
|
|
rel_info = relpath(info1, info2)
|
|
assert isinstance(rel_info, str)
|
|
assert rel_info == path1
|
|
|
|
|
|
def test_get_inode(tmp_dir):
|
|
tmp_dir.gen("foo", "foo content")
|
|
|
|
assert get_inode("foo") == get_inode(PathInfo("foo"))
|
|
|
|
|
|
def test_path_object_and_str_are_valid_types_get_mtime_and_size(tmp_dir):
|
|
tmp_dir.gen(
|
|
{"dir": {"dir_file": "dir file content"}, "file": "file_content"}
|
|
)
|
|
tree = LocalRemoteTree(
|
|
None, {"url": os.fspath(tmp_dir)}, use_dvcignore=True
|
|
)
|
|
|
|
time, size = get_mtime_and_size("dir", tree)
|
|
object_time, object_size = get_mtime_and_size(PathInfo("dir"), tree)
|
|
assert time == object_time
|
|
assert size == object_size
|
|
|
|
time, size = get_mtime_and_size("file", tree)
|
|
object_time, object_size = get_mtime_and_size(PathInfo("file"), tree)
|
|
assert time == object_time
|
|
assert size == object_size
|
|
|
|
|
|
def test_move(tmp_dir):
|
|
tmp_dir.gen({"foo": "foo content", "bar": "bar content"})
|
|
src = "foo"
|
|
src_info = PathInfo("bar")
|
|
dest = os.path.join("some", "directory")
|
|
dest_info = PathInfo(os.path.join("some", "path-like", "directory"))
|
|
|
|
os.makedirs(dest)
|
|
assert len(os.listdir(dest)) == 0
|
|
move(src, dest)
|
|
assert not os.path.isfile(src)
|
|
assert len(os.listdir(dest)) == 1
|
|
|
|
os.makedirs(dest_info)
|
|
assert len(os.listdir(dest_info)) == 0
|
|
move(src_info, dest_info)
|
|
assert not os.path.isfile(src_info)
|
|
assert len(os.listdir(dest_info)) == 1
|
|
|
|
|
|
def test_remove(tmp_dir):
|
|
tmp_dir.gen({"foo": "foo content", "bar": "bar content"})
|
|
path = "foo"
|
|
path_info = PathInfo("bar")
|
|
|
|
remove(path)
|
|
assert not os.path.isfile(path)
|
|
|
|
remove(path_info)
|
|
assert not os.path.isfile(path_info)
|
|
|
|
|
|
def test_path_isin_positive():
|
|
child = os.path.join("path", "to", "folder")
|
|
|
|
assert path_isin(child, os.path.join("path", "to", ""))
|
|
assert path_isin(child, os.path.join("path", "to"))
|
|
assert path_isin(child, os.path.join("path", ""))
|
|
assert path_isin(child, os.path.join("path"))
|
|
|
|
|
|
def test_path_isin_on_same_path():
|
|
path = os.path.join("path", "to", "folder")
|
|
path_with_sep = os.path.join(path, "")
|
|
|
|
assert not path_isin(path, path)
|
|
assert not path_isin(path, path_with_sep)
|
|
assert not path_isin(path_with_sep, path)
|
|
assert not path_isin(path_with_sep, path_with_sep)
|
|
|
|
|
|
def test_path_isin_on_common_substring_path():
|
|
path1 = os.path.join("path", "to", "folder1")
|
|
path2 = os.path.join("path", "to", "folder")
|
|
|
|
assert not path_isin(path1, path2)
|
|
|
|
|
|
def test_path_isin_accepts_pathinfo():
|
|
child = os.path.join("path", "to", "folder")
|
|
parent = PathInfo(child) / ".."
|
|
|
|
assert path_isin(child, parent)
|
|
# pylint: disable=arguments-out-of-order
|
|
assert not path_isin(parent, child)
|
|
|
|
|
|
def test_path_isin_with_absolute_path():
|
|
parent = os.path.abspath("path")
|
|
child = os.path.join(parent, "to", "folder")
|
|
|
|
assert path_isin(child, parent)
|
|
|
|
|
|
def test_makedirs(tmp_dir):
|
|
path = os.path.join(tmp_dir, "directory")
|
|
path_info = PathInfo(os.path.join(tmp_dir, "another", "directory"))
|
|
|
|
makedirs(path)
|
|
assert os.path.isdir(path)
|
|
|
|
makedirs(path_info)
|
|
assert os.path.isdir(path_info)
|
|
|
|
|
|
@pytest.mark.parametrize("path", ["file", "dir"])
|
|
def test_copyfile(path, tmp_dir):
|
|
tmp_dir.gen(
|
|
{
|
|
"foo": "foo content",
|
|
"bar": "bar content",
|
|
"file": "file content",
|
|
"dir": {},
|
|
}
|
|
)
|
|
src = "foo"
|
|
dest = path
|
|
src_info = PathInfo("bar")
|
|
dest_info = PathInfo(path)
|
|
|
|
copyfile(src, dest)
|
|
if os.path.isdir(dest):
|
|
assert filecmp.cmp(
|
|
src, os.path.join(dest, os.path.basename(src)), shallow=False
|
|
)
|
|
else:
|
|
assert filecmp.cmp(src, dest, shallow=False)
|
|
|
|
copyfile(src_info, dest_info)
|
|
if os.path.isdir(dest_info):
|
|
assert filecmp.cmp(
|
|
src_info,
|
|
os.path.join(dest_info, os.path.basename(src_info)),
|
|
shallow=False,
|
|
)
|
|
else:
|
|
assert filecmp.cmp(src_info, dest_info, shallow=False)
|
|
|
|
|
|
def test_copy_fobj_to_file(tmp_dir):
|
|
tmp_dir.gen({"foo": "foo content"})
|
|
src = tmp_dir / "foo"
|
|
dest = "path"
|
|
|
|
with open(src, "rb") as fobj:
|
|
copy_fobj_to_file(fobj, dest)
|
|
assert filecmp.cmp(src, dest)
|
|
|
|
|
|
def test_walk_files(tmp_dir):
|
|
assert list(walk_files(".")) == list(walk_files(tmp_dir))
|