diff --git a/dvc/command/experiments.py b/dvc/command/experiments.py index 5c29f3f53..067d4a598 100644 --- a/dvc/command/experiments.py +++ b/dvc/command/experiments.py @@ -356,7 +356,9 @@ class CmdExperimentsShow(CmdBase): class CmdExperimentsApply(CmdBase): def run(self): - self.repo.experiments.apply(self.args.experiment) + self.repo.experiments.apply( + self.args.experiment, force=self.args.force + ) return 0 @@ -448,6 +450,7 @@ class CmdExperimentsRun(CmdRepro): jobs=self.args.jobs, params=self.args.params, checkpoint_resume=self.args.checkpoint_resume, + tmp_dir=self.args.tmp_dir, **self._repro_kwargs, ) @@ -724,6 +727,12 @@ def add_parser(subparsers, parent_parser): help=EXPERIMENTS_APPLY_HELP, formatter_class=argparse.RawDescriptionHelpFormatter, ) + experiments_apply_parser.add_argument( + "-f", + "--force", + action="store_true", + help="Overwrite any conflicting changes.", + ) experiments_apply_parser.add_argument( "experiment", help="Experiment to be applied.", ) @@ -1094,3 +1103,13 @@ def _add_run_common(parser): help="Run the specified number of experiments at a time in parallel.", metavar="", ) + parser.add_argument( + "--temp", + action="store_true", + dest="tmp_dir", + help=( + "Run this experiment in a separate temporary directory instead of " + "your workspace. Only applies when running a single experiment " + "without --queue." + ), + ) diff --git a/dvc/repo/experiments/__init__.py b/dvc/repo/experiments/__init__.py index 24720f19a..20b2e8136 100644 --- a/dvc/repo/experiments/__init__.py +++ b/dvc/repo/experiments/__init__.py @@ -16,9 +16,12 @@ from dvc.exceptions import DvcException from dvc.path_info import PathInfo from dvc.stage.run import CheckpointKilledError from dvc.utils import relpath +from dvc.utils.fs import makedirs from .base import ( + EXEC_APPLY, EXEC_BASELINE, + EXEC_BRANCH, EXEC_CHECKPOINT, EXEC_HEAD, EXEC_MERGE, @@ -47,6 +50,22 @@ def scm_locked(f): return wrapper +def unlocked_repo(f): + @wraps(f) + def wrapper(exp, *args, **kwargs): + exp.repo.state.dump() + exp.repo.lock.unlock() + exp.repo._reset() # pylint: disable=protected-access + try: + ret = f(exp, *args, **kwargs) + finally: + exp.repo.lock.lock() + exp.repo.state.load() + return ret + + return wrapper + + class Experiments: """Class that manages experiments in a DVC repo. @@ -66,6 +85,7 @@ class Experiments: r"(?P-checkpoint)?$" ) LAST_CHECKPOINT = ":last" + EXEC_TMP_DIR = "exps" StashEntry = namedtuple( "StashEntry", ["index", "rev", "baseline_rev", "branch", "name"] @@ -98,7 +118,7 @@ class Experiments: @cached_property def args_file(self): - from .executor import BaseExecutor + from .executor.base import BaseExecutor return os.path.join(self.repo.tmp_dir, BaseExecutor.PACKED_ARGS_FILE) @@ -128,7 +148,7 @@ class Experiments: self, *args, params: Optional[dict] = None, - detach_rev: Optional[str] = None, + resume_rev: Optional[str] = None, baseline_rev: Optional[str] = None, branch: Optional[str] = None, name: Optional[str] = None, @@ -140,6 +160,7 @@ class Experiments: params: Optional dictionary of parameter values to be used. Values take priority over any parameters specified in the user's workspace. + resume_rev: Optional checkpoint resume rev. baseline_rev: Optional baseline rev for this experiment, defaults to the current SCM rev. branch: Optional experiment branch name. If specified, the @@ -149,38 +170,52 @@ class Experiments: the human-readable name in the experiment branch ref. Has no effect of branch is specified. """ - with self.scm.stash_workspace( - include_untracked=detach_rev or branch - ) as workspace: - # If we are not extending an existing branch, apply current - # workspace changes to be made in new branch - if not (branch or detach_rev) and workspace: - self.stash.apply(workspace) - self._prune_lockfiles() + with self.scm.detach_head() as orig_head: + stash_head = orig_head + if baseline_rev is None: + baseline_rev = orig_head - # checkout and detach at branch (or current HEAD) - if detach_rev: - head: Optional[str] = detach_rev - elif branch: - head = branch - else: - head = None + with self.scm.stash_workspace() as workspace: + try: + if workspace: + self.stash.apply(workspace) - try: - with self.scm.detach_head(head) as rev: - if baseline_rev is None: - baseline_rev = rev + if resume_rev: + # move HEAD to the resume rev so that the stashed diff + # only contains changes relative to resume rev + stash_head = resume_rev + self.scm.set_ref( + "HEAD", + resume_rev, + message=f"dvc: resume from HEAD {resume_rev[:7]}", + ) + self.scm.reset() + + self._prune_lockfiles() # update experiment params from command line if params: self._update_params(params) + if resume_rev: + if self.scm.is_dirty(): + logger.debug( + "Resume new checkpoint branch with " + "modifications" + ) + branch = None + else: + logger.debug( + "Resuming from tip of existing branch '%s'", + branch, + ) + # save additional repro command line arguments self._pack_args(*args, **kwargs) # save experiment as a stash commit msg = self._stash_msg( - rev, + stash_head, baseline_rev=baseline_rev, branch=branch, name=name, @@ -194,9 +229,13 @@ class Experiments: stash_rev[:7], baseline_rev[:7], ) - finally: - # Reset any changes before prior workspace is unstashed - self.scm.reset(hard=True) + finally: + # Reset any of our changes before prior unstashing + if resume_rev: + self.scm.set_ref( + "HEAD", orig_head, message="dvc: restore HEAD" + ) + self.scm.reset(hard=True) return stash_rev @@ -232,7 +271,7 @@ class Experiments: def _pack_args(self, *args, **kwargs): import pickle - from .executor import BaseExecutor + from .executor.base import BaseExecutor if os.path.exists(self.args_file) and self.scm.is_tracked( self.args_file @@ -281,39 +320,49 @@ class Experiments: # whether the file is dirty self.scm.add(list(params.keys())) - def reproduce_one(self, queue=False, **kwargs): + def reproduce_one(self, queue=False, tmp_dir=False, **kwargs): """Reproduce and checkout a single experiment.""" stash_rev = self.new(**kwargs) if queue: logger.info( - "Queued experiment '%s' for future execution.", stash_rev[:7] + "Queued experiment '%s' for future execution.", stash_rev[:7], ) return [stash_rev] - results = self.reproduce([stash_rev], keep_stash=False) + if tmp_dir or queue: + results = self._reproduce_revs(revs=[stash_rev], keep_stash=False) + else: + results = self._workspace_repro() exp_rev = first(results) if exp_rev is not None: - self._log_reproduced(results) + self._log_reproduced(results, tmp_dir=tmp_dir) return results def reproduce_queued(self, **kwargs): - results = self.reproduce(**kwargs) + results = self._reproduce_revs(**kwargs) if results: - self._log_reproduced(results) + self._log_reproduced(results, tmp_dir=True) return results - def _log_reproduced(self, revs: Iterable[str]): + def _log_reproduced(self, revs: Iterable[str], tmp_dir: bool = False): names = [] for rev in revs: name = self.get_exact_name(rev) names.append(name if name else rev[:7]) - fmt = ( - "\nReproduced experiment(s): %s\n" - "To promote an experiment to a Git branch run:\n\n" - "\tdvc exp branch \n\n" - "To apply the results of an experiment to your workspace run:\n\n" - "\tdvc exp apply " + logger.info("\nReproduced experiment(s): %s", ", ".join(names)) + if tmp_dir: + logger.info( + "To apply the results of an experiment to your workspace " + "run:\n\n" + "\tdvc exp apply " + ) + else: + logger.info( + "Experiment results have been applied to your workspace." + ) + logger.info( + "\nTo promote an experiment to a Git branch run:\n\n" + "\tdvc exp branch \n" ) - logger.info(fmt, ", ".join(names)) @scm_locked def new( @@ -346,6 +395,7 @@ class Experiments: resume_rev = self._get_last_checkpoint() else: resume_rev = self.scm.resolve_rev(checkpoint_resume) + allow_multiple = "params" in kwargs branch: Optional[str] = self.get_branch_by_rev( resume_rev, allow_multiple=allow_multiple @@ -356,25 +406,34 @@ class Experiments: f"'{checkpoint_resume}'" ) + last_applied = self.scm.get_ref(EXEC_APPLY) + if resume_rev != last_applied: + if last_applied is None: + msg = "Current workspace does not contain an experiment. " + else: + msg = ( + f"Checkpoint '{checkpoint_resume[:7]}' does not match the " + "most recently applied experiment in your workspace " + f"('{last_applied[:7]}')." + ) + + raise DvcException( + f"{msg}\n" + "To resume this experiment run:\n\n" + f"\tdvc exp apply {checkpoint_resume[:7]}\n\n" + "And then retry this 'dvc exp res' command." + ) + baseline_rev = self._get_baseline(branch) - if kwargs.get("params", None): - logger.debug( - "Branching from checkpoint '%s' with modified params, " - "baseline '%s'", - checkpoint_resume, - baseline_rev[:7], - ) - detach_rev = resume_rev - branch = None - else: - logger.debug( - "Continuing from tip of checkpoint '%s'", checkpoint_resume - ) - detach_rev = None + logger.debug( + "Resume from checkpoint '%s' with baseline '%s'", + checkpoint_resume, + baseline_rev, + ) return self._stash_exp( *args, - detach_rev=detach_rev, + resume_rev=resume_rev, baseline_rev=baseline_rev, branch=branch, **kwargs, @@ -387,7 +446,7 @@ class Experiments: raise DvcException("No existing checkpoint experiment to continue") @scm_locked - def reproduce( + def _reproduce_revs( self, revs: Optional[Iterable] = None, keep_stash: Optional[bool] = True, @@ -428,7 +487,7 @@ class Experiments: ) executors = self._init_executors(to_run) - exec_results = self._reproduce(executors, **kwargs) + exec_results = self._executors_repro(executors, **kwargs) if keep_stash: # only drop successfully run stashed experiments @@ -455,9 +514,12 @@ class Experiments: return result def _init_executors(self, to_run): - from .executor import LocalExecutor + from .executor.local import TempDirExecutor executors = {} + base_tmp_dir = os.path.join(self.repo.tmp_dir, self.EXEC_TMP_DIR) + if not os.path.exists(base_tmp_dir): + makedirs(base_tmp_dir) for stash_rev, item in to_run.items(): self.scm.set_ref(EXEC_HEAD, item.rev) self.scm.set_ref(EXEC_MERGE, stash_rev) @@ -469,11 +531,12 @@ class Experiments: # EXEC_MERGE - the unmerged changes (from our stash) # to be reproduced # EXEC_BASELINE - the baseline commit for this experiment - executor = LocalExecutor( + executor = TempDirExecutor( self.scm, self.dvc_dir, name=item.name, branch=item.branch, + tmp_dir=base_tmp_dir, cache_dir=self.repo.cache.local.cache_dir, ) executors[stash_rev] = executor @@ -483,7 +546,7 @@ class Experiments: return executors - def _reproduce( + def _executors_repro( self, executors: dict, jobs: Optional[int] = 1 ) -> Mapping[str, Mapping[str, str]]: """Run dvc repro for the specified BaseExecutors in parallel. @@ -504,8 +567,8 @@ class Experiments: future = workers.submit( executor.reproduce, executor.dvc_dir, - pid_q, rev, + queue=pid_q, name=executor.name, rel_cwd=rel_cwd, log_level=logger.getEffectiveLevel(), @@ -533,9 +596,9 @@ class Experiments: try: if exc is None: - exp_hash, force = future.result() + exec_result = future.result() result[rev].update( - self._collect_executor(executor, exp_hash, force) + self._collect_executor(executor, exec_result) ) else: # Checkpoint errors have already been logged @@ -556,9 +619,7 @@ class Experiments: return result - def _collect_executor( - self, executor, exp_hash, force - ) -> Mapping[str, str]: + def _collect_executor(self, executor, exec_result) -> Mapping[str, str]: # NOTE: GitPython Repo instances cannot be re-used # after process has received SIGINT or SIGTERM, so we # need this hack to re-instantiate git instances after @@ -575,15 +636,64 @@ class Experiments: raise ExperimentExistsError(ref_info.name) for ref in executor.fetch_exps( - self.scm, force=force, on_diverged=on_diverged, + self.scm, force=exec_result.force, on_diverged=on_diverged, ): exp_rev = self.scm.get_ref(ref) if exp_rev: logger.debug("Collected experiment '%s'.", exp_rev[:7]) - results[exp_rev] = exp_hash + results[exp_rev] = exec_result.exp_hash return results + @unlocked_repo + def _workspace_repro(self) -> Mapping[str, str]: + """Run the most recently stashed experiment in the workspace.""" + from .executor.base import BaseExecutor + + entry = first(self.stash_revs.values()) + assert entry.index == 0 + + # NOTE: the stash commit to be popped already contains all the current + # workspace changes plus CLI modifed --params changes. + # `reset --hard` here will not lose any data (pop without reset would + # result in conflict between workspace params and stashed CLI params). + self.scm.reset(hard=True) + with self.scm.detach_head(entry.rev): + rev = self.stash.pop() + self.scm.set_ref(EXEC_BASELINE, entry.baseline_rev) + if entry.branch: + self.scm.set_ref(EXEC_BRANCH, entry.branch, symbolic=True) + elif self.scm.get_ref(EXEC_BRANCH): + self.scm.remove_ref(EXEC_BRANCH) + try: + exec_result = BaseExecutor.reproduce( + None, + rev, + name=entry.name, + rel_cwd=relpath(os.getcwd(), self.scm.root_dir), + ) + + if not exec_result.exp_hash or not exec_result.ref_info: + raise DvcException( + f"Failed to reproduce experiment '{rev[:7]}'" + ) + exp_rev = self.scm.get_ref(str(exec_result.ref_info)) + self.scm.set_ref(EXEC_APPLY, exp_rev) + return {exp_rev: exec_result.exp_hash} + except CheckpointKilledError: + # Checkpoint errors have already been logged + return {} + except DvcException: + raise + except Exception as exc: + raise DvcException( + f"Failed to reproduce experiment '{rev[:7]}'" + ) from exc + finally: + self.scm.remove_ref(EXEC_BASELINE) + if entry.branch: + self.scm.remove_ref(EXEC_BRANCH) + def check_baseline(self, exp_rev): baseline_sha = self.repo.scm.get_rev() if exp_rev == baseline_sha: diff --git a/dvc/repo/experiments/apply.py b/dvc/repo/experiments/apply.py index bdfdac49a..3fb3b7114 100644 --- a/dvc/repo/experiments/apply.py +++ b/dvc/repo/experiments/apply.py @@ -6,15 +6,20 @@ from dvc.repo.scm_context import scm_context from dvc.scm.base import RevError from dvc.utils.fs import remove -from .base import BaselineMismatchError, InvalidExpRevError -from .executor import BaseExecutor +from .base import ( + EXEC_APPLY, + ApplyConflictError, + BaselineMismatchError, + InvalidExpRevError, +) +from .executor.base import BaseExecutor logger = logging.getLogger(__name__) @locked @scm_context -def apply(repo, rev, *args, **kwargs): +def apply(repo, rev, force=False, **kwargs): from git.exc import GitCommandError from dvc.repo.checkout import checkout as dvc_checkout @@ -22,18 +27,14 @@ def apply(repo, rev, *args, **kwargs): exps = repo.experiments try: - rev = repo.scm.resolve_rev(rev) - exps.check_baseline(rev) + exp_rev = repo.scm.resolve_rev(rev) + exps.check_baseline(exp_rev) except (RevError, BaselineMismatchError) as exc: raise InvalidExpRevError(rev) from exc - stash_rev = rev in exps.stash_revs - if stash_rev: - branch = rev - else: - branch = exps.get_branch_by_rev(rev) - if not branch: - raise InvalidExpRevError(rev) + stash_rev = exp_rev in exps.stash_revs + if not stash_rev and not exps.get_branch_by_rev(exp_rev): + raise InvalidExpRevError(exp_rev) # Note that we don't use stash_workspace() here since we need finer control # over the merge behavior when we unstash everything @@ -43,15 +44,21 @@ def apply(repo, rev, *args, **kwargs): else: workspace = None - repo.scm.gitpython.repo.git.merge(branch, squash=True, no_commit=True) + repo.scm.gitpython.repo.git.merge(exp_rev, squash=True, no_commit=True) if workspace: try: repo.scm.stash.apply(workspace) except GitCommandError: - # if stash apply returns merge conflicts, prefer experiment - # changes over prior stashed changes - repo.scm.gitpython.repo.git.checkout("--ours", "--", ".") + # Applied experiment conflicts with user's workspace changes + if force: + # prefer applied experiment changes over prior stashed changes + repo.scm.gitpython.repo.git.checkout("--ours", "--", ".") + else: + # revert applied changes and restore user's workspace + repo.scm.reset(hard=True) + repo.scm.stash.pop() + raise ApplyConflictError(rev) repo.scm.stash.drop() repo.scm.gitpython.repo.git.reset() @@ -62,6 +69,7 @@ def apply(repo, rev, *args, **kwargs): dvc_checkout(repo, **kwargs) + repo.scm.set_ref(EXEC_APPLY, rev) logger.info( "Changes for experiment '%s' have been applied to your current " "workspace.", diff --git a/dvc/repo/experiments/base.py b/dvc/repo/experiments/base.py index 2de9d1794..474ad4bb4 100644 --- a/dvc/repo/experiments/base.py +++ b/dvc/repo/experiments/base.py @@ -7,6 +7,7 @@ from dvc.exceptions import DvcException, InvalidArgumentError EXPS_NAMESPACE = "refs/exps" EXPS_STASH = f"{EXPS_NAMESPACE}/stash" EXEC_NAMESPACE = f"{EXPS_NAMESPACE}/exec" +EXEC_APPLY = f"{EXEC_NAMESPACE}/EXEC_APPLY" EXEC_CHECKPOINT = f"{EXEC_NAMESPACE}/EXEC_CHECKPOINT" EXEC_BRANCH = f"{EXEC_NAMESPACE}/EXEC_BRANCH" EXEC_BASELINE = f"{EXEC_NAMESPACE}/EXEC_BASELINE" @@ -81,6 +82,18 @@ class MultipleBranchError(DvcException): self.rev = rev +class ApplyConflictError(InvalidArgumentError): + def __init__(self, name): + super().__init__( + f"Experiment '{name}' cannot be applied to because your current " + "workspace contains changes which would be overwritten. Either " + "'git stash' your current changes before applying this " + "experiment, or re-run this command with '--force' to overwrite " + "your current changes." + ) + self.name = name + + class ExpRefInfo: namespace = EXPS_NAMESPACE @@ -94,6 +107,11 @@ class ExpRefInfo: def __str__(self): return "/".join(self.parts) + def __repr__(self): + baseline = f"'{self.baseline_sha}'" if self.baseline_sha else "None" + name = f"'{self.name}'" if self.name else "None" + return f"ExpRefInfo(baseline_sha={baseline}, name={name})" + @property def parts(self): return ( diff --git a/dvc/repo/experiments/executor/__init__.py b/dvc/repo/experiments/executor/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/dvc/repo/experiments/executor.py b/dvc/repo/experiments/executor/base.py similarity index 77% rename from dvc/repo/experiments/executor.py rename to dvc/repo/experiments/executor/base.py index cb0fe86cc..77b20a4a6 100644 --- a/dvc/repo/experiments/executor.py +++ b/dvc/repo/experiments/executor/base.py @@ -1,15 +1,20 @@ import logging import os import pickle -import sys +from abc import ABC, abstractmethod from functools import partial -from tempfile import TemporaryDirectory -from typing import TYPE_CHECKING, Callable, Iterable, Optional, Tuple, Union +from typing import ( + TYPE_CHECKING, + Callable, + Iterable, + NamedTuple, + Optional, + Union, +) from funcy import cached_property from dvc.path_info import PathInfo -from dvc.repo import Repo from dvc.repo.experiments.base import ( EXEC_BASELINE, EXEC_BRANCH, @@ -19,6 +24,8 @@ from dvc.repo.experiments.base import ( EXEC_NAMESPACE, EXPS_NAMESPACE, EXPS_STASH, + CheckpointExistsError, + ExperimentExistsError, ExpRefInfo, UnchangedExperimentError, ) @@ -36,7 +43,13 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) -class BaseExecutor: +class ExecutorResult(NamedTuple): + exp_hash: Optional[str] + ref_info: Optional["ExpRefInfo"] + force: bool + + +class BaseExecutor(ABC): """Base class for executing experiments in parallel. Args: @@ -97,8 +110,9 @@ class BaseExecutor: return SCM(self.root_dir) @property + @abstractmethod def git_url(self) -> str: - raise NotImplementedError + pass @property def dvc_dir(self) -> str: @@ -193,42 +207,50 @@ class BaseExecutor: @classmethod def reproduce( cls, - dvc_dir: str, - queue: "Queue", + dvc_dir: Optional[str], rev: str, + queue: Optional["Queue"] = None, rel_cwd: Optional[str] = None, name: Optional[str] = None, log_level: Optional[int] = None, - ) -> Tuple[Optional[str], bool]: + ) -> "ExecutorResult": """Run dvc repro and return the result. - Returns tuple of (exp_hash, force) where exp_hash is the experiment - hash (or None on error) and force is a bool specifying whether or - not this experiment should force overwrite any existing duplicates. + Returns tuple of (exp_hash, exp_ref, force) where exp_hash is the + experiment hash (or None on error), exp_ref is the experiment ref, + and force is a bool specifying whether or not this experiment + should force overwrite any existing duplicates. """ + from dvc.repo import Repo from dvc.repo.checkout import checkout as dvc_checkout from dvc.repo.reproduce import reproduce as dvc_reproduce unchanged = [] - queue.put((rev, os.getpid())) - cls._set_log_level(log_level) + if queue is not None: + queue.put((rev, os.getpid())) + if log_level is not None: + cls._set_log_level(log_level) def filter_pipeline(stages): unchanged.extend( [stage for stage in stages if isinstance(stage, PipelineStage)] ) - result: Optional[str] = None + exp_hash: Optional[str] = None + exp_ref: Optional["ExpRefInfo"] = None repro_force: bool = False try: dvc = Repo(dvc_dir) - old_cwd = os.getcwd() - if rel_cwd: - os.chdir(os.path.join(dvc.root_dir, rel_cwd)) + if dvc_dir is not None: + old_cwd = os.getcwd() + if rel_cwd: + os.chdir(os.path.join(dvc.root_dir, rel_cwd)) + else: + os.chdir(dvc.root_dir) else: - os.chdir(dvc.root_dir) + old_cwd = None logger.debug("Running repro in '%s'", os.getcwd()) args_path = os.path.join( @@ -242,6 +264,7 @@ class BaseExecutor: kwargs = {} repro_force = kwargs.get("force", False) + logger.debug("force = %s", str(repro_force)) # NOTE: for checkpoint experiments we handle persist outs slightly # differently than normal: @@ -256,7 +279,9 @@ class BaseExecutor: # experiment run dvc_checkout(dvc, force=True, quiet=True) - checkpoint_func = partial(cls.checkpoint_callback, dvc.scm, name) + checkpoint_func = partial( + cls.checkpoint_callback, dvc.scm, name, repro_force + ) stages = dvc_reproduce( dvc, *args, @@ -266,12 +291,19 @@ class BaseExecutor: ) exp_hash = cls.hash_exp(stages) - result = exp_hash - exp_rev = cls.commit(dvc.scm, exp_hash, exp_name=name) - if dvc.scm.get_ref(EXEC_CHECKPOINT): - dvc.scm.set_ref(EXEC_CHECKPOINT, exp_rev) - except UnchangedExperimentError: - pass + try: + cls.commit( + dvc.scm, + exp_hash, + exp_name=name, + force=repro_force, + checkpoint=any(stage.is_checkpoint for stage in stages), + ) + except UnchangedExperimentError: + pass + ref = dvc.scm.get_ref(EXEC_BRANCH, follow=False) + if ref: + exp_ref = ExpRefInfo.from_ref(ref) finally: if dvc: dvc.scm.close() @@ -281,26 +313,35 @@ class BaseExecutor: # ideally we would return stages here like a normal repro() call, but # stages is not currently picklable and cannot be returned across # multiprocessing calls - return result, repro_force + return ExecutorResult(exp_hash, exp_ref, repro_force) @classmethod def checkpoint_callback( cls, scm: "Git", name: Optional[str], + force: bool, unchanged: Iterable["PipelineStage"], stages: Iterable["PipelineStage"], ): try: exp_hash = cls.hash_exp(list(stages) + list(unchanged)) - exp_rev = cls.commit(scm, exp_hash, exp_name=name) - scm.set_ref(EXEC_CHECKPOINT, exp_rev) + exp_rev = cls.commit( + scm, exp_hash, exp_name=name, force=force, checkpoint=True + ) logger.info("Checkpoint experiment iteration '%s'.", exp_rev[:7]) except UnchangedExperimentError: pass @classmethod - def commit(cls, scm: "Git", exp_hash: str, exp_name: Optional[str] = None): + def commit( + cls, + scm: "Git", + exp_hash: str, + exp_name: Optional[str] = None, + force: bool = False, + checkpoint: bool = False, + ): """Commit stages as an experiment and return the commit SHA.""" rev = scm.get_rev() if not scm.is_dirty(untracked_files=True): @@ -314,8 +355,13 @@ class BaseExecutor: else: baseline_rev = scm.get_ref(EXEC_BASELINE) name = exp_name if exp_name else f"exp-{exp_hash[:5]}" - branch = str(ExpRefInfo(baseline_rev, name)) + ref_info = ExpRefInfo(baseline_rev, name) + branch = str(ref_info) old_ref = None + if not force and scm.get_ref(branch): + if checkpoint: + raise CheckpointExistsError(ref_info.name) + raise ExperimentExistsError(ref_info.name) logger.debug("Commit to new experiment branch '%s'", branch) scm.gitpython.repo.git.add(update=True) @@ -323,6 +369,8 @@ class BaseExecutor: new_rev = scm.get_rev() scm.set_ref(branch, new_rev, old_ref=old_ref) scm.set_ref(EXEC_BRANCH, branch, symbolic=True) + if checkpoint: + scm.set_ref(EXEC_CHECKPOINT, new_rev) return new_rev @staticmethod @@ -336,48 +384,3 @@ class BaseExecutor: disable_other_loggers() if level is not None: dvc_logger.setLevel(level) - - -class LocalExecutor(BaseExecutor): - """Local machine experiment executor.""" - - def __init__( - self, - *args, - tmp_dir: Optional[str] = None, - cache_dir: Optional[str] = None, - **kwargs, - ): - self._tmp_dir = TemporaryDirectory(dir=tmp_dir) - kwargs["root_dir"] = self._tmp_dir.name - super().__init__(*args, **kwargs) - if cache_dir: - self._config(cache_dir) - logger.debug( - "Init local executor in dir '%s'", self._tmp_dir, - ) - - def _config(self, cache_dir): - local_config = os.path.join(self.dvc_dir, "config.local") - logger.debug("Writing experiments local config '%s'", local_config) - with open(local_config, "w") as fobj: - fobj.write(f"[cache]\n dir = {cache_dir}") - - @property - def git_url(self) -> str: - root_dir = os.path.abspath(self.root_dir) - if os.name == "nt": - root_dir = root_dir.replace(os.sep, "/") - return f"file://{root_dir}" - - def cleanup(self): - super().cleanup() - logger.debug("Removing tmpdir '%s'", self._tmp_dir) - try: - self._tmp_dir.cleanup() - except PermissionError: - if os.name == "nt" and sys.version_info < (3, 8): - # see https://bugs.python.org/issue26660 - remove(self._tmp_dir.name) - return - raise diff --git a/dvc/repo/experiments/executor/local.py b/dvc/repo/experiments/executor/local.py new file mode 100644 index 000000000..df90159fb --- /dev/null +++ b/dvc/repo/experiments/executor/local.py @@ -0,0 +1,60 @@ +import logging +import os +import sys +from tempfile import TemporaryDirectory +from typing import Optional + +from dvc.utils.fs import remove + +from .base import BaseExecutor + +logger = logging.getLogger(__name__) + + +class BaseLocalExecutor(BaseExecutor): + """Base local machine executor.""" + + @property + def git_url(self) -> str: + root_dir = os.path.abspath(self.root_dir) + if os.name == "nt": + root_dir = root_dir.replace(os.sep, "/") + return f"file://{root_dir}" + + +class TempDirExecutor(BaseLocalExecutor): + """Temp directory experiment executor.""" + + def __init__( + self, + *args, + tmp_dir: Optional[str] = None, + cache_dir: Optional[str] = None, + **kwargs, + ): + self._tmp_dir = TemporaryDirectory(dir=tmp_dir) + kwargs["root_dir"] = self._tmp_dir.name + super().__init__(*args, **kwargs) + if cache_dir: + self._config(cache_dir) + logger.debug( + "Init temp dir executor in dir '%s'", self._tmp_dir, + ) + + def _config(self, cache_dir): + local_config = os.path.join(self.dvc_dir, "config.local") + logger.debug("Writing experiments local config '%s'", local_config) + with open(local_config, "w") as fobj: + fobj.write(f"[cache]\n dir = {cache_dir}") + + def cleanup(self): + super().cleanup() + logger.debug("Removing tmpdir '%s'", self._tmp_dir) + try: + self._tmp_dir.cleanup() + except PermissionError: + if os.name == "nt" and sys.version_info < (3, 8): + # see https://bugs.python.org/issue26660 + remove(self._tmp_dir.name) + return + raise diff --git a/dvc/repo/experiments/run.py b/dvc/repo/experiments/run.py index 54b43bc1f..0a26a8b5c 100644 --- a/dvc/repo/experiments/run.py +++ b/dvc/repo/experiments/run.py @@ -3,7 +3,6 @@ from typing import Iterable, Optional from dvc.exceptions import InvalidArgumentError from dvc.repo import locked -from dvc.repo.experiments.base import UnchangedExperimentError logger = logging.getLogger(__name__) @@ -41,6 +40,7 @@ def run( params: Optional[Iterable] = None, run_all: Optional[bool] = False, jobs: Optional[int] = 1, + tmp_dir: Optional[bool] = False, **kwargs, ) -> dict: """Reproduce the specified targets as an experiment. @@ -57,12 +57,6 @@ def run( params = _parse_params(params) else: params = [] - try: - return repo.experiments.reproduce_one( - targets=targets, params=params, **kwargs - ) - except UnchangedExperimentError: - # If experiment contains no changes, just run regular repro - kwargs.pop("queue", None) - kwargs.pop("checkpoint_resume", None) - return {None: repo.reproduce(targets=targets, **kwargs)} + return repo.experiments.reproduce_one( + targets=targets, params=params, tmp_dir=tmp_dir, **kwargs + ) diff --git a/dvc/scm/git/__init__.py b/dvc/scm/git/__init__.py index f35c83ccb..7373184f7 100644 --- a/dvc/scm/git/__init__.py +++ b/dvc/scm/git/__init__.py @@ -380,9 +380,19 @@ class Git(Base): finally: prefix = self.LOCAL_BRANCH_PREFIX if orig_head.startswith(prefix): - orig_head = orig_head[len(prefix) :] - logger.debug("Restore HEAD to '%s'", orig_head) - self.checkout(orig_head) + symbolic = True + name = orig_head[len(prefix) :] + else: + symbolic = False + name = orig_head + self.set_ref( + "HEAD", + orig_head, + symbolic=symbolic, + message=f"dvc: Restore HEAD to '{name}'", + ) + logger.debug("Restore HEAD to '%s'", name) + self.reset() @contextmanager def stash_workspace(self, **kwargs): diff --git a/dvc/scm/git/backend/dulwich.py b/dvc/scm/git/backend/dulwich.py index f721825ef..a8495d698 100644 --- a/dvc/scm/git/backend/dulwich.py +++ b/dvc/scm/git/backend/dulwich.py @@ -233,7 +233,7 @@ class DulwichBackend(BaseGitBackend): # pylint:disable=abstract-method message_b = message.encode("utf-8") if message else None if symbolic: return self.repo.refs.set_symbolic_ref( - name_b, new_ref_b, message=message + name_b, new_ref_b, message=message_b ) if not self.repo.refs.set_if_equals( name_b, old_ref_b, new_ref_b, message=message_b diff --git a/tests/func/experiments/conftest.py b/tests/func/experiments/conftest.py index 62d1f1d24..607c3e595 100644 --- a/tests/func/experiments/conftest.py +++ b/tests/func/experiments/conftest.py @@ -51,7 +51,16 @@ def exp_stage(tmp_dir, scm, dvc): params=["foo"], name="copy-file", ) - scm.add(["dvc.yaml", "dvc.lock", "copy.py", "params.yaml", "metrics.yaml"]) + scm.add( + [ + "dvc.yaml", + "dvc.lock", + "copy.py", + "params.yaml", + "metrics.yaml", + ".gitignore", + ] + ) scm.commit("init") return stage @@ -68,6 +77,6 @@ def checkpoint_stage(tmp_dir, scm, dvc): no_exec=True, name="checkpoint-file", ) - scm.add(["dvc.yaml", "checkpoint.py", "params.yaml"]) + scm.add(["dvc.yaml", "checkpoint.py", "params.yaml", ".gitignore"]) scm.commit("init") return stage diff --git a/tests/func/experiments/test_checkpoints.py b/tests/func/experiments/test_checkpoints.py index 366bc92e1..86dd92562 100644 --- a/tests/func/experiments/test_checkpoints.py +++ b/tests/func/experiments/test_checkpoints.py @@ -1,14 +1,20 @@ +import logging + import pytest from funcy import first from dvc.exceptions import DvcException from dvc.repo.experiments import Experiments, MultipleBranchError +from dvc.repo.experiments.base import EXEC_APPLY, EXEC_CHECKPOINT -def test_new_checkpoint(tmp_dir, scm, dvc, checkpoint_stage, mocker): +@pytest.mark.parametrize("workspace", [True, False]) +def test_new_checkpoint( + tmp_dir, scm, dvc, checkpoint_stage, mocker, workspace +): new_mock = mocker.spy(dvc.experiments, "new") results = dvc.experiments.run( - checkpoint_stage.addressing, params=["foo=2"] + checkpoint_stage.addressing, params=["foo=2"], tmp_dir=not workspace ) exp = first(results) @@ -22,33 +28,53 @@ def test_new_checkpoint(tmp_dir, scm, dvc, checkpoint_stage, mocker): with tree.open(tmp_dir / "metrics.yaml") as fobj: assert fobj.read().strip() == "foo: 2" + if workspace: + assert scm.get_ref(EXEC_APPLY) == exp + assert scm.get_ref(EXEC_CHECKPOINT) == exp + if workspace: + assert (tmp_dir / "foo").read_text().strip() == "5" + assert (tmp_dir / "metrics.yaml").read_text().strip() == "foo: 2" + @pytest.mark.parametrize( - "checkpoint_resume", [Experiments.LAST_CHECKPOINT, "foo"] + "checkpoint_resume, workspace", + [ + (Experiments.LAST_CHECKPOINT, True), + (Experiments.LAST_CHECKPOINT, False), + ("foo", True), + ("foo", False), + ], ) def test_resume_checkpoint( - tmp_dir, scm, dvc, checkpoint_stage, checkpoint_resume + tmp_dir, scm, dvc, checkpoint_stage, checkpoint_resume, workspace ): with pytest.raises(DvcException): dvc.experiments.run( checkpoint_stage=checkpoint_stage.addressing, checkpoint_resume=checkpoint_resume, + tmp_dir=not workspace, ) results = dvc.experiments.run( - checkpoint_stage.addressing, params=["foo=2"] + checkpoint_stage.addressing, params=["foo=2"], tmp_dir=not workspace ) with pytest.raises(DvcException): dvc.experiments.run( - checkpoint_stage.addressing, checkpoint_resume="abc1234", + checkpoint_stage.addressing, + checkpoint_resume="abc1234", + tmp_dir=not workspace, ) if checkpoint_resume != Experiments.LAST_CHECKPOINT: checkpoint_resume = first(results) + if not workspace: + dvc.experiments.apply(first(results)) results = dvc.experiments.run( - checkpoint_stage.addressing, checkpoint_resume=checkpoint_resume + checkpoint_stage.addressing, + checkpoint_resume=checkpoint_resume, + tmp_dir=not workspace, ) exp = first(results) @@ -61,21 +87,52 @@ def test_resume_checkpoint( with tree.open(tmp_dir / "metrics.yaml") as fobj: assert fobj.read().strip() == "foo: 2" + if workspace: + assert scm.get_ref(EXEC_APPLY) == exp + assert scm.get_ref(EXEC_CHECKPOINT) == exp -def test_reset_checkpoint(tmp_dir, scm, dvc, checkpoint_stage, caplog): + +@pytest.mark.parametrize("workspace", [True, False]) +def test_reset_checkpoint( + tmp_dir, scm, dvc, checkpoint_stage, caplog, workspace +): from dvc.repo.experiments.base import CheckpointExistsError - dvc.experiments.run(checkpoint_stage.addressing, name="foo") - scm.gitpython.repo.git.reset(hard=True) - scm.gitpython.repo.git.clean(force=True) + dvc.experiments.run( + checkpoint_stage.addressing, name="foo", tmp_dir=not workspace, + ) - with pytest.raises(CheckpointExistsError): - dvc.experiments.run( - checkpoint_stage.addressing, name="foo", params=["foo=2"] - ) + if workspace: + scm.reset(hard=True) + scm.gitpython.repo.git.clean(force=True) + + if workspace: + with caplog.at_level(logging.ERROR): + dvc.experiments.run( + checkpoint_stage.addressing, + name="foo", + params=["foo=2"], + tmp_dir=not workspace, + ) + assert "checkpoint experiment conflicts with existing" in caplog.text + + scm.reset(hard=True) + scm.gitpython.repo.git.clean(force=True) + else: + with pytest.raises(CheckpointExistsError): + dvc.experiments.run( + checkpoint_stage.addressing, + name="foo", + params=["foo=2"], + tmp_dir=not workspace, + ) results = dvc.experiments.run( - checkpoint_stage.addressing, params=["foo=2"], name="foo", force=True + checkpoint_stage.addressing, + params=["foo=2"], + name="foo", + force=True, + tmp_dir=not workspace, ) exp = first(results) @@ -88,22 +145,43 @@ def test_reset_checkpoint(tmp_dir, scm, dvc, checkpoint_stage, caplog): with tree.open(tmp_dir / "metrics.yaml") as fobj: assert fobj.read().strip() == "foo: 2" + if workspace: + assert scm.get_ref(EXEC_APPLY) == exp + assert scm.get_ref(EXEC_CHECKPOINT) == exp -def test_resume_branch(tmp_dir, scm, dvc, checkpoint_stage): + +@pytest.mark.parametrize("workspace", [True, False]) +def test_resume_branch(tmp_dir, scm, dvc, checkpoint_stage, workspace): results = dvc.experiments.run( - checkpoint_stage.addressing, params=["foo=2"] + checkpoint_stage.addressing, params=["foo=2"], tmp_dir=not workspace ) branch_rev = first(results) - - results = dvc.experiments.run( - checkpoint_stage.addressing, checkpoint_resume=branch_rev - ) - checkpoint_a = first(results) + if not workspace: + dvc.experiments.apply(branch_rev) + results = dvc.experiments.run( + checkpoint_stage.addressing, + checkpoint_resume=branch_rev, + tmp_dir=not workspace, + ) + checkpoint_a = first(results) + if not workspace: + dvc.experiments.apply(checkpoint_a, force=True) + + with pytest.raises(DvcException): + results = dvc.experiments.run( + checkpoint_stage.addressing, + checkpoint_resume=branch_rev, + params=["foo=100"], + tmp_dir=not workspace, + ) + + dvc.experiments.apply(branch_rev, force=True) results = dvc.experiments.run( checkpoint_stage.addressing, checkpoint_resume=branch_rev, params=["foo=100"], + tmp_dir=not workspace, ) checkpoint_b = first(results) diff --git a/tests/func/experiments/test_experiments.py b/tests/func/experiments/test_experiments.py index e0525e094..c704e6366 100644 --- a/tests/func/experiments/test_experiments.py +++ b/tests/func/experiments/test_experiments.py @@ -11,13 +11,18 @@ from dvc.utils.serialize import PythonFileCorruptedError from tests.func.test_repro_multistage import COPY_SCRIPT -@pytest.mark.parametrize("name", [None, "foo"]) -def test_new_simple(tmp_dir, scm, dvc, exp_stage, mocker, name): +@pytest.mark.parametrize( + "name,workspace", + [(None, True), (None, False), ("foo", True), ("foo", False)], +) +def test_new_simple(tmp_dir, scm, dvc, exp_stage, mocker, name, workspace): baseline = scm.get_rev() tmp_dir.gen("params.yaml", "foo: 2") new_mock = mocker.spy(dvc.experiments, "new") - results = dvc.experiments.run(exp_stage.addressing, name=name) + results = dvc.experiments.run( + exp_stage.addressing, name=name, tmp_dir=not workspace + ) exp = first(results) ref_info = first(exp_refs_by_rev(scm, exp)) assert ref_info and ref_info.baseline_sha == baseline @@ -27,21 +32,39 @@ def test_new_simple(tmp_dir, scm, dvc, exp_stage, mocker, name): with tree.open(tmp_dir / "metrics.yaml") as fobj: assert fobj.read().strip() == "foo: 2" + if workspace: + assert (tmp_dir / "metrics.yaml").read_text().strip() == "foo: 2" + exp_name = name if name else ref_info.name assert dvc.experiments.get_exact_name(exp) == exp_name assert scm.resolve_rev(exp_name) == exp -def test_experiment_exists(tmp_dir, scm, dvc, exp_stage, mocker): +@pytest.mark.parametrize("workspace", [True, False]) +def test_experiment_exists(tmp_dir, scm, dvc, exp_stage, mocker, workspace): from dvc.repo.experiments.base import ExperimentExistsError - dvc.experiments.run(exp_stage.addressing, name="foo", params=["foo=2"]) + dvc.experiments.run( + exp_stage.addressing, + name="foo", + params=["foo=2"], + tmp_dir=not workspace, + ) with pytest.raises(ExperimentExistsError): - dvc.experiments.run(exp_stage.addressing, name="foo", params=["foo=3"]) + dvc.experiments.run( + exp_stage.addressing, + name="foo", + params=["foo=3"], + tmp_dir=not workspace, + ) results = dvc.experiments.run( - exp_stage.addressing, name="foo", params=["foo=3"], force=True + exp_stage.addressing, + name="foo", + params=["foo=3"], + force=True, + tmp_dir=not workspace, ) exp = first(results) @@ -72,7 +95,7 @@ def test_failed_exp(tmp_dir, scm, dvc, exp_stage, mocker, caplog): return_value=ReproductionError(exp_stage.relpath), ) with caplog.at_level(logging.ERROR): - dvc.experiments.run(exp_stage.addressing) + dvc.experiments.run(exp_stage.addressing, tmp_dir=True) assert "Failed to reproduce experiment" in caplog.text @@ -127,15 +150,16 @@ def test_modify_params(tmp_dir, scm, dvc, mocker, changes, expected): @pytest.mark.parametrize("queue", [True, False]) def test_apply(tmp_dir, scm, dvc, exp_stage, queue): from dvc.exceptions import InvalidArgumentError + from dvc.repo.experiments.base import ApplyConflictError metrics_original = (tmp_dir / "metrics.yaml").read_text().strip() results = dvc.experiments.run( - exp_stage.addressing, params=["foo=2"], queue=queue + exp_stage.addressing, params=["foo=2"], queue=queue, tmp_dir=True ) exp_a = first(results) results = dvc.experiments.run( - exp_stage.addressing, params=["foo=3"], queue=queue + exp_stage.addressing, params=["foo=3"], queue=queue, tmp_dir=True ) exp_b = first(results) @@ -150,7 +174,17 @@ def test_apply(tmp_dir, scm, dvc, exp_stage, queue): else "foo: 2" ) - dvc.experiments.apply(exp_b) + with pytest.raises(ApplyConflictError): + dvc.experiments.apply(exp_b) + # failed apply should revert everything to prior state + assert (tmp_dir / "params.yaml").read_text().strip() == "foo: 2" + assert ( + (tmp_dir / "metrics.yaml").read_text().strip() == metrics_original + if queue + else "foo: 2" + ) + + dvc.experiments.apply(exp_b, force=True) assert (tmp_dir / "params.yaml").read_text().strip() == "foo: 3" assert ( (tmp_dir / "metrics.yaml").read_text().strip() == metrics_original @@ -172,7 +206,6 @@ def test_get_baseline(tmp_dir, scm, dvc, exp_stage): dvc.experiments.run(exp_stage.addressing, params=["foo=3"], queue=True) assert dvc.experiments.get_baseline(f"{EXPS_STASH}@{{0}}") == init_rev - dvc.experiments.apply(exp_rev) scm.add(["dvc.yaml", "dvc.lock", "copy.py", "params.yaml", "metrics.yaml"]) scm.commit("promote exp") promote_rev = scm.get_rev() @@ -200,7 +233,9 @@ def test_update_py_params(tmp_dir, scm, dvc): scm.add(["dvc.yaml", "dvc.lock", "copy.py", "params.py", "metrics.py"]) scm.commit("init") - results = dvc.experiments.run(stage.addressing, params=["params.py:INT=2"]) + results = dvc.experiments.run( + stage.addressing, params=["params.py:INT=2"], tmp_dir=True + ) exp_a = first(results) tree = scm.get_tree(exp_a) @@ -227,6 +262,7 @@ def test_update_py_params(tmp_dir, scm, dvc): results = dvc.experiments.run( stage.addressing, params=["params.py:FLOAT=0.1,Train.seed=2121,Klass.a=222"], + tmp_dir=True, ) exp_a = first(results) @@ -261,7 +297,9 @@ def test_update_py_params(tmp_dir, scm, dvc): scm.commit("init") with pytest.raises(PythonFileCorruptedError): - dvc.experiments.run(stage.addressing, params=["params.py:INT=2a"]) + dvc.experiments.run( + stage.addressing, params=["params.py:INT=2a"], tmp_dir=True + ) def test_detached_parent(tmp_dir, scm, dvc, exp_stage, mocker): @@ -278,8 +316,6 @@ def test_detached_parent(tmp_dir, scm, dvc, exp_stage, mocker): exp_rev = first(results) assert dvc.experiments.get_baseline(exp_rev) == detached_rev - - dvc.experiments.apply(exp_rev) assert (tmp_dir / "params.yaml").read_text().strip() == "foo: 3" @@ -328,23 +364,19 @@ def test_no_scm(tmp_dir): dvc = DvcRepo.init(no_scm=True) - with pytest.raises(NoSCMError): - dvc.experiments.run() - - with pytest.raises(NoSCMError): - dvc.experiments.apply() - - with pytest.raises(NoSCMError): - dvc.experiments.branch() - - with pytest.raises(NoSCMError): - dvc.experiments.diff() - - with pytest.raises(NoSCMError): - dvc.experiments.show() - - with pytest.raises(NoSCMError): - dvc.experiments.gc() + for cmd in [ + "apply", + "branch", + "diff", + "show", + "run", + "gc", + "push", + "pull", + "ls", + ]: + with pytest.raises(NoSCMError): + getattr(dvc.experiments, cmd)() def test_untracked(tmp_dir, scm, dvc, caplog): @@ -361,13 +393,17 @@ def test_untracked(tmp_dir, scm, dvc, caplog): # copy.py is untracked with caplog.at_level(logging.ERROR): - results = dvc.experiments.run(stage.addressing, params=["foo=2"]) + results = dvc.experiments.run( + stage.addressing, params=["foo=2"], tmp_dir=True + ) assert "Failed to reproduce experiment" in caplog.text assert not results # copy.py is staged as new file but not committed scm.add(["copy.py"]) - results = dvc.experiments.run(stage.addressing, params=["foo=2"]) + results = dvc.experiments.run( + stage.addressing, params=["foo=2"], tmp_dir=True + ) exp = first(results) tree = scm.get_tree(exp) assert tree.exists("copy.py") @@ -375,7 +411,8 @@ def test_untracked(tmp_dir, scm, dvc, caplog): assert fobj.read().strip() == "foo: 2" -def test_dirty_lockfile(tmp_dir, scm, dvc, exp_stage): +@pytest.mark.parametrize("workspace", [True, False]) +def test_dirty_lockfile(tmp_dir, scm, dvc, exp_stage, workspace): from dvc.dvcfile import LockfileCorruptedError tmp_dir.gen("dvc.lock", "foo") @@ -383,18 +420,21 @@ def test_dirty_lockfile(tmp_dir, scm, dvc, exp_stage): with pytest.raises(LockfileCorruptedError): dvc.reproduce(exp_stage.addressing) - results = dvc.experiments.run(exp_stage.addressing, params=["foo=2"]) + results = dvc.experiments.run( + exp_stage.addressing, params=["foo=2"], tmp_dir=not workspace + ) exp = first(results) tree = scm.get_tree(exp) with tree.open(tmp_dir / "metrics.yaml") as fobj: assert fobj.read().strip() == "foo: 2" - assert (tmp_dir / "dvc.lock").read_text() == "foo" + if not workspace: + assert (tmp_dir / "dvc.lock").read_text() == "foo" def test_packed_args_exists(tmp_dir, scm, dvc, exp_stage, caplog): - from dvc.repo.experiments.executor import BaseExecutor + from dvc.repo.experiments.executor.base import BaseExecutor tmp_dir.scm_gen( tmp_dir / ".dvc" / "tmp" / BaseExecutor.PACKED_ARGS_FILE, @@ -439,7 +479,8 @@ def test_list(tmp_dir, scm, dvc, exp_stage): } -def test_subdir(tmp_dir, scm, dvc): +@pytest.mark.parametrize("workspace", [True, False]) +def test_subdir(tmp_dir, scm, dvc, workspace): subdir = tmp_dir / "dir" subdir.gen("copy.py", COPY_SCRIPT) subdir.gen("params.yaml", "foo: 1") @@ -457,7 +498,9 @@ def test_subdir(tmp_dir, scm, dvc): ) scm.commit("init") - results = dvc.experiments.run(PIPELINE_FILE, params=["foo=2"]) + results = dvc.experiments.run( + PIPELINE_FILE, params=["foo=2"], tmp_dir=not workspace + ) assert results exp = first(results) @@ -473,7 +516,8 @@ def test_subdir(tmp_dir, scm, dvc): assert scm.resolve_rev(ref_info.name) == exp -def test_subrepo(tmp_dir, scm): +@pytest.mark.parametrize("workspace", [True, False]) +def test_subrepo(tmp_dir, scm, workspace): from tests.unit.tree.test_repo import make_subrepo subrepo = tmp_dir / "dir" / "repo" @@ -499,7 +543,9 @@ def test_subrepo(tmp_dir, scm): ) scm.commit("init") - results = subrepo.dvc.experiments.run(PIPELINE_FILE, params=["foo=2"]) + results = subrepo.dvc.experiments.run( + PIPELINE_FILE, params=["foo=2"], tmp_dir=not workspace + ) assert results exp = first(results) @@ -513,3 +559,22 @@ def test_subrepo(tmp_dir, scm): assert subrepo.dvc.experiments.get_exact_name(exp) == ref_info.name assert scm.resolve_rev(ref_info.name) == exp + + +def test_queue(tmp_dir, scm, dvc, exp_stage, mocker): + dvc.experiments.run(exp_stage.addressing, params=["foo=2"], queue=True) + dvc.experiments.run(exp_stage.addressing, params=["foo=3"], queue=True) + assert len(dvc.experiments.stash_revs) == 2 + + repro_mock = mocker.spy(dvc.experiments, "_reproduce_revs") + results = dvc.experiments.run(run_all=True) + assert len(results) == 2 + repro_mock.assert_called_with(jobs=1) + + expected = {"foo: 2", "foo: 3"} + metrics = set() + for exp in results: + tree = scm.get_tree(exp) + with tree.open(tmp_dir / "metrics.yaml") as fobj: + metrics.add(fobj.read().strip()) + assert expected == metrics diff --git a/tests/func/experiments/test_gc.py b/tests/func/experiments/test_gc.py index e5f9069ca..7a48a03e2 100644 --- a/tests/func/experiments/test_gc.py +++ b/tests/func/experiments/test_gc.py @@ -43,7 +43,7 @@ def test_all_commits(tmp_dir, scm, dvc, queued, expected): exp_rev = first(results) dvc.experiments.run(stage.addressing, params=["foo=3"], queue=True) - dvc.experiments.apply(exp_rev) + dvc.experiments.apply(exp_rev, force=True) scm.add(["dvc.yaml", "dvc.lock", "copy.py", "params.yaml", "metrics.yaml"]) scm.commit("v2") @@ -72,7 +72,7 @@ def test_all_branches(tmp_dir, scm, dvc, queued, expected): exp_rev = first(results) dvc.experiments.run(stage.addressing, params=["foo=3"], queue=True) - dvc.experiments.apply(exp_rev) + dvc.experiments.apply(exp_rev, force=True) scm.add(["dvc.yaml", "dvc.lock", "copy.py", "params.yaml", "metrics.yaml"]) scm.commit("v2") @@ -82,7 +82,7 @@ def test_all_branches(tmp_dir, scm, dvc, queued, expected): dvc.experiments.run(stage.addressing, params=["foo=5"], queue=True) exp_rev = first(results) - dvc.experiments.apply(exp_rev) + dvc.experiments.apply(exp_rev, force=True) scm.add(["dvc.yaml", "dvc.lock", "copy.py", "params.yaml", "metrics.yaml"]) scm.commit("v3") @@ -108,7 +108,7 @@ def test_all_tags(tmp_dir, scm, dvc, queued, expected): exp_rev = first(results) dvc.experiments.run(stage.addressing, params=["foo=3"], queue=True) - dvc.experiments.apply(exp_rev) + dvc.experiments.apply(exp_rev, force=True) scm.add(["dvc.yaml", "dvc.lock", "copy.py", "params.yaml", "metrics.yaml"]) scm.commit("v2") scm.tag("tag-v2") diff --git a/tests/func/experiments/test_show.py b/tests/func/experiments/test_show.py index cd2b3a895..915054311 100644 --- a/tests/func/experiments/test_show.py +++ b/tests/func/experiments/test_show.py @@ -1,6 +1,7 @@ import os from datetime import datetime +import pytest from funcy import first from dvc.main import main @@ -18,13 +19,16 @@ def test_show_simple(tmp_dir, scm, dvc, exp_stage): } -def test_show_experiment(tmp_dir, scm, dvc, exp_stage): +@pytest.mark.parametrize("workspace", [True, False]) +def test_show_experiment(tmp_dir, scm, dvc, exp_stage, workspace): baseline_rev = scm.get_rev() timestamp = datetime.fromtimestamp( scm.gitpython.repo.rev_parse(baseline_rev).committed_date ) - dvc.experiments.run(exp_stage.addressing, params=["foo=2"]) + dvc.experiments.run( + exp_stage.addressing, params=["foo=2"], tmp_dir=not workspace + ) results = dvc.experiments.show() expected_baseline = { @@ -77,10 +81,13 @@ def test_show_queued(tmp_dir, scm, dvc, exp_stage): assert exp["params"]["params.yaml"] == {"foo": 3} -def test_show_checkpoint(tmp_dir, scm, dvc, checkpoint_stage, capsys): +@pytest.mark.parametrize("workspace", [True, False]) +def test_show_checkpoint( + tmp_dir, scm, dvc, checkpoint_stage, capsys, workspace +): baseline_rev = scm.get_rev() results = dvc.experiments.run( - checkpoint_stage.addressing, params=["foo=2"] + checkpoint_stage.addressing, params=["foo=2"], tmp_dir=not workspace ) exp_rev = first(results) @@ -110,21 +117,30 @@ def test_show_checkpoint(tmp_dir, scm, dvc, checkpoint_stage, capsys): assert f"{tree} {name}" in cap.out -def test_show_checkpoint_branch(tmp_dir, scm, dvc, checkpoint_stage, capsys): +@pytest.mark.parametrize("workspace", [True, False]) +def test_show_checkpoint_branch( + tmp_dir, scm, dvc, checkpoint_stage, capsys, workspace +): results = dvc.experiments.run( - checkpoint_stage.addressing, params=["foo=2"] + checkpoint_stage.addressing, params=["foo=2"], tmp_dir=not workspace ) branch_rev = first(results) - - results = dvc.experiments.run( - checkpoint_stage.addressing, checkpoint_resume=branch_rev - ) - checkpoint_a = first(results) + if not workspace: + dvc.experiments.apply(branch_rev) + results = dvc.experiments.run( + checkpoint_stage.addressing, + checkpoint_resume=branch_rev, + tmp_dir=not workspace, + ) + checkpoint_a = first(results) + + dvc.experiments.apply(branch_rev, force=True) results = dvc.experiments.run( checkpoint_stage.addressing, checkpoint_resume=branch_rev, params=["foo=100"], + tmp_dir=not workspace, ) checkpoint_b = first(results) diff --git a/tests/unit/command/test_experiments.py b/tests/unit/command/test_experiments.py index 59aed9cbf..1311873f5 100644 --- a/tests/unit/command/test_experiments.py +++ b/tests/unit/command/test_experiments.py @@ -18,7 +18,7 @@ from .test_repro import default_arguments as repro_arguments def test_experiments_apply(dvc, scm, mocker): - cli_args = parse_args(["experiments", "apply", "exp_rev"]) + cli_args = parse_args(["experiments", "apply", "--force", "exp_rev"]) assert cli_args.func == CmdExperimentsApply cmd = cli_args.func(cli_args) @@ -26,7 +26,7 @@ def test_experiments_apply(dvc, scm, mocker): assert cmd.run() == 0 - m.assert_called_once_with(cmd.repo, "exp_rev") + m.assert_called_once_with(cmd.repo, "exp_rev", force=True) def test_experiments_diff(dvc, scm, mocker): @@ -94,6 +94,7 @@ def test_experiments_run(dvc, scm, mocker, args, resume): "run_all": False, "jobs": None, "checkpoint_resume": resume, + "tmp_dir": False, } default_arguments.update(repro_arguments)