mirror of
https://github.com/bigchaindb/bigchaindb.git
synced 2024-10-13 13:34:05 +00:00
b4063dd9ad
* Remove unnecessary import of rethinkdb in a test * Move test_run_query_util as a rethinkdb-specific test * Move rethinkdb-specific command tests to own file * Add __init__.py to new test folders * Move command tests to their own test module * Move fixtures out of command tests into separate conftest for commands * Fix some small flake8 issues with rethinkdb's test_run_query_util * Add pytest ignore hook to filter out test dirs that are not for the requested backend * Move backend-specific tests in tests/db/ to tests/backend/
125 lines
3.9 KiB
Python
125 lines
3.9 KiB
Python
import pytest
|
|
import rethinkdb as r
|
|
|
|
import bigchaindb
|
|
from bigchaindb import backend
|
|
from bigchaindb.backend.rethinkdb import schema
|
|
|
|
|
|
@pytest.mark.usefixtures('setup_database')
|
|
def test_init_creates_db_tables_and_indexes():
|
|
from bigchaindb.backend.schema import init_database
|
|
conn = backend.connect()
|
|
dbname = bigchaindb.config['database']['name']
|
|
|
|
# The db is set up by fixtures so we need to remove it
|
|
conn.run(r.db_drop(dbname))
|
|
|
|
init_database()
|
|
|
|
assert conn.run(r.db_list().contains(dbname)) is True
|
|
|
|
assert conn.run(r.db(dbname).table_list().contains('backlog', 'bigchain')) is True
|
|
|
|
assert conn.run(r.db(dbname).table('bigchain').index_list().contains(
|
|
'block_timestamp')) is True
|
|
|
|
assert conn.run(r.db(dbname).table('backlog').index_list().contains(
|
|
'assignee__transaction_timestamp')) is True
|
|
|
|
|
|
@pytest.mark.usefixtures('setup_database')
|
|
def test_init_database_fails_if_db_exists():
|
|
from bigchaindb.backend.schema import init_database
|
|
from bigchaindb.common import exceptions
|
|
|
|
conn = backend.connect()
|
|
dbname = bigchaindb.config['database']['name']
|
|
|
|
# The db is set up by fixtures
|
|
assert conn.run(r.db_list().contains(dbname)) is True
|
|
|
|
with pytest.raises(exceptions.DatabaseAlreadyExists):
|
|
init_database()
|
|
|
|
|
|
def test_create_database():
|
|
conn = backend.connect()
|
|
dbname = bigchaindb.config['database']['name']
|
|
schema.create_database(conn, dbname)
|
|
assert conn.run(r.db_list().contains(dbname)) is True
|
|
|
|
|
|
@pytest.mark.usefixtures('setup_database')
|
|
def test_create_tables():
|
|
conn = backend.connect()
|
|
dbname = bigchaindb.config['database']['name']
|
|
|
|
# The db is set up by fixtures so we need to remove it
|
|
# and recreate it just with one table
|
|
conn.run(r.db_drop(dbname))
|
|
schema.create_database(conn, dbname)
|
|
schema.create_tables(conn, dbname)
|
|
|
|
assert conn.run(r.db(dbname).table_list().contains('bigchain')) is True
|
|
assert conn.run(r.db(dbname).table_list().contains('backlog')) is True
|
|
assert conn.run(r.db(dbname).table_list().contains('votes')) is True
|
|
assert len(conn.run(r.db(dbname).table_list())) == 3
|
|
|
|
|
|
@pytest.mark.usefixtures('setup_database')
|
|
def test_create_secondary_indexes():
|
|
conn = backend.connect()
|
|
dbname = bigchaindb.config['database']['name']
|
|
|
|
# The db is set up by fixtures so we need to remove it
|
|
# and recreate it just with one table
|
|
conn.run(r.db_drop(dbname))
|
|
schema.create_database(conn, dbname)
|
|
schema.create_tables(conn, dbname)
|
|
schema.create_indexes(conn, dbname)
|
|
|
|
# Bigchain table
|
|
assert conn.run(r.db(dbname).table('bigchain').index_list().contains(
|
|
'block_timestamp')) is True
|
|
assert conn.run(r.db(dbname).table('bigchain').index_list().contains(
|
|
'transaction_id')) is True
|
|
assert conn.run(r.db(dbname).table('bigchain').index_list().contains(
|
|
'asset_id')) is True
|
|
|
|
# Backlog table
|
|
assert conn.run(r.db(dbname).table('backlog').index_list().contains(
|
|
'assignee__transaction_timestamp')) is True
|
|
|
|
# Votes table
|
|
assert conn.run(r.db(dbname).table('votes').index_list().contains(
|
|
'block_and_voter')) is True
|
|
|
|
|
|
@pytest.mark.usefixtures('setup_database')
|
|
def test_drop():
|
|
conn = backend.connect()
|
|
dbname = bigchaindb.config['database']['name']
|
|
|
|
# The db is set up by fixtures
|
|
assert conn.run(r.db_list().contains(dbname)) is True
|
|
|
|
schema.drop_database(conn, dbname)
|
|
assert conn.run(r.db_list().contains(dbname)) is False
|
|
|
|
|
|
@pytest.mark.usefixtures('setup_database')
|
|
def test_drop_non_existent_db_raises_an_error():
|
|
from bigchaindb.common import exceptions
|
|
|
|
conn = backend.connect()
|
|
dbname = bigchaindb.config['database']['name']
|
|
|
|
# The db is set up by fixtures
|
|
assert conn.run(r.db_list().contains(dbname)) is True
|
|
|
|
schema.drop_database(conn, dbname)
|
|
|
|
with pytest.raises(exceptions.DatabaseDoesNotExist):
|
|
schema.drop_database(conn, dbname)
|