deps: bump dvc-data to >=2.20.0

* daemon: ipc to log pid from the child

* fix tests to use pid

* also test for updater/analytics keyword

dvc/analytics.py

@@ -2,7 +2,7 @@ import json
 import logging
 import os
 
-from .env import DVC_NO_ANALYTICS
+from .env import DVC_ANALYTICS_ENDPOINT, DVC_NO_ANALYTICS
 
 logger = logging.getLogger(__name__)
 
@@ -34,6 +34,10 @@ def collect_and_send_report(args=None, return_code=None):
 
     with tempfile.NamedTemporaryFile(delete=False, mode="w") as fobj:
         json.dump(report, fobj)
+
+    logger.trace(  # type: ignore[attr-defined]
+        "Saving analytics report to %s", fobj.name
+    )
     daemon(["analytics", fobj.name])
 
 
@@ -65,7 +69,7 @@ def send(path):
     """
     import requests
 
-    url = "https://analytics.dvc.org"
+    url = os.environ.get(DVC_ANALYTICS_ENDPOINT, "https://analytics.dvc.org")
     headers = {"content-type": "application/json"}
 
     with open(path, encoding="utf-8") as fobj:
@@ -73,11 +77,16 @@ def send(path):
 
     report.update(_runtime_info())
 
+    logger.debug("uploading report to %s", url)
+    logger.trace("Sending %s to %s", report, url)  # type: ignore[attr-defined]
+
     try:
         requests.post(url, json=report, headers=headers, timeout=5)
-    except requests.exceptions.RequestException:
-        logger.debug("failed to send analytics report", exc_info=True)
+    except requests.exceptions.RequestException as e:
+        logger.trace("", exc_info=True)  # type: ignore[attr-defined]
+        logger.debug("failed to send analytics report %s", str(e))
 
+    logger.trace("removing report %s", path)  # type: ignore[attr-defined]
     os.remove(path)
 
 

dvc/api/experiments.py

@@ -53,7 +53,7 @@ def _postprocess(exp_rows):
                 except ValueError:
                     exp_row[k] = v_str
 
-            if not exp_row[k]:
+            elif not exp_row[k]:
                 exp_row[k] = None
 
     return exp_rows

dvc/cli/__init__.py

@@ -165,18 +165,18 @@ def main(argv=None):  # noqa: C901, PLR0912, PLR0915
     # the copied parent's standard file descriptors. If we make any logging
     # calls in this state it will cause an exception due to writing to a closed
     # file descriptor.
-    if sys.stderr.closed:  # pylint: disable=using-constant-test
+    if not sys.stderr or sys.stderr.closed:  # pylint: disable=using-constant-test
         logging.disable()
-    elif sys.stdout.closed:  # pylint: disable=using-constant-test
+    elif not sys.stdout or sys.stdout.closed:  # pylint: disable=using-constant-test
         logging.disable(logging.INFO)
 
     args = None
 
     outer_log_level = logger.level
+    level = None
     try:
         args = parse_args(argv)
 
-        level = None
         if args.quiet:
             level = logging.CRITICAL
         elif args.verbose == 1:
@@ -199,7 +199,7 @@ def main(argv=None):  # noqa: C901, PLR0912, PLR0915
 
         logger.trace(args)  # type: ignore[attr-defined]
 
-        if not sys.stdout.closed and not args.quiet:
+        if sys.stdout and not sys.stdout.closed and not args.quiet:
             from dvc.ui import ui
 
             ui.enable()
@@ -235,11 +235,17 @@ def main(argv=None):  # noqa: C901, PLR0912, PLR0915
         ret = _log_exceptions(exc) or 255
 
     try:
+        import os
+
         from dvc import analytics
 
         if analytics.is_enabled():
             analytics.collect_and_send_report(args, ret)
 
+        logger.trace(  # type: ignore[attr-defined]
+            "Process %s exiting with %s", os.getpid(), ret
+        )
+
         return ret
     finally:
         logger.setLevel(outer_log_level)

dvc/cli/utils.py

@@ -2,6 +2,10 @@ import argparse
 
 
 class DictAction(argparse.Action):
+    def __init__(self, *args, **kwargs):
+        kwargs.setdefault("metavar", "<name>=<value>")
+        super().__init__(*args, **kwargs)
+
     def __call__(self, parser, args, values, option_string=None):  # noqa: ARG002
         d = getattr(args, self.dest) or {}
 

dvc/commands/daemon.py

@@ -1,7 +1,11 @@
+import logging
+
 from dvc.cli import completion
 from dvc.cli.command import CmdBaseNoRepo
 from dvc.cli.utils import fix_subparsers
 
+logger = logging.getLogger(__name__)
+
 
 class CmdDaemonBase(CmdBaseNoRepo):
     pass
@@ -21,6 +25,8 @@ class CmdDaemonUpdater(CmdDaemonBase):
         config = Config(dvc_dir, validate=False)
         hardlink_lock = config.get("core", {}).get("hardlink_lock", False)
         updater = Updater(tmp_dir, hardlink_lock=hardlink_lock)
+
+        logger.info("Starting updater to fetch the latest version")
         updater.fetch(detach=False)
 
         return 0
@@ -30,6 +36,7 @@ class CmdDaemonAnalytics(CmdDaemonBase):
     def run(self):
         from dvc import analytics
 
+        logger.info("Sending analytics")
         analytics.send(self.args.target)
 
         return 0

dvc/commands/experiments/apply.py

@@ -4,13 +4,21 @@ import logging
 from dvc.cli import completion
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
+from dvc.ui import ui
 
 logger = logging.getLogger(__name__)
 
 
 class CmdExperimentsApply(CmdBase):
     def run(self):
-        self.repo.experiments.apply(self.args.experiment, force=self.args.force)
+        if not self.args.force:
+            ui.write(
+                "The --no-force option is deprecated and will be removed in a future"
+                " DVC release. To revert the result of 'exp apply', run:\n"
+                "\n\tgit reset --hard\n"
+                "\tgit stash apply refs/exps/apply/stash\n"
+            )
+        self.repo.experiments.apply(self.args.experiment)
 
         return 0
 
@@ -25,11 +33,10 @@ def add_parser(experiments_subparsers, parent_parser):
         formatter_class=argparse.RawDescriptionHelpFormatter,
     )
     experiments_apply_parser.add_argument(
-        "-f",
-        "--force",
-        action="store_true",
-        default=False,
-        help="Do not prompt when removing working directory files.",
+        "--no-force",
+        action="store_false",
+        dest="force",
+        help="Fail if this command would overwrite conflicting changes.",
     )
     experiments_apply_parser.add_argument(
         "experiment", help="Experiment to be applied."

dvc/config_schema.py

@@ -256,6 +256,9 @@ SCHEMA = {
                     "exclude_shared_token_cache_credential": Bool,
                     "exclude_managed_identity_credential": Bool,
                     Optional("verify", default=False): Bool,
+                    "timeout": int,
+                    "read_timeout": int,
+                    "connection_timeout": int,
                     **REMOTE_COMMON,
                 },
                 "oss": {

dvc/daemon.py

@@ -3,66 +3,67 @@
 import inspect
 import logging
 import os
+import subprocess  # nosec B404
 import sys
-from subprocess import Popen  # nosec B404
-from typing import List
+from contextlib import nullcontext
+from typing import TYPE_CHECKING, Any, List, Mapping, Optional, Sequence, Union
 
-from dvc.env import DVC_DAEMON
+if TYPE_CHECKING:
+    from typing import ContextManager
+
+from dvc.env import DVC_DAEMON, DVC_DAEMON_LOGFILE
 from dvc.utils import fix_env, is_binary
+from dvc.utils.collections import ensure_list
 
 logger = logging.getLogger(__name__)
 
 
-def _suppress_resource_warning(popen: Popen):
+def _suppress_resource_warning(popen: subprocess.Popen) -> None:
     """Sets the returncode to avoid ResourceWarning when popen is garbage collected."""
     # only use for daemon processes.
     # See https://bugs.python.org/issue38890.
     popen.returncode = 0
 
 
-def _popen(cmd, **kwargs) -> Popen:
-    prefix = [sys.executable]
-    if not is_binary():
-        main_entrypoint = os.path.join(
-            os.path.abspath(os.path.dirname(__file__)), "__main__.py"
-        )
-        prefix += [main_entrypoint]
-    return Popen(
-        prefix + cmd, close_fds=True, shell=False, **kwargs  # nosec B603  # noqa: S603
+def _win_detached_subprocess(args: Sequence[str], **kwargs) -> int:
+    assert os.name == "nt"
+
+    from subprocess import (  # type: ignore[attr-defined] # nosec B404
+        CREATE_NEW_PROCESS_GROUP,
+        CREATE_NO_WINDOW,
+        STARTF_USESHOWWINDOW,
+        STARTUPINFO,
     )
 
+    # https://stackoverflow.com/a/7006424
+    # https://bugs.python.org/issue41619
+    creationflags = CREATE_NEW_PROCESS_GROUP | CREATE_NO_WINDOW
 
-def _spawn_windows(cmd, env):
-    if sys.platform == "win32":
-        from subprocess import (  # nosec B404
-            CREATE_NEW_PROCESS_GROUP,
-            CREATE_NO_WINDOW,
-            STARTF_USESHOWWINDOW,
-            STARTUPINFO,
-        )
-
-        # https://stackoverflow.com/a/7006424
-        # https://bugs.python.org/issue41619
-        creationflags = CREATE_NEW_PROCESS_GROUP | CREATE_NO_WINDOW
-
-        startupinfo = STARTUPINFO()
-        startupinfo.dwFlags |= STARTF_USESHOWWINDOW
-
-        popen = _popen(
-            cmd, env=env, creationflags=creationflags, startupinfo=startupinfo
-        )
-        _suppress_resource_warning(popen)
+    startupinfo = STARTUPINFO()
+    startupinfo.dwFlags |= STARTF_USESHOWWINDOW
+    popen = subprocess.Popen(
+        args,
+        close_fds=True,
+        shell=False,  # noqa: S603 # nosec B603
+        startupinfo=startupinfo,
+        creationflags=creationflags,
+        **kwargs,
+    )
+    _suppress_resource_warning(popen)
+    return popen.pid
 
 
-def _spawn_posix(cmd, env):
-    from dvc.cli import main
+def _get_dvc_args() -> List[str]:
+    args = [sys.executable]
+    if not is_binary():
+        root_dir = os.path.abspath(os.path.dirname(__file__))
+        main_entrypoint = os.path.join(root_dir, "__main__.py")
+        args.append(main_entrypoint)
+    return args
 
-    # `fork` will copy buffers, so we need to flush them before forking.
-    # Otherwise, we will get duplicated outputs.
-    if sys.stdout and not sys.stdout.closed:
-        sys.stdout.flush()
-    if sys.stderr and not sys.stderr.closed:
-        sys.stderr.flush()
+
+def _fork_process() -> int:
+    assert os.name == "posix"
 
     # NOTE: using os._exit instead of sys.exit, because dvc built
     # with PyInstaller has trouble with SystemExit exception and throws
@@ -71,7 +72,7 @@ def _spawn_posix(cmd, env):
         # pylint: disable-next=no-member
         pid = os.fork()  # type: ignore[attr-defined]
         if pid > 0:
-            return
+            return pid
     except OSError:
         logger.exception("failed at first fork")
         os._exit(1)  # pylint: disable=protected-access
@@ -87,48 +88,100 @@ def _spawn_posix(cmd, env):
         logger.exception("failed at second fork")
         os._exit(1)  # pylint: disable=protected-access
 
-    sys.stdin.close()
-    sys.stdout.close()
-    sys.stderr.close()
-    os.closerange(0, 3)
-
-    os.environ.update(env)
-    main(cmd)
-
-    os._exit(0)  # pylint: disable=protected-access
+    # disconnect from the terminal
+    fd = os.open(os.devnull, os.O_RDWR)
+    for fd2 in range(3):
+        os.dup2(fd, fd2)
+    os.close(fd)
+    return pid
 
 
-def _spawn(cmd, env):
-    logger.debug("Trying to spawn '%s'", cmd)
+def _posix_detached_subprocess(args: Sequence[str], **kwargs) -> int:
+    # double fork and execute a subprocess so that there are no zombies
+    read_end, write_end = os.pipe()
+    pid = _fork_process()
+    if pid > 0:  # in parent
+        os.close(write_end)
+        pid_str = os.read(read_end, 32).decode("utf8")
+        os.close(read_end)
+        return int(pid_str)
+
+    proc = subprocess.Popen(
+        args,
+        shell=False,  # noqa: S603 # nosec B603
+        close_fds=True,
+        **kwargs,
+    )
+    os.close(read_end)
+    os.write(write_end, str(proc.pid).encode("utf8"))
+    os.close(write_end)
+
+    exit_code = proc.wait()
+    os._exit(exit_code)
+
+
+def _detached_subprocess(args: Sequence[str], **kwargs) -> int:
+    """Run in a detached subprocess."""
+    kwargs.setdefault("stdin", subprocess.DEVNULL)
+    kwargs.setdefault("stdout", subprocess.DEVNULL)
+    kwargs.setdefault("stderr", subprocess.DEVNULL)
 
     if os.name == "nt":
-        _spawn_windows(cmd, env)
-    elif os.name == "posix":
-        _spawn_posix(cmd, env)
-    else:
-        raise NotImplementedError
-
-    logger.debug("Spawned '%s'", cmd)
+        return _win_detached_subprocess(args, **kwargs)
+    return _posix_detached_subprocess(args, **kwargs)
 
 
-def daemon(args):
+def _map_log_level_to_flag() -> Optional[str]:
+    flags = {logging.DEBUG: "-v", logging.TRACE: "-vv"}  # type: ignore[attr-defined]
+    return flags.get(logger.getEffectiveLevel())
+
+
+def daemon(args: List[str]) -> None:
     """Launch a `dvc daemon` command in a detached process.
 
     Args:
         args (list): list of arguments to append to `dvc daemon` command.
     """
-    daemonize(["daemon", "-q", *args])
+    if flag := _map_log_level_to_flag():
+        args = [*args, flag]
+    daemonize(["daemon", *args])
 
 
-def daemonize(cmd: List[str]):
+def _spawn(
+    args: List[str],
+    executable: Optional[Union[str, List[str]]] = None,
+    env: Optional[Mapping[str, str]] = None,
+    output_file: Optional[str] = None,
+) -> int:
+    file: "ContextManager[Any]" = nullcontext()
+    kwargs = {}
+    if output_file:
+        file = open(output_file, "ab")  # noqa: SIM115
+        kwargs = {"stdout": file, "stderr": file}
+
+    if executable is None:
+        executable = _get_dvc_args()
+    else:
+        executable = ensure_list(executable)
+
+    with file:
+        return _detached_subprocess(executable + args, env=env, **kwargs)
+
+
+def daemonize(args: List[str], executable: Union[None, str, List[str]] = None) -> None:
+    if os.name not in ("posix", "nt"):
+        return None
+
     if os.environ.get(DVC_DAEMON):
         logger.debug("skipping launching a new daemon.")
         return
 
     env = fix_env()
+    env[DVC_DAEMON] = "1"
     if not is_binary():
         file_path = os.path.abspath(inspect.stack()[0][1])
         env["PYTHONPATH"] = os.path.dirname(os.path.dirname(file_path))
-    env[DVC_DAEMON] = "1"
 
-    _spawn(cmd, env)
+    logger.debug("Trying to spawn %r", args)
+    pid = _spawn(args, executable, env, output_file=env.get(DVC_DAEMON_LOGFILE))
+    logger.debug("Spawned %r with pid %s", args, pid)

dvc/env.py

@@ -1,4 +1,6 @@
+DVC_ANALYTICS_ENDPOINT = "DVC_ANALYTICS_ENDPOINT"
 DVC_DAEMON = "DVC_DAEMON"
+DVC_DAEMON_LOGFILE = "DVC_DAEMON_LOGFILE"
 DVC_EXP_AUTO_PUSH = "DVC_EXP_AUTO_PUSH"
 DVC_EXP_BASELINE_REV = "DVC_EXP_BASELINE_REV"
 DVC_EXP_GIT_REMOTE = "DVC_EXP_GIT_REMOTE"
@@ -15,3 +17,4 @@ DVC_STUDIO_REPO_URL = "DVC_STUDIO_REPO_URL"
 DVC_STUDIO_TOKEN = "DVC_STUDIO_TOKEN"  # noqa: S105 # nosec B105
 DVC_STUDIO_URL = "DVC_STUDIO_URL"
 DVC_SYSTEM_CONFIG_DIR = "DVC_SYSTEM_CONFIG_DIR"
+DVC_UPDATER_ENDPOINT = "DVC_UPDATER_ENDPOINT"

dvc/logger.py

@@ -182,9 +182,11 @@ def set_loggers_level(level: int = logging.INFO) -> None:
 
 
 def setup(level: int = logging.INFO, log_colors: bool = True) -> None:
+    from dvc.utils import isatty
+
     colorama.init()
 
-    formatter = ColorFormatter(log_colors=log_colors and sys.stdout.isatty())
+    formatter = ColorFormatter(log_colors=log_colors and isatty(sys.stdout))
 
     console_info = LoggerHandler(sys.stdout)
     console_info.setLevel(logging.INFO)
@@ -205,7 +207,7 @@ def setup(level: int = logging.INFO, log_colors: bool = True) -> None:
 
     show_traceback = bool(os.environ.get(DVC_SHOW_TRACEBACK))
     err_formatter = ColorFormatter(
-        log_colors=log_colors and sys.stderr.isatty(), show_traceback=show_traceback
+        log_colors=log_colors and isatty(sys.stderr), show_traceback=show_traceback
     )
     console_errors = LoggerHandler(sys.stderr)
     console_errors.setLevel(logging.WARNING)

dvc/prompt.py

@@ -1,7 +1,6 @@
 """Manages user prompts."""
 
 import logging
-import sys
 from getpass import getpass
 from typing import Collection, Optional
 
@@ -9,7 +8,9 @@ logger = logging.getLogger(__name__)
 
 
 def ask(prompt: str, limited_to: Optional[Collection[str]] = None):
-    if not sys.stdout.isatty():
+    from dvc.ui import Console
+
+    if not Console.isatty():
         return None
 
     while True:

dvc/render/converter/vega.py

@@ -114,10 +114,6 @@ class VegaConverter(Converter):
         self.plot_id = plot_id
         self.inferred_properties: Dict = {}
 
-        # TODO we should be handling that in `convert`,
-        #      to avoid stateful `self.inferred_properties`
-        self._infer_x_y()
-
     def _infer_y_from_data(self):
         if self.plot_id in self.data:
             for lst in _lists(self.data[self.plot_id]):
@@ -296,16 +292,11 @@ class VegaConverter(Converter):
     ):
         """
         Convert the data. Fill necessary fields ('x', 'y') and return both
