How to use the dvc.system.System function in dvc

To help you get started, we’ve selected a few dvc examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github iterative / dvc / tests / func / test_import.py View on Github external
def test_cache_type_is_properly_overridden(git, dvc_repo, erepo):
    erepo.dvc.config.set(
        Config.SECTION_CACHE, Config.SECTION_CACHE_TYPE, "symlink"
    )
    erepo.dvc.scm.add([erepo.dvc.config.config_file])
    erepo.dvc.scm.commit("set source repo cache type to symlinks")

    src = erepo.FOO
    dst = erepo.FOO + "_imported"

    dvc_repo.imp(erepo.root_dir, src, dst)

    assert not System.is_symlink(dst)
    assert os.path.exists(dst)
    assert os.path.isfile(dst)
    assert filecmp.cmp(erepo.FOO, dst, shallow=False)
    assert git.git.check_ignore(dst)
github iterative / dvc / tests / test_cmd_target.py View on Github external
os.chdir(self._test_git_dir)

        os.mkdir(os.path.join('data', 'dir1'))
        os.mkdir(os.path.join('.dvc', 'cache', 'dir1'))

        self.file1_cache = os.path.join('.dvc', 'cache', 'dir1', 'file1')
        self.file1_data = os.path.join('data', 'dir1', 'file1')

        open(self.file1_cache, 'w').write('ddfdff')
        System.symlink(self.file1_cache, self.file1_data)

        self.file2_cache = os.path.join('.dvc', 'cache', 'file2')
        self.file2_data = os.path.join('data', 'file2')

        open(self.file2_cache, 'w').write('ddfdff')
        System.symlink(self.file2_cache, self.file2_data)
github iterative / dvc / tests / test_data_file_obj.py View on Github external
def setUp(self):
        BasicEnvironment.init_environment(self, test_dir=os.path.join(os.path.sep, 'tmp', 'ntx_unit_test'))

        git = GitWrapperI(git_dir=self._test_git_dir, commit='ad45ba8')
        config = ConfigI('data')
        self.path_factory = PathFactory(git, config)

        self.data_file = os.path.join('data', 'file.txt')
        self.cache_file = os.path.join(ConfigI.CONFIG_DIR, ConfigI.CACHE_DIR, 'fsymlinc.txt')

        fd = open(self.cache_file, 'w+')
        fd.write('some text')
        fd.close()

        os.chdir('data')
        System.symlink(os.path.join('..', self.cache_file), 'file.txt')
        os.chdir('..')
        pass
github iterative / dvc / tests / func / test_checkout.py View on Github external
old_foo_link = readlink(self.FOO)

        self.assertTrue(System.is_symlink(self.DATA))
        old_data_link = readlink(self.DATA)

        old_cache_dir = self.dvc.cache.local.cache_dir
        new_cache_dir = old_cache_dir + "_new"
        os.rename(old_cache_dir, new_cache_dir)

        ret = main(["cache", "dir", new_cache_dir])
        self.assertEqual(ret, 0)

        ret = main(["checkout", "-f"])
        self.assertEqual(ret, 0)

        self.assertTrue(System.is_symlink(self.FOO))
        new_foo_link = readlink(self.FOO)

        self.assertTrue(System.is_symlink(self.DATA))
        new_data_link = readlink(self.DATA)

        self.assertEqual(
            relpath(old_foo_link, old_cache_dir),
            relpath(new_foo_link, new_cache_dir),
        )

        self.assertEqual(
            relpath(old_data_link, old_cache_dir),
            relpath(new_data_link, new_cache_dir),
        )
github iterative / dvc / dvc / state_file.py View on Github external
def get_dvc_path(self):
        pwd = System.get_cwd()
        if not pwd.startswith(self.settings.git.git_dir_abs):
            raise StateFileError('the file cannot be created outside of a git repository')

        return os.path.relpath(pwd, self.settings.git.git_dir_abs)
github iterative / dvc / dvc / system.py View on Github external
def reflink(source, link_name):
        import platform
        from dvc.exceptions import DvcException

        source, link_name = fspath(source), fspath(link_name)

        system = platform.system()
        try:
            if system == "Windows":
                ret = System._reflink_windows(source, link_name)
            elif system == "Darwin":
                ret = System._reflink_darwin(source, link_name)
            elif system == "Linux":
                ret = System._reflink_linux(source, link_name)
            else:
                ret = -1
        except IOError:
            ret = -1

        if ret != 0:
            raise DvcException("reflink is not supported")
github iterative / dvc / dvc / dagascii.py View on Github external
else:
                        line.append(" ")
                assert len(line) == (smaxcol + 1)
                screen.print_at("".join(line), 0, y)

            screen.refresh()

            # NOTE: get_event() doesn't block by itself,
            # so we have to do the blocking ourselves.
            #
            # NOTE: using this workaround while waiting for PR [1]
            # to get merged and released. After that need to adjust
            # asciimatics version requirements.
            #
            # [1] https://github.com/peterbrittain/asciimatics/pull/188
            System.wait_for_input(self.TIMEOUT)

            event = screen.get_event()
            if not isinstance(event, KeyboardEvent):
                continue

            k = event.key_code
            if k == screen.KEY_DOWN or k == ord("s"):
                offset_y += 1
            elif k == screen.KEY_PAGE_DOWN or k == ord("S"):
                offset_y += smaxrow
            elif k == screen.KEY_UP or k == ord("w"):
                offset_y -= 1
            elif k == screen.KEY_PAGE_UP or k == ord("W"):
                offset_y -= smaxrow
            elif k == screen.KEY_RIGHT or k == ord("d"):
                offset_x += 1
github iterative / dvc / dvc / system.py View on Github external
def inode(path):
        path = fspath(path)

        if System.is_unix():
            import ctypes

            inode = os.lstat(path).st_ino
            # NOTE: See https://bugs.python.org/issue29619 and
            # https://stackoverflow.com/questions/34643289/
            # pythons-os-stat-is-returning-wrong-inode-value
            inode = ctypes.c_ulong(inode).value
        else:
            # getdirinfo from ntfsutils works on both files and dirs
            info = System._getdirinfo(path)
            inode = abs(
                hash(
                    (
                        info.dwVolumeSerialNumber,
                        info.nFileIndexHigh,
                        info.nFileIndexLow,
                    )
                )
            )
        assert inode >= 0
        assert inode < 2 ** 64
        return inode
github iterative / dvc / dvc / remote / local.py View on Github external
def copy(self, from_info, to_info):
        tmp_info = to_info.parent / tmp_fname(to_info.name)
        try:
            System.copy(from_info, tmp_info)
            os.rename(fspath_py35(tmp_info), fspath_py35(to_info))
        except Exception:
            self.remove(tmp_info)
            raise