dvc/tests/func/test_data_cloud.py

import copy
import logging
import os
import shutil
import uuid
from unittest import SkipTest

import pytest
from flaky.flaky_decorator import flaky

from dvc.cache import NamedCache
from dvc.data_cloud import DataCloud
from dvc.external_repo import clean_repos
from dvc.main import main
from dvc.remote.azure import AzureRemoteTree
from dvc.remote.base import STATUS_DELETED, STATUS_NEW, STATUS_OK
from dvc.remote.gdrive import GDriveRemoteTree
from dvc.remote.gs import GSRemoteTree
from dvc.remote.hdfs import HDFSRemoteTree
from dvc.remote.http import HTTPRemoteTree
from dvc.remote.local import LocalRemoteTree
from dvc.remote.oss import OSSRemoteTree
from dvc.remote.s3 import S3RemoteTree
from dvc.remote.ssh import SSHRemoteTree
from dvc.stage.exceptions import StageNotFound
from dvc.utils import file_md5
from dvc.utils.fs import remove
from dvc.utils.yaml import dump_yaml, load_yaml
from tests.basic_env import TestDvc
from tests.remotes import (
    GCP,
    HDFS,
    HTTP,
    OSS,
    S3,
    TEST_CONFIG,
    TEST_GCP_CREDS_FILE,
    TEST_REMOTE,
    Azure,
    GDrive,
    Local,
    SSHMocked,
)


class TestDataCloud(TestDvc):
    def _test_cloud(self, config, cl):
        self.dvc.config = config
        cloud = DataCloud(self.dvc)
        self.assertIsInstance(cloud.get_remote().tree, cl)

    def test(self):
        config = copy.deepcopy(TEST_CONFIG)

        clist = [
            ("s3://mybucket/", S3RemoteTree),
            ("gs://mybucket/", GSRemoteTree),
            ("ssh://user@localhost:/", SSHRemoteTree),
            ("http://localhost:8000/", HTTPRemoteTree),
            ("azure://ContainerName=mybucket;conn_string;", AzureRemoteTree),
            ("oss://mybucket/", OSSRemoteTree),
            (TestDvc.mkdtemp(), LocalRemoteTree),
        ]

        for scheme, cl in clist:
            remote_url = scheme + str(uuid.uuid4())
            config["remote"][TEST_REMOTE]["url"] = remote_url
            self._test_cloud(config, cl)


