split models and db from init
Move models into their own file Move db functions into their own file Update tests
This commit is contained in:
parent
76528d1500
commit
58b3567887
11 changed files with 1064 additions and 981 deletions
|
|
@ -5,29 +5,18 @@ from tinydb.storages import MemoryStorage
|
|||
from tinydb import TinyDB, Query
|
||||
import bumper
|
||||
import os
|
||||
import datetime, time
|
||||
import platform
|
||||
import json
|
||||
import asyncio
|
||||
from testfixtures import LogCapture
|
||||
|
||||
|
||||
def test_get_milli_time():
|
||||
assert (
|
||||
bumper.get_milli_time(
|
||||
datetime.datetime(
|
||||
2018, 1, 1, 1, 0, 0, 0, tzinfo=datetime.timezone.utc
|
||||
).timestamp()
|
||||
)
|
||||
== 1514768400000
|
||||
)
|
||||
|
||||
|
||||
def test_strtobool():
|
||||
assert bumper.strtobool("t") == True
|
||||
assert bumper.strtobool("f") == False
|
||||
assert bumper.strtobool(0) == False
|
||||
|
||||
|
||||
async def test_start_stop():
|
||||
with LogCapture() as l:
|
||||
if os.path.exists("tests/tmp.db"):
|
||||
|
|
@ -36,12 +25,12 @@ async def test_start_stop():
|
|||
b = bumper
|
||||
b.db = "tests/tmp.db" # Set db location for testing
|
||||
b.conf1_listen_address = "127.0.0.1"
|
||||
b.conf1_listen_port = 444
|
||||
b.conf1_listen_port = 444
|
||||
asyncio.create_task(b.start())
|
||||
await asyncio.sleep(0.1)
|
||||
l.check_present(("bumper", "INFO", "Starting Bumper"))
|
||||
l.clear()
|
||||
|
||||
|
||||
asyncio.create_task(b.shutdown())
|
||||
await asyncio.sleep(0.1)
|
||||
l.check_present(
|
||||
|
|
@ -71,142 +60,3 @@ async def test_start_stop_debug():
|
|||
)
|
||||
assert b.shutting_down == True
|
||||
|
||||
|
||||
def test_db_path():
|
||||
bumper.db = None
|
||||
assert bumper.db_file() == os.path.join(bumper.data_dir, "bumper.db")
|
||||
|
||||
|
||||
def test_user_db():
|
||||
|
||||
bumper.db = "tests/tmp.db" # Set db location for testing
|
||||
bumper.user_add("testuser") # Add testuser
|
||||
|
||||
assert (
|
||||
bumper.user_get("testuser")["userid"] == "testuser"
|
||||
) # Test that testuser was created and returned
|
||||
|
||||
bumper.user_add_device("testuser", "dev_1234") # Add device to testuser
|
||||
|
||||
assert (
|
||||
bumper.user_by_deviceid("dev_1234")["userid"] == "testuser"
|
||||
) # Test that testuser was found by deviceid
|
||||
|
||||
bumper.user_remove_device("testuser", "dev_1234") # Remove device from testuser
|
||||
|
||||
assert "dev_1234" not in bumper.user_get("testuser")["devices"]
|
||||
# Test that dev_1234 was not found in testuser devices
|
||||
|
||||
bumper.user_add_bot("testuser", "bot_1234") # Add bot did to testuser
|
||||
|
||||
assert "bot_1234" in bumper.user_get("testuser")["bots"]
|
||||
# Test that bot was found in testuser's bot list
|
||||
|
||||
bumper.user_remove_bot("testuser", "bot_1234") # Remove bot did from testuser
|
||||
|
||||
assert "bot_1234" not in bumper.user_get("testuser")["bots"]
|
||||
# Test that bot was not found in testuser's bot list
|
||||
|
||||
bumper.user_add_token("testuser", "token_1234") # Add token to testuser
|
||||
|
||||
assert bumper.check_token("testuser", "token_1234")
|
||||
# Test that token was found for testuser
|
||||
|
||||
assert bumper.user_get_token("testuser", "token_1234")
|
||||
# Test that token was returned for testuser
|
||||
|
||||
bumper.user_add_authcode(
|
||||
"testuser", "token_1234", "auth_1234"
|
||||
) # Add authcode to token_1234 for testuser
|
||||
assert bumper.check_authcode("testuser", "auth_1234")
|
||||
# Test that authcode was found for testuser
|
||||
|
||||
bumper.user_revoke_authcode(
|
||||
"testuser", "token_1234", "auth_1234"
|
||||
) # Remove authcode from testuser
|
||||
assert bumper.check_authcode("testuser", "auth_1234") == False
|
||||
# Test that authcode was not found for testuser
|
||||
bumper.user_revoke_token("testuser", "token_1234") # Remove token from testuser
|
||||
assert (
|
||||
bumper.check_token("testuser", "token_1234") == False
|
||||
) # Test that token was not found for testuser
|
||||
bumper.user_add_token("testuser", "token_1234") # Add token_1234
|
||||
bumper.user_add_token("testuser", "token_4321") # Add token_4321
|
||||
assert len(bumper.user_get_tokens("testuser")) == 2 # Test 2 tokens are available
|
||||
bumper.user_revoke_all_tokens("testuser") # Revoke all tokens
|
||||
assert len(bumper.user_get_tokens("testuser")) == 0 # Test 0 tokens are available
|
||||
|
||||
db = TinyDB("tests/tmp.db")
|
||||
tokens = db.table("tokens")
|
||||
tokens.insert(
|
||||
{
|
||||
"userid": "testuser",
|
||||
"token": "token_1234",
|
||||
"expiration": "{}".format(
|
||||
datetime.datetime.now() + datetime.timedelta(seconds=-10)
|
||||
),
|
||||
}
|
||||
) # Add expired token
|
||||
db.close()
|
||||
assert len(bumper.user_get_tokens("testuser")) == 1 # Test 1 tokens are available
|
||||
bumper.user_revoke_expired_tokens("testuser") # Revoke expired tokens
|
||||
assert len(bumper.user_get_tokens("testuser")) == 0 # Test 0 tokens are available
|
||||
|
||||
db = TinyDB("tests/tmp.db")
|
||||
tokens = db.table("tokens")
|
||||
tokens.insert(
|
||||
{
|
||||
"userid": "testuser",
|
||||
"token": "token_1234",
|
||||
"expiration": "{}".format(
|
||||
datetime.datetime.now() + datetime.timedelta(seconds=-10)
|
||||
),
|
||||
}
|
||||
) # Add expired token
|
||||
db.close()
|
||||
assert len(bumper.user_get_tokens("testuser")) == 1 # Test 1 tokens are available
|
||||
bumper.revoke_expired_tokens() # Revoke expired tokens
|
||||
assert len(bumper.user_get_tokens("testuser")) == 0 # Test 0 tokens are available
|
||||
|
||||
|
||||
def test_bot_db():
|
||||
bumper.db = "tests/tmp.db" # Set db location for testing
|
||||
bumper.bot_add("sn_123", "did_123", "dev_123", "res_123", "co_123")
|
||||
assert bumper.bot_get("did_123") # Test that bot was added to db
|
||||
|
||||
bumper.bot_set_nick("did_123", "nick_123")
|
||||
assert (
|
||||
bumper.bot_get("did_123")["nick"] == "nick_123"
|
||||
) # Test that nick was added to bot
|
||||
|
||||
bumper.bot_set_mqtt("did_123", True)
|
||||
assert bumper.bot_get("did_123")[
|
||||
"mqtt_connection"
|
||||
] # Test that mqtt was set True for bot
|
||||
|
||||
bumper.bot_set_xmpp("did_123", True)
|
||||
assert bumper.bot_get("did_123")[
|
||||
"xmpp_connection"
|
||||
] # Test that xmpp was set True for bot
|
||||
|
||||
bumper.bot_remove("did_123")
|
||||
assert bumper.bot_get("did_123") == None # Test that bot is no longer in db
|
||||
|
||||
|
||||
def test_client_db():
|
||||
bumper.db = "tests/tmp.db" # Set db location for testing
|
||||
bumper.client_add("user_123", "realm_123", "resource_123")
|
||||
assert bumper.client_get("resource_123") # Test client was added
|
||||
|
||||
bumper.client_set_mqtt("resource_123", True)
|
||||
assert bumper.client_get("resource_123")[
|
||||
"mqtt_connection"
|
||||
] # Test that mqtt was set True for client
|
||||
|
||||
bumper.client_set_xmpp("resource_123", False)
|
||||
assert (
|
||||
bumper.client_get("resource_123")["xmpp_connection"] == False
|
||||
) # Test that xmpp was set False for client
|
||||
assert (
|
||||
len(bumper.get_disconnected_xmpp_clients()) == 1
|
||||
) # Test len of connected xmpp clients is 1
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue