Commit b17beb55 authored by Samuel GAIST's avatar Samuel GAIST
Browse files

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

parent 0bf3201f
......@@ -37,128 +37,89 @@
# Basic tests for the command line beat program: algorithms
import nose.tools
from click.testing import CliRunner
from beat.core.test.utils import slow, cleanup, skipif
from beat.core.test.utils import slow
from beat.core.algorithm import Storage
from beat.core.dataformat import Storage as DFStorage
from beat.cmdline.scripts import main_cli
from . import core
from . import platform, disconnected, prefix, tmp_prefix, user, token
def setup():
"""Create default dataformat for algorithm prototype loading"""
from .test_dataformats import call as df_call
obj = "user/integers/1"
storage = DFStorage(tmp_prefix, obj)
if not storage.exists():
nose.tools.eq_(df_call("create", obj, prefix=tmp_prefix), 0)
if not disconnected:
nose.tools.eq_(df_call("push", obj, prefix=tmp_prefix), 0)
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")
runner = CliRunner()
with runner.isolated_filesystem():
result = runner.invoke(
main_cli.main,
[
"--platform",
use_platform,
"--user",
user,
"--token",
token,
"--prefix",
use_prefix,
"--cache",
use_cache,
"--test-mode",
"algorithms",
]
+ list(args),
catch_exceptions=False,
)
return result.exit_code, result.output
@slow
@nose.tools.with_setup(teardown=cleanup)
@skipif(disconnected, "missing test platform (%s)" % platform)
def test_remote_list():
exit_code, outputs = call("list", "--remote")
nose.tools.eq_(exit_code, 0, msg=outputs)
@slow
@nose.tools.with_setup(teardown=cleanup)
@skipif(disconnected, "missing test platform (%s)" % platform)
def test_pull_one():
obj = "user/integers_add/1"
exit_code, outputs = call("pull", obj, prefix=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
s = Storage(tmp_prefix, obj)
nose.tools.assert_true(s.exists())
@slow
@nose.tools.with_setup(teardown=cleanup)
@skipif(disconnected, "missing test platform (%s)" % platform)
def test_pull_all():
exit_code, outputs = call("pull", prefix=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
@slow
@nose.tools.with_setup(teardown=cleanup)
@skipif(disconnected, "missing test platform (%s)" % platform)
def test_diff():
obj = "user/integers_add/1"
exit_code, outputs = call("pull", obj, prefix=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
# quickly modify the user algorithm by emptying it
storage = Storage(tmp_prefix, obj)
storage.code.save("class Algorithm:\n pass")
exit_code, outputs = call("diff", obj, prefix=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
@slow
@nose.tools.with_setup(teardown=cleanup)
@skipif(disconnected, "missing test platform (%s)" % platform)
def test_status():
test_diff()
test_pull_one()
exit_code, outputs = call("status", prefix=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
@nose.tools.with_setup(setup=setup, teardown=cleanup)
@skipif(disconnected, "missing test platform (%s)" % platform)
def test_push_and_delete():
obj = "user/newobject/1"
TestAlgorithmLocal.create(obj)
# now push the new object and then delete it remotely
exit_code, outputs = call("push", obj, prefix=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
exit_code, outputs = call("rm", "--remote", obj, prefix=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
from . import tmp_prefix
class TestOnlinAlgorithms(core.OnlineAssetTestCase):
asset_type = "algorithm"
@slow
@core.skip_disconnected
def test_remote_list(self):
exit_code, outputs = self.call("list", "--remote")
nose.tools.eq_(exit_code, 0, msg=outputs)
@slow
@core.skip_disconnected
def test_pull_one(self):
obj = "user/integers_add/1"
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())
@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/integers_add/1"
exit_code, outputs = self.call("pull", obj, prefix=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
# quickly modify the user algorithm by emptying it
storage = Storage(tmp_prefix, obj)
storage.code.save("class Algorithm:\n pass")
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)
def create_dataformat(self):
obj = "user/integers/1"
storage = DFStorage(tmp_prefix, obj)
if not storage.exists():
exit_code, outputs = TestAlgorithmLocal.call(
"create", obj, prefix=tmp_prefix, asset_type="dataformat"
)
nose.tools.eq_(exit_code, 0, outputs)
exit_code, outputs = self.call(
"push", obj, prefix=tmp_prefix, asset_type="dataformat"
)
nose.tools.eq_(exit_code, 0, outputs)
@slow
@core.skip_disconnected
def test_push_and_delete(self):
self.create_dataformat()
obj = "user/newobject/1"
TestAlgorithmLocal.create(obj)
# now push the new object and then delete it remotely
exit_code, outputs = self.call("push", obj, prefix=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
exit_code, outputs = self.call("rm", "--remote", obj, prefix=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
class TestAlgorithmLocal(core.AssetLocalTest):
......@@ -180,8 +141,3 @@ class TestAlgorithmLocal(core.AssetLocalTest):
"create", obj, prefix=tmp_prefix, asset_type="dataformat"
)
nose.tools.eq_(exit_code, 0, outputs)
if not disconnected:
exit_code, outputs = self.call(
"push", obj, prefix=tmp_prefix, asset_type="dataformat"
)
nose.tools.eq_(exit_code, 0, 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