class TestDataCloudBase(TestDvc):
    def _get_cloud_class(self):
        return None

    @staticmethod
    def should_test():
        return False

    @staticmethod
    def get_url():
        return NotImplementedError

    def _get_keyfile(self):
        return None

    def _ensure_should_run(self):
        if not self.should_test():
            raise SkipTest(f"Test {self.__class__.__name__} is disabled")

    def _setup_cloud(self):
        self._ensure_should_run()

        repo = self.get_url()
        keyfile = self._get_keyfile()

        config = copy.deepcopy(TEST_CONFIG)
        config["remote"][TEST_REMOTE] = {"url": repo, "keyfile": keyfile}
        self.dvc.config = config
        self.cloud = DataCloud(self.dvc)

        self.assertIsInstance(
            self.cloud.get_remote().tree, self._get_cloud_class()
        )

    def _test_cloud(self):
        self._setup_cloud()

        stages = self.dvc.add(self.FOO)
        self.assertEqual(len(stages), 1)
        stage = stages[0]
        self.assertTrue(stage is not None)
        out = stage.outs[0]
        cache = out.cache_path
        md5 = out.checksum
        info = out.get_used_cache()

        stages = self.dvc.add(self.DATA_DIR)
        self.assertEqual(len(stages), 1)
        stage_dir = stages[0]
        self.assertTrue(stage_dir is not None)
        out_dir = stage_dir.outs[0]
        cache_dir = out_dir.cache_path
        name_dir = str(out_dir)
        md5_dir = out_dir.checksum
        info_dir = NamedCache.make(out_dir.scheme, md5_dir, name_dir)

        with self.cloud.repo.state:
            # Check status
            status = self.cloud.status(info, show_checksums=True)
            expected = {md5: {"name": md5, "status": STATUS_NEW}}
            self.assertEqual(status, expected)

            status_dir = self.cloud.status(info_dir, show_checksums=True)
            expected = {md5_dir: {"name": md5_dir, "status": STATUS_NEW}}
            self.assertEqual(status_dir, expected)

            # Push and check status
            self.cloud.push(info)
            self.assertTrue(os.path.exists(cache))
            self.assertTrue(os.path.isfile(cache))

            self.cloud.push(info_dir)
            self.assertTrue(os.path.isfile(cache_dir))

            status = self.cloud.status(info, show_checksums=True)
            expected = {md5: {"name": md5, "status": STATUS_OK}}
            self.assertEqual(status, expected)

            status_dir = self.cloud.status(info_dir, show_checksums=True)
            expected = {md5_dir: {"name": md5_dir, "status": STATUS_OK}}
            self.assertEqual(status_dir, expected)

            # Remove and check status
            remove(self.dvc.cache.local.cache_dir)

            status = self.cloud.status(info, show_checksums=True)
            expected = {md5: {"name": md5, "status": STATUS_DELETED}}
            self.assertEqual(status, expected)

            status_dir = self.cloud.status(info_dir, show_checksums=True)
            expected = {md5_dir: {"name": md5_dir, "status": STATUS_DELETED}}
            self.assertEqual(status_dir, expected)

            # Pull and check status
            self.cloud.pull(info)
            self.assertTrue(os.path.exists(cache))
            self.assertTrue(os.path.isfile(cache))
            with open(cache) as fd:
                self.assertEqual(fd.read(), self.FOO_CONTENTS)

            self.cloud.pull(info_dir)
            self.assertTrue(os.path.isfile(cache_dir))

            status = self.cloud.status(info, show_checksums=True)
            expected = {md5: {"name": md5, "status": STATUS_OK}}
            self.assertEqual(status, expected)

            status_dir = self.cloud.status(info_dir, show_checksums=True)
            expected = {md5_dir: {"name": md5_dir, "status": STATUS_OK}}
            self.assertTrue(status_dir, expected)

    def test(self):
        self._ensure_should_run()
        self._test_cloud()


class TestS3Remote(S3, TestDataCloudBase):
    def _get_cloud_class(self):
        return S3RemoteTree


class TestGDriveRemote(GDrive, TestDataCloudBase):
    def _setup_cloud(self):
        self._ensure_should_run()

        url = self.get_url()
        self.create_dir(self.dvc, url)

        config = copy.deepcopy(TEST_CONFIG)
        config["remote"][TEST_REMOTE] = {
            "url": url,
            "gdrive_service_account_email": "test",
            "gdrive_service_account_p12_file_path": "test.p12",
            "gdrive_use_service_account": True,
        }

        self.dvc.config = config
        self.cloud = DataCloud(self.dvc)
        remote = self.cloud.get_remote()
        self.assertIsInstance(remote.tree, self._get_cloud_class())

    def _get_cloud_class(self):
        return GDriveRemoteTree


class TestGSRemote(GCP, TestDataCloudBase):
    def _setup_cloud(self):
        self._ensure_should_run()

        repo = self.get_url()

        config = copy.deepcopy(TEST_CONFIG)
        config["remote"][TEST_REMOTE] = {
            "url": repo,
            "credentialpath": TEST_GCP_CREDS_FILE,
        }
        self.dvc.config = config
        self.cloud = DataCloud(self.dvc)

        self.assertIsInstance(
            self.cloud.get_remote().tree, self._get_cloud_class()
        )

    def _get_cloud_class(self):
        return GSRemoteTree


class TestAzureRemote(Azure, TestDataCloudBase):
    def _get_cloud_class(self):
        return AzureRemoteTree


