dvc/tests/unit/remote/test_base.py

import mock
from unittest import TestCase

from dvc.remote import REMOTES, RemoteLOCAL, RemoteSSH, RemoteHDFS
from dvc.remote.base import RemoteBase, RemoteCmdError, RemoteMissingDepsError


class TestMissingDeps(TestCase):
    def test(self):
        remotes = REMOTES + [RemoteLOCAL, RemoteBase]
        for remote_class in remotes:
            REQUIRES = {"foo": None, "bar": None, "mock": mock}
            with mock.patch.object(remote_class, "REQUIRES", REQUIRES):
                with self.assertRaises(RemoteMissingDepsError):
                    remote_class(None, {})


class TestCmdError(TestCase):
    def test(self):
        for remote_class in [RemoteSSH, RemoteHDFS]:
            repo = None
            config = {}

            remote_name = remote_class.scheme
            cmd = "sed 'hello'"
            ret = "1"
            err = "sed: expression #1, char 2: extra characters after command"

            with mock.patch.object(
                remote_class,
                "remove",
                side_effect=RemoteCmdError(remote_name, cmd, ret, err),
            ):
                with self.assertRaises(RemoteCmdError):
                    remote_class(repo, config).remove("file")


class TestCacheExists(TestCase):
    def test(self):
        for remote_class in REMOTES:
            config = {
                "url": remote_class.scheme + "://example/prefix",
                "connection_string": "1234567",
            }
            remote = remote_class(None, config)

            with mock.patch.object(remote, "changed_cache", return_value=True):
                with mock.patch.object(remote, "copy") as cp:
                    remote.save(
                        remote.path_info, {remote.PARAM_CHECKSUM: "1234567890"}
                    )
                    cp.assert_called_once()

            with mock.patch.object(
                remote, "changed_cache", return_value=False
            ):
                with mock.patch.object(remote, "copy") as cp:
                    remote.save(
                        remote.path_info, {remote.PARAM_CHECKSUM: "1234567890"}
                    )
                    cp.assert_not_called()