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

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

parent ab70826e
......@@ -37,18 +37,15 @@
# Basic tests for the command line beat program: databases
import nose.tools
import click
from click.testing import CliRunner
from beat.backend.python.test.test_database import INTEGERS_DBS
from beat.backend.python.protocoltemplate import Storage as PTStorage
from beat.core.test.utils import slow, cleanup, skipif
from beat.core.test.utils import slow
from beat.core.database import Storage, Database, get_first_procotol_template
from beat.cmdline.scripts import main_cli
from . import core
from . import platform, disconnected, prefix, tmp_prefix, user, token
from . import prefix, tmp_prefix
class TestDatabaseLocal(core.AssetLocalTest):
......@@ -72,168 +69,130 @@ class TestDatabaseLocal(core.AssetLocalTest):
nose.tools.eq_(exit_code, 0, outputs)
def index_integer_db():
call("index", "integers_db/1")
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)
runner = CliRunner()
with runner.isolated_filesystem():
result = runner.invoke(
main_cli.main,
[
"--test-mode",
"--prefix",
use_prefix,
"--token",
token,
"--user",
user,
"--platform",
use_platform,
"databases",
]
+ list(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)
@nose.tools.with_setup(teardown=cleanup)
def test_local_list():
nose.tools.eq_(call("list"), 0)
@nose.tools.nottest
@slow
@nose.tools.with_setup(teardown=cleanup)
@skipif(disconnected, "missing test platform (%s)" % platform)
def test_pull_one():
obj = "simple/1"
nose.tools.eq_(call("pull", obj, prefix=tmp_prefix), 0)
s = Storage(tmp_prefix, obj)
nose.tools.assert_true(s.exists())
@nose.tools.nottest
@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 = "simple/1"
nose.tools.eq_(call("pull", obj, prefix=tmp_prefix), 0)
# quickly modify the user algorithm by emptying it
f = Database(tmp_prefix, obj)
nose.tools.eq_(
len(f.errors), 0, "Failed to load Database: \n%s" % "\n".join(f.errors)
)
f.data["root_folder"] = "/a/different/path"
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()
nose.tools.eq_(call("status", prefix=tmp_prefix), 0)
@nose.tools.with_setup(setup=index_integer_db, teardown=cleanup)
def test_view_good():
nose.tools.eq_(call("view", "integers_db/1/double/double"), 0)
@nose.tools.with_setup(setup=index_integer_db, teardown=cleanup)
def test_view_unknown_protocol():
nose.tools.eq_(call("view", "integers_db/1/single/double"), 1)
@nose.tools.with_setup(setup=index_integer_db, teardown=cleanup)
def test_view_unknown_set():
nose.tools.eq_(call("view", "integers_db/1/double/single"), 1)
@nose.tools.with_setup(setup=index_integer_db, teardown=cleanup)
def test_view_bad():
nose.tools.eq_(call("view", "integers_db/1/two_sets"), 1)
@nose.tools.with_setup(setup=index_integer_db, teardown=cleanup)
def test_view_invalid():
nose.tools.eq_(call("view", "invalid/1/default/set"), 1)
class TestOnlinDatabases(core.OnlineAssetTestCase):
asset_type = "database"
def test_index_unknown_database():
nose.tools.eq_(call("index", "foobar/1"), 1)
@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 = "simple/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 = "simple/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
f = Database(tmp_prefix, obj)
nose.tools.eq_(
len(f.errors), 0, "Failed to load Database: \n%s" % "\n".join(f.errors)
)
f.data["root_folder"] = "/a/different/path"
f.write()
exit_code, outputs = self.call("diff", obj, prefix=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
def test_index_method():
for db_name in INTEGERS_DBS:
yield index_good, db_name
@slow
@core.skip_disconnected
def test_status(self):
self.test_diff()
exit_code, outputs = self.call("status", prefix=tmp_prefix)
nose.tools.eq_(exit_code, 0, outputs)
@nose.tools.with_setup(teardown=cleanup)
def index_good(db_name):
nose.tools.eq_(call("index", db_name), 0)
class TestView(core.AssetBaseTest):
asset_type = "database"
def setUp(self):
self.index_integer_db()
def test_list_index_good():
for db_name in INTEGERS_DBS:
yield list_index_good, db_name
def index_integer_db(self):
self.call("index", "integers_db/1")
def test_view_good(self):
exit_code, outputs = self.call("view", "integers_db/1/double/double")
nose.tools.eq_(exit_code, 0, outputs)
@nose.tools.with_setup(teardown=cleanup)
def list_index_good(db_name):
nose.tools.eq_(call("index", db_name), 0)
nose.tools.eq_(call("index", "--list", db_name), 0)
def test_view_unknown_protocol(self):
exit_code, outputs = self.call("view", "integers_db/1/single/double")
nose.tools.eq_(exit_code, 1, outputs)
def test_view_unknown_set(self):
exit_code, outputs = self.call("view", "integers_db/1/double/single")
nose.tools.eq_(exit_code, 1, outputs)
def test_delete_index_good():
for db_name in INTEGERS_DBS:
yield delete_index_good, db_name
def test_view_bad(self):
exit_code, outputs = self.call("view", "integers_db/1/two_sets")
nose.tools.eq_(exit_code, 1, outputs)
def test_view_invalid(self):
exit_code, outputs = self.call("view", "invalid/1/default/set")
nose.tools.eq_(exit_code, 1, outputs)
@nose.tools.with_setup(teardown=cleanup)
def delete_index_good(db_name):
nose.tools.eq_(call("index", db_name), 0)
nose.tools.eq_(call("index", "--delete", db_name), 0)
class TestIndex(core.AssetBaseTest):
asset_type = "database"
@nose.tools.with_setup(teardown=cleanup)
def test_index_all(): # bad and good, return != 0
expected_errors = 16
existing_errors = call("index")
nose.tools.assert_true(
existing_errors >= expected_errors,
(
"There should be at least %d "
"errors on installed databases, but I've found only %d"
% (expected_errors, existing_errors)
),
)
def test_index_unknown_database(self):
exit_code, outputs = self.call("index", "foobar/1")
nose.tools.eq_(exit_code, 1, outputs)
def test_index_method(self):
for db_name in INTEGERS_DBS:
yield self.index_good, db_name
def index_good(self, db_name):
exit_code, outputs = self.call("index", db_name)
nose.tools.eq_(exit_code, 0, outputs)
def test_list_index_good(self):
for db_name in INTEGERS_DBS:
yield self.list_index_good, db_name
def list_index_good(self, db_name):
exit_code, outputs = self.call("index", db_name)
nose.tools.eq_(exit_code, 0, outputs)
exit_code, outputs = self.call("index", "--list", db_name)
nose.tools.eq_(exit_code, 0, outputs)
def test_delete_index_good(self):
for db_name in INTEGERS_DBS:
yield self.delete_index_good, db_name
def delete_index_good(self, db_name):
exit_code, outputs = self.call("index", db_name)
nose.tools.eq_(exit_code, 0, outputs)
exit_code, outputs = self.call("index", "--delete", db_name)
nose.tools.eq_(exit_code, 0, outputs)
def test_index_all(self): # bad and good, return != 0
expected_errors = 16
existing_errors, outputs = self.call("index")
nose.tools.assert_true(
existing_errors >= expected_errors,
(
"There should be at least %d "
"errors on installed databases, but I've found only %d"
% (expected_errors, existing_errors)
),
)
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