experiments: auto push experiments (#10323)

* add auto_push for experiment on run and on save.
add config.exp.auto_push and config.exp.git_remote

* add func tests

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* separate auto_push behaviour out of commit

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Saugat Pachhai (सौगात) <[email protected]>

dvc/config_schema.py

@@ -336,6 +336,8 @@ SCHEMA = {
         "params": str,
         "plots": str,
         "live": str,
+        "auto_push": Bool,
+        "git_remote": str,
     },
     "parsing": {
         "bool": All(Lower, Choices("store_true", "boolean_optional")),

dvc/repo/experiments/executor/base.py

@@ -297,13 +297,16 @@ class BaseExecutor(ABC):
             exp_hash = cls.hash_exp(stages)
             if include_untracked:
                 dvc.scm.add(include_untracked, force=True)  # type: ignore[call-arg]
-            cls.commit(
-                dvc.scm,  # type: ignore[arg-type]
-                exp_hash,
-                exp_name=info.name,
-                force=force,
-                message=message,
-            )
+
+            with cls.auto_push(dvc):
+                cls.commit(
+                    dvc.scm,  # type: ignore[arg-type]
+                    exp_hash,
+                    exp_name=info.name,
+                    force=force,
+                    message=message,
+                )
+
             ref: Optional[str] = dvc.scm.get_ref(EXEC_BRANCH, follow=False)
             exp_ref = ExpRefInfo.from_ref(ref) if ref else None
             untracked = dvc.scm.untracked_files()
@@ -460,9 +463,6 @@ class BaseExecutor(ABC):
         from dvc.repo.checkout import checkout as dvc_checkout
         from dvc.ui import ui
 
-        auto_push = env2bool(DVC_EXP_AUTO_PUSH)
-        git_remote = os.getenv(DVC_EXP_GIT_REMOTE, None)
-
         if queue is not None:
             queue.put((rev, os.getpid()))
         if log_errors and log_level is not None:
@@ -483,9 +483,6 @@ class BaseExecutor(ABC):
             message=message,
             **kwargs,
         ) as dvc:
-            if auto_push:
-                cls._validate_remotes(dvc, git_remote)
-
             args, kwargs = cls._repro_args(dvc)
             if args:
                 targets: Optional[Union[list, str]] = args[0]
@@ -519,8 +516,6 @@ class BaseExecutor(ABC):
                     dvc,
                     info,
                     exp_hash,
-                    auto_push,
-                    git_remote,
                     repro_force,
                     message=message,
                 )
@@ -550,20 +545,18 @@ class BaseExecutor(ABC):
         dvc,
         info,
         exp_hash,
-        auto_push,
-        git_remote,
         repro_force,
         message: Optional[str] = None,
     ) -> tuple[Optional[str], Optional["ExpRefInfo"], bool]:
-        cls.commit(
-            dvc.scm,
-            exp_hash,
-            exp_name=info.name,
-            force=repro_force,
-            message=message,
-        )
-        if auto_push:
-            cls._auto_push(dvc, dvc.scm, git_remote)
+        with cls.auto_push(dvc):
+            cls.commit(
+                dvc.scm,
+                exp_hash,
+                exp_name=info.name,
+                force=repro_force,
+                message=message,
+            )
+
         ref: Optional[str] = dvc.scm.get_ref(EXEC_BRANCH, follow=False)
         exp_ref: Optional["ExpRefInfo"] = ExpRefInfo.from_ref(ref) if ref else None
         if cls.WARN_UNTRACKED:
@@ -672,15 +665,30 @@ class BaseExecutor(ABC):
             kwargs = {}
         return args, kwargs
 
+    @classmethod
+    @contextmanager
+    def auto_push(cls, dvc: "Repo") -> Iterator[None]:
+        exp_config = dvc.config.get("exp", {})
+        auto_push = env2bool(DVC_EXP_AUTO_PUSH, exp_config.get("auto_push", False))
+        if not auto_push:
+            yield
+            return
+
+        git_remote = os.getenv(
+            DVC_EXP_GIT_REMOTE, exp_config.get("git_remote", "origin")
+        )
+        cls._validate_remotes(dvc, git_remote)
+        yield
+        cls._auto_push(dvc, git_remote)
+
     @staticmethod
     def _auto_push(
         dvc: "Repo",
-        scm: "Git",
         git_remote: Optional[str],
         push_cache=True,
         run_cache=True,
     ):
-        branch = scm.get_ref(EXEC_BRANCH, follow=False)
+        branch = dvc.scm.get_ref(EXEC_BRANCH, follow=False)
         try:
             dvc.experiments.push(
                 git_remote,
@@ -708,6 +716,7 @@ class BaseExecutor(ABC):
         message: Optional[str] = None,
     ):
         """Commit stages as an experiment and return the commit SHA."""
+
         rev = scm.get_rev()
         if not scm.is_dirty(untracked_files=False):
             logger.debug("No changes to commit")

dvc/testing/api_tests.py

@@ -33,9 +33,9 @@ class TestAPI:
         [
             {},
             {"url": "{path}"},
-            {"url": "{path}", "rev": "master"},
+            {"url": "{path}", "rev": "{default_branch}"},
             {"url": "file://{posixpath}"},
-            {"url": "file://{posixpath}", "rev": "master"},
+            {"url": "file://{posixpath}", "rev": "{default_branch}"},
         ],
         ids=["current", "local", "local_rev", "git", "git_rev"],
     )
@@ -61,6 +61,8 @@ class TestAPI:
 
         if url := fs_kwargs.get("url"):
             fs_kwargs["url"] = url.format(path=tmp_dir, posixpath=tmp_dir.as_posix())
+        if rev := fs_kwargs.get("rev"):
+            fs_kwargs["rev"] = rev.format(default_branch=scm.active_branch())
 
         fs = DVCFileSystem(**fs_kwargs)
 