class TestOSSRemote(OSS, TestDataCloudBase):
    def _get_cloud_class(self):
        return OSSRemoteTree


class TestLocalRemote(Local, TestDataCloudBase):
    def _get_cloud_class(self):
        return LocalRemoteTree


@pytest.mark.usefixtures("ssh_server")
class TestSSHRemoteMocked(SSHMocked, TestDataCloudBase):
    @pytest.fixture(autouse=True)
    def setup_method_fixture(self, request, ssh_server):
        self.ssh_server = ssh_server
        self.method_name = request.function.__name__

    def _setup_cloud(self):
        self._ensure_should_run()

        repo = self.get_url()
        keyfile = self._get_keyfile()

        self._get_cloud_class().CAN_TRAVERSE = False
        config = copy.deepcopy(TEST_CONFIG)
        config["remote"][TEST_REMOTE] = {
            "url": repo,
            "keyfile": keyfile,
        }
        self.dvc.config = config
        self.cloud = DataCloud(self.dvc)

        self.assertIsInstance(
            self.cloud.get_remote().tree, self._get_cloud_class()
        )

    def get_url(self):
        user = self.ssh_server.test_creds["username"]
        return super().get_url(user, self.ssh_server.port)

    def _get_keyfile(self):
        return self.ssh_server.test_creds["key_filename"]

    def _get_cloud_class(self):
        return SSHRemoteTree


class TestHDFSRemote(HDFS, TestDataCloudBase):
    def _get_cloud_class(self):
        return HDFSRemoteTree


@pytest.mark.usefixtures("http_server")
class TestHTTPRemote(HTTP, TestDataCloudBase):
    @pytest.fixture(autouse=True)
    def setup_method_fixture(self, request, http_server):
        self.http_server = http_server
        self.method_name = request.function.__name__

    def get_url(self):
        return super().get_url(self.http_server.server_port)

    def _get_cloud_class(self):
        return HTTPRemoteTree


class TestDataCloudCLIBase(TestDvc):
    def main(self, args):
        ret = main(args)
        self.assertEqual(ret, 0)

    @staticmethod
    def should_test():
        return False

    @staticmethod
    def get_url():
        raise NotImplementedError

    def _setup_cloud(self):
        pass

    def _test_cloud(self, remote=None):
        self._setup_cloud()

        args = ["-v", "-j", "2"]
        if remote:
            args += ["-r", remote]
        else:
            args += []

        stages = self.dvc.add(self.FOO)
        self.assertEqual(len(stages), 1)
        stage = stages[0]
        self.assertTrue(stage is not None)
        cache = stage.outs[0].cache_path

        stages = self.dvc.add(self.DATA_DIR)
        self.assertEqual(len(stages), 1)
        stage_dir = stages[0]
        self.assertTrue(stage_dir is not None)
        cache_dir = stage_dir.outs[0].cache_path

        # FIXME check status output

        self.main(["push"] + args)
        self.assertTrue(os.path.exists(cache))
        self.assertTrue(os.path.isfile(cache))
        self.assertTrue(os.path.isfile(cache_dir))

        remove(self.dvc.cache.local.cache_dir)

        self.main(["fetch"] + args)
        self.assertTrue(os.path.exists(cache))
        self.assertTrue(os.path.isfile(cache))
        self.assertTrue(os.path.isfile(cache_dir))

        self.main(["pull"] + args)
        self.assertTrue(os.path.exists(cache))
        self.assertTrue(os.path.isfile(cache))
        self.assertTrue(os.path.isfile(cache_dir))
        self.assertTrue(os.path.isfile(self.FOO))
        self.assertTrue(os.path.isdir(self.DATA_DIR))

        with open(cache) as fd:
            self.assertEqual(fd.read(), self.FOO_CONTENTS)
        self.assertTrue(os.path.isfile(cache_dir))

        # NOTE: check if remote gc works correctly on directories
        self.main(["gc", "-cw", "-f"] + args)
        shutil.move(
            self.dvc.cache.local.cache_dir,
            self.dvc.cache.local.cache_dir + ".back",
        )

        self.main(["fetch"] + args)

        self.main(["pull", "-f"] + args)
        self.assertTrue(os.path.exists(cache))
        self.assertTrue(os.path.isfile(cache))
        self.assertTrue(os.path.isfile(cache_dir))
        self.assertTrue(os.path.isfile(self.FOO))
        self.assertTrue(os.path.isdir(self.DATA_DIR))

    def _test(self):
        pass

    def test(self):
        if not self.should_test():
            raise SkipTest(f"Test {self.__class__.__name__} is disabled")
        self._test()


