mia/bumper
1
0
Fork 0
Code Issues 47 Pull requests 11 Projects Releases Packages Wiki Activity Actions

Start adding tests for confserver

Browse source 
Start adding tests for confserver
This commit is contained in:
Brian Martin 2019-03-14 09:35:37 -04:00
parent 616729e97f
commit 747eeda4a5
3 changed files with 203 additions and 66 deletions
Download patch file Download diff file Expand all files Collapse all files

7
bumper/__init__.py
View file

@ -66,9 +66,9 @@ def db_get():
db = TinyDB(db_file())
# Will create the tables if they don't exist
users_table = db.table("users")
clients_table = db.table("clients")
bots_table = db.table("bots")
users_table = db.table("users", cache_size=0)
clients_table = db.table("clients", cache_size=0)
bots_table = db.table("bots", cache_size=0)
return db
@ -169,6 +169,7 @@ def user_add_token(userid, token):
tokens = db_get().table("tokens")
tmptoken = tokens.get((Query().userid == userid) & (Query().token == token))
if not tmptoken:
bumperlog.debug("Adding token {} for userid {}".format(token, userid))
tokens.insert(
{
"userid": userid,

132
bumper/confserver.py
View file

@ -42,6 +42,7 @@ class ConfServer:
self.usessl = usessl
self.address = address
self.confthread = None
self.app = None
def run(self, run_async=False):
try:
@ -72,62 +73,61 @@ class ConfServer:
loop = asyncio.new_event_loop()
try:
self.confserver_app()
loop.run_until_complete(self.start_server())
loop.run_forever()
except Exception as e:
confserverlog.exception("{}".format(e))
def confserver_app(self):
self.app = web.Application()
self.app.add_routes(
[
web.get("", self.handle_base),
web.get(
"/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/login",
self.handle_login,
),
web.get(
"/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/checkLogin",
self.handle_login,
),
web.get(
"/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/logout",
self.handle_logout,
),
web.get(
"/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/getAuthCode",
self.handle_getAuthCode,
),
web.get(
"/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/checkAgreement",
self.handle_checkAgreement,
),
web.get(
"/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/common/checkVersion",
self.handle_checkVersion,
),
web.get(
"/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/campaign/homePageAlert",
self.handle_homePageAlert,
),
web.post("/api/users/user.do", self.handle_usersapi),
web.get("/api/users/user.do", self.handle_usersapi),
web.post(
"/api/pim/product/getProductIotMap", self.handle_getProductIotMap
),
web.post("/api/iot/devmanager.do", self.handle_devmanager_botcommand),
web.post("/lookup.do", self.handle_lookup),
]
)
# Direct register from app:
# /{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/directRegister
async def start_server(self):
try:
app = web.Application()
app.add_routes(
[
web.get("", self.handle_base),
web.get(
"/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/login",
self.handle_login,
),
web.get(
"/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/checkLogin",
self.handle_login,
),
web.get(
"/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/logout",
self.handle_logout,
),
web.get(
"/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/getAuthCode",
self.handle_getAuthCode,
),
web.get(
"/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/checkAgreement",
self.handle_checkAgreement,
),
web.get(
"/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/common/checkVersion",
self.handle_checkVersion,
),
web.get(
"/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/campaign/homePageAlert",
self.handle_homePageAlert,
),
web.post("/api/users/user.do", self.handle_usersapi),
web.get("/api/users/user.do", self.handle_usersapi),
web.post(
"/api/pim/product/getProductIotMap",
self.handle_getProductIotMap,
),
web.post(
"/api/iot/devmanager.do", self.handle_devmanager_botcommand
),
web.post("/lookup.do", self.handle_lookup),
]
)
# Direct register from app:
# /{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/directRegister
runner = web.AppRunner(app)
runner = web.AppRunner(self.app)
await runner.setup()
if self.usessl:
@ -200,7 +200,9 @@ class ConfServer:
"username": "fusername_{}".format(user["userid"]),
},
"msg": "操作成功",
"time": bumper.get_milli_time(datetime.datetime.utcnow().timestamp()),
"time": bumper.get_milli_time(
datetime.utcnow().timestamp()
),
}
return web.json_response(body)
@ -208,7 +210,9 @@ class ConfServer:
"code": bumper.ERR_USER_NOT_ACTIVATED,
"data": None,
"msg": "当前密码错误",
"time": bumper.get_milli_time(datetime.datetime.utcnow().timestamp()),
"time": bumper.get_milli_time(
datetime.utcnow().timestamp()
),
}
return web.json_response(body)
@ -233,7 +237,7 @@ class ConfServer:
"username": "fusername_{}".format(user["userid"]),
},
"msg": "操作成功",
"time": bumper.get_milli_time(datetime.datetime.utcnow().timestamp()),
"time": bumper.get_milli_time(datetime.utcnow().timestamp()),
}
return web.json_response(body)
@ -242,7 +246,7 @@ class ConfServer:
"code": bumper.ERR_TOKEN_INVALID,
"data": None,
"msg": "当前密码错误",
"time": bumper.get_milli_time(datetime.datetime.utcnow().timestamp()),
"time": bumper.get_milli_time(datetime.utcnow().timestamp()),
}
return web.json_response(body)
@ -276,7 +280,7 @@ class ConfServer:
if "checkLogin" in request.path: # If request was to check a token do so
checkToken = self.check_token(
countrycode, user, request.query["accessToken"]
countrycode, tmpuser, request.query["accessToken"]
)
isGood = json.loads(checkToken.text)
if isGood["code"] == "0000":
@ -295,7 +299,7 @@ class ConfServer:
"username": "fusername_{}".format(tmpuser["userid"]),
},
"msg": "操作成功",
"time": bumper.get_milli_time(datetime.datetime.utcnow().timestamp()),
"time": bumper.get_milli_time(datetime.utcnow().timestamp()),
}
return body
@ -319,7 +323,7 @@ class ConfServer:
"code": bumper.RETURN_API_SUCCESS,
"data": None,
"msg": "操作成功",
"time": bumper.get_milli_time(datetime.datetime.utcnow().timestamp()),
"time": bumper.get_milli_time(datetime.utcnow().timestamp()),
}
return web.json_response(body)
@ -355,7 +359,9 @@ class ConfServer:
"ecovacsUid": request.query["uid"],
},
"msg": "操作成功",
"time": bumper.get_milli_time(datetime.datetime.utcnow().timestamp()),
"time": bumper.get_milli_time(
datetime.utcnow().timestamp()
),
}
return web.json_response(body)
@ -363,7 +369,7 @@ class ConfServer:
"code": bumper.ERR_TOKEN_INVALID,
"data": None,
"msg": "当前密码错误",
"time": bumper.get_milli_time(datetime.datetime.utcnow().timestamp()),
"time": bumper.get_milli_time(datetime.utcnow().timestamp()),
}
return web.json_response(body)
@ -385,7 +391,7 @@ class ConfServer:
"v": None,
},
"msg": "操作成功",
"time": bumper.get_milli_time(datetime.datetime.utcnow().timestamp()),
"time": bumper.get_milli_time(datetime.utcnow().timestamp()),
}
return web.json_response(body)
@ -399,7 +405,7 @@ class ConfServer:
"code": bumper.RETURN_API_SUCCESS,
"data": [],
"msg": "操作成功",
"time": bumper.get_milli_time(datetime.datetime.utcnow().timestamp()),
"time": bumper.get_milli_time(datetime.utcnow().timestamp()),
}
return web.json_response(body)
@ -421,10 +427,12 @@ class ConfServer:
"hasCampaign": "N",
"imageUrl": None,
"nextAlertTime": nextAlert,
"serverTime": bumper.get_milli_time(datetime.datetime.utcnow().timestamp()),
"serverTime": bumper.get_milli_time(
datetime.utcnow().timestamp()
),
},
"msg": "操作成功",
"time": bumper.get_milli_time(datetime.datetime.utcnow().timestamp()),
"time": bumper.get_milli_time(datetime.utcnow().timestamp()),
}
return web.json_response(body)

128
tests/test_confserver.py Normal file
View file

@ -0,0 +1,128 @@
from nose.tools import *
import mock
import bumper
import asyncio
import os
import json
import tinydb
from aiohttp.test_utils import TestClient, TestServer, loop_context
from aiohttp import request
confserver = bumper.ConfServer("127.0.0.1:11111", False, None)
confserver.confserver_app()
app = confserver.app
def test_base():
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_base():
resp = await client.get("/")
assert resp.status == 200
text = await resp.text()
assert "Bumper!" in text
loop.run_until_complete(test_handle_base()) # Test handle_base
loop.run_until_complete(
client.close()
) # Close test server after all tests are done
def test_login():
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_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()
loginresp = json.loads(text)
if loginresp:
assert loginresp["code"] == "0000"
assert "accessToken" in loginresp["data"]
assert "uid" in loginresp["data"]
assert "username" in loginresp["data"]
else:
assert loginresp
loop.run_until_complete(test_handle_login()) # Test handle_login
#Add a user to db and test with existing users
bumper.user_add("testuser")
loop.run_until_complete(test_handle_login()) # Test handle_login with user in db
#Add a bot to db that will be added to user
bumper.bot_add("sn_123", "did_123", "dev_123", "res_123", "com_123")
loop.run_until_complete(test_handle_login()) # Test handle_login with user in db
loop.run_until_complete(
client.close()
) # Close test server after all tests are done
def test_check_login():
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_checkLogin_nouser():
resp = await client.get("/1/private/us/en/dev_1234/ios/1/0/0/user/checkLogin?accessToken=token_1234")
assert resp.status == 200
text = await resp.text()
loginresp = json.loads(text)
if loginresp:
assert loginresp["code"] == "0000"
assert "accessToken" in loginresp["data"]
assert loginresp["data"]["accessToken"] != "token_1234"
assert "uid" in loginresp["data"]
assert "username" in loginresp["data"]
else:
assert loginresp
async def test_handle_checkLogin_withuser():
resp = await client.get("/1/private/us/en/dev_1234/ios/1/0/0/user/checkLogin?accessToken=token_1234")
assert resp.status == 200
text = await resp.text()
loginresp = json.loads(text)
if loginresp:
assert loginresp["code"] == "0000"
assert "accessToken" in loginresp["data"]
assert loginresp["data"]["accessToken"] == "token_1234"
assert "uid" in loginresp["data"]
assert "username" in loginresp["data"]
else:
assert loginresp
loop.run_until_complete(test_handle_checkLogin_nouser()) # Test handle_login no user
#Add a user to db and test with existing users
bumper.user_add("testuser")
loop.run_until_complete(test_handle_checkLogin_nouser()) # Test handle_login with user in db
#Remove dev from tmpuser
bumper.user_remove_device("tmpuser","dev_1234")
#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")
loop.run_until_complete(test_handle_checkLogin_withuser()) # Test handle_login with user in db
loop.run_until_complete(
client.close()
) # Close test server after all tests are done