-        generated datapoints and updated properties. If `x` is not provided,
-        leave it as None, fronteds should handle it.
-
-        NOTE: Studio uses this method.
-              The only thing studio FE handles is filling `x` and `y`.
-              `x/y_label` should be filled here.
-
-              Datapoints are not stripped according to config, because users
-              might be utilizing other fields in their custom plots.
+        generated datapoints and updated properties. `x`, `y` values and labels
+        are inferred and always provided.
         """
+        self._infer_x_y()
+
         datapoints = self._find_datapoints()
         properties = {**self.properties, **self.inferred_properties}
 

dvc/repo/experiments/apply.py

@@ -50,6 +50,7 @@ def apply(repo: "Repo", rev: str, **kwargs):  # noqa: C901
         raise InvalidExpRevError(rev) from exc
 
     _apply(repo, exp_rev, name=rev, is_stash=is_stash)
+    kwargs["force"] = True
     dvc_checkout(repo, **kwargs)
 
     repo.scm.set_ref(EXEC_APPLY, exp_rev)

dvc/repo/fetch.py

@@ -59,10 +59,9 @@ def _collect_indexes(  # noqa: PLR0913
                 types=types,
             )
 
-            data = idx.data["repo"]
-            data.onerror = _make_index_onerror(onerror, rev)
+            idx.data["repo"].onerror = _make_index_onerror(onerror, rev)
 
-            indexes[idx.data_tree.hash_info.value] = data
+            indexes[rev or "workspace"] = idx
         except Exception as exc:  # pylint: disable=broad-except
             if onerror:
                 onerror(rev, None, exc)
@@ -138,7 +137,10 @@ def fetch(  # noqa: C901, PLR0913
         onerror=onerror,
     )
 
-    cache_key = ("fetch", tokenize(sorted(indexes.keys())))
+    cache_key = (
+        "fetch",
+        tokenize(sorted(idx.data_tree.hash_info.value for idx in indexes.values())),
+    )
 
     with ui.progress(
         desc="Collecting",
@@ -146,7 +148,7 @@ def fetch(  # noqa: C901, PLR0913
         leave=True,
     ) as pb:
         data = collect(
-            indexes.values(),
+            [idx.data["repo"] for idx in indexes.values()],
             "remote",
             cache_index=self.data_index,
             cache_key=cache_key,

dvc/repo/index.py

@@ -169,8 +169,9 @@ def _load_storage_from_out(storage_map, key, out):
                 FileStorage(
                     key=key,
                     fs=remote.fs,
-                    path=remote.fs.path.join(remote.path, *key),
+                    path=remote.path,
                     index=remote.index,
+                    prefix=(),
                 )
             )
         else:

dvc/repo/push.py

@@ -1,17 +1,36 @@
 from contextlib import suppress
-from typing import TYPE_CHECKING, Optional, Sequence
 
-from dvc.config import NoRemoteError
 from dvc.exceptions import InvalidArgumentError, UploadError
-from dvc.utils import glob_targets
+from dvc.ui import ui
 
 from . import locked
 
-if TYPE_CHECKING:
-    from dvc.data_cloud import Remote
-    from dvc.repo import Repo
-    from dvc.types import TargetType
-    from dvc_objects.db import ObjectDB
+
+def _update_meta(index, **kwargs):
+    from dvc.repo.index import build_data_index
+    from dvc.repo.worktree import _merge_push_meta, worktree_view_by_remotes
+
+    stages = set()
+    for remote_name, idx in worktree_view_by_remotes(index, push=True, **kwargs):
+        remote = index.repo.cloud.get_remote(remote_name)
+
+        with ui.progress("Collecting", unit="entry") as pb:
+            new = build_data_index(
+                idx,
+                remote.path,
+                remote.fs,
+                callback=pb.as_callback(),
+            )
+
+        for out in idx.outs:
+            if not remote.fs.version_aware:
+                continue
+
+            _merge_push_meta(out, new, remote.name)
+            stages.add(out.stage)
+
+    for stage in stages:
+        stage.dump(with_files=True, update_pipeline=False)
 
 
 @locked
@@ -28,92 +47,93 @@ def push(  # noqa: C901, PLR0913
     run_cache=False,
     revs=None,
     glob=False,
-    odb: Optional["ObjectDB"] = None,
-    include_imports=False,
 ):
-    worktree_remote: Optional["Remote"] = None
+    from fsspec.utils import tokenize
+
+    from dvc.config import NoRemoteError
+    from dvc.utils import glob_targets
+    from dvc_data.index.fetch import collect
+    from dvc_data.index.push import push as ipush
+
+    from .fetch import _collect_indexes
+
+    failed_count = 0
+    transferred_count = 0
+
     with suppress(NoRemoteError):
         _remote = self.cloud.get_remote(name=remote)
-        if _remote and (_remote.worktree or _remote.fs.version_aware):
-            worktree_remote = _remote
+        if (
+            _remote
+            and (_remote.worktree or _remote.fs.version_aware)
+            and (revs or all_branches or all_tags or all_commits)
+        ):
+            raise InvalidArgumentError(
+                "Multiple rev push is unsupported for cloud versioned remotes"
+            )
 
-    pushed = 0
-    used_run_cache = self.stage_cache.push(remote, odb=odb) if run_cache else []
-    pushed += len(used_run_cache)
+    used_run_cache = self.stage_cache.push(remote) if run_cache else []
+    transferred_count += len(used_run_cache)
 
     if isinstance(targets, str):
         targets = [targets]
 
-    expanded_targets = glob_targets(targets, glob=glob)
+    indexes = _collect_indexes(
+        self,
+        targets=glob_targets(targets, glob=glob),
+        remote=remote,
+        all_branches=all_branches,
+        with_deps=with_deps,
+        all_tags=all_tags,
+        recursive=recursive,
+        all_commits=all_commits,
+        revs=revs,
+    )
 
-    if worktree_remote is not None:
-        pushed += _push_worktree(
-            self,
-            worktree_remote,
-            revs=revs,
-            all_branches=all_branches,
-            all_tags=all_tags,
-            all_commits=all_commits,
-            targets=expanded_targets,
-            jobs=jobs,
-            with_deps=with_deps,
-            recursive=recursive,
-        )
-    else:
-        used = self.used_objs(
-            expanded_targets,
-            all_branches=all_branches,
-            all_tags=all_tags,
-            all_commits=all_commits,
-            with_deps=with_deps,
-            force=True,
-            remote=remote,
-            jobs=jobs,
-            recursive=recursive,
-            used_run_cache=used_run_cache,
-            revs=revs,
+    cache_key = (
+        "push",
+        tokenize(sorted(idx.data_tree.hash_info.value for idx in indexes.values())),
+    )
+
+    with ui.progress(
+        desc="Collecting",
+        unit="entry",
+    ) as pb:
+        data = collect(
+            [idx.data["repo"] for idx in indexes.values()],
+            "remote",
+            cache_index=self.data_index,
+            cache_key=cache_key,
+            callback=pb.as_callback(),
             push=True,
         )
 
-        if odb:
-            all_ids = set()
-            for dest_odb, obj_ids in used.items():
-                if not include_imports and dest_odb and dest_odb.read_only:
-                    continue
-                all_ids.update(obj_ids)
-            result = self.cloud.push(all_ids, jobs, remote=remote, odb=odb)
-            if result.failed:
-                raise UploadError(len(result.failed))
-            pushed += len(result.transferred)
-        else:
-            for dest_odb, obj_ids in used.items():
-                if dest_odb and dest_odb.read_only:
-                    continue
-                result = self.cloud.push(
-                    obj_ids, jobs, remote=remote, odb=odb or dest_odb
-                )
-                if result.failed:
-                    raise UploadError(len(result.failed))
-                pushed += len(result.transferred)
-    return pushed
+    try:
+        with ui.progress(
+            desc="Pushing",
+            unit="file",
+        ) as pb:
+            push_transferred, push_failed = ipush(
+                data,
+                jobs=jobs,
+                callback=pb.as_callback(),
+            )  # pylint: disable=assignment-from-no-return
+    finally:
+        ws_idx = indexes.get("workspace")
+        if ws_idx is not None:
+            _update_meta(
+                self.index,
+                targets=glob_targets(targets, glob=glob),
+                remote=remote,
+                with_deps=with_deps,
+                recursive=recursive,
+            )
 
+        for fs_index in data:
+            fs_index.close()
 
-def _push_worktree(
-    repo: "Repo",
-    remote: "Remote",
-    revs: Optional[Sequence[str]] = None,
-    all_branches: bool = False,
-    all_tags: bool = False,
-    all_commits: bool = False,
-    targets: Optional["TargetType"] = None,
-    jobs: Optional[int] = None,
-    **kwargs,
-) -> int:
-    from dvc.repo.worktree import push_worktree
+    transferred_count += push_transferred
+    failed_count += push_failed
+    if failed_count:
+        raise UploadError(failed_count)
 
-    if revs or all_branches or all_tags or all_commits:
-        raise InvalidArgumentError(
-            "Multiple rev push is unsupported for cloud versioned remotes"
-        )
-
-    return push_worktree(repo, remote, targets=targets, jobs=jobs, **kwargs)
+    return transferred_count

dvc/repo/worktree.py

@@ -1,10 +1,9 @@
 import logging
 from functools import partial
-from typing import TYPE_CHECKING, Any, Dict, Iterable, Optional, Set, Tuple, Union
+from typing import TYPE_CHECKING, Any, Dict, Iterable, Optional, Tuple, Union
 
 from funcy import first
 
-from dvc.exceptions import DvcException
 from dvc.fs.callbacks import Callback
 from dvc.stage.exceptions import StageUpdateError
 
@@ -104,90 +103,6 @@ def _get_remote(
     return repo.cloud.get_remote(name, command)
 
 
-def push_worktree(
-    repo: "Repo",
-    remote: "Remote",
-    targets: Optional["TargetType"] = None,
-    jobs: Optional[int] = None,
-    **kwargs: Any,
-) -> int:
-    from dvc.repo.index import build_data_index
-    from dvc_data.index.checkout import VersioningNotSupported, apply, compare
-
-    pushed = 0
-    stages: Set["Stage"] = set()
-
-    for remote_name, view in worktree_view_by_remotes(
-        repo.index, push=True, targets=targets, **kwargs
-    ):
-        remote_obj = _get_remote(repo, remote_name, remote, "push")
-        new_index = view.data["repo"]
-        if remote_obj.worktree:
-            logger.debug("indexing latest worktree for '%s'", remote_obj.path)
-            old_index = build_data_index(view, remote_obj.path, remote_obj.fs)
-            logger.debug("Pushing worktree changes to '%s'", remote_obj.path)
-        else:
-            old_index = None
-            logger.debug("Pushing version-aware files to '%s'", remote_obj.path)
-
-        if remote_obj.worktree:
-            diff_kwargs: Dict[str, Any] = {
-                "meta_only": True,
-                "meta_cmp_key": partial(_meta_checksum, remote_obj.fs),
-            }
-        else:
-            diff_kwargs = {}
-
-        with Callback.as_tqdm_callback(
-            unit="entry",
-            desc=f"Comparing indexes for remote {remote_obj.name!r}",
-        ) as cb:
-            diff = compare(
-                old_index,
-                new_index,
-                callback=cb,
-                delete=remote_obj.worktree,
-                **diff_kwargs,
-            )
-
-        total = len(new_index)
-        with Callback.as_tqdm_callback(
-            unit="file",
-            desc=f"Pushing to remote {remote_obj.name!r}",
-            disable=total == 0,
-        ) as cb:
-            cb.set_size(total)
-            try:
-                apply(
-                    diff,
-                    remote_obj.path,
-                    remote_obj.fs,
-                    callback=cb,
-                    latest_only=remote_obj.worktree,
-                    jobs=jobs,
-                )
-                pushed += len(diff.files_create)
-            except VersioningNotSupported:
-                logger.exception("")
-                raise DvcException(
-                    f"remote {remote_obj.name!r} does not support versioning"
-                ) from None
-
-        if remote_obj.index is not None:
-            for key, entry in new_index.iteritems():
-                remote_obj.index[key] = entry
-            remote_obj.index.commit()
-
-        for out in view.outs:
-            workspace, _key = out.index_key
-            _merge_push_meta(out, repo.index.data[workspace], remote_obj.name)
-            stages.add(out.stage)
-
-    for stage in stages:
-        stage.dump(with_files=True, update_pipeline=False)
-    return pushed
-
-
 def _merge_push_meta(
     out: "Output",
     index: Union["DataIndex", "DataIndexView"],
@@ -210,7 +125,9 @@ def _merge_push_meta(
         assert isinstance(old_tree, Tree)
         entry.hash_info = old_tree.hash_info
         entry.meta = out.meta
+        entries = [entry]
         for subkey, entry in index.iteritems(key):
+            entries.append(entry)
             if entry.meta is not None and entry.meta.isdir:
                 continue
             fs_path = repo.fs.path.join(repo.root_dir, *subkey)
@@ -225,6 +142,10 @@ def _merge_push_meta(
                 # this dir (entry will have the latest remote version
                 # ID after checkout)
                 entry.meta = meta
+
+        for entry in entries:
+            index.add(entry)
+
         tree_meta, new_tree = build_tree(index, key)
         out.obj = new_tree
         out.hash_info = new_tree.hash_info

dvc/testing/fixtures.py

@@ -1,9 +1,12 @@
 import os
+import pathlib
 import subprocess  # nosec B404
 from typing import Dict, Tuple
 
 import pytest
 
+from .scripts import copy_script
+
 # pylint: disable=redefined-outer-name,unused-argument
 
 __all__ = [
@@ -28,6 +31,8 @@ __all__ = [
     "local_workspace",
     "docker_compose_project_name",
     "docker_services",
+    "copy_script",
+    "run_copy",
 ]
 
 CACHE: Dict[Tuple[bool, bool, bool], str] = {}
@@ -261,3 +266,20 @@ def docker_services(
         executor.execute(docker_setup)
         # note: we are not tearing down the containers here
         return Services(executor)
+
+
+@pytest.fixture
+def run_copy(tmp_dir, copy_script, dvc):  # noqa: ARG001
+    def run_copy(src, dst, **run_kwargs):
+        wdir = pathlib.Path(run_kwargs.get("wdir", "."))
+        wdir = pathlib.Path("../" * len(wdir.parts))
+        script_path = wdir / "copy.py"
+
+        return dvc.run(
+            cmd=f"python {script_path} {src} {dst}",
+            outs=[dst],
+            deps=[src, f"{script_path}"],
+            **run_kwargs,
+        )
+
+    return run_copy

dvc/testing/remote_tests.py

@@ -169,22 +169,40 @@ class TestRemote:
 
 
 class TestRemoteVersionAware:
-    def test_file(self, tmp_dir, dvc, remote_version_aware):  # pylint: disable=W0613
+    def test_file(
+        self, tmp_dir, dvc, run_copy, remote_version_aware
+    ):  # pylint: disable=W0613
         (stage,) = tmp_dir.dvc_gen("foo", "foo")
+        run_copy("foo", "foo_copy", name="copy")
 
         dvc.push()
-        assert "version_id" in (tmp_dir / "foo.dvc").read_text()
+        foo_dvc = (tmp_dir / "foo.dvc").read_text()
+        assert "version_id" in foo_dvc
         stage = stage.reload()
         out = stage.outs[0]
         assert out.meta.version_id
+        dvc_lock = (tmp_dir / "dvc.lock").read_text()
 
         remove(dvc.cache.local.path)
         remove(tmp_dir / "foo")
 
         dvc.pull()
         assert (tmp_dir / "foo").read_text() == "foo"
+        assert (tmp_dir / "foo.dvc").read_text() == foo_dvc
+        assert (tmp_dir / "dvc.lock").read_text() == dvc_lock
 
-    def test_dir(self, tmp_dir, dvc, remote_version_aware):  # pylint: disable=W0613
+        dvc.push()
+        assert (tmp_dir / "foo.dvc").read_text() == foo_dvc
+        assert (tmp_dir / "dvc.lock").read_text() == dvc_lock
+
+        dvc.reproduce()
+        dvc.push()
+        assert (tmp_dir / "foo.dvc").read_text() == foo_dvc
+        assert (tmp_dir / "dvc.lock").read_text() == dvc_lock
+
+    def test_dir(
+        self, tmp_dir, dvc, run_copy, remote_version_aware
+    ):  # pylint: disable=W0613
         (stage,) = tmp_dir.dvc_gen(
             {
                 "data_dir": {
@@ -196,8 +214,10 @@ class TestRemoteVersionAware:
         )
 
         dvc.push()
-        assert "files" in (tmp_dir / "data_dir.dvc").read_text()
-        assert "version_id" in (tmp_dir / "data_dir.dvc").read_text()
+
+        data_dir_dvc = (tmp_dir / "data_dir.dvc").read_text()
+        assert "files" in data_dir_dvc
+        assert "version_id" in data_dir_dvc
         stage = stage.reload()
         out = stage.outs[0]
         assert out.files
@@ -213,6 +233,19 @@ class TestRemoteVersionAware:
         assert (
             tmp_dir / "data_dir" / "data_sub_dir" / "data_sub"
         ).read_text() == "data_sub"
+        assert (tmp_dir / "data_dir.dvc").read_text() == data_dir_dvc
+
+        run_copy("data_dir", "data_dir_copy", name="copy")
+        dvc_lock = (tmp_dir / "dvc.lock").read_text()
+
+        dvc.push()
+        assert (tmp_dir / "data_dir.dvc").read_text() == data_dir_dvc
+        assert (tmp_dir / "dvc.lock").read_text() != dvc_lock
+        dvc_lock = (tmp_dir / "dvc.lock").read_text()
+
+        dvc.push()
+        assert (tmp_dir / "data_dir.dvc").read_text() == data_dir_dvc
+        assert (tmp_dir / "dvc.lock").read_text() == dvc_lock
 
 
 class TestRemoteWorktree:

dvc/testing/scripts.py

@@ -0,0 +1,28 @@
+import pytest
+
+COPY_SCRIPT = """
+import os
+import shutil
+import sys
+
+if os.path.isfile(sys.argv[1]):
+    shutil.copyfile(sys.argv[1], sys.argv[2])
+else:
+    shutil.copytree(sys.argv[1], sys.argv[2])
+""".replace(
+    "\r\n", "\n"
+)
+
+
+def _add_script(tmp_dir, path, contents=""):
+    script, *_ = tmp_dir.gen(path, contents.encode("utf-8"))
+    if hasattr(tmp_dir, "scm"):
+        tmp_dir.scm_add(path, commit=f"add {path}")
+    return script.fs_path
+
+
+@pytest.fixture
+def copy_script(
+    tmp_dir,
+):
+    return _add_script(tmp_dir, "copy.py", COPY_SCRIPT)

dvc/ui/__init__.py

@@ -340,10 +340,13 @@ class Console:
     def status(self, status: str, **kwargs: Any) -> "Status":
         return self.error_console.status(status, **kwargs)
 
-    def isatty(self) -> bool:
+    @staticmethod
+    def isatty() -> bool:
         import sys
 
-        return sys.stdout.isatty()
+        from dvc import utils
+
+        return utils.isatty(sys.stdout)
 
     def open_browser(self, file: "StrPath") -> int:
         import webbrowser

dvc/ui/pager.py

@@ -3,7 +3,6 @@
 import logging
 import os
 import pydoc
-import sys
 
 from rich.pager import Pager
 
@@ -51,7 +50,9 @@ def make_pager(cmd=None):
 
 
 def find_pager():
-    if not sys.stdout.isatty():
+    from . import Console
+
+    if not Console.isatty():
         return None
 
     # pylint: disable=redefined-outer-name

dvc/updater.py

@@ -1,12 +1,12 @@
 import logging
 import os
-import sys
 import time
 from typing import TYPE_CHECKING, Optional
 
 from packaging import version
 
 from dvc import PKG, __version__
+from dvc.env import DVC_UPDATER_ENDPOINT
 
 if TYPE_CHECKING:
     from dvc.ui import RichText
@@ -46,6 +46,7 @@ class Updater:
             with self.lock:
                 func()
         except LockError:
+            logger.trace("", exc_info=True)  # type: ignore[attr-defined]
             logger.debug(
                 "Failed to acquire '%s' before %s updates",
                 self.lock.lockfile,
@@ -77,6 +78,7 @@ class Updater:
                 info = json.load(fobj)
                 latest = info["version"]
             except Exception as e:  # noqa: BLE001  # pylint: disable=W0703
+                logger.trace("", exc_info=True)  # type: ignore[attr-defined]
                 logger.debug("'%s' is not a valid json: %s", self.updater_file, e)
                 self.fetch()
                 return
@@ -98,20 +100,29 @@ class Updater:
 
         import requests
 
+        url = os.environ.get(DVC_UPDATER_ENDPOINT, self.URL)
+        logger.debug("Checking updates in %s", url)
         try:
-            resp = requests.get(self.URL, timeout=self.TIMEOUT_GET)
+            resp = requests.get(url, timeout=self.TIMEOUT_GET)
             info = resp.json()
         except requests.exceptions.RequestException as exc:
+            logger.trace("", exc_info=True)  # type: ignore[attr-defined]
             logger.debug("Failed to retrieve latest version: %s", exc)
             return
 
+        logger.trace(  # type: ignore[attr-defined]
+            "received payload: %s (status=%s)", info, resp.status_code
+        )
         with open(self.updater_file, "w+", encoding="utf-8") as fobj:
+            logger.trace(  # type: ignore[attr-defined]
+                "Saving latest version info to %s", self.updater_file
+            )
             json.dump(info, fobj)
 
     def _notify(self, latest: str, pkg: Optional[str] = PKG) -> None:
         from dvc.ui import ui
 
-        if not sys.stdout.isatty():
+        if not ui.isatty():
             return
 
         message = self._get_message(latest, pkg=pkg)

dvc/utils/__init__.py

@@ -11,6 +11,8 @@ from typing import TYPE_CHECKING, Dict, Iterable, Iterator, List, Optional, Tupl
 import colorama
 
 if TYPE_CHECKING:
+    from typing import TextIO
+
     from dvc.fs import FileSystem
 
 logger = logging.getLogger(__name__)
@@ -424,3 +426,9 @@ def expand_paths(fs: "FileSystem", paths: Iterable[str]) -> Iterator[str]:
             yield from fs.find(path)
         else:
             yield path
+
+
+def isatty(stream: "Optional[TextIO]") -> bool:
+    if stream is None:
+        return False
+    return stream.isatty()

pyproject.toml

@@ -34,7 +34,7 @@ dependencies = [
     "configobj>=5.0.6",
     "distro>=1.3",
     "dpath<3,>=2.1.0",
-    "dvc-data>=2.18.1,<2.19.0",
+    "dvc-data>=2.20.0,<2.21.0",
     "dvc-http>=2.29.0",
     "dvc-render>=0.3.1,<1",
     "dvc-studio-client>=0.13.0,<1",
@@ -70,7 +70,7 @@ dependencies = [
 
 [project.optional-dependencies]
 all = ["dvc[azure,gdrive,gs,hdfs,oss,s3,ssh,webdav,webhdfs]"]
-azure = ["dvc-azure>=2.21.2"]
+azure = ["dvc-azure>=2.23.0"]
 dev = ["dvc[azure,gdrive,gs,hdfs,lint,oss,s3,ssh,tests,webdav,webhdfs]"]
 gdrive = ["dvc-gdrive==2.20"]
 gs = ["dvc-gs==2.22.1"]

tests/dir_helpers.py

@@ -45,12 +45,10 @@ from global repo template to creating everything inplace, which:
 
 
 import os
-import pathlib
 
 import pytest
 
 __all__ = [
-    "run_copy",
     "run_head",
     "erepo_dir",
     "git_dir",
@@ -59,23 +57,6 @@ __all__ = [
 ]
 
 
-@pytest.fixture
-def run_copy(tmp_dir, copy_script, dvc):
-    def run_copy(src, dst, **run_kwargs):
-        wdir = pathlib.Path(run_kwargs.get("wdir", "."))
-        wdir = pathlib.Path("../" * len(wdir.parts))
-        script_path = wdir / "copy.py"
-
-        return dvc.run(
-            cmd=f"python {script_path} {src} {dst}",
-            outs=[dst],
-            deps=[src, f"{script_path}"],
-            **run_kwargs,
-        )
-
-    return run_copy
-
-
 @pytest.fixture
 def run_head(tmp_dir, head_script, dvc):
     script = os.path.abspath(tmp_dir / "head.py")

tests/func/api/test_experiments.py

@@ -20,6 +20,8 @@ def test_exp_save(tmp_dir, dvc, scm):
 
 
 def test_exp_show(tmp_dir, dvc, scm, exp_stage):  # noqa: F811
+    with open("params.yaml", "a") as fobj:
+        fobj.write("\nbar: 0")
     exps = api.exp_show()
 
     assert len(exps) == 2
@@ -32,3 +34,5 @@ def test_exp_show(tmp_dir, dvc, scm, exp_stage):  # noqa: F811
     assert exps[0]["State"] is None
     # Postprocessing empty string as `None`
     assert exps[0]["Experiment"] is None
+    # Postprocessing 0 as float
+    assert exps[0]["bar"] == 0.0

tests/func/experiments/test_experiments.py

@@ -25,8 +25,8 @@ from dvc.repo.experiments.refs import CELERY_STASH
 from dvc.repo.experiments.utils import exp_refs_by_rev
 from dvc.scm import SCMError, resolve_rev
 from dvc.stage.exceptions import StageFileDoesNotExistError
+from dvc.testing.scripts import COPY_SCRIPT
 from dvc.utils.serialize import PythonFileCorruptedError
-from tests.scripts import COPY_SCRIPT
 
 
 @pytest.mark.parametrize("name", [None, "foo"])

tests/func/test_daemon.py

@@ -0,0 +1,156 @@
+import json
+import os
+import re
+import subprocess
+import sys
+from collections import defaultdict
+from contextlib import contextmanager
+from http.server import HTTPServer, SimpleHTTPRequestHandler
+from pathlib import Path
+from threading import Thread
+from typing import Dict, Iterator
+
+import psutil
+import pytest
+
+from dvc import version_tuple
+from dvc.daemon import _get_dvc_args
+from dvc.env import (
+    DVC_ANALYTICS_ENDPOINT,
+    DVC_DAEMON_LOGFILE,
+    DVC_NO_ANALYTICS,
+    DVC_UPDATER_ENDPOINT,
+)
+from dvc.updater import Updater
+
+version = ".".join(map(str, version_tuple[:3]))
+UPDATER_INFO = {
+    "version": version,
+    "packages": {
+        "linux": {
+            "deb": f"https://dvc.org/download/linux-deb/dvc-{version}",
+            "rpm": f"https://dvc.org/download/linux-rpm/dvc-{version}",
+        },
+        "windows": {"exe": f"https://dvc.org/download/win/dvc-{version}"},
+        "osx": {"pkg": f"https://dvc.org/download/osx/dvc-{version}"},
+    },
+}
+
+UPDATER_INFO_STR = json.dumps(UPDATER_INFO).encode("utf8")
+
+
+def make_request_handler():
+    class RequestHandler(SimpleHTTPRequestHandler):
+        # save requests count for each method
+        hits: Dict[str, int] = defaultdict(int)
+
+        def log_message(self, format, *args) -> None:  # noqa: A002
+            super().log_message(format, *args)
+            if length := self.headers.get("Content-Length"):
+                data = self.rfile.read(int(length)).decode("utf8")
+                sys.stderr.write(f"{data}\n")
+
+        def do_POST(self):  # noqa: N802
+            # analytics endpoint
+            self.hits["POST"] += 1
+            self.send_response(200)
+            super().end_headers()
+
+        def do_GET(self):  # noqa: N802
+            # updater endpoint
+            self.hits["GET"] += 1
+            self.send_response(200)
+            self.send_header("Content-type", "application/json")
+            self.end_headers()
+            self.wfile.write(UPDATER_INFO_STR)
+
+    return RequestHandler
+
+
+@contextmanager
+def make_server(port: int = 0) -> Iterator["HTTPServer"]:
+    with HTTPServer(("localhost", port), make_request_handler()) as httpd:
+        yield httpd
+
+
+@pytest.fixture
+def server():
+    with make_server() as httpd:
+        thread = Thread(target=httpd.serve_forever)
+        thread.daemon = True
+        thread.start()
+        yield httpd
+
+
+def test_analytics(tmp_path, server):
+    addr = server.server_address
+    logfile = tmp_path / "logfile"
+
+    env = {
+        **os.environ,
+        DVC_DAEMON_LOGFILE: str(logfile),
+        DVC_ANALYTICS_ENDPOINT: "http://{}:{}".format(*addr),
+    }
+    env.pop("DVC_TEST", None)
+    env.pop("DVC_NO_ANALYTICS", None)
+
+    output = subprocess.check_output(
+        [*_get_dvc_args(), "config", "-l", "-vv"],
+        env=env,
+        text=True,
+    )
+
+    match = re.search(r".*Saving analytics report to (.*)", output, flags=re.M)
+    assert match, "no match for the report file"
+    report_file = match.group(1).strip()
+
+    match = re.search(r".*Spawned .*analytics.* with pid (.*)", output, flags=re.M)
+    assert match, "no match for the pid"
+    pid = int(match.group(1).strip())
+
+    psutil.Process(pid).wait(timeout=10)
+    assert not os.path.exists(report_file)
+    assert f"Process {pid} exiting with 0" in logfile.read_text(encoding="utf8")
+    assert server.RequestHandlerClass.hits == {"POST": 1}
+
+
+def test_updater(tmp_dir, dvc, server):
+    addr = server.server_address
+    logfile = tmp_dir / "logfile"
+
+    env = {
+        **os.environ,
+        DVC_DAEMON_LOGFILE: str(logfile),
+        DVC_UPDATER_ENDPOINT: "http://{}:{}".format(*addr),
+        # prevent running analytics daemon
+        DVC_NO_ANALYTICS: "true",
+    }
+    env.pop("DVC_TEST", None)
+    env.pop("CI", None)
+
+    output = subprocess.check_output(
+        [*_get_dvc_args(), "version", "-vv"],
+        env=env,
+        text=True,
+    )
+
+    match = re.search(r".*Spawned .*updater.* with pid (.*)", output, flags=re.M)
+    assert match, "no match for the pid"
+    pid = int(match.group(1).strip())
+
+    psutil.Process(pid).wait(timeout=10)
+    assert f"Process {pid} exiting with 0" in logfile.read_text(encoding="utf8")
+    assert server.RequestHandlerClass.hits == {"GET": 1}
+    # check that the file is saved correctly
+    updater_file = Path(dvc.tmp_dir) / Updater.UPDATER_FILE
+    assert json.loads(updater_file.read_text(encoding="utf8")) == UPDATER_INFO
+
+
+if __name__ == "__main__":
+    # python -m tests.func.test_daemon [<port>]
+    port = int(sys.argv[1]) if len(sys.argv) >= 2 else 0
+    with make_server(port) as httpd:
+        print(  # noqa:  T201
+            "Running server on http://{}:{}".format(*httpd.server_address)
+        )
+        httpd.serve_forever()

tests/func/test_data_cloud.py

@@ -148,7 +148,7 @@ def test_hash_recalculation(mocker, dvc, tmp_dir, local_remote):
     assert ret == 0
     ret = main(["push"])
     assert ret == 0
-    assert test_file_md5.mock.call_count == 1
+    assert test_file_md5.mock.call_count == 3
 
 
 def test_missing_cache(tmp_dir, dvc, local_remote, caplog):
@@ -486,7 +486,7 @@ def test_pull_partial(tmp_dir, dvc, local_remote):
     clean(["foo"], dvc)
 
     stats = dvc.pull(os.path.join("foo", "bar"))
-    assert stats["fetched"] == 3
+    assert stats["fetched"] == 2
     assert (tmp_dir / "foo").read_text() == {"bar": {"baz": "baz"}}
 
 

tests/func/test_virtual_directory.py

@@ -182,7 +182,7 @@ def test_partial_checkout_and_update(M, tmp_dir, dvc, remote):
 
     assert dvc.pull("dir/subdir") == M.dict(
         added=[join("dir", "")],
-        fetched=3,
+        fetched=2,
     )
     assert (tmp_dir / "dir").read_text() == {"subdir": {"lorem": "lorem"}}
 

tests/scripts.py

@@ -1,24 +1,6 @@
 import pytest
 
-COPY_SCRIPT = """
-import os
-import shutil
-import sys
-
-if os.path.isfile(sys.argv[1]):
-    shutil.copyfile(sys.argv[1], sys.argv[2])
-else:
-    shutil.copytree(sys.argv[1], sys.argv[2])
-""".replace(
-    "\r\n", "\n"
-)
-
-
-def _add_script(tmp_dir, path, contents=""):
-    script, *_ = tmp_dir.gen(path, contents.encode("utf-8"))
-    if hasattr(tmp_dir, "scm"):
-        tmp_dir.scm_add(path, commit=f"add {path}")
-    return script.fs_path
+from dvc.testing.scripts import _add_script
 
 
 @pytest.fixture
@@ -38,13 +20,6 @@ with Path(sys.argv[1]).open("a+", encoding="utf-8") as f:
     )
 
 
-@pytest.fixture
-def copy_script(
-    tmp_dir,
-):
-    return _add_script(tmp_dir, "copy.py", COPY_SCRIPT)
-
-
 @pytest.fixture
 def head_script(tmp_dir):
     """Output first line of each file to different file with '-1' appended.

