deps: bump dvc-data to >=0.42.3

Makes for a much simpler resource management elsewhere.

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)),
@@ -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/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/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
@@ -388,17 +397,17 @@ 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")
-            os.makedirs(index_dir, exist_ok=True)
+            if self.index_db_dir:
+                index_dir = os.path.join(self.index_db_dir, "index", "data")
+                os.makedirs(index_dir, exist_ok=True)
 
-            self._data_index = DataIndex.open(os.path.join(index_dir, "db.db"))
+                self._data_index = DataIndex.open(os.path.join(index_dir, "db.db"))
+            else:
+                self._data_index = DataIndex()
 
         return self._data_index
 
@@ -463,6 +472,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,

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/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/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)
@@ -233,7 +236,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 +340,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/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.42.3,<0.43",
     "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",
     "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",

tests/conftest.py

@@ -174,22 +174,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 +207,8 @@ defaultBranch=master
     import pygit2
 
     pygit2.settings.search_path[pygit2.GIT_CONFIG_LEVEL_GLOBAL] = str(home_dir)
+    yield
+    monkeypatch.undo()
 
 
 @pytest.fixture
@@ -247,10 +247,7 @@ def run_copy_metrics(tmp_dir, copy_script):
 
 
 @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
-
+def gc_collect_on_dvc_close_on_win(mocker):
     from dvc.repo import Repo
 
     close = Repo.close
@@ -259,4 +256,5 @@ def gc_collect_on_dvc_close_on_win_311(mocker):
         close(repo)
         gc.collect()
 
-    mocker.patch("dvc.repo.Repo.close", wrapped)
+    if os.name == "nt":
+        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_experiments.py

@@ -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",

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/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/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/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),