From 7fb9dcec12f3b324b197bcc2080610e483598877 Mon Sep 17 00:00:00 2001 From: karajan1001 Date: Tue, 23 Aug 2022 15:18:27 +0800 Subject: [PATCH] Add tests for executor status 1. Move some basic test script from function tests to unit test. 2. Add success/failed tests for the status change of `tempdir`, `celery`, `workspace` running case. --- tests/func/experiments/conftest.py | 118 +----------------- tests/unit/repo/experiments/conftest.py | 112 +++++++++++++++++ .../repo/experiments/test_executor_status.py | 95 ++++++++++++++ 3 files changed, 211 insertions(+), 114 deletions(-) create mode 100644 tests/unit/repo/experiments/test_executor_status.py diff --git a/tests/func/experiments/conftest.py b/tests/func/experiments/conftest.py index cc115ee8e..8f0228b7f 100644 --- a/tests/func/experiments/conftest.py +++ b/tests/func/experiments/conftest.py @@ -1,122 +1,12 @@ -from textwrap import dedent - import pytest -from tests.func.test_repro_multistage import COPY_SCRIPT -from tests.unit.repo.experiments.conftest import ( # noqa pylint disable=unused-argument +from tests.unit.repo.experiments.conftest import ( # noqa, pylint disable=unused-argument + checkpoint_stage, + exp_stage, + failed_exp_stage, test_queue, ) -DEFAULT_ITERATIONS = 2 -CHECKPOINT_SCRIPT_FORMAT = dedent( - """\ - import os - import sys - import shutil - - from dvc.api import make_checkpoint - - checkpoint_file = {} - checkpoint_iterations = int({}) - if os.path.exists(checkpoint_file): - with open(checkpoint_file) as fobj: - try: - value = int(fobj.read()) - except ValueError: - value = 0 - else: - with open(checkpoint_file, "w"): - pass - value = 0 - - shutil.copyfile({}, {}) - - if os.getenv("DVC_CHECKPOINT"): - for _ in range(checkpoint_iterations): - value += 1 - with open(checkpoint_file, "w") as fobj: - fobj.write(str(value)) - make_checkpoint() -""" -) -CHECKPOINT_SCRIPT = CHECKPOINT_SCRIPT_FORMAT.format( - "sys.argv[1]", "sys.argv[2]", "sys.argv[3]", "sys.argv[4]" -) - - -@pytest.fixture -def exp_stage(tmp_dir, scm, dvc): - tmp_dir.gen("copy.py", COPY_SCRIPT) - tmp_dir.gen("params.yaml", "foo: 1") - stage = dvc.run( - cmd="python copy.py params.yaml metrics.yaml", - metrics_no_cache=["metrics.yaml"], - params=["foo"], - name="copy-file", - deps=["copy.py"], - ) - scm.add( - [ - "dvc.yaml", - "dvc.lock", - "copy.py", - "params.yaml", - "metrics.yaml", - ".gitignore", - ] - ) - scm.commit("init") - return stage - - -@pytest.fixture -def checkpoint_stage(tmp_dir, scm, dvc, mocker): - mocker.patch("dvc.stage.run.Monitor.AWAIT", 0.01) - - tmp_dir.gen("checkpoint.py", CHECKPOINT_SCRIPT) - tmp_dir.gen("params.yaml", "foo: 1") - stage = dvc.run( - cmd=( - f"python checkpoint.py foo {DEFAULT_ITERATIONS} " - "params.yaml metrics.yaml" - ), - metrics_no_cache=["metrics.yaml"], - params=["foo"], - checkpoints=["foo"], - deps=["checkpoint.py"], - no_exec=True, - name="checkpoint-file", - ) - scm.add(["dvc.yaml", "checkpoint.py", "params.yaml", ".gitignore"]) - scm.commit("init") - stage.iterations = DEFAULT_ITERATIONS - return stage - - -@pytest.fixture -def failed_exp_stage(tmp_dir, scm, dvc): - tmp_dir.gen("copy.py", COPY_SCRIPT) - tmp_dir.gen("params.yaml", "foo: 1") - stage = dvc.stage.add( - cmd="python -c 'import sys; sys.exit(1)'", - metrics_no_cache=["failed-metrics.yaml"], - params=["foo"], - name="failed-copy-file", - deps=["copy.py"], - ) - scm.add( - [ - "dvc.yaml", - "dvc.lock", - "copy.py", - "params.yaml", - "failed-metrics.yaml", - ".gitignore", - ] - ) - scm.commit("init") - return stage - @pytest.fixture def http_auth_patch(mocker): diff --git a/tests/unit/repo/experiments/conftest.py b/tests/unit/repo/experiments/conftest.py index 2100049b4..3aba50586 100644 --- a/tests/unit/repo/experiments/conftest.py +++ b/tests/unit/repo/experiments/conftest.py @@ -1,8 +1,120 @@ from contextlib import contextmanager +from textwrap import dedent import pytest from dvc.repo.experiments.queue.celery import LocalCeleryQueue +from tests.func.test_repro_multistage import COPY_SCRIPT + +DEFAULT_ITERATIONS = 2 +CHECKPOINT_SCRIPT_FORMAT = dedent( + """\ + import os + import sys + import shutil + + from dvc.api import make_checkpoint + + checkpoint_file = {} + checkpoint_iterations = int({}) + if os.path.exists(checkpoint_file): + with open(checkpoint_file) as fobj: + try: + value = int(fobj.read()) + except ValueError: + value = 0 + else: + with open(checkpoint_file, "w"): + pass + value = 0 + + shutil.copyfile({}, {}) + + if os.getenv("DVC_CHECKPOINT"): + for _ in range(checkpoint_iterations): + value += 1 + with open(checkpoint_file, "w") as fobj: + fobj.write(str(value)) + make_checkpoint() +""" +) +CHECKPOINT_SCRIPT = CHECKPOINT_SCRIPT_FORMAT.format( + "sys.argv[1]", "sys.argv[2]", "sys.argv[3]", "sys.argv[4]" +) + + +@pytest.fixture +def exp_stage(tmp_dir, scm, dvc): + tmp_dir.gen("copy.py", COPY_SCRIPT) + tmp_dir.gen("params.yaml", "foo: 1") + stage = dvc.run( + cmd="python copy.py params.yaml metrics.yaml", + metrics_no_cache=["metrics.yaml"], + params=["foo"], + name="copy-file", + deps=["copy.py"], + ) + scm.add( + [ + "dvc.yaml", + "dvc.lock", + "copy.py", + "params.yaml", + "metrics.yaml", + ".gitignore", + ] + ) + scm.commit("init") + return stage + + +@pytest.fixture +def checkpoint_stage(tmp_dir, scm, dvc, mocker): + mocker.patch("dvc.stage.run.Monitor.AWAIT", 0.01) + + tmp_dir.gen("checkpoint.py", CHECKPOINT_SCRIPT) + tmp_dir.gen("params.yaml", "foo: 1") + stage = dvc.run( + cmd=( + f"python checkpoint.py foo {DEFAULT_ITERATIONS} " + "params.yaml metrics.yaml" + ), + metrics_no_cache=["metrics.yaml"], + params=["foo"], + checkpoints=["foo"], + deps=["checkpoint.py"], + no_exec=True, + name="checkpoint-file", + ) + scm.add(["dvc.yaml", "checkpoint.py", "params.yaml", ".gitignore"]) + scm.commit("init") + stage.iterations = DEFAULT_ITERATIONS + return stage + + +@pytest.fixture +def failed_exp_stage(tmp_dir, scm, dvc): + tmp_dir.gen("copy.py", COPY_SCRIPT) + tmp_dir.gen("params.yaml", "foo: 1") + stage = dvc.stage.add( + cmd="python -c 'import sys; sys.exit(1)'", + metrics_no_cache=["failed-metrics.yaml"], + params=["foo"], + name="failed-copy-file", + deps=["copy.py"], + ) + scm.add( + [ + "dvc.yaml", + "dvc.lock", + "copy.py", + "params.yaml", + "failed-metrics.yaml", + ".gitignore", + ] + ) + scm.commit("init") + return stage @contextmanager diff --git a/tests/unit/repo/experiments/test_executor_status.py b/tests/unit/repo/experiments/test_executor_status.py new file mode 100644 index 000000000..7af673e74 --- /dev/null +++ b/tests/unit/repo/experiments/test_executor_status.py @@ -0,0 +1,95 @@ +import os + +import pytest + +from dvc.exceptions import DvcException +from dvc.repo.experiments.executor.base import ExecutorInfo, TaskStatus +from dvc.repo.experiments.queue.tasks import collect_exp, setup_exp + + +def test_celery_queue_success_status(dvc, scm, test_queue, exp_stage): + queue_entry = test_queue._stash_exp( + params={"params.yaml": ["foo=1"]}, + targets=exp_stage.addressing, + name="success", + ) + infofile = test_queue.get_infofile_path(queue_entry.stash_rev) + setup_exp.s(queue_entry.asdict())() + executor_info = ExecutorInfo.load_json(infofile) + assert executor_info.status == TaskStatus.PREPARING + + cmd = ["dvc", "exp", "exec-run", "--infofile", infofile] + proc_dict = test_queue.proc.run_signature( + cmd, name=queue_entry.stash_rev + )() + + executor_info = ExecutorInfo.load_json(infofile) + assert executor_info.status == TaskStatus.SUCCESS + + collect_exp.s(proc_dict, queue_entry.asdict())() + executor_info = ExecutorInfo.load_json(infofile) + assert executor_info.status == TaskStatus.FINISHED + + +def test_celery_queue_failure_status(dvc, scm, test_queue, failed_exp_stage): + queue_entry = test_queue._stash_exp( + params={"params.yaml": ["foo=1"]}, + targets=failed_exp_stage.addressing, + name="failed", + ) + infofile = test_queue.get_infofile_path(queue_entry.stash_rev) + setup_exp.s(queue_entry.asdict())() + cmd = ["dvc", "exp", "exec-run", "--infofile", infofile] + test_queue.proc.run_signature(cmd, name=queue_entry.stash_rev)() + executor_info = ExecutorInfo.load_json(infofile) + assert executor_info.status == TaskStatus.FAILED + + +@pytest.mark.parametrize("queue_type", ["workspace_queue", "tempdir_queue"]) +def test_workspace_executor_success_status(dvc, scm, exp_stage, queue_type): + workspace_queue = getattr(dvc.experiments, queue_type) + queue_entry = workspace_queue.put( + params={"params.yaml": ["foo=1"]}, + targets=exp_stage.addressing, + name="success", + ) + name = workspace_queue._EXEC_NAME or queue_entry.stash_rev + infofile = workspace_queue.get_infofile_path(name) + entry, executor = workspace_queue.get() + rev = entry.stash_rev + exec_result = executor.reproduce( + info=executor.info, + rev=rev, + infofile=infofile, + ) + executor_info = ExecutorInfo.load_json(infofile) + assert executor_info.status == TaskStatus.SUCCESS + if exec_result.ref_info: + workspace_queue.collect_executor( + dvc.experiments, executor, exec_result, infofile + ) + + executor_info = ExecutorInfo.load_json(infofile) + assert executor_info.status == TaskStatus.FINISHED + + +@pytest.mark.parametrize("queue_type", ["workspace_queue", "tempdir_queue"]) +def test_workspace_executor_failed_status( + dvc, scm, failed_exp_stage, queue_type +): + workspace_queue = getattr(dvc.experiments, queue_type) + queue_entry = workspace_queue.put( + params={"params.yaml": ["foo=1"]}, + targets=failed_exp_stage.addressing, + name="failed", + ) + name = workspace_queue._EXEC_NAME or queue_entry.stash_rev + + infofile = workspace_queue.get_infofile_path(name) + with pytest.raises(DvcException): + workspace_queue.reproduce() + if queue_type == "workspace_queue": + assert not os.path.exists(infofile) + else: + executor_info = ExecutorInfo.load_json(infofile) + assert executor_info.status == TaskStatus.FAILED