class TestLocalRemoteCLI(Local, TestDataCloudCLIBase):
    def _test(self):
        url = self.get_url()

        self.main(["remote", "add", TEST_REMOTE, url])

        self._test_cloud(TEST_REMOTE)


class TestRemoteHDFSCLI(HDFS, TestDataCloudCLIBase):
    def _test(self):
        url = self.get_url()

        self.main(["remote", "add", TEST_REMOTE, url])

        self._test_cloud(TEST_REMOTE)


class TestS3RemoteCLI(S3, TestDataCloudCLIBase):
    def _test(self):
        url = self.get_url()

        self.main(["remote", "add", TEST_REMOTE, url])

        self._test_cloud(TEST_REMOTE)


class TestGDriveRemoteCLI(GDrive, TestDataCloudCLIBase):
    def _test(self):
        url = self.get_url()

        self.create_dir(self.dvc, url)

        self.main(["remote", "add", TEST_REMOTE, url])
        self.main(
            [
                "remote",
                "modify",
                TEST_REMOTE,
                "gdrive_service_account_email",
                "test",
            ]
        )
        self.main(
            [
                "remote",
                "modify",
                TEST_REMOTE,
                "gdrive_service_account_p12_file_path",
                "test.p12",
            ]
        )
        self.main(
            [
                "remote",
                "modify",
                TEST_REMOTE,
                "gdrive_use_service_account",
                "True",
            ]
        )

        self._test_cloud(TEST_REMOTE)


class TestGSRemoteCLI(GCP, TestDataCloudCLIBase):
    def _test(self):
        url = self.get_url()

        self.main(["remote", "add", TEST_REMOTE, url])
        self.main(
            [
                "remote",
                "modify",
                TEST_REMOTE,
                "credentialpath",
                TEST_GCP_CREDS_FILE,
            ]
        )

        self._test_cloud(TEST_REMOTE)


class TestAzureRemoteCLI(Azure, TestDataCloudCLIBase):
    def _test(self):
        url = self.get_url()

        self.main(["remote", "add", TEST_REMOTE, url])

        self._test_cloud(TEST_REMOTE)


class TestOSSRemoteCLI(OSS, TestDataCloudCLIBase):
    def _test(self):
        url = self.get_url()

        self.main(["remote", "add", TEST_REMOTE, url])

        self._test_cloud(TEST_REMOTE)


class TestDataCloudErrorCLI(TestDvc):
    def main_fail(self, args):
        ret = main(args)
        self.assertNotEqual(ret, 0)

    def test_error(self):
        f = "non-existing-file"
        self.main_fail(["status", "-c", f])
        self.main_fail(["push", f])
        self.main_fail(["pull", f])
        self.main_fail(["fetch", f])


class TestWarnOnOutdatedStage(TestDvc):
    def main(self, args):
        ret = main(args)
        self.assertEqual(ret, 0)

    def _test(self):
        url = Local.get_url()
        self.main(["remote", "add", "-d", TEST_REMOTE, url])

        stage = self.dvc.run(
            outs=["bar"], cmd="echo bar > bar", single_stage=True
        )
        self.main(["push"])

        stage_file_path = stage.relpath
        content = load_yaml(stage_file_path)
        del content["outs"][0]["md5"]
        dump_yaml(stage_file_path, content)

        with self._caplog.at_level(logging.WARNING, logger="dvc"):
            self._caplog.clear()
            self.main(["status", "-c"])
            expected_warning = (
                "Output 'bar'(stage: 'bar.dvc') is missing version info. "
                "Cache for it will not be collected. "
                "Use `dvc repro` to get your pipeline up to date."
            )

            assert expected_warning in self._caplog.text

    def test(self):
        self._test()


class TestRecursiveSyncOperations(Local, TestDataCloudBase):
    def main(self, args):
        ret = main(args)
        self.assertEqual(ret, 0)

    def _get_cloud_class(self):
        return LocalRemoteTree

    def _prepare_repo(self):
        remote = self.cloud.get_remote()
        self.main(
            ["remote", "add", "-d", TEST_REMOTE, remote.path_info.fspath]
        )

        self.dvc.add(self.DATA)
        self.dvc.add(self.DATA_SUB)

    def _remove_local_data_files(self):
        os.remove(self.DATA)
        os.remove(self.DATA_SUB)

    def _test_recursive_pull(self):
        self._remove_local_data_files()
        self._clear_local_cache()

        self.assertFalse(os.path.exists(self.DATA))
        self.assertFalse(os.path.exists(self.DATA_SUB))

        self.main(["pull", "-R", self.DATA_DIR])

        self.assertTrue(os.path.exists(self.DATA))
        self.assertTrue(os.path.exists(self.DATA_SUB))

    def _clear_local_cache(self):
        remove(self.dvc.cache.local.cache_dir)

    def _test_recursive_fetch(self, data_md5, data_sub_md5):
        self._clear_local_cache()

        local_cache = self.dvc.cache.local
        local_cache_data_path = local_cache.hash_to_path_info(data_md5)
        local_cache_data_sub_path = local_cache.hash_to_path_info(data_sub_md5)

        self.assertFalse(os.path.exists(local_cache_data_path))
        self.assertFalse(os.path.exists(local_cache_data_sub_path))

        self.main(["fetch", "-R", self.DATA_DIR])

        self.assertTrue(os.path.exists(local_cache_data_path))
        self.assertTrue(os.path.exists(local_cache_data_sub_path))

    def _test_recursive_push(self, data_md5, data_sub_md5):
        remote = self.cloud.get_remote()
        cloud_data_path = remote.hash_to_path_info(data_md5)
        cloud_data_sub_path = remote.hash_to_path_info(data_sub_md5)

        self.assertFalse(os.path.exists(cloud_data_path))
        self.assertFalse(os.path.exists(cloud_data_sub_path))

        self.main(["push", "-R", self.DATA_DIR])

        self.assertTrue(os.path.exists(cloud_data_path))
        self.assertTrue(os.path.exists(cloud_data_sub_path))

    def test(self):
        self._setup_cloud()
        self._prepare_repo()

        data_md5 = file_md5(self.DATA)[0]
        data_sub_md5 = file_md5(self.DATA_SUB)[0]

        self._test_recursive_push(data_md5, data_sub_md5)

        self._test_recursive_fetch(data_md5, data_sub_md5)

        self._test_recursive_pull()


def test_hash_recalculation(mocker, dvc, tmp_dir):
    tmp_dir.gen({"foo": "foo"})
    test_get_file_hash = mocker.spy(LocalRemoteTree, "get_file_hash")
    url = Local.get_url()
    ret = main(["remote", "add", "-d", TEST_REMOTE, url])
    assert ret == 0
    ret = main(["config", "cache.type", "hardlink"])
    assert ret == 0
    ret = main(["add", "foo"])
    assert ret == 0
    ret = main(["push"])
    assert ret == 0
    ret = main(["run", "--single-stage", "-d", "foo", "echo foo"])
    assert ret == 0
    assert test_get_file_hash.mock.call_count == 1


