Commit 6ec7fddc authored by Samuel GAIST's avatar Samuel GAIST
Browse files

[test][experiments] Refactor tests to use new live server based test class

parent e981f5c6
......@@ -39,297 +39,283 @@
import os
import logging
import nose.tools
import click
from click.testing import CliRunner
from . import platform, disconnected, prefix, tmp_prefix, user, token
from .utils import index_experiment_dbs, MockLoggingHandler
from ..common import Selector
from beat.cmdline.scripts import main_cli
from beat.core.test.utils import slow, cleanup, skipif
from beat.core.experiment import Storage, Experiment
from beat.core.test.utils import slow
from beat.core.test.test_execution import create_conda_environment
from beat.core.test.test_execution import clear_conda_environment
from beat.core.experiment import Storage, Experiment
def setup_experiments():
index_experiment_dbs("user/user/double_triangle/1/double_triangle")
index_experiment_dbs("user/user/integers_addition/1/integers_addition")
def call(*args, **kwargs):
"""A central mechanism to call the main routine with the right parameters"""
use_prefix = kwargs.get("prefix", prefix)
use_platform = kwargs.get("platform", platform)
use_cache = kwargs.get("cache", "cache")
environment = kwargs.pop("environment", None)
cli_args = [
"--test-mode",
"--prefix",
use_prefix,
"--token",
token,
"--user",
user,
"--platform",
use_platform,
"--cache",
use_cache,
"experiments",
]
cli_args += list(args)
if environment:
cli_args += ["--environment", environment]
runner = CliRunner()
with runner.isolated_filesystem():
result = runner.invoke(main_cli.main, cli_args, catch_exceptions=False)
if result.exit_code != 0:
click.echo(result.output)
return result.exit_code
@slow
@nose.tools.with_setup(teardown=cleanup)
@skipif(disconnected, "missing test platform (%s)" % platform)
def test_remote_list():
nose.tools.eq_(call("list", "--remote"), 0)
@slow
@nose.tools.with_setup(teardown=cleanup)
@skipif(disconnected, "missing test platform (%s)" % platform)
def test_pull_one(obj=None):
obj = obj or "user/user/single/1/single"
nose.tools.eq_(call("pull", obj, prefix=tmp_prefix), 0)
s = Storage(tmp_prefix, obj)
nose.tools.assert_true(s.exists())
return s
@slow
@nose.tools.with_setup(teardown=cleanup)
@skipif(disconnected, "missing test platform (%s)" % platform)
def test_pull_all():
nose.tools.eq_(call("pull", prefix=tmp_prefix), 0)
@slow
@nose.tools.with_setup(teardown=cleanup)
@skipif(disconnected, "missing test platform (%s)" % platform)
def test_diff():
obj = "user/user/single/1/single"
nose.tools.eq_(call("pull", obj, prefix=tmp_prefix), 0)
s = Storage(tmp_prefix, obj)
nose.tools.assert_true(s.exists())
# quickly modify the user experiment:
f = Experiment(tmp_prefix, obj)
f.data["globals"]["queue"] = "another_queue"
f.write()
nose.tools.eq_(call("diff", obj, prefix=tmp_prefix), 0)
@slow
@nose.tools.with_setup(teardown=cleanup)
@skipif(disconnected, "missing test platform (%s)" % platform)
def test_status():
test_diff()
test_pull_one()
nose.tools.eq_(call("status", prefix=tmp_prefix), 0)
def test_check_valid():
obj = "user/user/single/1/single"
nose.tools.eq_(call("check", obj), 0)
def test_check_invalid():
obj = "user/user/single/1/does_not_exist"
nose.tools.eq_(call("check", obj), 1)
@nose.tools.with_setup(teardown=cleanup)
@skipif(disconnected, "missing test platform (%s)" % platform)
def test_fork(obj=None, obj2=None):
obj = obj or "user/user/single/1/single"
test_pull_one(obj)
obj2 = obj2 or "user/user/single/1/different"
nose.tools.eq_(call("fork", obj, obj2, prefix=tmp_prefix), 0)
s = Storage(tmp_prefix, obj2)
nose.tools.assert_true(s.exists())
# check fork status
with Selector(tmp_prefix) as selector:
nose.tools.eq_(selector.forked_from("experiment", obj2), obj)
@nose.tools.with_setup(teardown=cleanup)
@skipif(disconnected, "missing test platform (%s)" % platform)
def test_delete_local():
obj = "user/user/single/1/single"
storage = test_pull_one(obj)
nose.tools.eq_(call("rm", obj, prefix=tmp_prefix), 0)
nose.tools.assert_false(storage.exists())
from ..common import Selector
@slow
@nose.tools.with_setup(setup=setup_experiments, teardown=cleanup)
def test_run_integers_addition_1():
obj = "user/user/integers_addition/1/integers_addition"
nose.tools.eq_(call("run", obj, cache=tmp_prefix), 0)
from .utils import index_experiment_dbs, MockLoggingHandler
from . import core
from . import tmp_prefix
@slow
@nose.tools.with_setup(setup=setup_experiments, teardown=cleanup)
def test_run_integers_addition_1_different_environment():
environment_path = create_conda_environment(["beat.backend.python"])
obj = "user/user/integers_addition/1/integers_addition"
status = call(
"run",
obj,
cache=tmp_prefix,
environment=os.path.join(environment_path, "bin", "python"),
)
clear_conda_environment(environment_path)
nose.tools.eq_(status, 0)
@slow
@nose.tools.with_setup(setup=setup_experiments, teardown=cleanup)
def test_list_integers_addition_1_cache():
obj = "user/user/integers_addition/1/integers_addition"
nose.tools.eq_(call("run", obj, cache=tmp_prefix), 0)
nose.tools.eq_(call("caches", "--list", obj, cache=tmp_prefix), 0)
@slow
@nose.tools.with_setup(setup=setup_experiments, teardown=cleanup)
def test_checksum_integers_addition_1_cache():
obj = "user/user/integers_addition/1/integers_addition"
nose.tools.eq_(call("run", obj, cache=tmp_prefix), 0)
nose.tools.eq_(call("caches", "--checksum", obj, cache=tmp_prefix), 0)
@slow
@nose.tools.with_setup(setup=setup_experiments, teardown=cleanup)
def test_delete_integers_addition_1_cache():
obj = "user/user/integers_addition/1/integers_addition"
nose.tools.eq_(call("run", obj, cache=tmp_prefix), 0)
nose.tools.eq_(call("caches", "--delete", obj, cache=tmp_prefix), 0)
def setup_experiments():
index_experiment_dbs("user/user/double_triangle/1/double_triangle")
index_experiment_dbs("user/user/integers_addition/1/integers_addition")
@slow
@nose.tools.with_setup(setup=setup_experiments, teardown=cleanup)
def test_run_integers_addition_1_twice():
log_handler = MockLoggingHandler(level="DEBUG")
logging.getLogger().addHandler(log_handler)
log_messages = log_handler.messages
obj = "user/user/integers_addition/1/integers_addition"
nose.tools.eq_(call("run", obj, cache=tmp_prefix), 0)
info_len = len(log_messages["info"])
nose.tools.assert_greater(info_len, 4)
nose.tools.assert_true(log_messages["info"][info_len - 1].startswith(" Results"))
nose.tools.eq_(call("run", obj, cache=tmp_prefix), 0)
info_len = len(log_messages["info"])
nose.tools.assert_greater(info_len, 6)
nose.tools.assert_true(log_messages["info"][info_len - 1].startswith(" Results"))
@slow
@nose.tools.with_setup(setup=setup_experiments, teardown=cleanup)
def test_run_double_triangle_1():
obj = "user/user/double_triangle/1/double_triangle"
nose.tools.eq_(call("run", obj, cache=tmp_prefix), 0)
@slow
@nose.tools.with_setup(setup=setup_experiments, teardown=cleanup)
def test_run_single_error_1_local():
# When running locally, the module with the error is loaded
# inside the currently running process and will return '1'.
obj = "user/user/single/1/single_error"
nose.tools.eq_(call("run", obj, "--local", cache=tmp_prefix), 1)
@slow
@nose.tools.with_setup(setup=setup_experiments, teardown=cleanup)
def test_run_single_error_twice_local():
# This one makes sure our output reset is working properly. Both tries should
# give out the same error.
obj = "user/user/single/1/single_error"
nose.tools.eq_(call("run", obj, "--local", cache=tmp_prefix), 1)
nose.tools.eq_(call("run", obj, "--local", cache=tmp_prefix), 1)
@nose.tools.with_setup(teardown=cleanup)
@skipif(disconnected, "missing test platform (%s)" % platform)
def test_push_and_delete():
obj = "user/user/single/1/single"
obj2 = "user/user/single/1/different"
test_fork(obj, obj2)
# now push the fork and then delete it remotely
nose.tools.eq_(call("push", obj2, prefix=tmp_prefix), 0)
nose.tools.eq_(call("rm", "--remote", obj2, prefix=tmp_prefix), 0)
@nose.tools.with_setup(teardown=cleanup)
@skipif(disconnected, "missing test platform (%s)" % platform)
def test_draw():
obj = "user/user/double_triangle/1/double_triangle"
test_pull_one(obj)
# now push the new object and then delete it remotely
nose.tools.eq_(call("draw", "--path=%s" % tmp_prefix, prefix=tmp_prefix), 0)
nose.tools.assert_true(
os.path.exists(os.path.join(tmp_prefix, "experiments", obj + ".dot"))
)
nose.tools.assert_true(
os.path.exists(os.path.join(tmp_prefix, "experiments", obj + ".png"))
)
@slow
@nose.tools.with_setup(teardown=cleanup)
@skipif(disconnected, "missing test platform (%s)" % platform)
def test_start():
obj = "user/user/double_triangle/1/double_triangle"
nose.tools.eq_(call("start", obj), 0)
@slow
@nose.tools.with_setup(teardown=cleanup)
@skipif(disconnected, "missing test platform (%s)" % platform)
def test_cancel():
obj = "user/user/double_triangle/1/double_triangle"
nose.tools.eq_(call("cancel", obj), 0)
@slow
@nose.tools.with_setup(teardown=cleanup)
@skipif(disconnected, "missing test platform (%s)" % platform)
def test_reset():
obj = "user/user/double_triangle/1/double_triangle"
nose.tools.eq_(call("reset", obj), 0)
@slow
@nose.tools.with_setup(teardown=cleanup)
@skipif(disconnected, "missing test platform (%s)" % platform)
def test_runstatus():
obj = "user/user/double_triangle/1/double_triangle"
nose.tools.eq_(call("runstatus", obj), 0)
class TestOnlineExperiments(core.OnlineAssetTestCase):
asset_type = "experiment"
@slow
@core.skip_disconnected
def test_remote_list(self):
exit_code, outputs = self.call("list", "--remote")
nose.tools.eq_(exit_code, 0, outputs)
@slow
@core.skip_disconnected
def test_pull_one(self, obj=None):
obj = obj or "user/user/single/1/single"
exit_code, outputs = self.call("pull", obj, prefix=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
s = Storage(tmp_prefix, obj)
nose.tools.assert_true(s.exists())
return s
@slow
@core.skip_disconnected
def test_pull_all(self):
exit_code, outputs = self.call("pull", prefix=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
@slow
@core.skip_disconnected
def test_diff(self):
obj = "user/user/single/1/single"
exit_code, outputs = self.call("pull", obj, prefix=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
s = Storage(tmp_prefix, obj)
nose.tools.assert_true(s.exists())
# quickly modify the user experiment:
f = Experiment(tmp_prefix, obj)
f.data["globals"]["queue"] = "another_queue"
f.write()
exit_code, outputs = self.call("diff", obj, prefix=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
@slow
@core.skip_disconnected
def test_status(self):
self.test_diff()
self.test_pull_one()
exit_code, outputs = self.call("status", prefix=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
@core.skip_disconnected
def test_fork(self, obj=None, obj2=None):
obj = obj or "user/user/single/1/single"
self.test_pull_one(obj)
obj2 = obj2 or "user/user/single/1/different"
exit_code, outputs = self.call("fork", obj, obj2, prefix=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
s = Storage(tmp_prefix, obj2)
nose.tools.assert_true(s.exists())
# check fork status
with Selector(tmp_prefix) as selector:
nose.tools.eq_(selector.forked_from("experiment", obj2), obj)
@core.skip_disconnected
def test_delete_local(self):
obj = "user/user/single/1/single"
storage = self.test_pull_one(obj)
exit_code, outputs = self.call("rm", obj, prefix=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
nose.tools.assert_false(storage.exists())
@core.skip_disconnected
def test_push_and_delete(self):
obj = "user/user/double_triangle/1/double_triangle"
obj2 = "user/user/double_triangle/1/different"
self.test_fork(obj, obj2)
# now push the fork and then delete it remotely
exit_code, outputs = self.call("push", obj2, prefix=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
exit_code, outputs = self.call("rm", "--remote", obj2, prefix=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
@core.skip_disconnected
def test_draw(self):
obj = "user/user/double_triangle/1/double_triangle"
self.test_pull_one(obj)
# now push the new object and then delete it remotely
exit_code, outputs = self.call(
"draw", "--path=%s" % tmp_prefix, prefix=tmp_prefix
)
nose.tools.eq_(exit_code, 0, outputs)
nose.tools.assert_true(
os.path.exists(os.path.join(tmp_prefix, "experiments", obj + ".dot"))
)
nose.tools.assert_true(
os.path.exists(os.path.join(tmp_prefix, "experiments", obj + ".png"))
)
@slow
@core.skip_disconnected
def test_start(self):
obj = "user/user/double_triangle/1/double_triangle"
exit_code, outputs = self.call("start", obj)
nose.tools.eq_(exit_code, 0, outputs)
@slow
@core.skip_disconnected
def test_cancel(self):
obj = "user/user/double_triangle/1/double_triangle"
exit_code, outputs = self.call("cancel", obj)
nose.tools.eq_(exit_code, 0, outputs)
@slow
@core.skip_disconnected
def test_reset(self):
obj = "user/user/double_triangle/1/double_triangle"
exit_code, outputs = self.call("reset", obj)
nose.tools.eq_(exit_code, 0, outputs)
@slow
@core.skip_disconnected
def test_runstatus(self):
obj = "user/user/double_triangle/1/double_triangle"
exit_code, outputs = self.call("runstatus", obj)
nose.tools.eq_(exit_code, 0, outputs)
class TestValidity(core.AssetBaseTest):
asset_type = "experiment"
def test_check_valid(self):
obj = "user/user/single/1/single"
exit_code, outputs = self.call("check", obj)
nose.tools.eq_(exit_code, 0, outputs)
def test_check_invalid(self):
obj = "user/user/single/1/does_not_exist"
exit_code, outputs = self.call("check", obj)
nose.tools.eq_(exit_code, 1, outputs)
class TestCache(core.AssetBaseTest):
asset_type = "experiment"
def setUp(self):
super().setUp()
setup_experiments()
@slow
def test_list_integers_addition_1_cache(self):
obj = "user/user/integers_addition/1/integers_addition"
exit_code, outputs = self.call("run", obj, cache=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
exit_code, outputs = self.call("caches", "--list", obj, cache=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
@slow
def test_checksum_integers_addition_1_cache(self):
obj = "user/user/integers_addition/1/integers_addition"
exit_code, outputs = self.call("run", obj, cache=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
exit_code, outputs = self.call("caches", "--checksum", obj, cache=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
@slow
def test_delete_integers_addition_1_cache(self):
obj = "user/user/integers_addition/1/integers_addition"
exit_code, outputs = self.call("run", obj, cache=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
exit_code, outputs = self.call("caches", "--delete", obj, cache=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
class TestRun(core.AssetBaseTest):
asset_type = "experiment"
def setUp(self):
super().setUp()
setup_experiments()
@slow
def test_run_integers_addition_1(self):
obj = "user/user/integers_addition/1/integers_addition"
exit_code, outputs = self.call("run", obj, cache=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
@slow
def test_run_integers_addition_1_different_environment(self):
environment_path = create_conda_environment(["beat.backend.python"])
obj = "user/user/integers_addition/1/integers_addition"
exit_code, status = self.call(
"run",
obj,
cache=tmp_prefix,
environment=os.path.join(environment_path, "bin", "python"),
)
clear_conda_environment(environment_path)
nose.tools.eq_(exit_code, 0)
@slow
def test_run_integers_addition_1_twice(self):
log_handler = MockLoggingHandler(level="DEBUG")
logging.getLogger().addHandler(log_handler)
log_messages = log_handler.messages
obj = "user/user/integers_addition/1/integers_addition"
exit_code, outputs = self.call("run", obj, cache=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
info_len = len(log_messages["info"])
nose.tools.assert_greater(info_len, 4)
nose.tools.assert_true(
log_messages["info"][info_len - 1].startswith(" Results")
)
exit_code, outputs = self.call("run", obj, cache=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
info_len = len(log_messages["info"])
nose.tools.assert_greater(info_len, 6)
nose.tools.assert_true(
log_messages["info"][info_len - 1].startswith(" Results")
)
@slow
def test_run_double_triangle_1(self):
obj = "user/user/double_triangle/1/double_triangle"
exit_code, outputs = self.call("run", obj, cache=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
@slow
def test_run_single_error_1_local(self):
# When running locally, the module with the error is loaded
# inside the currently running process and will return '1'.
obj = "user/user/single/1/single_error"
exit_code, outputs = self.call("run", obj, "--local", cache=tmp_prefix)
nose.tools.eq_(exit_code, 1, outputs)
@slow
def test_run_single_error_twice_local(self):
# This one makes sure our output reset is working properly. Both tries should
# give out the same error.
obj = "user/user/single/1/single_error"
exit_code, outputs = self.call("run", obj, "--local", cache=tmp_prefix)
nose.tools.eq_(exit_code, 1, outputs)
exit_code, outputs = self.call("run", obj, "--local", cache=tmp_prefix)
nose.tools.eq_(exit_code, 1, outputs)
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment