Reworking tests
Reworking tests with pytest
This commit is contained in:
Brian Martin
parent
c78202c347
commit
d4414bd150
7 changed files with 556 additions and 439 deletions
|
|
@ -1,16 +1,22 @@
|
|||
from nose.tools import *
|
||||
import mock
|
||||
import bumper
|
||||
import asyncio
|
||||
import pytest
|
||||
import os
|
||||
import json
|
||||
import tinydb
|
||||
from aiohttp.test_utils import TestClient, TestServer, loop_context
|
||||
from aiohttp import request
|
||||
import pytest_aiohttp
|
||||
from aiohttp import web
|
||||
|
||||
confserver = bumper.ConfServer("127.0.0.1:11111", False, mock.MagicMock)
|
||||
confserver.confserver_app()
|
||||
app = confserver.app
|
||||
|
||||
def create_confserver():
|
||||
return bumper.ConfServer("127.0.0.1:11111", False, mock.MagicMock)
|
||||
|
||||
|
||||
def create_app(loop):
|
||||
confserver = bumper.ConfServer("127.0.0.1:11111", False, mock.MagicMock)
|
||||
confserver.confserver_app(loop=loop)
|
||||
return confserver.app
|
||||
|
||||
|
||||
def async_return(result):
|
||||
|
|
@ -19,151 +25,115 @@ def async_return(result):
|
|||
return f
|
||||
|
||||
|
||||
def test_disconnect():
|
||||
async def test_disconnect_async():
|
||||
await confserver.disconnect()
|
||||
|
||||
loop = asyncio.get_event_loop()
|
||||
# Test
|
||||
loop.run_until_complete(test_disconnect_async())
|
||||
|
||||
|
||||
def test_base():
|
||||
async def test_base(aiohttp_client):
|
||||
if os.path.exists("tests/tmp.db"):
|
||||
os.remove("tests/tmp.db") # Remove existing db
|
||||
bumper.db = "tests/tmp.db" # Set db location for testing
|
||||
loop = asyncio.get_event_loop()
|
||||
client = TestClient(TestServer(app), loop=loop)
|
||||
loop.run_until_complete(client.start_server())
|
||||
root = "http://{}".format(confserver.address)
|
||||
client = await aiohttp_client(create_app)
|
||||
|
||||
async def test_handle_base():
|
||||
resp = await client.get("/")
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
assert "Bumper!" in text
|
||||
|
||||
# Test
|
||||
loop.run_until_complete(test_handle_base())
|
||||
|
||||
loop.run_until_complete(
|
||||
client.close()
|
||||
) # Close test server after all tests are done
|
||||
resp = await client.get("/")
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
assert "Bumper!" in text
|
||||
|
||||
|
||||
def test_login():
|
||||
async def test_login(aiohttp_client):
|
||||
if os.path.exists("tests/tmp.db"):
|
||||
os.remove("tests/tmp.db") # Remove existing db
|
||||
bumper.db = "tests/tmp.db" # Set db location for testing
|
||||
loop = asyncio.get_event_loop()
|
||||
client = TestClient(TestServer(app), loop=loop)
|
||||
loop.run_until_complete(client.start_server())
|
||||
root = "http://{}".format(confserver.address)
|
||||
client = await aiohttp_client(create_app)
|
||||
|
||||
async def test_handle_login():
|
||||
resp = await client.get("/1/private/us/en/dev_1234/ios/1/0/0/user/login")
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
if jsonresp:
|
||||
assert jsonresp["code"] == bumper.RETURN_API_SUCCESS
|
||||
assert "accessToken" in jsonresp["data"]
|
||||
assert "uid" in jsonresp["data"]
|
||||
assert "username" in jsonresp["data"]
|
||||
else:
|
||||
assert jsonresp
|
||||
|
||||
# Test
|
||||
loop.run_until_complete(test_handle_login())
|
||||
# Test without user
|
||||
resp = await client.get("/1/private/us/en/dev_1234/ios/1/0/0/user/login")
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
assert jsonresp["code"] == bumper.RETURN_API_SUCCESS
|
||||
assert "accessToken" in jsonresp["data"]
|
||||
assert "uid" in jsonresp["data"]
|
||||
assert "username" in jsonresp["data"]
|
||||
|
||||
# Add a user to db and test with existing users
|
||||
bumper.user_add("testuser")
|
||||
# Test
|
||||
loop.run_until_complete(test_handle_login())
|
||||
resp = await client.get("/1/private/us/en/dev_1234/ios/1/0/0/user/login")
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
assert jsonresp["code"] == bumper.RETURN_API_SUCCESS
|
||||
assert "accessToken" in jsonresp["data"]
|
||||
assert "uid" in jsonresp["data"]
|
||||
assert "username" in jsonresp["data"]
|
||||
|
||||
# Add a bot to db that will be added to user
|
||||
bumper.bot_add("sn_123", "did_123", "dev_123", "res_123", "com_123")
|
||||
# Test
|
||||
loop.run_until_complete(test_handle_login())
|
||||
|
||||
loop.run_until_complete(
|
||||
client.close()
|
||||
) # Close test server after all tests are done
|
||||
resp = await client.get("/1/private/us/en/dev_1234/ios/1/0/0/user/login")
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
assert jsonresp["code"] == bumper.RETURN_API_SUCCESS
|
||||
assert "accessToken" in jsonresp["data"]
|
||||
assert "uid" in jsonresp["data"]
|
||||
assert "username" in jsonresp["data"]
|
||||
|
||||
|
||||
def test_logout():
|
||||
async def test_logout(aiohttp_client):
|
||||
if os.path.exists("tests/tmp.db"):
|
||||
os.remove("tests/tmp.db") # Remove existing db
|
||||
bumper.db = "tests/tmp.db" # Set db location for testing
|
||||
loop = asyncio.get_event_loop()
|
||||
client = TestClient(TestServer(app), loop=loop)
|
||||
loop.run_until_complete(client.start_server())
|
||||
root = "http://{}".format(confserver.address)
|
||||
|
||||
async def test_handle_logout(token=None):
|
||||
resp = await client.get(
|
||||
"/1/private/us/en/dev_1234/ios/1/0/0/user/logout?accessToken={}".format(
|
||||
token
|
||||
)
|
||||
)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
if jsonresp:
|
||||
assert jsonresp["code"] == bumper.RETURN_API_SUCCESS
|
||||
else:
|
||||
assert jsonresp
|
||||
client = await aiohttp_client(create_app)
|
||||
|
||||
# Add a token to user and test
|
||||
bumper.user_add("testuser")
|
||||
bumper.user_add_device("testuser", "dev_1234")
|
||||
bumper.user_add_token("testuser", "token_1234")
|
||||
# Test
|
||||
loop.run_until_complete(test_handle_logout(token="token_1234"))
|
||||
resp = await client.get(
|
||||
"/1/private/us/en/dev_1234/ios/1/0/0/user/logout?accessToken={}".format(
|
||||
"token_1234"
|
||||
)
|
||||
)
|
||||
|
||||
loop.run_until_complete(
|
||||
client.close()
|
||||
) # Close test server after all tests are done
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
assert jsonresp["code"] == bumper.RETURN_API_SUCCESS
|
||||
|
||||
|
||||
def test_checkLogin():
|
||||
async def test_checkLogin(aiohttp_client):
|
||||
if os.path.exists("tests/tmp.db"):
|
||||
os.remove("tests/tmp.db") # Remove existing db
|
||||
bumper.db = "tests/tmp.db" # Set db location for testing
|
||||
loop = asyncio.get_event_loop()
|
||||
client = TestClient(TestServer(app), loop=loop)
|
||||
loop.run_until_complete(client.start_server())
|
||||
root = "http://{}".format(confserver.address)
|
||||
client = await aiohttp_client(create_app)
|
||||
|
||||
async def test_handle_checkLogin(token=None):
|
||||
resp = await client.get(
|
||||
"/1/private/us/en/dev_1234/ios/1/0/0/user/checkLogin?accessToken={}".format(
|
||||
token
|
||||
)
|
||||
# Test without token
|
||||
resp = await client.get(
|
||||
"/1/private/us/en/dev_1234/ios/1/0/0/user/checkLogin?accessToken={}".format(
|
||||
None
|
||||
)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
if jsonresp:
|
||||
assert jsonresp["code"] == bumper.RETURN_API_SUCCESS
|
||||
assert "accessToken" in jsonresp["data"]
|
||||
if not token:
|
||||
assert jsonresp["data"]["accessToken"] != "token_1234"
|
||||
else:
|
||||
assert jsonresp["data"]["accessToken"] == "token_1234"
|
||||
|
||||
assert "uid" in jsonresp["data"]
|
||||
assert "username" in jsonresp["data"]
|
||||
else:
|
||||
assert jsonresp
|
||||
|
||||
# Test
|
||||
loop.run_until_complete(test_handle_checkLogin())
|
||||
)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
assert jsonresp["code"] == bumper.RETURN_API_SUCCESS
|
||||
assert "accessToken" in jsonresp["data"]
|
||||
assert jsonresp["data"]["accessToken"] != "token_1234"
|
||||
assert "uid" in jsonresp["data"]
|
||||
assert "username" in jsonresp["data"]
|
||||
|
||||
# Add a user to db and test with existing users
|
||||
bumper.user_add("testuser")
|
||||
# Test
|
||||
loop.run_until_complete(test_handle_checkLogin())
|
||||
resp = await client.get(
|
||||
"/1/private/us/en/dev_1234/ios/1/0/0/user/checkLogin?accessToken={}".format(
|
||||
None
|
||||
)
|
||||
)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
assert jsonresp["code"] == bumper.RETURN_API_SUCCESS
|
||||
assert "accessToken" in jsonresp["data"]
|
||||
assert jsonresp["data"]["accessToken"] != "token_1234"
|
||||
assert "uid" in jsonresp["data"]
|
||||
assert "username" in jsonresp["data"]
|
||||
|
||||
# Remove dev from tmpuser
|
||||
bumper.user_remove_device("tmpuser", "dev_1234")
|
||||
|
|
@ -172,231 +142,156 @@ def test_checkLogin():
|
|||
bumper.user_add("testuser")
|
||||
bumper.user_add_device("testuser", "dev_1234")
|
||||
bumper.user_add_token("testuser", "token_1234")
|
||||
# Test
|
||||
loop.run_until_complete(test_handle_checkLogin(token="token_1234"))
|
||||
|
||||
loop.run_until_complete(
|
||||
client.close()
|
||||
) # Close test server after all tests are done
|
||||
resp = await client.get(
|
||||
"/1/private/us/en/dev_1234/ios/1/0/0/user/checkLogin?accessToken={}".format(
|
||||
"token_1234"
|
||||
)
|
||||
)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
assert jsonresp["code"] == bumper.RETURN_API_SUCCESS
|
||||
assert "accessToken" in jsonresp["data"]
|
||||
assert jsonresp["data"]["accessToken"] == "token_1234"
|
||||
assert "uid" in jsonresp["data"]
|
||||
assert "username" in jsonresp["data"]
|
||||
|
||||
|
||||
def test_getAuthCode():
|
||||
async def test_getAuthCode(aiohttp_client):
|
||||
if os.path.exists("tests/tmp.db"):
|
||||
os.remove("tests/tmp.db") # Remove existing db
|
||||
bumper.db = "tests/tmp.db" # Set db location for testing
|
||||
loop = asyncio.get_event_loop()
|
||||
client = TestClient(TestServer(app), loop=loop)
|
||||
loop.run_until_complete(client.start_server())
|
||||
root = "http://{}".format(confserver.address)
|
||||
|
||||
async def test_handle_getAuthCode(uid=None, token=None):
|
||||
resp = await client.get(
|
||||
"/1/private/us/en/dev_1234/ios/1/0/0/user/getAuthCode?uid={}&accessToken={}".format(
|
||||
uid, token
|
||||
)
|
||||
)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
if jsonresp:
|
||||
if token:
|
||||
assert jsonresp["code"] == bumper.RETURN_API_SUCCESS
|
||||
assert "authCode" in jsonresp["data"]
|
||||
assert "ecovacsUid" in jsonresp["data"]
|
||||
else:
|
||||
assert jsonresp["code"] == bumper.ERR_TOKEN_INVALID
|
||||
else:
|
||||
assert jsonresp
|
||||
client = await aiohttp_client(create_app)
|
||||
|
||||
# Test without user or token
|
||||
loop.run_until_complete(test_handle_getAuthCode())
|
||||
resp = await client.get(
|
||||
"/1/private/us/en/dev_1234/ios/1/0/0/user/getAuthCode?uid={}&accessToken={}".format(
|
||||
None, None
|
||||
)
|
||||
)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
assert jsonresp["code"] == bumper.ERR_TOKEN_INVALID
|
||||
|
||||
# Add a token to user and test
|
||||
bumper.user_add("testuser")
|
||||
bumper.user_add_device("testuser", "dev_1234")
|
||||
bumper.user_add_token("testuser", "token_1234")
|
||||
# Test
|
||||
loop.run_until_complete(test_handle_getAuthCode(uid="testuser", token="token_1234"))
|
||||
resp = await client.get(
|
||||
"/1/private/us/en/dev_1234/ios/1/0/0/user/getAuthCode?uid={}&accessToken={}".format(
|
||||
"testuser", "token_1234"
|
||||
)
|
||||
)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
assert jsonresp["code"] == bumper.RETURN_API_SUCCESS
|
||||
assert "authCode" in jsonresp["data"]
|
||||
assert "ecovacsUid" in jsonresp["data"]
|
||||
|
||||
# The above should have added an authcode to token, try again to test with existing authcode
|
||||
# Test
|
||||
loop.run_until_complete(test_handle_getAuthCode(uid="testuser", token="token_1234"))
|
||||
|
||||
loop.run_until_complete(
|
||||
client.close()
|
||||
) # Close test server after all tests are done
|
||||
|
||||
|
||||
def test_checkAgreement():
|
||||
if os.path.exists("tests/tmp.db"):
|
||||
os.remove("tests/tmp.db") # Remove existing db
|
||||
bumper.db = "tests/tmp.db" # Set db location for testing
|
||||
loop = asyncio.get_event_loop()
|
||||
client = TestClient(TestServer(app), loop=loop)
|
||||
loop.run_until_complete(client.start_server())
|
||||
root = "http://{}".format(confserver.address)
|
||||
|
||||
async def test_handle_checkAgreement():
|
||||
resp = await client.get(
|
||||
"/1/private/us/en/dev_1234/ios/1/0/0/user/checkAgreement"
|
||||
resp = await client.get(
|
||||
"/1/private/us/en/dev_1234/ios/1/0/0/user/getAuthCode?uid={}&accessToken={}".format(
|
||||
"testuser", "token_1234"
|
||||
)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
if jsonresp:
|
||||
assert jsonresp["code"] == bumper.RETURN_API_SUCCESS
|
||||
else:
|
||||
assert jsonresp
|
||||
|
||||
# Test
|
||||
loop.run_until_complete(test_handle_checkAgreement())
|
||||
|
||||
loop.run_until_complete(
|
||||
client.close()
|
||||
) # Close test server after all tests are done
|
||||
)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
assert jsonresp["code"] == bumper.RETURN_API_SUCCESS
|
||||
assert "authCode" in jsonresp["data"]
|
||||
assert "ecovacsUid" in jsonresp["data"]
|
||||
|
||||
|
||||
def test_homePageAlert():
|
||||
async def test_checkAgreement(aiohttp_client):
|
||||
if os.path.exists("tests/tmp.db"):
|
||||
os.remove("tests/tmp.db") # Remove existing db
|
||||
bumper.db = "tests/tmp.db" # Set db location for testing
|
||||
loop = asyncio.get_event_loop()
|
||||
client = TestClient(TestServer(app), loop=loop)
|
||||
loop.run_until_complete(client.start_server())
|
||||
root = "http://{}".format(confserver.address)
|
||||
client = await aiohttp_client(create_app)
|
||||
|
||||
async def test_handle_homePageAlert():
|
||||
resp = await client.get(
|
||||
"/1/private/us/en/dev_1234/ios/1/0/0/campaign/homePageAlert"
|
||||
)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
if jsonresp:
|
||||
assert jsonresp["code"] == bumper.RETURN_API_SUCCESS
|
||||
else:
|
||||
assert jsonresp
|
||||
|
||||
# Test
|
||||
loop.run_until_complete(test_handle_homePageAlert())
|
||||
|
||||
loop.run_until_complete(
|
||||
client.close()
|
||||
) # Close test server after all tests are done
|
||||
resp = await client.get("/1/private/us/en/dev_1234/ios/1/0/0/user/checkAgreement")
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
assert jsonresp["code"] == bumper.RETURN_API_SUCCESS
|
||||
|
||||
|
||||
def test_checkVersion():
|
||||
async def test_homePageAlert(aiohttp_client):
|
||||
if os.path.exists("tests/tmp.db"):
|
||||
os.remove("tests/tmp.db") # Remove existing db
|
||||
bumper.db = "tests/tmp.db" # Set db location for testing
|
||||
loop = asyncio.get_event_loop()
|
||||
client = TestClient(TestServer(app), loop=loop)
|
||||
loop.run_until_complete(client.start_server())
|
||||
root = "http://{}".format(confserver.address)
|
||||
client = await aiohttp_client(create_app)
|
||||
|
||||
async def test_handle_checkVersion():
|
||||
resp = await client.get(
|
||||
"/1/private/us/en/dev_1234/ios/1/0/0/common/checkVersion"
|
||||
)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
if jsonresp:
|
||||
assert jsonresp["code"] == bumper.RETURN_API_SUCCESS
|
||||
else:
|
||||
assert jsonresp
|
||||
|
||||
# Test
|
||||
loop.run_until_complete(test_handle_checkVersion())
|
||||
|
||||
loop.run_until_complete(
|
||||
client.close()
|
||||
) # Close test server after all tests are done
|
||||
resp = await client.get(
|
||||
"/1/private/us/en/dev_1234/ios/1/0/0/campaign/homePageAlert"
|
||||
)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
assert jsonresp["code"] == bumper.RETURN_API_SUCCESS
|
||||
|
||||
|
||||
def test_getProductIotMap():
|
||||
async def test_checkVersion(aiohttp_client):
|
||||
if os.path.exists("tests/tmp.db"):
|
||||
os.remove("tests/tmp.db") # Remove existing db
|
||||
bumper.db = "tests/tmp.db" # Set db location for testing
|
||||
loop = asyncio.get_event_loop()
|
||||
client = TestClient(TestServer(app), loop=loop)
|
||||
loop.run_until_complete(client.start_server())
|
||||
root = "http://{}".format(confserver.address)
|
||||
client = await aiohttp_client(create_app)
|
||||
|
||||
async def test_handle_getProductIotMap():
|
||||
resp = await client.post("/api/pim/product/getProductIotMap")
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
if jsonresp:
|
||||
assert jsonresp["code"] == bumper.RETURN_API_SUCCESS
|
||||
else:
|
||||
assert jsonresp
|
||||
|
||||
# Test
|
||||
loop.run_until_complete(test_handle_getProductIotMap())
|
||||
|
||||
loop.run_until_complete(
|
||||
client.close()
|
||||
) # Close test server after all tests are done
|
||||
resp = await client.get("/1/private/us/en/dev_1234/ios/1/0/0/common/checkVersion")
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
assert jsonresp["code"] == bumper.RETURN_API_SUCCESS
|
||||
|
||||
|
||||
def test_getUsersAPI():
|
||||
async def test_getProductIotMap(aiohttp_client):
|
||||
if os.path.exists("tests/tmp.db"):
|
||||
os.remove("tests/tmp.db") # Remove existing db
|
||||
bumper.db = "tests/tmp.db" # Set db location for testing
|
||||
loop = asyncio.get_event_loop()
|
||||
client = TestClient(TestServer(app), loop=loop)
|
||||
loop.run_until_complete(client.start_server())
|
||||
root = "http://{}".format(confserver.address)
|
||||
client = await aiohttp_client(create_app)
|
||||
|
||||
async def test_handle_getUsersApi():
|
||||
resp = await client.get("/api/users/user.do")
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
if jsonresp:
|
||||
assert jsonresp["result"] == "fail"
|
||||
else:
|
||||
assert jsonresp
|
||||
|
||||
# Test
|
||||
loop.run_until_complete(test_handle_getUsersApi())
|
||||
|
||||
loop.run_until_complete(
|
||||
client.close()
|
||||
) # Close test server after all tests are done
|
||||
resp = await client.post("/api/pim/product/getProductIotMap")
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
assert jsonresp["code"] == bumper.RETURN_API_SUCCESS
|
||||
|
||||
|
||||
def test_postUsersAPI():
|
||||
async def test_getUsersAPI(aiohttp_client):
|
||||
if os.path.exists("tests/tmp.db"):
|
||||
os.remove("tests/tmp.db") # Remove existing db
|
||||
bumper.db = "tests/tmp.db" # Set db location for testing
|
||||
loop = asyncio.get_event_loop()
|
||||
client = TestClient(TestServer(app), loop=loop)
|
||||
loop.run_until_complete(client.start_server())
|
||||
root = "http://{}".format(confserver.address)
|
||||
client = await aiohttp_client(create_app)
|
||||
|
||||
async def test_handle_postUsersApi(postbody=None):
|
||||
resp = await client.post("/api/users/user.do", json=postbody)
|
||||
resp = await client.get("/api/users/user.do")
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
assert jsonresp["result"] == "fail"
|
||||
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
if jsonresp:
|
||||
assert jsonresp["result"] == "ok"
|
||||
else:
|
||||
assert jsonresp
|
||||
|
||||
async def test_postUsersAPI(aiohttp_client):
|
||||
if os.path.exists("tests/tmp.db"):
|
||||
os.remove("tests/tmp.db") # Remove existing db
|
||||
bumper.db = "tests/tmp.db" # Set db location for testing
|
||||
client = await aiohttp_client(create_app)
|
||||
|
||||
# Test FindBest
|
||||
postbody = {"todo": "FindBest", "service": "EcoMsgNew"}
|
||||
# Test
|
||||
loop.run_until_complete(test_handle_postUsersApi(postbody))
|
||||
resp = await client.post("/api/users/user.do", json=postbody)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
assert jsonresp["result"] == "ok"
|
||||
|
||||
# Test EcoUpdate
|
||||
postbody = {"todo": "FindBest", "service": "EcoUpdate"}
|
||||
# Test
|
||||
loop.run_until_complete(test_handle_postUsersApi(postbody))
|
||||
resp = await client.post("/api/users/user.do", json=postbody)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
assert jsonresp["result"] == "ok"
|
||||
|
||||
# Test loginByItToken - Uses the authcode
|
||||
bumper.user_add("testuser")
|
||||
|
|
@ -415,7 +310,11 @@ def test_postUsersAPI():
|
|||
"token": "auth_1234",
|
||||
"userId": "testuser",
|
||||
}
|
||||
loop.run_until_complete(test_handle_postUsersApi(postbody))
|
||||
resp = await client.post("/api/users/user.do", json=postbody)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
assert jsonresp["result"] == "ok"
|
||||
|
||||
# Test GetDeviceList
|
||||
postbody = {
|
||||
|
|
@ -429,7 +328,11 @@ def test_postUsersAPI():
|
|||
"todo": "GetDeviceList",
|
||||
"userid": "testuser",
|
||||
}
|
||||
loop.run_until_complete(test_handle_postUsersApi(postbody))
|
||||
resp = await client.post("/api/users/user.do", json=postbody)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
assert jsonresp["result"] == "ok"
|
||||
|
||||
# Test SetDeviceNick
|
||||
postbody = {
|
||||
|
|
@ -444,7 +347,11 @@ def test_postUsersAPI():
|
|||
"nick": "botnick",
|
||||
"did": "did_1234",
|
||||
}
|
||||
loop.run_until_complete(test_handle_postUsersApi(postbody))
|
||||
resp = await client.post("/api/users/user.do", json=postbody)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
assert jsonresp["result"] == "ok"
|
||||
|
||||
# Test AddOneDevice - Same as set nick for some bots
|
||||
postbody = {
|
||||
|
|
@ -459,7 +366,11 @@ def test_postUsersAPI():
|
|||
"nick": "botnick",
|
||||
"did": "did_1234",
|
||||
}
|
||||
loop.run_until_complete(test_handle_postUsersApi(postbody))
|
||||
resp = await client.post("/api/users/user.do", json=postbody)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
assert jsonresp["result"] == "ok"
|
||||
|
||||
# Test DeleteOneDevice - remove bot
|
||||
postbody = {
|
||||
|
|
@ -473,79 +384,50 @@ def test_postUsersAPI():
|
|||
"todo": "DeleteOneDevice",
|
||||
"did": "did_1234",
|
||||
}
|
||||
loop.run_until_complete(test_handle_postUsersApi(postbody))
|
||||
|
||||
loop.run_until_complete(
|
||||
client.close()
|
||||
) # Close test server after all tests are done
|
||||
resp = await client.post("/api/users/user.do", json=postbody)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
assert jsonresp["result"] == "ok"
|
||||
|
||||
|
||||
def test_postLookup():
|
||||
async def test_postLookup(aiohttp_client):
|
||||
if os.path.exists("tests/tmp.db"):
|
||||
os.remove("tests/tmp.db") # Remove existing db
|
||||
bumper.db = "tests/tmp.db" # Set db location for testing
|
||||
loop = asyncio.get_event_loop()
|
||||
client = TestClient(TestServer(app), loop=loop)
|
||||
loop.run_until_complete(client.start_server())
|
||||
root = "http://{}".format(confserver.address)
|
||||
|
||||
async def test_handle_lookup(postbody=None):
|
||||
resp = await client.post("/lookup.do", json=postbody)
|
||||
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
if jsonresp:
|
||||
assert jsonresp["result"] == "ok"
|
||||
else:
|
||||
assert jsonresp
|
||||
client = await aiohttp_client(create_app)
|
||||
|
||||
# Test FindBest
|
||||
postbody = {"todo": "FindBest", "service": "EcoMsgNew"}
|
||||
# Test
|
||||
loop.run_until_complete(test_handle_lookup(postbody))
|
||||
resp = await client.post("/lookup.do", json=postbody)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
test_resp = json.loads(text)
|
||||
assert test_resp["result"] == "ok"
|
||||
|
||||
# Test EcoUpdate
|
||||
postbody = {"todo": "FindBest", "service": "EcoUpdate"}
|
||||
# Test
|
||||
loop.run_until_complete(test_handle_lookup(postbody))
|
||||
|
||||
loop.run_until_complete(
|
||||
client.close()
|
||||
) # Close test server after all tests are done
|
||||
resp = await client.post("/lookup.do", json=postbody)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
test_resp = json.loads(text)
|
||||
assert test_resp["result"] == "ok"
|
||||
|
||||
|
||||
def test_devmgr():
|
||||
async def test_devmgr(aiohttp_client):
|
||||
if os.path.exists("tests/tmp.db"):
|
||||
os.remove("tests/tmp.db") # Remove existing db
|
||||
bumper.db = "tests/tmp.db" # Set db location for testing
|
||||
loop = asyncio.get_event_loop()
|
||||
client = TestClient(TestServer(app), loop=loop)
|
||||
loop.run_until_complete(client.start_server())
|
||||
root = "http://{}".format(confserver.address)
|
||||
|
||||
async def test_devmanager(postbody=None, command=False):
|
||||
resp = await client.post("/api/iot/devmanager.do", json=postbody)
|
||||
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
if jsonresp:
|
||||
if not command:
|
||||
assert jsonresp["ret"] == "ok"
|
||||
else:
|
||||
if "ret" in jsonresp:
|
||||
if jsonresp["ret"] == "ok":
|
||||
assert jsonresp["resp"]
|
||||
else:
|
||||
assert jsonresp["errno"]
|
||||
else:
|
||||
assert jsonresp
|
||||
confserver = create_confserver()
|
||||
client = await aiohttp_client(create_app)
|
||||
|
||||
# Test PollSCResult
|
||||
postbody = {"td": "PollSCResult"}
|
||||
# Test
|
||||
loop.run_until_complete(test_devmanager(postbody, command=False))
|
||||
resp = await client.post("/api/dim/devmanager.do", json=postbody)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
test_resp = json.loads(text)
|
||||
assert test_resp["ret"] == "ok"
|
||||
|
||||
# Test BotCommand
|
||||
bumper.bot_add("sn_1234", "did_1234", "dev_1234", "res_1234", "eco-ng")
|
||||
|
|
@ -561,25 +443,99 @@ def test_devmgr():
|
|||
confserver.helperbot.send_command = mock.MagicMock(
|
||||
return_value=async_return(command_getstatus_resp)
|
||||
)
|
||||
# Test
|
||||
loop.run_until_complete(test_devmanager(postbody, command=True))
|
||||
resp = await client.post("/api/dim/devmanager.do", json=postbody)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
test_resp = json.loads(text)
|
||||
assert test_resp["ret"] == "ok"
|
||||
|
||||
# Test return fail timeout
|
||||
command_timeout_resp = {"id": "resp_1234", "errno": "timeout", "ret": "fail"}
|
||||
confserver.helperbot.send_command = mock.MagicMock(
|
||||
return_value=async_return(command_timeout_resp)
|
||||
)
|
||||
# Test
|
||||
loop.run_until_complete(test_devmanager(postbody, command=True))
|
||||
resp = await client.post("/api/dim/devmanager.do", json=postbody)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
test_resp = json.loads(text)
|
||||
assert test_resp["ret"] == "fail"
|
||||
|
||||
# Set bot not on mqtt
|
||||
bumper.bot_set_mqtt("did_1234", False)
|
||||
confserver.helperbot.send_command = mock.MagicMock(
|
||||
return_value=async_return(command_getstatus_resp)
|
||||
)
|
||||
# Test
|
||||
loop.run_until_complete(test_devmanager(postbody, command=True))
|
||||
resp = await client.post("/api/dim/devmanager.do", json=postbody)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
test_resp = json.loads(text)
|
||||
assert test_resp["ret"] == "fail"
|
||||
|
||||
|
||||
async def test_dim_devmanager(aiohttp_client):
|
||||
if os.path.exists("tests/tmp.db"):
|
||||
os.remove("tests/tmp.db") # Remove existing db
|
||||
bumper.db = "tests/tmp.db" # Set db location for testing
|
||||
confserver = create_confserver()
|
||||
client = await aiohttp_client(create_app)
|
||||
|
||||
# Test PollSCResult
|
||||
postbody = {"td": "PollSCResult"}
|
||||
resp = await client.post("/api/dim/devmanager.do", json=postbody)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
test_resp = json.loads(text)
|
||||
assert test_resp["ret"] == "ok"
|
||||
|
||||
# Test HasUnreadMsg
|
||||
postbody = {"td": "HasUnreadMsg"}
|
||||
resp = await client.post("/api/dim/devmanager.do", json=postbody)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
test_resp = json.loads(text)
|
||||
assert test_resp["ret"] == "ok"
|
||||
assert test_resp["unRead"] == False
|
||||
|
||||
# Test BotCommand
|
||||
bumper.bot_add("sn_1234", "did_1234", "dev_1234", "res_1234", "eco-ng")
|
||||
bumper.bot_set_mqtt("did_1234", True)
|
||||
postbody = {"toId": "did_1234"}
|
||||
|
||||
# Test return get status
|
||||
command_getstatus_resp = {
|
||||
"id": "resp_1234",
|
||||
"resp": "<ctl ret='ok' status='idle'/>",
|
||||
"ret": "ok",
|
||||
}
|
||||
confserver.helperbot.send_command = mock.MagicMock(
|
||||
return_value=async_return(command_getstatus_resp)
|
||||
)
|
||||
resp = await client.post("/api/dim/devmanager.do", json=postbody)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
test_resp = json.loads(text)
|
||||
assert test_resp["ret"] == "ok"
|
||||
|
||||
# Test return fail timeout
|
||||
command_timeout_resp = {"id": "resp_1234", "errno": "timeout", "ret": "fail"}
|
||||
confserver.helperbot.send_command = mock.MagicMock(
|
||||
return_value=async_return(command_timeout_resp)
|
||||
)
|
||||
resp = await client.post("/api/dim/devmanager.do", json=postbody)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
test_resp = json.loads(text)
|
||||
assert test_resp["ret"] == "fail"
|
||||
assert test_resp["errno"] == "timeout"
|
||||
|
||||
# Set bot not on mqtt
|
||||
bumper.bot_set_mqtt("did_1234", False)
|
||||
confserver.helperbot.send_command = mock.MagicMock(
|
||||
return_value=async_return(command_getstatus_resp)
|
||||
)
|
||||
resp = await client.post("/api/dim/devmanager.do", json=postbody)
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
test_resp = json.loads(text)
|
||||
assert test_resp["ret"] == "fail"
|
||||
|
||||
loop.run_until_complete(
|
||||
client.close()
|
||||
) # Close test server after all tests are done
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue