mirror of
https://github.com/Stone-Red-Code/dvc.git
synced 2026-11-01 23:08:43 +01:00
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.
This commit is contained in:
karajan1001
@@ -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):
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user