dvc/testing/benchmarks/cli/commands/test_get.py

@@ -1,7 +1,7 @@
 import pytest
 
 
-@pytest.mark.flaky(max_runs=3, min_passes=1)
+@pytest.mark.flaky(reruns=3)
 def test_get(bench_dvc, tmp_dir, scm, dvc, make_dataset, remote):
     dataset = make_dataset(
         cache=False, files=False, dvcfile=True, commit=True, remote=True

dvc/testing/benchmarks/cli/commands/test_import.py

@@ -1,7 +1,7 @@
 import pytest
 
 
-@pytest.mark.flaky(max_runs=3, min_passes=1)
+@pytest.mark.flaky(reruns=3)
 def test_import(bench_dvc, tmp_dir, scm, dvc, make_dataset, remote):
     dataset = make_dataset(
         cache=False, files=False, dvcfile=True, commit=True, remote=True

dvc/testing/fixtures.py

@@ -221,15 +221,8 @@ def docker_compose_project_name():
 
 
 @pytest.fixture(scope="session")
-def docker_services(
-    tmp_path_factory,
-    docker_compose_command,
-    docker_compose_file,
-    docker_compose_project_name,
-    docker_setup,
-):
+def docker_services(tmp_path_factory, request):
     from filelock import FileLock
-    from pytest_docker.plugin import DockerComposeExecutor, Services
 
     if os.environ.get("CI") and os.name == "nt":
         pytest.skip("disabled for Windows on CI")
@@ -251,18 +244,12 @@ def docker_services(
         out = (err.output or b"").decode("utf-8")
         pytest.skip(f"docker-compose is not installed: {out}")
 
-    executor = DockerComposeExecutor(
-        docker_compose_command, docker_compose_file, docker_compose_project_name
-    )
-
     # making sure we don't accidentally launch docker-compose in parallel,
     # as it might result in network conflicts. Inspired by:
     # https://github.com/pytest-dev/pytest-xdist#making-session-scoped-fixtures-execute-only-once
     lockfile = tmp_path_factory.getbasetemp().parent / "docker-compose.lock"
     with FileLock(os.fspath(lockfile)):
-        executor.execute(docker_setup)
-        # note: we are not tearing down the containers here
-        return Services(executor)
+        return request.getfixturevalue("docker_services")
 
 
 @pytest.fixture

pyproject.toml

@@ -102,8 +102,8 @@ tests = [
     "beautifulsoup4>=4.4",
     "dvc-ssh",
     "filelock",
-    "flaky",
     "pytest>=7,<9",
+    "pytest-rerunfailures",
     "pytest-cov>=4.1.0",
     "pytest-docker>=1,<4",
     "pytest-mock",

tests/conftest.py

@@ -123,7 +123,7 @@ def pytest_runtest_setup(item):
     if "CI" in os.environ and item.get_closest_marker("needs_internet") is not None:
         # remotes that need internet connection might be flaky,
         # so we rerun them in case it fails.
-        item.add_marker(pytest.mark.flaky(max_runs=5, min_passes=1))
+        item.add_marker(pytest.mark.flaky(reruns=5))
 
 
 @pytest.fixture(scope="session")

tests/func/experiments/test_remote.py

@@ -372,3 +372,40 @@ def test_push_pull_invalid_workspace(
         dvc.experiments.push(git_upstream.remote, push_cache=True)
         dvc.experiments.pull(git_upstream.remote, pull_cache=True)
         assert "failed to collect" not in caplog.text
+
+
+@pytest.mark.parametrize(
+    "auto_push, expected_key", [(True, "up_to_date"), (False, "success")]
+)
+def test_auto_push_on_run(
+    tmp_dir, scm, dvc, git_upstream, local_remote, exp_stage, auto_push, expected_key
+):
+    remote = git_upstream.remote
+
+    with dvc.config.edit() as conf:
+        conf["exp"]["auto_push"] = auto_push
+        conf["exp"]["git_remote"] = remote
+
+    exp_name = "foo"
+    dvc.experiments.run(exp_stage.addressing, params=["foo=2"], name=exp_name)
+
+    assert first(dvc.experiments.push(name=exp_name, git_remote=remote)) == expected_key
+
+
+@pytest.mark.parametrize(
+    "auto_push, expected_key", [(True, "up_to_date"), (False, "success")]
+)
+def test_auto_push_on_save(
+    tmp_dir, scm, dvc, git_upstream, local_remote, exp_stage, auto_push, expected_key
+):
+    remote = git_upstream.remote
+    exp_name = "foo"
+    dvc.experiments.run(exp_stage.addressing, params=["foo=2"], name=exp_name)
+
+    with dvc.config.edit() as conf:
+        conf["exp"]["auto_push"] = auto_push
+        conf["exp"]["git_remote"] = remote
+
+    dvc.experiments.save(name=exp_name, force=True)
+
+    assert first(dvc.experiments.push(name=exp_name, git_remote=remote)) == expected_key

tests/func/test_data_cloud.py

@@ -3,7 +3,6 @@ import os
 import shutil
 
 import pytest
-from flaky.flaky_decorator import flaky
 
 import dvc_data
 from dvc.cli import main
@@ -207,7 +206,7 @@ def test_verify_hashes(tmp_dir, scm, dvc, mocker, tmp_path_factory, local_remote
     assert hash_spy.call_count == 10
 
 
-@flaky(max_runs=3, min_passes=1)
+@pytest.mark.flaky(reruns=3)
 @pytest.mark.parametrize("erepo_type", ["git_dir", "erepo_dir"])
 def test_pull_git_imports(request, tmp_dir, dvc, scm, erepo_type):
     erepo = request.getfixturevalue(erepo_type)