class TestShouldWarnOnNoChecksumInLocalAndRemoteCache(TestDvc):
    def setUp(self):
        super().setUp()

        cache_dir = self.mkdtemp()
        ret = main(["add", self.FOO])
        self.assertEqual(0, ret)

        ret = main(["add", self.BAR])
        self.assertEqual(0, ret)

        # purge cache
        remove(self.dvc.cache.local.cache_dir)

        ret = main(["remote", "add", "remote_name", "-d", cache_dir])
        self.assertEqual(0, ret)

        checksum_foo = file_md5(self.FOO)[0]
        checksum_bar = file_md5(self.BAR)[0]
        self.message_header = (
            "Some of the cache files do not exist neither locally "
            "nor on remote. Missing cache files: "
        )
        self.message_bar_part = "name: {}, md5: {}".format(
            self.BAR, checksum_bar
        )
        self.message_foo_part = "name: {}, md5: {}".format(
            self.FOO, checksum_foo
        )

    def test(self):
        self._caplog.clear()
        main(["push"])
        assert self.message_header in self._caplog.text
        assert self.message_foo_part in self._caplog.text
        assert self.message_bar_part in self._caplog.text

        self._caplog.clear()
        main(["pull"])
        assert self.message_header in self._caplog.text
        assert self.message_foo_part in self._caplog.text
        assert self.message_bar_part in self._caplog.text

        self._caplog.clear()
        main(["status", "-c"])
        assert self.message_header in self._caplog.text
        assert self.message_foo_part in self._caplog.text
        assert self.message_bar_part in self._caplog.text


def test_verify_hashes(
    tmp_dir, scm, dvc, mocker, tmp_path_factory, setup_remote
):

    setup_remote(dvc, name="upstream")
    tmp_dir.dvc_gen({"file": "file1 content"}, commit="add file")
    tmp_dir.dvc_gen({"dir": {"subfile": "file2 content"}}, commit="add dir")
    dvc.push()

    # remove artifacts and cache to trigger fetching
    remove("file")
    remove("dir")
    remove(dvc.cache.local.cache_dir)

    hash_spy = mocker.spy(dvc.cache.local.tree, "get_file_hash")

    dvc.pull()
    assert hash_spy.call_count == 0

    # Removing cache will invalidate existing state entries
    remove(dvc.cache.local.cache_dir)

    dvc.config["remote"]["upstream"]["verify"] = True

    dvc.pull()
    assert hash_spy.call_count == 3


@flaky(max_runs=3, min_passes=1)
@pytest.mark.parametrize("erepo", ["git_dir", "erepo_dir"])
def test_pull_git_imports(request, tmp_dir, dvc, scm, erepo):
    erepo = request.getfixturevalue(erepo)
    with erepo.chdir():
        erepo.scm_gen({"dir": {"bar": "bar"}}, commit="second")
        erepo.scm_gen("foo", "foo", commit="first")

    dvc.imp(os.fspath(erepo), "foo")
    dvc.imp(os.fspath(erepo), "dir", out="new_dir", rev="HEAD~")

    assert dvc.pull()["fetched"] == 0

    for item in ["foo", "new_dir", dvc.cache.local.cache_dir]:
        remove(item)
    os.makedirs(dvc.cache.local.cache_dir, exist_ok=True)
    clean_repos()

    assert dvc.pull(force=True)["fetched"] == 2

    assert (tmp_dir / "foo").exists()
    assert (tmp_dir / "foo").read_text() == "foo"

    assert (tmp_dir / "new_dir").exists()
    assert (tmp_dir / "new_dir" / "bar").read_text() == "bar"


