Commit 4cef1f92 authored by Samuel GAIST's avatar Samuel GAIST
Browse files

[test][algorithms] Moved local test to AssetLocalTest

parent 272c41a3
......@@ -37,185 +37,151 @@
# Basic tests for the command line beat program: algorithms
import nose.tools
import os
import click
from click.testing import CliRunner
import shutil
import json
from beat.core.test.utils import slow, cleanup, skipif
from beat.core.algorithm import Storage
from beat.cmdline.scripts import main_cli
from beat.core.dataformat import Storage as DFStorage
from .. import common
from beat.cmdline.scripts import main_cli
from . import platform, disconnected, prefix, tmp_prefix, user, token, temp_cwd
from . import core
from . import platform, disconnected, prefix, tmp_prefix, user, token
def setup():
"""Create default dataformat for algorithm prototype loading"""
"""Create default dataformat for algorithm prototype loading"""
from .test_dataformats import test_create as df_test_create
from .test_dataformats import call as df_call
obj = 'user/integers/1'
from .test_dataformats import call as df_call
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)
obj = "user/integers/1"
storage = DFStorage(tmp_prefix, obj)
def call(*args, **kwargs):
'''A central mechanism to call the main routine with the right parameters'''
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)
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
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)
@nose.tools.with_setup(teardown=cleanup)
def test_local_list():
exit_code, outputs = call('list')
nose.tools.eq_(exit_code, 0, outputs)
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)
assert s.exists()
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)
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)
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')
# 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)
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)
def test_check_valid():
obj = 'legacy/valid_algorithm/1'
exit_code, outputs = call('check', obj)
nose.tools.eq_(exit_code, 0, outputs)
def test_check_invalid():
obj = 'legacy/no_inputs_declarations/1'
exit_code, outputs = call('check', obj)
nose.tools.eq_(exit_code, 1, outputs)
@nose.tools.with_setup(setup=setup, teardown=cleanup)
def test_create(obj=None):
obj = obj or 'legacy/algorithm/1'
exit_code, outputs = call('create', obj, prefix=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
s = Storage(tmp_prefix, obj)
assert s.exists()
return s
@nose.tools.with_setup(setup=setup, teardown=cleanup)
def test_new_version():
obj = 'legacy/algorithm/1'
test_create(obj)
obj2 = 'legacy/algorithm/2'
exit_code, outputs = call('version', obj, prefix=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
s = Storage(tmp_prefix, obj2)
assert s.exists()
# check version status
with common.Selector(tmp_prefix) as selector:
assert selector.version_of('algorithm', obj2) == obj
@nose.tools.with_setup(setup=setup, teardown=cleanup)
def test_fork():
obj = 'legacy/algorithm/1'
test_create(obj)
obj2 = 'legacy/different/1'
exit_code, outputs = call('fork', obj, obj2, prefix=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
s = Storage(tmp_prefix, obj2)
assert s.exists()
# check fork status
with common.Selector(tmp_prefix) as selector:
assert selector.forked_from('algorithm', obj2) == obj
@nose.tools.with_setup(setup=setup, teardown=cleanup)
def test_delete_local():
obj = 'legacy/algorithm/1'
storage = test_create(obj)
# quickly make sure it exists
storage = Storage(tmp_prefix, obj)
assert storage.exists()
exit_code, outputs = call('rm', obj, prefix=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
assert not storage.exists()
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'
test_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)
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)
class TestAlgorithmLocal(core.AssetLocalTest):
storage_cls = Storage
asset_type = "algorithm"
object_map = {
"valid": "legacy/valid_algorithm/1",
"invalid": "user/invalid/1",
"create": "legacy/new_algorithm/1",
"new": "legacy/new_algorithm/2",
"fork": "legacy/forked_algorithm/1",
}
def setup(self):
obj = "user/integers/1"
storage = DFStorage(tmp_prefix, obj)
if not storage.exists():
exit_code, outputs = self.call(
"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)
Markdown is supported
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