install: pre-commit: install all deps by default

Fixes #9178

dvc/cli/parser.py

@@ -2,6 +2,7 @@
 import argparse
 import logging
 import os
+from functools import lru_cache
 
 from dvc import __version__
 from dvc.commands import (
@@ -150,6 +151,7 @@ def get_parent_parser():
     return parent_parser
 
 
+@lru_cache(maxsize=1)
 def get_main_parser():
     parent_parser = get_parent_parser()
 

dvc/commands/experiments/clean.py

@@ -14,7 +14,7 @@ class CmdExperimentsClean(CmdBase):
 
 
 def add_parser(experiments_subparsers, parent_parser):
-    EXPERIMENTS_CLEAN_HELP = "Cleanup dvc exp internal tempfiles."
+    EXPERIMENTS_CLEAN_HELP = "Cleanup experiments temporary internal files."
     experiments_clean_parser = experiments_subparsers.add_parser(
         "clean",
         parents=[parent_parser],

dvc/commands/experiments/diff.py

@@ -52,7 +52,7 @@ class CmdExperimentsDiff(CmdBase):
 
 
 def add_parser(experiments_subparsers, parent_parser):
-    EXPERIMENTS_DIFF_HELP = "Show changes between experiments in the DVC repository."
+    EXPERIMENTS_DIFF_HELP = "Show changes between experiments."
 
     experiments_diff_parser = experiments_subparsers.add_parser(
         "diff",

dvc/commands/experiments/gc.py

@@ -78,7 +78,8 @@ class CmdExperimentsGC(CmdBase):
 def add_parser(experiments_subparsers, parent_parser):
     EXPERIMENTS_GC_HELP = "Garbage collect unneeded experiments."
     EXPERIMENTS_GC_DESCRIPTION = (
-        "Removes all experiments which are not derived from the specifiedGit revisions."
+        "Removes all experiments which are not derived"
+        " from the specified Git revisions."
     )
     experiments_gc_parser = experiments_subparsers.add_parser(
         "gc",

dvc/commands/experiments/save.py

@@ -35,7 +35,7 @@ class CmdExperimentsSave(CmdBase):
 
 
 def add_parser(experiments_subparsers, parent_parser):
-    EXPERIMENTS_SAVE_HELP = "Save current workspace as a dvc experiment."
+    EXPERIMENTS_SAVE_HELP = "Save current workspace as an experiment."
     save_parser = experiments_subparsers.add_parser(
         "save",
         parents=[parent_parser],

dvc/commands/plots.py

@@ -19,7 +19,7 @@ def _show_json(renderers, split=False):
     from dvc.render.convert import to_json
 
     result = {renderer.name: to_json(renderer, split) for renderer in renderers}
-    ui.write_json(result)
+    ui.write_json(result, highlight=False)
 
 
 def _adjust_vega_renderers(renderers):
@@ -112,7 +112,7 @@ class CmdPlots(CmdBase):
                 props=self._props(),
             )
 
-            if not plots_data:
+            if not plots_data and not self.args.json:
                 ui.error_write(
                     "No plots were loaded, visualization file will not be created."
                 )

dvc/config.py

@@ -117,7 +117,7 @@ class Config(dict):
 
     @classmethod
     def get_dir(cls, level):
-        from appdirs import site_config_dir, user_config_dir
+        from platformdirs import site_config_dir, user_config_dir
 
         assert level in ("global", "system")
 

dvc/config_schema.py

@@ -1,8 +1,21 @@
+import logging
 import os
 from urllib.parse import urlparse
 
-from funcy import walk_values
-from voluptuous import All, Any, Coerce, Invalid, Lower, Optional, Range, Schema
+from funcy import once, walk_values
+from voluptuous import (
+    REMOVE_EXTRA,
+    All,
+    Any,
+    Coerce,
+    Invalid,
+    Lower,
+    Optional,
+    Range,
+    Schema,
+)
+
+logger = logging.getLogger(__name__)
 
 Bool = All(
     Lower,
@@ -65,6 +78,27 @@ class RelPath(str):
     pass
 
 
+class FeatureSchema(Schema):
+    def __init__(self, schema, required=False):
+        super().__init__(schema, required=required, extra=REMOVE_EXTRA)
+
+    @staticmethod
+    @once
+    def _log_deprecated(keys):
+        # only run this once per session
+        message = "%s config option%s unsupported"
+        paths = ", ".join(f"'feature.{key}'" for key in keys)
+        pluralize = " is" if len(keys) == 1 else "s are"
+        logger.warning(message, paths, pluralize)
+
+    def __call__(self, data):
+        ret = super().__call__(data)
+        extra_keys = data.keys() - ret.keys()
+        if extra_keys:
+            self._log_deprecated(sorted(extra_keys))
+        return ret
+
+
 REMOTE_COMMON = {
     "url": str,
     "checksum_jobs": All(Coerce(int), Range(1)),
@@ -239,12 +273,12 @@ SCHEMA = {
         )
     },
     "state": {
-        "dir": str,
+        "dir": str,  # obsoleted
         "row_limit": All(Coerce(int), Range(1)),  # obsoleted
         "row_cleanup_quota": All(Coerce(int), Range(0, 100)),  # obsoleted
     },
     "index": {
-        "dir": str,
+        "dir": str,  # obsoleted
     },
     "machine": {
         str: {
@@ -262,12 +296,12 @@ SCHEMA = {
         },
     },
     # section for experimental features
-    "feature": {
-        Optional("machine", default=False): Bool,
-        # enabled by default. It's of no use, kept for backward compatibility.
-        Optional("data_index_cache", default=False): Bool,
-        Optional("parametrization", default=True): Bool,
-    },
+    # only specified keys are validated, others get logged and then ignored/removed
+    "feature": FeatureSchema(
+        {
+            Optional("machine", default=False): Bool,
+        },
+    ),
     "plots": {
         "html_template": str,
         Optional("auto_open", default=False): Bool,

dvc/daemon.py

@@ -14,7 +14,14 @@ from dvc.utils import fix_env, is_binary
 logger = logging.getLogger(__name__)
 
 
-def _popen(cmd, **kwargs):
+def _suppress_resource_warning(popen: Popen):
+    """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(
@@ -40,7 +47,10 @@ def _spawn_windows(cmd, env):
         startupinfo = STARTUPINFO()
         startupinfo.dwFlags |= STARTF_USESHOWWINDOW
 
-        _popen(cmd, env=env, creationflags=creationflags, startupinfo=startupinfo)
+        popen = _popen(
+            cmd, env=env, creationflags=creationflags, startupinfo=startupinfo
+        )
+        _suppress_resource_warning(popen)
 
 
 def _spawn_posix(cmd, env):

dvc/data_cloud.py

@@ -75,7 +75,7 @@ class DataCloud:
                     config["version_aware"] = True
 
             fs = cls(**config)
-            config["tmp_dir"] = self.repo.index_db_dir
+            config["tmp_dir"] = self.repo.site_cache_dir
             if self.repo.data_index is not None:
                 index = self.repo.data_index.view(("remote", name))
             else:

dvc/fs/__init__.py

@@ -121,7 +121,7 @@ def _get_cloud_fs(repo_config, **kwargs):
 
     remote_conf = get_fs_config(repo_config, **kwargs)
     try:
-        remote_conf = SCHEMA["remote"][str](remote_conf)
+        remote_conf = SCHEMA["remote"][str](remote_conf)  # type: ignore[index]
     except Invalid as exc:
         raise RepoConfigError(str(exc)) from None
 

dvc/info.py

@@ -54,6 +54,7 @@ def get_dvc_info():
             fs_root = get_fs_type(os.path.abspath(root_directory))
             info.append(f"Workspace directory: {fs_root}")
             info.append(f"Repo: {_get_dvc_repo_info(repo)}")
+            info.append(f"Repo.site_cache_dir: {repo.site_cache_dir}")
     except NotDvcRepoError:
         pass
     except SCMError:

dvc/output.py

@@ -2,6 +2,7 @@ import logging
 import os
 import posixpath
 from collections import defaultdict
+from contextlib import suppress
 from operator import itemgetter
 from typing import TYPE_CHECKING, Dict, List, Optional, Set, Tuple, Type
 from urllib.parse import urlparse
@@ -1017,7 +1018,8 @@ class Output:
         except FileNotFoundError:
             if self.remote:
                 kwargs["remote"] = self.remote
-            self.repo.cloud.pull([obj.hash_info], **kwargs)
+            with suppress(Exception):
+                self.repo.cloud.pull([obj.hash_info], **kwargs)
 
         if self.obj:
             return self.obj

dvc/proc/exceptions.py

@@ -1,23 +0,0 @@
-from dvc.exceptions import DvcException
-
-
-class ProcessNotTerminatedError(DvcException):
-    def __init__(self, name):
-        super().__init__(f"Managed process '{name}' has not been terminated.")
-
-
-class ProcessNotFoundError(DvcException):
-    def __init__(self, name):
-        super().__init__(f"Managed process '{name}' does not exist.")
-
-
-class TimeoutExpired(DvcException):
-    def __init__(self, cmd, timeout):
-        super().__init__(f"'{cmd}' did not complete before timeout '{timeout}'")
-        self.cmd = cmd
-        self.timeout = timeout
-
-
-class UnsupportedSignalError(DvcException):
-    def __init__(self, sig):
-        super().__init__(f"Unsupported signal: {sig}")

dvc/proc/manager.py

@@ -1,149 +0,0 @@
-"""Serverless process manager."""
-
-import json
-import logging
-import os
-import signal
-import sys
-from typing import Generator, List, Optional, Tuple, Union
-
-from funcy.flow import reraise
-from shortuuid import uuid
-
-from .exceptions import ProcessNotTerminatedError, UnsupportedSignalError
-from .process import ManagedProcess, ProcessInfo
-
-logger = logging.getLogger(__name__)
-
-
-class ProcessManager:
-    """Manager for controlling background ManagedProcess(es).
-
-    Spawned process entries are kept in the manager directory until they
-    are explicitly removed (with remove() or cleanup()) so that return
-    value and log information can be accessed after a process has completed.
-    """
-
-    def __init__(self, wdir: Optional[str] = None):
-        self.wdir = wdir or "."
-
-    def __iter__(self) -> Generator[str, None, None]:
-        if not os.path.exists(self.wdir):
-            return
-        yield from os.listdir(self.wdir)
-
-    def __getitem__(self, key: str) -> "ProcessInfo":
-        info_path = os.path.join(self.wdir, key, f"{key}.json")
-        try:
-            with open(info_path, encoding="utf-8") as fobj:
-                return ProcessInfo.from_dict(json.load(fobj))
-        except FileNotFoundError:
-            raise KeyError  # noqa: B904
-
-    @reraise(FileNotFoundError, KeyError)
-    def __setitem__(self, key: str, value: "ProcessInfo"):
-        info_path = os.path.join(self.wdir, key, f"{key}.json")
-        with open(info_path, "w", encoding="utf-8") as fobj:
-            return json.dump(value.asdict(), fobj)
-
-    def __delitem__(self, key: str) -> None:
-        from dvc.utils.fs import remove
-
-        path = os.path.join(self.wdir, key)
-        if os.path.exists(path):
-            remove(path)
-
-    def get(self, key: str, default=None):
-        try:
-            return self[key]
-        except KeyError:
-            return default
-
-    def processes(self) -> Generator[Tuple[str, "ProcessInfo"], None, None]:
-        for name in self:
-            try:
-                yield name, self[name]
-            except KeyError:
-                continue
-
-    def spawn(self, args: Union[str, List[str]], name: Optional[str] = None):
-        """Run the given command in the background."""
-        name = name or uuid()
-        pid = ManagedProcess.spawn(
-            args,
-            wdir=os.path.join(self.wdir, name),
-            name=name,
-        )
-        logger.debug(
-            "Spawned managed process '%s' (PID: '%d')",
-            name,
-            pid,
-        )
-
-    def send_signal(self, name: str, sig: int):
-        """Send `signal` to the specified named process."""
-        process_info = self[name]
-        if sys.platform == "win32" and sig not in (
-            signal.SIGTERM,
-            signal.CTRL_C_EVENT,  # pylint: disable=no-member
-            signal.CTRL_BREAK_EVENT,  # pylint: disable=no-member
-        ):
-            raise UnsupportedSignalError(sig)
-
-        def handle_closed_process():
-            logging.warning("Process %s had already aborted unexpectedly.", name)
-            process_info.returncode = -1
-            self[name] = process_info
-
-        if process_info.returncode is None:
-            try:
-                os.kill(process_info.pid, sig)
-            except ProcessLookupError:
-                handle_closed_process()
-                raise
-            except OSError as exc:
-                if getattr(exc, "winerror", None) == 87:
-                    handle_closed_process()
-                    raise ProcessLookupError from exc
-                raise
-
-    def terminate(self, name: str):
-        """Terminate the specified named process."""
-        self.send_signal(name, signal.SIGTERM)
-
-    def kill(self, name: str):
-        """Kill the specified named process."""
-        if sys.platform == "win32":
-            self.send_signal(name, signal.SIGTERM)
-        else:
-            self.send_signal(name, signal.SIGKILL)  # pylint: disable=no-member
-
-    def remove(self, name: str, force: bool = False):
-        """Remove the specified named process from this manager.
-
-        If the specified process is still running, it will be forcefully killed
-        if `force` is True`, otherwise an exception will be raised.
-
-        Raises:
-            ProcessNotTerminatedError if the specified process is still
-            running and was not forcefully killed.
-        """
-        try:
-            process_info = self[name]
-        except KeyError:
-            return
-        if process_info.returncode is None and not force:
-            raise ProcessNotTerminatedError(name)
-        try:
-            self.kill(name)
-        except ProcessLookupError:
-            pass
-        del self[name]
-
-    def cleanup(self, force: bool = False):
-        """Remove stale (terminated) processes from this manager."""
-        for name in self:
-            try:
-                self.remove(name, force)
-            except ProcessNotTerminatedError:
-                continue

dvc/proc/process.py

@@ -1,179 +0,0 @@
-import json
-import logging
-import os
-import shlex
-import subprocess
-from contextlib import AbstractContextManager
-from dataclasses import asdict, dataclass
-from typing import BinaryIO, List, Optional, Union
-
-from shortuuid import uuid
-
-from dvc.utils.objects import cached_property
-
-from .exceptions import TimeoutExpired
-
-logger = logging.getLogger(__name__)
-
-
-@dataclass
-class ProcessInfo:
-    pid: int
-    stdin: Optional[str]
-    stdout: Optional[str]
-    stderr: Optional[str]
-    returncode: Optional[int]
-
-    @classmethod
-    def from_dict(cls, d):
-        return cls(**d)
-
-    def asdict(self):
-        return asdict(self)
-
-
-class ManagedProcess(AbstractContextManager):
-    """Run the specified command with redirected output.
-
-    stdout and stderr will both be redirected to <name>.out.
-    Interactive processes (requiring stdin input) are currently unsupported.
-
-    Parameters:
-        args: Command to be run.
-        wdir: If specified, redirected output files will be placed in `wdir`.
-        name: Name to use for this process, if not specified a UUID will be
-            generated instead.
-    """
-
-    def __init__(
-        self,
-        args: Union[str, List[str]],
-        wdir: Optional[str] = None,
-        name: Optional[str] = None,
-    ):
-        self.args: List[str] = (
-            shlex.split(args, posix=os.name == "posix")
-            if isinstance(args, str)
-            else list(args)
-        )
-        self.wdir = wdir
-        self.name = name or uuid()
-        self.returncode: Optional[int] = None
-        self._stdout: Optional[BinaryIO] = None
-        self._stderr: Optional[BinaryIO] = None
-
-        self._run()
-
-    def __exit__(self, exc_type, exc_value, traceback):
-        self.wait()
-
-    def _close_fds(self):
-        if self._stdout:
-            self._stdout.close()
-            self._stdout = None
-        if self._stderr:
-            self._stderr.close()
-            self._stderr = None
-
-    def _make_path(self, path: str) -> str:
-        return os.path.join(self.wdir, path) if self.wdir else path
-
-    @cached_property
-    def stdout_path(self) -> str:
-        return self._make_path(f"{self.name}.out")
-
-    @cached_property
-    def info_path(self) -> str:
-        return self._make_path(f"{self.name}.json")
-
-    @cached_property
-    def pidfile_path(self) -> str:
-        return self._make_path(f"{self.name}.pid")
-
-    @property
-    def info(self) -> "ProcessInfo":
-        return ProcessInfo(
-            pid=self.pid,
-            stdin=None,
-            stdout=self.stdout_path,
-            stderr=None,
-            returncode=self.returncode,
-        )
-
-    def _make_wdir(self):
-        if self.wdir:
-            os.makedirs(self.wdir, exist_ok=True)
-
-    def _dump(self):
-        self._make_wdir()
-        with open(self.info_path, "w", encoding="utf-8") as fobj:
-            json.dump(self.info.asdict(), fobj)
-        with open(self.pidfile_path, "w", encoding="utf-8") as fobj:
-            fobj.write(str(self.pid))
-
-    def _run(self) -> None:
-        self._make_wdir()
-        logger.debug(
-            "Appending output to '%s'",
-            self.stdout_path,
-        )
-        self._stdout = open(self.stdout_path, "ab")  # noqa: SIM115
-        try:
-            self._proc = subprocess.Popen(
-                self.args,
-                stdin=subprocess.DEVNULL,
-                stdout=self._stdout,
-                stderr=subprocess.STDOUT,
-                close_fds=True,
-                shell=False,
-            )
-            self.pid: int = self._proc.pid
-            self._dump()
-        except Exception:
-            if self._proc is not None:
-                self._proc.kill()
-            self._close_fds()
-            raise
-
-    def wait(self, timeout: Optional[int] = None) -> Optional[int]:
-        """Block until a process started with `run` has completed.
-
-        Raises:
-            TimeoutExpired if `timeout` was set and the process
-            did not terminate after `timeout` seconds.
-        """
-        if self.returncode is not None:
-            return self.returncode
-        try:
-            self._proc.wait(timeout=timeout)
-        except subprocess.TimeoutExpired as exc:
-            raise TimeoutExpired(exc.cmd, exc.timeout) from exc
-        self.returncode = self._proc.returncode
-        self._close_fds()
-        self._dump()
-        return self.returncode
-
-    @classmethod
-    def spawn(cls, *args, **kwargs) -> Optional[int]:
-        """Spawn a ManagedProcess command in the background.
-
-        Returns: The spawned process PID.
-        """
-        import multiprocessing as mp
-
-        proc = mp.Process(
-            target=cls._spawn,
-            args=args,
-            kwargs=kwargs,
-            daemon=True,
-        )
-        proc.start()
-        # Do not terminate the child daemon when the main process exits
-        # pylint: disable=protected-access
-        mp.process._children.discard(proc)  # type: ignore[attr-defined]
-        return proc.pid
-
-    @classmethod
-    def _spawn(cls, *args, **kwargs):
-        with cls(*args, **kwargs):
-            pass

dvc/repo/__init__.py

@@ -3,7 +3,16 @@ import os
 from collections import defaultdict
 from contextlib import contextmanager
 from functools import wraps
-from typing import TYPE_CHECKING, Callable, Iterable, List, Optional, Tuple, Union
+from typing import (
+    TYPE_CHECKING,
+    Callable,
+    ContextManager,
+    Iterable,
+    List,
+    Optional,
+    Tuple,
+    Union,
+)
 
 from dvc.exceptions import FileMissingError
 from dvc.exceptions import IsADirectoryError as DvcIsADirectoryError
@@ -125,38 +134,6 @@ class Repo:
         assert root_dir
         return root_dir, dvc_dir
 
-    def _get_database_dir(self, db_name: str) -> Optional[str]:
-        from dvc.fs import localfs
-
-        # NOTE: by default, store SQLite-based remote indexes and state's
-        # `links` and `md5s` caches in the repository itself to avoid any
-        # possible state corruption in 'shared cache dir' scenario, but allow
-        # user to override this through config when, say, the repository is
-        # located on a mounted volume — see
-        # https://github.com/iterative/dvc/issues/4420
-        base_db_dir = self.config.get(db_name, {}).get("dir", None)
-        if not base_db_dir:
-            return self.tmp_dir
-
-        import hashlib
-
-        if self.local_dvc_dir:
-            fs: "FileSystem" = localfs
-            local_root = fs.path.parent(self.local_dvc_dir)
-        else:
-            fs = self.fs
-            local_root = self.root_dir
-        root_dir_hash = hashlib.sha224(local_root.encode("utf-8")).hexdigest()
-
-        db_dir = fs.path.join(
-            base_db_dir,
-            self.DVC_DIR,
-            f"{fs.path.name(local_root)}-{root_dir_hash[0:7]}",
-        )
-
-        fs.makedirs(db_dir, exist_ok=True)
-        return db_dir
-
     def __init__(  # noqa: PLR0915
         self,
         root_dir: Optional[str] = None,
@@ -232,8 +209,8 @@ class Repo:
                     hardlink_lock=self.config["core"].get("hardlink_lock", False),
                     friendly=True,
                 )
-                state_db_dir = self._get_database_dir("state")
-                self.state = State(self.root_dir, state_db_dir, self.dvcignore)
+                os.makedirs(self.site_cache_dir, exist_ok=True)
+                self.state = State(self.root_dir, self.site_cache_dir, self.dvcignore)
             else:
                 self.lock = LockNoop()
                 self.state = StateNoop()
@@ -388,16 +365,12 @@ class Repo:
         self._reset()
 
     @property
-    def data_index(self) -> Optional["DataIndex"]:
+    def data_index(self) -> "DataIndex":
         from dvc_data.index import DataIndex
 
-        if not self.index_db_dir:
-            return None
-
         if self._data_index is None:
-            index_dir = os.path.join(self.index_db_dir, "index", "data")
+            index_dir = os.path.join(self.site_cache_dir, "index", "data")
             os.makedirs(index_dir, exist_ok=True)
-
             self._data_index = DataIndex.open(os.path.join(index_dir, "db.db"))
 
         return self._data_index
@@ -463,6 +436,11 @@ class Repo:
 
         return brancher(self, *args, **kwargs)
 
+    def switch(self, rev: str) -> ContextManager[str]:
+        from dvc.repo.brancher import switch
+
+        return switch(self, rev)
+
     def used_objs(  # noqa: PLR0913
         self,
         targets=None,
@@ -576,8 +554,27 @@ class Repo:
         return DVCFileSystem(repo=self, subrepos=self.subrepos, **self._fs_conf)
 
     @cached_property
-    def index_db_dir(self):
-        return self._get_database_dir("index")
+    def site_cache_dir(self) -> str:
+        import hashlib
+
+        import platformdirs
+
+        from dvc.fs import GitFileSystem
+
+        cache_dir = platformdirs.site_cache_dir("dvc", "iterative", opinion=True)
+
+        if isinstance(self.fs, GitFileSystem):
+            relparts = ()
+            if self.root_dir != "/":
+                # subrepo
+                relparts = self.fs.path.relparts(self.root_dir, "/")
+            root_dir = os.path.join(self.scm.root_dir, *relparts)
+        else:
+            root_dir = self.root_dir
+
+        repo_token = hashlib.md5(os.fsencode(root_dir)).hexdigest()  # noqa: S324
+
+        return os.path.join(cache_dir, "repo", repo_token)
 
     @contextmanager
     def open_by_relpath(self, path, remote=None, mode="r", encoding=None):

dvc/repo/brancher.py

@@ -1,8 +1,15 @@
 import logging
-from typing import Optional
+from contextlib import contextmanager
+from typing import TYPE_CHECKING, Iterator, Optional, Tuple
 
+from scmrepo.git import Git
+
+from dvc.exceptions import NotDvcRepoError
 from dvc.scm import iter_revs
 
+if TYPE_CHECKING:
+    from dvc.repo import Repo
+
 logger = logging.getLogger(__name__)
 
 
@@ -83,23 +90,68 @@ def brancher(  # noqa: E302
     )
 
     try:
-        from dvc.fs import GitFileSystem
-
         for sha, names in found_revs.items():
-            logger.trace(  # type: ignore[attr-defined]
-                "switching fs to revision %s [%s]", sha[:7], ", ".join(names)
-            )
-            self.fs = GitFileSystem(scm=scm, rev=sha)
-            self.root_dir = self.fs.path.join("/", *repo_root_parts)
-
-            if cwd_parts:
-                cwd = self.fs.path.join("/", *cwd_parts)  # type: ignore[unreachable]
-                self.fs.path.chdir(cwd)
-
-            # ignore revs that don't contain repo root
-            # (i.e. revs from before a subdir=True repo was init'ed)
-            if self.fs.exists(self.root_dir):
+            try:
+                _switch_fs(self, sha, repo_root_parts, cwd_parts)
                 yield sha if sha_only else ",".join(names)
+            except NotDvcRepoError:
+                # ignore revs that don't contain repo root
+                # (i.e. revs from before a subdir=True repo was init'ed)
+                pass
     finally:
         self.fs = saved_fs
         self.root_dir = saved_root
+
+
+def _switch_fs(
+    repo: "Repo",
+    rev: str,
+    repo_root_parts: Tuple[str, ...],
+    cwd_parts: Tuple[str, ...],
+):
+    from dvc.fs import GitFileSystem, LocalFileSystem
+
+    if rev == "workspace":
+        logger.trace("switching fs to workspace")  # type: ignore[attr-defined]
+        repo.fs = LocalFileSystem(url=repo.root_dir)
+        return
+
+    logger.trace("switching fs to revision %s", rev[:7])  # type: ignore[attr-defined]
+    assert isinstance(repo.scm, Git)
+    fs = GitFileSystem(scm=repo.scm, rev=rev)
+    root_dir = repo.fs.path.join("/", *repo_root_parts)
+    if not fs.exists(root_dir):
+        raise NotDvcRepoError(f"Commit '{rev[:7]}' does not contain a DVC repo")
+
+    repo.fs = fs
+    repo.root_dir = root_dir
+
+    if cwd_parts:
+        cwd = repo.fs.path.join("/", *cwd_parts)
+        repo.fs.path.chdir(cwd)
+
+
+@contextmanager
+def switch(repo: "Repo", rev: str) -> Iterator[str]:
+    """Switch to a specific revision."""
+    from dvc.scm import resolve_rev
+
+    if rev != "workspace":
+        rev = resolve_rev(repo.scm, rev)
+
+    repo_root_parts = ()
+    if repo.fs.path.isin(repo.root_dir, repo.scm.root_dir):
+        repo_root_parts = repo.fs.path.relparts(repo.root_dir, repo.scm.root_dir)
+
+    cwd_parts = ()
+    if repo.fs.path.isin(repo.fs.path.getcwd(), repo.scm.root_dir):
+        cwd_parts = repo.fs.path.relparts(repo.fs.path.getcwd(), repo.scm.root_dir)
+
+    saved_fs = repo.fs
+    saved_root = repo.root_dir
+    try:
+        _switch_fs(repo, rev, repo_root_parts, cwd_parts)
+        yield rev
+    finally:
+        repo.fs = saved_fs
+        repo.root_dir = saved_root

dvc/repo/data.py

@@ -182,10 +182,7 @@ def _diff_head_to_index(
 ) -> Dict[str, List[str]]:
     index = repo.index.data["repo"]
 
-    for rev in repo.brancher(revs=[head]):
-        if rev == "workspace":
-            continue
-
+    with repo.switch(head):
         head_index = repo.index.data["repo"]
 
     with ui.status("Calculating diff between head/index"):

dvc/repo/experiments/__init__.py

@@ -65,6 +65,11 @@ class Experiments:
 
     @property
     def scm(self):
+        from dvc.scm import SCMError
+
+        if self.repo.scm.no_commits:
+            raise SCMError("Empty Git repo. Add a commit to use experiments.")
+
         return self.repo.scm
 
     @cached_property

dvc/repo/experiments/executor/base.py

@@ -21,6 +21,7 @@ from typing import (
     Union,
 )
 
+from funcy import get_in
 from scmrepo.exceptions import SCMError
 
 from dvc.env import DVC_EXP_AUTO_PUSH, DVC_EXP_GIT_REMOTE
@@ -590,6 +591,7 @@ class BaseExecutor(ABC):
                     info.name,
                     "dvc",
                     experiment_rev=dvc.experiments.scm.get_ref(EXEC_BRANCH),
+                    metrics=get_in(dvc.metrics.show(), ["", "data"]),
                 )
 
                 if infofile is not None:

dvc/repo/experiments/queue/base.py

@@ -17,6 +17,8 @@ from typing import (
     Union,
 )
 
+from dvc_studio_client.env import STUDIO_REPO_URL
+from dvc_studio_client.post_live_metrics import get_studio_repo_url
 from funcy import retry
 
 from dvc.dependency import ParamsDependency
@@ -415,6 +417,10 @@ class BaseStashQueue(ABC):
                     run_env[DVC_EXP_NAME] = name
                     if resume_rev:
                         run_env[DVCLIVE_RESUME] = "1"
+                    studio_repo_url = get_studio_repo_url()
+                    if studio_repo_url is not None:
+                        run_env[STUDIO_REPO_URL] = studio_repo_url
+
                     self._pack_args(*args, run_env=run_env, **kwargs)
 
                     # save experiment as a stash commit

dvc/repo/experiments/show.py

@@ -19,7 +19,7 @@ from typing import (
 from scmrepo.exceptions import SCMError as InnerScmError
 
 from dvc.exceptions import DvcException
-from dvc.scm import Git, RevError, SCMError, iter_revs, resolve_rev
+from dvc.scm import Git, SCMError, iter_revs, resolve_rev
 
 from .refs import ExpRefInfo
 from .serialize import SerializableError, SerializableExp
@@ -81,7 +81,10 @@ def collect_experiment_commit(
     if exp_rev != "workspace" and not (force or param_deps):
         cached_exp = cache.get(exp_rev)
         if cached_exp:
-            if status == ExpStatus.Running:
+            if status == ExpStatus.Running or (
+                isinstance(cached_exp, SerializableExp)
+                and cached_exp.status == ExpStatus.Running.name
+            ):
                 # expire cached queued exp entry once we start running it
                 cache.delete(exp_rev)
             elif isinstance(cached_exp, SerializableError):
@@ -97,7 +100,12 @@ def collect_experiment_commit(
             force=force,
             **kwargs,
         )
-        if exp_rev != "workspace" and not param_deps and not exp.contains_error:
+        if not (
+            exp_rev == "workspace"
+            or status == ExpStatus.Running
+            or param_deps
+            or exp.contains_error
+        ):
             cache.put(exp, force=True)
         return _format_exp(exp)
     except Exception as exc:  # noqa: BLE001, pylint: disable=broad-except
@@ -114,10 +122,8 @@ def _collect_from_repo(
     **kwargs,
 ) -> "SerializableExp":
     running = running or {}
-    for rev in repo.brancher(revs=[exp_rev]):
+    with repo.switch(exp_rev) as rev:
         if rev == "workspace":
-            if exp_rev != "workspace":
-                continue
             timestamp: Optional[datetime] = None
         else:
             commit = repo.scm.resolve_commit(rev)
@@ -143,7 +149,6 @@ def _collect_from_repo(
             executor=executor,
             error=error,
         )
-    raise RevError(f"nonexistent exp rev: '{exp_rev}'")
 
 
 def _collect_complete_experiment(

dvc/repo/experiments/utils.py

@@ -35,6 +35,7 @@ from .refs import (
 
 if TYPE_CHECKING:
     from dvc.repo import Repo
+    from dvc.scm import NoSCM
 
 
 EXEC_TMP_DIR = "exps"
@@ -108,7 +109,9 @@ def exp_refs_by_rev(scm: "Git", rev: str) -> Generator[ExpRefInfo, None, None]:
 
 
 def exp_refs_by_baseline(
-    scm: "Git", revs: Optional[Set[str]] = None, url: Optional[str] = None
+    scm: "Git",
+    revs: Optional[Set[str]] = None,
+    url: Optional[str] = None,
 ) -> Mapping[str, List[ExpRefInfo]]:
     """Iterate over all experiment refs with the specified baseline."""
     all_exp_refs = exp_refs(scm, url)
@@ -212,8 +215,6 @@ def exp_commits(
     refs = ref_infos if ref_infos else exp_refs(scm)
     for ref_info in refs:
         shas.update(scm.branch_revs(str(ref_info), ref_info.baseline_sha))
-        if ref_info.baseline_sha:
-            shas.add(ref_info.baseline_sha)
     yield from shas
 
 
@@ -233,7 +234,7 @@ def remove_exp_refs(scm: "Git", ref_infos: Iterable[ExpRefInfo]):
         scm.remove_ref(str(ref_info))
 
 
-def fix_exp_head(scm: "Git", ref: Optional[str]) -> Optional[str]:
+def fix_exp_head(scm: Union["Git", "NoSCM"], ref: Optional[str]) -> Optional[str]:
     if ref:
         name, tail = Git.split_ref_pattern(ref)
         if name == "HEAD" and scm.get_ref(EXEC_BASELINE):
@@ -337,7 +338,7 @@ def to_studio_params(dvc_params):
     if not dvc_params:
         return result
     for rev_data in dvc_params.values():
-        for file_name, file_data in rev_data["data"].items():
+        for file_name, file_data in rev_data.get("data", {}).items():
             result[file_name] = file_data["data"]
 
     return result

dvc/repo/index.py

@@ -354,42 +354,34 @@ class Index:
 
     @cached_property
     def data(self) -> "Dict[str, DataIndex]":
-        from dvc_data.index import DataIndex
-
         prefix: "DataIndexKey"
         loaded = False
 
         index = self.repo.data_index
-        if index is None:
-            index = DataIndex()
-
         prefix = ("tree", self.data_tree.hash_info.value)
         if index.has_node(prefix):
             loaded = True
 
-        try:
-            if not loaded:
-                _load_data_from_outs(index, prefix, self.outs)
-                index.commit()
+        if not loaded:
+            _load_data_from_outs(index, prefix, self.outs)
+            index.commit()
 
-            by_workspace = {}
-            by_workspace["repo"] = index.view((*prefix, "repo"))
-            by_workspace["local"] = index.view((*prefix, "local"))
+        by_workspace = {}
+        by_workspace["repo"] = index.view((*prefix, "repo"))
+        by_workspace["local"] = index.view((*prefix, "local"))
 
-            for out in self.outs:
-                if not out.use_cache:
-                    continue
+        for out in self.outs:
+            if not out.use_cache:
+                continue
 
-                ws, key = out.index_key
-                if ws not in by_workspace:
-                    by_workspace[ws] = index.view((*prefix, ws))
+            ws, key = out.index_key
+            if ws not in by_workspace:
+                by_workspace[ws] = index.view((*prefix, ws))
 
-                data_index = by_workspace[ws]
-                _load_storage_from_out(data_index.storage_map, key, out)
+            data_index = by_workspace[ws]
+            _load_storage_from_out(data_index.storage_map, key, out)
 
-            return by_workspace
-        finally:
-            index.close()
+        return by_workspace
 
     @staticmethod
     def _hash_targets(

dvc/repo/init.py

@@ -12,7 +12,7 @@ from dvc.utils.fs import remove
 logger = logging.getLogger(__name__)
 
 
-def init(root_dir=os.curdir, no_scm=False, force=False, subdir=False):
+def init(root_dir=os.curdir, no_scm=False, force=False, subdir=False):  # noqa: C901
     """
     Creates an empty repo on the given directory -- basically a
     `.dvc` directory with subdirectories for configuration and cache.
@@ -75,6 +75,11 @@ def init(root_dir=os.curdir, no_scm=False, force=False, subdir=False):
 
     proj = Repo(root_dir)
 
+    if os.path.isdir(proj.site_cache_dir):
+        proj.close()
+        remove(proj.site_cache_dir)
+        proj = Repo(root_dir)
+
     with proj.scm_context(autostage=True) as context:
         files = [
             config.files["repo"],

dvc/repo/install.py

@@ -20,16 +20,19 @@ def pre_commit_install(scm: "Git") -> None:
             "hooks": [
                 {
                     "id": "dvc-pre-commit",
+                    "additional_dependencies": [".[all]"],
                     "language_version": "python3",
                     "stages": ["commit"],
                 },
                 {
                     "id": "dvc-pre-push",
+                    "additional_dependencies": [".[all]"],
                     "language_version": "python3",
                     "stages": ["push"],
                 },
                 {
                     "id": "dvc-post-checkout",
+                    "additional_dependencies": [".[all]"],
                     "language_version": "python3",
                     "stages": ["post-checkout"],
                     "always_run": True,

dvc/repo/push.py

@@ -1,5 +1,7 @@
+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
 
@@ -30,9 +32,10 @@ def push(  # noqa: C901, PLR0913
     include_imports=False,
 ):
     worktree_remote: Optional["Remote"] = None
-    _remote = self.cloud.get_remote(name=remote)
-    if _remote.worktree or _remote.fs.version_aware:
-        worktree_remote = _remote
+    with suppress(NoRemoteError):
+        _remote = self.cloud.get_remote(name=remote)
+        if _remote and (_remote.worktree or _remote.fs.version_aware):
+            worktree_remote = _remote
 
     pushed = 0
     used_run_cache = self.stage_cache.push(remote, odb=odb) if run_cache else []

dvc/scm.py

@@ -165,7 +165,7 @@ def clone(url: str, to_path: str, **kwargs):
             raise CloneError("SCM error") from exc
 
 
-def resolve_rev(scm: "Git", rev: str) -> str:
+def resolve_rev(scm: Union["Git", "NoSCM"], rev: str) -> str:
     from scmrepo.exceptions import RevError as InternalRevError
 
     from dvc.repo.experiments.utils import fix_exp_head
@@ -173,6 +173,7 @@ def resolve_rev(scm: "Git", rev: str) -> str:
     try:
         return scm.resolve_rev(fix_exp_head(scm, rev))
     except InternalRevError as exc:
+        assert isinstance(scm, Git)
         # `scm` will only resolve git branch and tag names,
         # if rev is not a sha it may be an abbreviated experiment name
         if not (rev == "HEAD" or rev.startswith("refs/")):

dvc/stage/cache.py

@@ -1,6 +1,5 @@
 import logging
 import os
-import tempfile
 from contextlib import contextmanager
 from typing import TYPE_CHECKING, List, Optional, Tuple
 
@@ -167,9 +166,10 @@ class StageCache:
         COMPILED_LOCK_FILE_STAGE_SCHEMA(cache)
 
         path = self._get_cache_path(cache_key, cache_value)
-        parent = self.repo.cache.local.fs.path.parent(path)
+        local_fs = self.repo.cache.local.fs
+        parent = local_fs.path.parent(path)
         self.repo.cache.local.makedirs(parent)
-        tmp = tempfile.NamedTemporaryFile(delete=False, dir=parent).name
+        tmp = local_fs.path.join(parent, fs.utils.tmp_fname())
         assert os.path.exists(parent)
         assert os.path.isdir(parent)
         dump_yaml(tmp, cache)

dvc/ui/__init__.py

@@ -17,7 +17,9 @@ from dvc.utils.objects import cached_property
 
 if TYPE_CHECKING:
     from rich.console import Console as RichConsole
+    from rich.console import JustifyMethod, OverflowMethod
     from rich.status import Status
+    from rich.style import Style
     from rich.text import Text as RichText
 
     from dvc.progress import Tqdm
@@ -105,8 +107,9 @@ class Console:
         if indent is None and self.isatty():
             indent = 2
 
-        console = self.error_console if stderr else self.rich_console
-        return console.print_json(
+        from rich.json import JSON
+
+        json = JSON.from_data(
             data=data,
             indent=indent,
             highlight=bool(highlight),
@@ -117,6 +120,50 @@ class Console:
             default=default,
             sort_keys=sort_keys,
         )
+        if not highlight:
+            return self.write(json.text, stderr=stderr)
+        return self.rich_print(json, stderr=stderr, soft_wrap=True)
+
+    def rich_print(
+        self,
+        *objects: Any,
+        sep: str = " ",
+        end: str = "\n",
+        stderr: bool = False,
+        style: Optional[Union[str, "Style"]] = None,
+        justify: Optional["JustifyMethod"] = None,
+        overflow: Optional["OverflowMethod"] = None,
+        no_wrap: Optional[bool] = None,
+        emoji: Optional[bool] = None,
+        markup: Optional[bool] = None,
+        highlight: Optional[bool] = None,
+        width: Optional[int] = None,
+        height: Optional[int] = None,
+        crop: bool = True,
+        soft_wrap: Optional[bool] = None,
+        new_line_start: bool = False,
+    ) -> None:
+        if stderr:
+            console = self.error_console
+        else:
+            console = self.rich_console
+        return console.print(
+            *objects,
+            sep=sep,
+            end=end,
+            style=style,
+            justify=justify,
+            overflow=overflow,
+            no_wrap=no_wrap,
+            emoji=emoji,
+            markup=markup,
+            highlight=highlight,
+            width=width,
+            height=height,
+            crop=crop,
+            soft_wrap=soft_wrap,
+            new_line_start=new_line_start,
+        )
 
     def write(
         self,
@@ -142,10 +189,21 @@ class Console:
         with Tqdm.external_write_mode(file=file):
             # if we are inside pager context, send the output to rich's buffer
             if styled or self._paginate.get():
-                console = self.error_console if stderr else self.rich_console
                 if styled:
-                    return console.print(*objects, sep=sep, end=end)
-                return console.out(*objects, sep=sep, end=end, highlight=False)
+                    return self.rich_print(*objects, sep=sep, end=end, stderr=stderr)
+                return self.rich_print(
+                    sep.join(str(_object) for _object in objects),
+                    style=None,
+                    highlight=False,
+                    emoji=False,
+                    markup=False,
+                    no_wrap=True,
+                    overflow="ignore",
+                    crop=False,
+                    sep=sep,
+                    end=end,
+                    stderr=stderr,
+                )
 
             values = (self.formatter.format(obj, style) for obj in objects)
             return print(*values, sep=sep, end=end, file=file)

pyproject.toml

@@ -30,15 +30,14 @@ classifiers = [
 ]
 dynamic = ["version"]
 dependencies = [
-    "appdirs>=1.4.3",
     "colorama>=0.3.9",
     "configobj>=5.0.6",
     "distro>=1.3",
     "dpath<3,>=2.1.0",
-    "dvc-data>=0.42.1,<0.43",
+    "dvc-data>=0.43.0,<0.44",
     "dvc-http",
     "dvc-render>=0.1.2",
-    "dvc-studio-client>=0.1.1",
+    "dvc-studio-client>=0.5.0,<1",
     "dvc-task>=0.2.0,<1",
     "flatten_dict<1,>=0.4.1",
     "flufl.lock>=5",
@@ -49,6 +48,7 @@ dependencies = [
     "networkx>=2.5",
     "packaging>=19",
     "pathspec>=0.10.3",
+    "platformdirs<4,>=3.1.1",
     "psutil>=5.8",
     "pydot>=1.2.4",
     "pygtrie>=2.3.2",
@@ -73,9 +73,8 @@ gdrive = ["dvc-gdrive==2.19.1"]
 gs = ["dvc-gs==2.22.0"]
 hdfs = ["dvc-hdfs==2.19"]
 lint = [
-    "mypy==1.0.1",
-    "pylint==2.16.3",
-    "types-appdirs",
+    "mypy==1.1.1",
+    "pylint==2.17.0",
     "types-colorama",
     "types-psutil",
     "types-requests",
@@ -133,6 +132,8 @@ profile = "black"
 [tool.pytest.ini_options]
 addopts = "-ra --cov-config pyproject.toml --dist worksteal"
 filterwarnings = [
+    "error::ResourceWarning",
+    "error::pytest.PytestUnraisableExceptionWarning",
     "error::pytest_mock.PytestMockWarning",
     # remove when aiobotocore supports botocore>=1.29.13
     "ignore:'cgi' is deprecated and slated for removal in Python 3.13:DeprecationWarning",
@@ -204,6 +205,7 @@ module = [
     "celery.*",
     "configobj.*",
     "dpath.*",
+    "distro",
     "dvc_http",
     "dvc_render.*",
     "dvc_ssh",
@@ -218,9 +220,12 @@ module = [
     "networkx.*",  # https://github.com/networkx/networkx/issues/3988
     "pygtrie.*",
     "pyinstrument",
+    "pyparsing",
     "pytest_docker.plugin",
     "ruamel.*",
     "ruamel.yaml.*",
+    "shortuuid",
+    "shtab",
     "tpi.*",
     "viztracer",
     "voluptuous",

tests/conftest.py

@@ -1,4 +1,3 @@
-import gc
 import json
 import os
 import sys
@@ -174,22 +173,20 @@ def mocked_webbrowser_open(mocker):
     mocker.patch("webbrowser.open")
 
 
-@pytest.fixture(autouse=True)
-def isolate(tmp_path_factory, monkeypatch) -> None:
+@pytest.fixture(scope="session", autouse=True)
+def isolate(tmp_path_factory):
     path = tmp_path_factory.mktemp("mock")
     home_dir = path / "home"
     home_dir.mkdir()
 
+    monkeypatch = pytest.MonkeyPatch()
     if sys.platform == "win32":
         home_drive, home_path = os.path.splitdrive(home_dir)
         monkeypatch.setenv("USERPROFILE", str(home_dir))
         monkeypatch.setenv("HOMEDRIVE", home_drive)
         monkeypatch.setenv("HOMEPATH", home_path)
 
-        for env_var, sub_path in (
-            ("APPDATA", "Roaming"),
-            ("LOCALAPPDATA", "Local"),
-        ):
+        for env_var, sub_path in (("APPDATA", "Roaming"), ("LOCALAPPDATA", "Local")):
             path = home_dir / "AppData" / sub_path
             path.mkdir(parents=True)
             monkeypatch.setenv(env_var, os.fspath(path))
@@ -209,6 +206,8 @@ defaultBranch=master
     import pygit2
 
     pygit2.settings.search_path[pygit2.GIT_CONFIG_LEVEL_GLOBAL] = str(home_dir)
+    yield
+    monkeypatch.undo()
 
 
 @pytest.fixture
@@ -244,19 +243,3 @@ def run_copy_metrics(tmp_dir, copy_script):
         return stage
 
     return run
-
-
-@pytest.fixture(autouse=True)
-def gc_collect_on_dvc_close_on_win_311(mocker):
-    if sys.version_info < (3, 11) and os.name != "nt":
-        return
-
-    from dvc.repo import Repo
-
-    close = Repo.close
-
-    def wrapped(repo):
-        close(repo)
-        gc.collect()
-
-    mocker.patch("dvc.repo.Repo.close", wrapped)

tests/func/experiments/conftest.py

@@ -5,6 +5,9 @@ from tests.unit.repo.experiments.conftest import (  # noqa, pylint disable=unuse
     exp_stage,
     failed_checkpoint_stage,
     failed_exp_stage,
+    session_app,
+    session_queue,
+    session_worker,
     test_queue,
 )
 
@@ -24,7 +27,7 @@ def http_auth_patch(mocker):
 
 
 @pytest.fixture(params=[True, False])
-def workspace(request, test_queue) -> bool:  # noqa: F811
+def workspace(request, session_queue) -> bool:  # noqa: F811
     return request.param
 
 

tests/func/experiments/test_apply.py

@@ -26,7 +26,7 @@ def test_apply(tmp_dir, scm, dvc, exp_stage):
     assert (tmp_dir / "metrics.yaml").read_text().strip() == "foo: 3"
 
 
-def test_apply_failed(tmp_dir, scm, dvc, failed_exp_stage, test_queue):
+def test_apply_failed(tmp_dir, scm, dvc, failed_exp_stage, session_queue):
     from dvc.exceptions import InvalidArgumentError
 
     dvc.experiments.run(failed_exp_stage.addressing, params=["foo=2"], queue=True)

tests/func/experiments/test_checkpoints.py

@@ -25,9 +25,7 @@ def test_new_checkpoint(tmp_dir, scm, dvc, checkpoint_stage, mocker, workspace,
     exp = first(results)
 
     new_mock.assert_called_once()
-    for rev in dvc.brancher([exp]):
-        if rev == "workspace":
-            continue
+    with dvc.switch(exp):
         fs = dvc.dvcfs
         with fs.open("foo") as fobj:
             assert fobj.read().strip() == str(checkpoint_stage.iterations)
@@ -69,9 +67,7 @@ def test_resume_checkpoint(
     )
     exp = first(results)
 
-    for rev in dvc.brancher([exp]):
-        if rev == "workspace":
-            continue
+    with dvc.switch(exp):
         fs = dvc.dvcfs
         with fs.open("foo") as fobj:
             assert fobj.read().strip() == str(2 * checkpoint_stage.iterations)
@@ -94,9 +90,7 @@ def test_reset_checkpoint(tmp_dir, scm, dvc, checkpoint_stage, caplog, workspace
     )
     exp = first(results)
 
-    for rev in dvc.brancher([exp]):
-        if rev == "workspace":
-            continue
+    with dvc.switch(exp):
         fs = dvc.dvcfs
         with fs.open("foo") as fobj:
             assert fobj.read().strip() == str(checkpoint_stage.iterations)
@@ -132,18 +126,14 @@ def test_resume_branch(tmp_dir, scm, dvc, checkpoint_stage, workspace):
     )
     checkpoint_b = first(results)
 
-    for rev in dvc.brancher([checkpoint_a]):
-        if rev == "workspace":
-            continue
+    with dvc.switch(checkpoint_a):
         fs = dvc.dvcfs
         with fs.open("foo") as fobj:
             assert fobj.read().strip() == str(2 * checkpoint_stage.iterations)
         with fs.open("metrics.yaml") as fobj:
             assert fobj.read().strip() == "foo: 2"
 
-    for rev in dvc.brancher([checkpoint_b]):
-        if rev == "workspace":
-            continue
+    with dvc.switch(checkpoint_b):
         fs = dvc.dvcfs
         with fs.open("foo") as fobj:
             assert fobj.read().strip() == str(2 * checkpoint_stage.iterations)

tests/func/experiments/test_diff.py

@@ -1,16 +1,6 @@
 from funcy import first
 
 
-def test_diff_no_commits(tmp_dir):
-    from dvc.repo import Repo
-    from dvc.scm import Git
-
-    git = Git.init(tmp_dir.fs_path)
-    assert git.no_commits
-
-    assert Repo.init().experiments.diff() == {}
-
-
 def test_diff_empty(tmp_dir, scm, dvc, exp_stage):
     assert dvc.experiments.diff() == {"params": {}, "metrics": {}}
 

tests/func/experiments/test_experiments.py

@@ -13,7 +13,7 @@ from dvc.repo.experiments.exceptions import ExperimentExistsError
 from dvc.repo.experiments.queue.base import BaseStashQueue
 from dvc.repo.experiments.refs import CELERY_STASH
 from dvc.repo.experiments.utils import exp_refs_by_rev
-from dvc.scm import resolve_rev
+from dvc.scm import SCMError, resolve_rev
 from dvc.stage.exceptions import StageFileDoesNotExistError
 from dvc.utils.serialize import PythonFileCorruptedError
 from tests.scripts import COPY_SCRIPT
@@ -131,7 +131,7 @@ def test_get_baseline(tmp_dir, scm, dvc, exp_stage):
     assert dvc.experiments.get_baseline(f"{CELERY_STASH}@{{1}}") == init_rev
 
 
-def test_update_py_params(tmp_dir, scm, dvc, test_queue, copy_script):
+def test_update_py_params(tmp_dir, scm, dvc, session_queue, copy_script):
     tmp_dir.gen("params.py", "INT = 1\n")
     stage = dvc.run(
         cmd="python copy.py params.py metrics.py",
@@ -597,17 +597,21 @@ def test_experiments_workspace_not_log_exception(caplog, dvc, scm):
     assert not caplog.text
 
 
+@pytest.mark.vscode
 def test_run_env(tmp_dir, dvc, scm, mocker):
     dump_run_env = dedent(
         """\
         import os
+        from dvc_studio_client.env import STUDIO_REPO_URL
         from dvc.env import DVC_EXP_BASELINE_REV, DVC_EXP_NAME
-        with open("DVC_EXP_BASELINE_REV", "w") as f:
-            f.write(os.environ.get(DVC_EXP_BASELINE_REV, ""))
-        with open("DVC_EXP_NAME", "w") as f:
-            f.write(os.environ.get(DVC_EXP_NAME, ""))
+        for v in (DVC_EXP_BASELINE_REV, DVC_EXP_NAME, STUDIO_REPO_URL):
+            with open(v, "w") as f:
+                f.write(os.environ.get(v, ""))
         """
     )
+    mocker.patch(
+        "dvc.repo.experiments.queue.base.get_studio_repo_url", return_value="REPO_URL"
+    )
     (tmp_dir / "dump_run_env.py").write_text(dump_run_env)
     baseline = scm.get_rev()
     dvc.stage.add(
@@ -617,10 +621,12 @@ def test_run_env(tmp_dir, dvc, scm, mocker):
     dvc.experiments.run()
     assert (tmp_dir / "DVC_EXP_BASELINE_REV").read_text().strip() == baseline
     assert (tmp_dir / "DVC_EXP_NAME").read_text().strip() != ""
+    assert (tmp_dir / "STUDIO_REPO_URL").read_text().strip() == "REPO_URL"
 
     dvc.experiments.run(name="foo")
     assert (tmp_dir / "DVC_EXP_BASELINE_REV").read_text().strip() == baseline
     assert (tmp_dir / "DVC_EXP_NAME").read_text().strip() == "foo"
+    assert (tmp_dir / "STUDIO_REPO_URL").read_text().strip() == "REPO_URL"
 
 
 def test_experiment_unchanged(tmp_dir, scm, dvc, exp_stage):
@@ -640,3 +646,20 @@ def test_clean(tmp_dir, scm, dvc, mocker):
     clean = mocker.spy(dvc.experiments.celery_queue.celery, "clean")
     dvc.experiments.clean()
     clean.assert_called_once_with()
+
+
+def test_experiment_no_commit(tmp_dir):
+    from scmrepo.git import Git
+
+    from dvc.repo import Repo
+
+    Git.init(tmp_dir.fs_path).close()
+
+    repo = Repo.init()
+    assert repo.scm.no_commits
+
+    try:
+        with pytest.raises(SCMError):  # noqa: PT011
+            repo.experiments.ls()
+    finally:
+        repo.close()

tests/func/experiments/test_queue.py

@@ -19,6 +19,7 @@ def test_celery_logs(
     failed_exp_stage,
     follow,
     capsys,
+    test_queue,
 ):
     celery_queue = dvc.experiments.celery_queue
     dvc.experiments.run(failed_exp_stage.addressing, queue=True)
@@ -33,10 +34,15 @@ def test_celery_logs(
     assert "failed to reproduce 'failed-copy-file'" in captured.out
 
 
+@pytest.mark.xfail(
+    strict=False,
+    reason="https://github.com/iterative/dvc/issues/9143",
+)
 def test_queue_remove_done(
     dvc,
     exp_stage,
     failed_exp_stage,
+    test_queue,
 ):
     queue_length = 3
     success_tasks = []

tests/func/experiments/test_show.py

@@ -41,17 +41,6 @@ def make_executor_info(**kwargs):
     return ExecutorInfo(**kwargs)
 
 
-@pytest.mark.vscode
-def test_show_no_commits(tmp_dir):
-    from dvc.repo import Repo
-    from dvc.scm import Git
-
-    git = Git.init(tmp_dir.fs_path)
-    assert git.no_commits
-
-    assert Repo.init().experiments.show() == {}
-
-
 @pytest.mark.vscode
 def test_show_branch_and_tag_name(tmp_dir, scm, dvc, exp_stage):
     with tmp_dir.branch("new/branch", new=True):

tests/func/test_gc.py

@@ -276,8 +276,7 @@ def test_gc_all_experiments(tmp_dir, scm, dvc):
     (foo,) = tmp_dir.dvc_gen("foo", "foo", commit="foo")
     foo_hash = foo.outs[0].hash_info.value
 
-    (bar,) = tmp_dir.dvc_gen("foo", "bar", commit="bar")
-    bar_hash = bar.outs[0].hash_info.value
+    tmp_dir.dvc_gen("foo", "bar", commit="bar")
     baseline = scm.get_rev()
 
     (baz,) = tmp_dir.dvc_gen("foo", "baz", commit="baz")
@@ -288,12 +287,7 @@ def test_gc_all_experiments(tmp_dir, scm, dvc):
 
     dvc.gc(all_experiments=True, force=True)
 
-    # all_experiments will include the experiment commit (baz) plus baseline
-    # commit (bar)
     assert not (tmp_dir / ".dvc" / "cache" / foo_hash[:2] / foo_hash[2:]).exists()
-    assert (
-        tmp_dir / ".dvc" / "cache" / bar_hash[:2] / bar_hash[2:]
-    ).read_text() == "bar"
     assert (
         tmp_dir / ".dvc" / "cache" / baz_hash[:2] / baz_hash[2:]
     ).read_text() == "baz"

tests/func/test_state.py

@@ -1,7 +1,5 @@
 import os
-import re
 
-from dvc.repo import Repo
 from dvc_data.hashfile.hash import file_md5
 from dvc_data.hashfile.hash_info import HashInfo
 from dvc_data.hashfile.state import State
@@ -71,15 +69,3 @@ def test_get_unused_links(tmp_dir, dvc):
             dvc.fs,
         )
     ) == {"bar"}
-
-
-def test_state_dir_config(make_tmp_dir, dvc):
-    assert dvc.state.tmp_dir == dvc.tmp_dir
-
-    index_dir = str(make_tmp_dir("tmp_index"))
-    repo = Repo(config={"state": {"dir": index_dir}})
-    assert os.path.dirname(repo.state.tmp_dir) == os.path.join(index_dir, ".dvc")
-    assert re.match(
-        r"^test_state_dir_config0-([0-9a-f]+)$",
-        os.path.basename(repo.state.tmp_dir),
-    )

tests/integration/test_studio_live_experiments.py

@@ -38,4 +38,5 @@ def test_post_to_studio(tmp_dir, dvc, scm, exp_stage, mocker, monkeypatch, tmp):
         "name": name,
         "client": "dvc",
         "experiment_rev": exp_rev,
+        "metrics": {"metrics.yaml": {"data": {"foo": 1}}},
     }

tests/unit/fs/test_data.py

@@ -89,10 +89,7 @@ def test_open_in_history(tmp_dir, scm, dvc):
     dvc.scm.add(["foo.dvc", ".gitignore"])
     dvc.scm.commit("foofoo")
 
-    for rev in dvc.brancher(revs=["HEAD~1"]):
-        if rev == "workspace":
-            continue
-
+    with dvc.switch("HEAD~1"):
         fs = DataFileSystem(index=dvc.index.data["repo"])
         with fs.open("foo", "r") as fobj:
             assert fobj.read() == "foo"

tests/unit/fs/test_dvc.py

@@ -59,10 +59,7 @@ def test_open_in_history(tmp_dir, scm, dvc):
     dvc.scm.add(["foo.dvc", ".gitignore"])
     dvc.scm.commit("foofoo")
 
-    for rev in dvc.brancher(revs=["HEAD~1"]):
-        if rev == "workspace":
-            continue
-
+    with dvc.switch("HEAD~1"):
         fs = DVCFileSystem(repo=dvc)
         with fs.open("foo", "r") as fobj:
             assert fobj.read() == "foo"

tests/unit/output/test_output.py

@@ -116,7 +116,7 @@ def test_remote_missing_dependency_on_dir_pull(tmp_dir, scm, dvc, mocker):
     remove(dvc.cache.local.path)
 
     mocker.patch(
-        "dvc.data_cloud.DataCloud.get_remote_odb",
+        "dvc.data_cloud.DataCloud.get_remote",
         side_effect=RemoteMissingDepsError(dvc.fs, "azure", "azure://", []),
     )
     with pytest.raises(RemoteMissingDepsError):

tests/unit/proc/test_manager.py

@@ -1,114 +0,0 @@
-import json
-import os
-import signal
-import sys
-
-import pytest
-
-from dvc.proc.exceptions import ProcessNotTerminatedError, UnsupportedSignalError
-from dvc.proc.manager import ProcessManager
-from dvc.proc.process import ProcessInfo
-
-PID_FINISHED = 1234
-PID_RUNNING = 5678
-
-
-def create_process(root: str, name: str, pid: int, returncode=None):
-    info_path = os.path.join(root, name, f"{name}.json")
-    os.makedirs(os.path.join(root, name))
-    process_info = ProcessInfo(
-        pid=pid, stdin=None, stdout=None, stderr=None, returncode=returncode
-    )
-    with open(info_path, "w", encoding="utf-8") as fobj:
-        json.dump(process_info.asdict(), fobj)
-
-
-@pytest.fixture
-def finished_process(tmp_dir):
-    key = "finished"
-    create_process(tmp_dir, key, PID_FINISHED, 0)
-    return key
-
-
-@pytest.fixture
-def running_process(tmp_dir):
-    key = "running"
-    create_process(tmp_dir, key, PID_RUNNING)
-    return key
-
-
-def test_send_signal(tmp_dir, mocker, finished_process, running_process):
-    m = mocker.patch("os.kill")
-    process_manager = ProcessManager(tmp_dir)
-    process_manager.send_signal(running_process, signal.SIGTERM)
-    m.assert_called_once_with(PID_RUNNING, signal.SIGTERM)
-
-    m.reset_mock()
-    process_manager.send_signal(finished_process, signal.SIGTERM)
-    m.assert_not_called()
-
-    if sys.platform == "win32":
-        with pytest.raises(UnsupportedSignalError):
-            process_manager.send_signal(finished_process, signal.SIGABRT)
-
-
-def test_dead_process(tmp_dir, mocker, running_process):
-    process_manager = ProcessManager(tmp_dir)
-
-    def side_effect(*args):
-        if sys.platform == "win32":
-            err = OSError()
-            err.winerror = 87
-            raise err
-        raise ProcessLookupError()
-
-    mocker.patch("os.kill", side_effect=side_effect)
-    with pytest.raises(ProcessLookupError):
-        process_manager.send_signal(running_process, signal.SIGTERM)
-    assert process_manager[running_process].returncode == -1
-
-
-def test_kill(tmp_dir, mocker, finished_process, running_process):
-    m = mocker.patch("os.kill")
-    process_manager = ProcessManager(tmp_dir)
-    process_manager.kill(running_process)
-    if sys.platform == "win32":
-        m.assert_called_once_with(PID_RUNNING, signal.SIGTERM)
-    else:
-        m.assert_called_once_with(PID_RUNNING, signal.SIGKILL)
-
-    m.reset_mock()
-    process_manager.kill(finished_process)
-    m.assert_not_called()
-
-
-def test_terminate(tmp_dir, mocker, running_process, finished_process):
-    m = mocker.patch("os.kill")
-    process_manager = ProcessManager(tmp_dir)
-    process_manager.terminate(running_process)
-    m.assert_called_once_with(PID_RUNNING, signal.SIGTERM)
-
-    m.reset_mock()
-    process_manager.terminate(finished_process)
-    m.assert_not_called()
-
-
-def test_remove(mocker, tmp_dir, running_process, finished_process):
-    mocker.patch("os.kill", return_value=None)
-    process_manager = ProcessManager(tmp_dir)
-    process_manager.remove(finished_process)
-    assert not (tmp_dir / finished_process).exists()
-    with pytest.raises(ProcessNotTerminatedError):
-        process_manager.remove(running_process)
-    assert (tmp_dir / running_process).exists()
-    process_manager.remove(running_process, True)
-    assert not (tmp_dir / running_process).exists()
-
-
-@pytest.mark.parametrize("force", [True, False])
-def test_cleanup(mocker, tmp_dir, running_process, finished_process, force):
-    mocker.patch("os.kill", return_value=None)
-    process_manager = ProcessManager(tmp_dir)
-    process_manager.cleanup(force)
-    assert (tmp_dir / running_process).exists() != force
-    assert not (tmp_dir / finished_process).exists()

tests/unit/proc/test_process.py

@@ -1,61 +0,0 @@
-import json
-import subprocess
-
-import pytest
-
-from dvc.proc.process import ManagedProcess, ProcessInfo
-
-TEST_PID = 1234
-
-
-@pytest.fixture(autouse=True)
-def mock_popen(mocker):
-    mocker.patch(
-        "subprocess.Popen",
-        return_value=mocker.MagicMock(pid=TEST_PID, returncode=None),
-    )
-
-
-@pytest.mark.parametrize(
-    "args",
-    [
-        "/bin/foo -o option",
-        ["/bin/foo", "-o", "option"],
-    ],
-)
-def test_init_args(tmp_dir, request, args, mocker):
-    expected = ["/bin/foo", "-o", "option"]
-    proc = ManagedProcess(args)
-    request.addfinalizer(proc._close_fds)
-    assert expected == proc.args
-
-
-def test_run(tmp_dir, request, mocker):
-    proc = ManagedProcess("/bin/foo")
-
-    request.addfinalizer(proc._close_fds)
-    assert proc.pid == TEST_PID
-
-    with open(proc.info_path, encoding="utf-8") as fobj:
-        info = ProcessInfo.from_dict(json.load(fobj))
-        assert info.pid == TEST_PID
-
-
-def test_wait(tmp_dir, mocker):
-    from dvc.proc.exceptions import TimeoutExpired
-
-    proc = ManagedProcess("/bin/foo")
-    proc._proc.wait = mocker.Mock(side_effect=subprocess.TimeoutExpired("/bin/foo", 5))
-    with pytest.raises(TimeoutExpired):
-        proc.wait(5)
-
-    proc._proc.wait = mocker.Mock(return_value=None)
-    proc._proc.returncode = 0
-    assert proc.wait() == 0
-    proc._proc.wait.assert_called_once()
-
-    # once subprocess return code is set, future ManagedProcess.wait() calls
-    # should not block
-    proc._proc.wait.reset_mock()
-    assert proc.wait() == 0
-    proc._proc.wait.assert_not_called()

tests/unit/remote/test_remote.py

@@ -1,11 +1,8 @@
-import os
-
 import pytest
 from dvc_gs import GSFileSystem
 from dvc_s3 import S3FileSystem
 
 from dvc.fs import get_cloud_fs
-from dvc_data.hashfile.db import get_index
 
 
 def test_remote_with_hash_jobs(dvc):
@@ -57,18 +54,3 @@ def test_makedirs_not_create_for_top_level_path(fs_cls, dvc, mocker):
 
     fs.makedirs(url)
     assert not mocked_client.called
-
-
-def test_remote_index_dir_config(make_tmp_dir, dvc):
-    index_dir = str(make_tmp_dir("tmp_index"))
-    with dvc.config.edit() as conf:
-        conf["index"]["dir"] = index_dir
-        conf["remote"]["s3"] = {"url": "s3://bucket/name"}
-
-    dvc.root_dir = "/usr/local/test_repo"
-    dvc.dvc_dir = "/usr/local/test_repo/.dvc"
-    dvc.__dict__.pop("local_dvc_dir")
-
-    assert os.path.dirname(
-        get_index(dvc.cloud.get_remote_odb(name="s3")).index_dir
-    ) == os.path.join(index_dir, ".dvc", "test_repo-a473718", "index")

tests/unit/repo/experiments/conftest.py

@@ -3,6 +3,8 @@ from textwrap import dedent
 
 import pytest
 
+from dvc_task.app import FSApp
+
 DEFAULT_ITERATIONS = 2
 CHECKPOINT_SCRIPT_FORMAT = dedent(
     """\
@@ -147,11 +149,60 @@ def _thread_worker(app, **kwargs):
     return worker.start_worker(app, pool="threads", **kwargs)
 
 
+@pytest.fixture(scope="session")
+def session_app(tmp_path_factory) -> FSApp:
+    """Session scoped experiments queue celery app."""
+    from kombu.transport.filesystem import Channel
+
+    # related to https://github.com/iterative/dvc-task/issues/61
+    Channel.QoS.restore_at_shutdown = False
+
+    from dvc_task.app import FSApp
+
+    wdir = tmp_path_factory.mktemp("dvc-test-celery")
+    app = FSApp(
+        "dvc-exp-local",
+        wdir=wdir,
+        mkdir=True,
+        include=[
+            "dvc.repo.experiments.queue.tasks",
+            "dvc_task.proc.tasks",
+        ],
+    )
+    app.conf.update({"task_acks_late": True, "result_expires": None})
+    return app
+
+
+@pytest.fixture(scope="session")
+def session_worker(session_app):
+    """Session scoped celery worker that runs in separate thread(s)."""
+    with _thread_worker(
+        session_app,
+        concurrency=1,
+        ping_task_timeout=20,
+        loglevel="DEBUG",
+    ) as worker:
+        yield worker
+
+
+@pytest.fixture
+def session_queue(tmp_dir, dvc, scm, mocker, session_app, session_worker):
+    """Patches experiments celery queue for pytest testing.
+
+    Uses session-scoped celery worker.
+    """
+    queue = dvc.experiments.celery_queue
+    queue.celery = session_app
+    queue.worker = session_worker
+    mocker.patch.object(queue, "_spawn_worker")
+    return queue
+
+
 @pytest.fixture
 def test_queue(tmp_dir, dvc, scm, mocker):
     """Patches experiments celery queue for pytest testing.
 
-    Test queue worker runs for the duration of the test in separate thread(s).
+    Uses function-scoped celery worker which runs in separate thread(s).
     """
     import celery
 

tests/unit/repo/experiments/queue/test_celery.py

@@ -49,10 +49,10 @@ def test_shutdown_with_kill(test_queue, mocker):
     )
 
 
-def test_post_run_after_kill(test_queue):
+def test_post_run_after_kill(session_queue):
     from celery import chain
 
-    sig_bar = test_queue.proc.run_signature(
+    sig_bar = session_queue.proc.run_signature(
         ["python3", "-c", "import time; time.sleep(10)"], name="bar"
     )
     sig_bar.freeze()
@@ -65,7 +65,7 @@ def test_post_run_after_kill(test_queue):
 
     while True:
         try:
-            test_queue.proc.kill("bar")
+            session_queue.proc.kill("bar")
             assert result_foo.status == "PENDING"
             break
         except ProcessLookupError:

tests/unit/repo/experiments/test_utils.py

@@ -2,7 +2,7 @@ import pytest
 
 from dvc.exceptions import InvalidArgumentError
 from dvc.repo.experiments.refs import EXPS_NAMESPACE, ExpRefInfo
-from dvc.repo.experiments.utils import check_ref_format, resolve_name
+from dvc.repo.experiments.utils import check_ref_format, resolve_name, to_studio_params
 
 
 def commit_exp_ref(tmp_dir, scm, file="foo", contents="foo", name="foo"):
@@ -53,3 +53,20 @@ def test_run_check_ref_format(scm, name, result):
     else:
         with pytest.raises(InvalidArgumentError):
             check_ref_format(scm, ref)
+
+
+@pytest.mark.parametrize(
+    "params,expected",
+    [
+        (
+            {"workspace": {"data": {"params.yaml": {"data": {"foo": 1}}}}},
+            {"params.yaml": {"foo": 1}},
+        ),
+        (
+            {"workspace": {"error": "something went wrong"}},
+            {},
+        ),
+    ],
+)
+def test_to_studio_params(params, expected):
+    assert to_studio_params(params) == expected

tests/unit/scm/test_scm.py

@@ -64,7 +64,6 @@ def test_iter_revs(
     assert gen == {
         rev_new: [rev_new],
         rev_old: [rev_old],
-        rev_root: [rev_root],
     }
 
     def _resolve_commit(rev):

tests/unit/test_config.py

@@ -1,3 +1,4 @@
+import logging
 import os
 import textwrap
 
@@ -94,3 +95,22 @@ def test_load_configob_error(tmp_dir, dvc, mocker):
     with pytest.raises(ConfigError):
         with config.edit():
             pass
+
+
+def test_feature_section_supports_arbitrary_values(caplog):
+    with caplog.at_level(logging.WARNING, logger="dvc.config_schema"):
+        data = Config.validate(
+            {
+                "feature": {
+                    "random_key_1": "random_value_1",
+                    "random_key_2": 42,
+                }
+            }
+        )
+
+    assert "random_key_1" not in data
+    assert "random_key_2" not in data
+    assert (
+        "'feature.random_key_1', 'feature.random_key_2' "
+        "config options are unsupported"
+    ) in caplog.text

tests/unit/test_ignore.py

@@ -87,7 +87,7 @@ def mock_dvcignore(dvcignore_path, patterns):
         ("to_ignore.txt", ["/*.txt"], True),
         (os.path.join("path", "to_ignore.txt"), ["/*.txt"], False),
         (os.path.join("data", "file.txt"), ["data/*"], True),
-        # (os.path.join("data", "subdir", "file.txt"), ["data/*"], False),
+        (os.path.join("data", "subdir", "file.txt"), ["data/*"], True),
         (os.path.join("data", "file.txt"), ["data/"], True),
         (os.path.join("data", "subdir", "file.txt"), ["data/"], True),
         (os.path.join("data", "subdir", "file.txt"), ["subdir/"], True),