def test_pull_external_dvc_imports(tmp_dir, dvc, scm, erepo_dir):
    with erepo_dir.chdir():
        erepo_dir.dvc_gen({"dir": {"bar": "bar"}}, commit="second")
        erepo_dir.dvc_gen("foo", "foo", commit="first")

        os.remove("foo")
        shutil.rmtree("dir")

    dvc.imp(os.fspath(erepo_dir), "foo")
    dvc.imp(os.fspath(erepo_dir), "dir", out="new_dir", rev="HEAD~")

    assert dvc.pull()["fetched"] == 0

    clean(["foo", "new_dir"], dvc)

    assert dvc.pull(force=True)["fetched"] == 2

    assert (tmp_dir / "foo").exists()
    assert (tmp_dir / "foo").read_text() == "foo"

    assert (tmp_dir / "new_dir").exists()
    assert (tmp_dir / "new_dir" / "bar").read_text() == "bar"


def clean(outs, dvc=None):
    if dvc:
        outs = outs + [dvc.cache.local.cache_dir]
    for path in outs:
        print(path)
        remove(path)
    if dvc:
        os.makedirs(dvc.cache.local.cache_dir, exist_ok=True)
        clean_repos()


def recurse_list_dir(d):
    return [
        os.path.join(d, f) for _, _, filenames in os.walk(d) for f in filenames
    ]


def test_dvc_pull_pipeline_stages(tmp_dir, dvc, run_copy, setup_remote):
    setup_remote(dvc)
    (stage0,) = tmp_dir.dvc_gen("foo", "foo")
    stage1 = run_copy("foo", "bar", single_stage=True)
    stage2 = run_copy("bar", "foobar", name="copy-bar-foobar")
    dvc.push()

    outs = ["foo", "bar", "foobar"]

    clean(outs, dvc)
    dvc.pull()
    assert all((tmp_dir / file).exists() for file in outs)

    for out, stage in zip(outs, [stage0, stage1, stage2]):
        for target in [stage.addressing, out]:
            clean(outs, dvc)
            stats = dvc.pull([target])
            assert stats["fetched"] == 1
            assert stats["added"] == [out]
            assert os.path.exists(out)
            assert not any(os.path.exists(out) for out in set(outs) - {out})

    clean(outs, dvc)
    stats = dvc.pull([stage2.addressing], with_deps=True)
    assert len(stats["added"]) == 3
    assert set(stats["added"]) == set(outs)

    clean(outs, dvc)
    stats = dvc.pull([os.curdir], recursive=True)
    assert set(stats["added"]) == set(outs)


def test_pipeline_file_target_ops(tmp_dir, dvc, run_copy, setup_remote):
    remote_path = setup_remote(dvc)
    tmp_dir.dvc_gen("foo", "foo")
    run_copy("foo", "bar", single_stage=True)

    tmp_dir.dvc_gen("lorem", "lorem")
    run_copy("lorem", "lorem2", name="copy-lorem-lorem2")

    tmp_dir.dvc_gen("ipsum", "ipsum")
    run_copy("ipsum", "baz", name="copy-ipsum-baz")

    outs = ["foo", "bar", "lorem", "ipsum", "baz", "lorem2"]

    remove(dvc.stage_cache.cache_dir)

    dvc.push()

    outs = ["foo", "bar", "lorem", "ipsum", "baz", "lorem2"]

    # each one's a copy of other, hence 3
    assert len(recurse_list_dir(remote_path)) == 3

    clean(outs, dvc)
    assert set(dvc.pull(["dvc.yaml"])["added"]) == {"lorem2", "baz"}

    clean(outs, dvc)
    assert set(dvc.pull()["added"]) == set(outs)

    # clean everything in remote and push
    from tests.dir_helpers import TmpDir

    clean(TmpDir(remote_path).iterdir())
    dvc.push(["dvc.yaml:copy-ipsum-baz"])
    assert len(recurse_list_dir(remote_path)) == 1

    clean(TmpDir(remote_path).iterdir())
    dvc.push(["dvc.yaml"])
    assert len(recurse_list_dir(remote_path)) == 2

    with pytest.raises(StageNotFound):
        dvc.push(["dvc.yaml:StageThatDoesNotExist"])

    with pytest.raises(StageNotFound):
        dvc.pull(["dvc.yaml:StageThatDoesNotExist"])