tests/unit/command/test_experiments.py

@@ -27,19 +27,7 @@ def test_experiments_apply(dvc, scm, mocker):
 
     assert cmd.run() == 0
 
-    m.assert_called_once_with(cmd.repo, "exp_rev", force=False)
-
-
-def test_experiments_apply_force(dvc, scm, mocker):
-    cli_args = parse_args(["experiments", "apply", "exp_rev", "--force"])
-    assert cli_args.func == CmdExperimentsApply
-
-    cmd = cli_args.func(cli_args)
-    m = mocker.patch("dvc.repo.experiments.apply.apply", return_value={})
-
-    assert cmd.run() == 0
-
-    m.assert_called_once_with(cmd.repo, "exp_rev", force=True)
+    m.assert_called_once_with(cmd.repo, "exp_rev")
 
 
 def test_experiments_diff(dvc, scm, mocker):

tests/unit/render/test_vega_converter.py

@@ -24,6 +24,7 @@ def test_finding_lists(dictionary, expected_result):
     assert list(result) == expected_result
 
 
+@pytest.mark.studio
 @pytest.mark.parametrize(
     "input_data,properties,expected_datapoints,expected_properties",
     [

tests/unit/test_daemon.py

@@ -5,20 +5,12 @@ from dvc import daemon
 
 
 def test_daemon(mocker):
-    mock_windows = mocker.patch("dvc.daemon._spawn_windows")
-    mock_posix = mocker.patch("dvc.daemon._spawn_posix")
+    mock = mocker.patch("dvc.daemon._spawn")
     daemon.daemon(["updater"])
 
-    if os.name == "nt":
-        mock_posix.assert_not_called()
-        mock_windows.assert_called()
-        args = mock_windows.call_args[0]
-    else:
-        mock_windows.assert_not_called()
-        mock_posix.assert_called()
-        args = mock_posix.call_args[0]
-
-    env = args[1]
+    mock.assert_called()
+    args = mock.call_args[0]
+    env = args[2]
     assert "PYTHONPATH" in env
 
     file_path = os.path.abspath(inspect.stack()[0][1])