@pytest.mark.parametrize(
    "fs, msg",
    [
        ({"foo": "foo", "bar": "bar"}, "2 files pushed"),
        ({"foo": "foo"}, "1 file pushed"),
        ({}, "Everything is up to date"),
    ],
)
def test_push_stats(tmp_dir, dvc, fs, msg, caplog, setup_remote):
    setup_remote(dvc)
    tmp_dir.dvc_gen(fs)

    caplog.clear()
    with caplog.at_level(level=logging.INFO, logger="dvc"):
        main(["push"])
    assert msg in caplog.text


@pytest.mark.parametrize(
    "fs, msg",
    [
        ({"foo": "foo", "bar": "bar"}, "2 files fetched"),
        ({"foo": "foo"}, "1 file fetched"),
        ({}, "Everything is up to date."),
    ],
)
def test_fetch_stats(tmp_dir, dvc, fs, msg, caplog, setup_remote):
    setup_remote(dvc)
    tmp_dir.dvc_gen(fs)
    dvc.push()
    clean(list(fs.keys()), dvc)
    caplog.clear()
    with caplog.at_level(level=logging.INFO, logger="dvc"):
        main(["fetch"])
    assert msg in caplog.text


def test_pull_stats(tmp_dir, dvc, caplog, setup_remote):
    setup_remote(dvc)
    tmp_dir.dvc_gen({"foo": "foo", "bar": "bar"})
    dvc.push()
    clean(["foo", "bar"], dvc)
    (tmp_dir / "bar").write_text("foobar")
    caplog.clear()
    with caplog.at_level(level=logging.INFO, logger="dvc"):
        main(["pull", "--force"])
    assert "M\tbar" in caplog.text
    assert "A\tfoo" in caplog.text
    assert "2 files fetched" in caplog.text
    assert "1 file added" in caplog.text
    assert "1 file modified" in caplog.text
    with caplog.at_level(level=logging.INFO, logger="dvc"):
        main(["pull"])
    assert "Everything is up to date." in caplog.text


@pytest.mark.parametrize(
    "key,expected", [("all_tags", 2), ("all_branches", 3), ("all_commits", 3)]
)
def test_push_pull_all(tmp_dir, scm, dvc, setup_remote, key, expected):
    setup_remote(dvc)
    tmp_dir.dvc_gen({"foo": "foo"}, commit="first")
    scm.tag("v1")
    dvc.remove("foo.dvc")
    tmp_dir.dvc_gen({"bar": "bar"}, commit="second")
    scm.tag("v2")
    with tmp_dir.branch("branch", new=True):
        dvc.remove("bar.dvc")
        tmp_dir.dvc_gen({"baz": "baz"}, commit="branch")

    assert dvc.push(**{key: True}) == expected

    clean(["foo", "bar", "baz"], dvc)
    assert dvc.pull(**{key: True})["fetched"] == expected


def test_push_pull_fetch_pipeline_stages(tmp_dir, dvc, run_copy, setup_remote):
    remote_path = setup_remote(dvc)
    tmp_dir.dvc_gen("foo", "foo")
    run_copy("foo", "bar", no_commit=True, name="copy-foo-bar")

    dvc.push("copy-foo-bar")
    assert len(recurse_list_dir(remote_path)) == 1
    # pushing everything so as we can check pull/fetch only downloads
    # from specified targets
    dvc.push()
    clean(["foo", "bar"], dvc)

    dvc.pull("copy-foo-bar")
    assert (tmp_dir / "bar").exists()
    assert len(recurse_list_dir(dvc.cache.local.cache_dir)) == 1
    clean(["bar"], dvc)

    dvc.fetch("copy-foo-bar")
    assert len(recurse_list_dir(dvc.cache.local.cache_dir)) == 1