Brian Martin
GitHub
commit
386f04c831
10 changed files with 942 additions and 123 deletions
12
.travis.yml
Normal file
12
.travis.yml
Normal file
|
|
@ -0,0 +1,12 @@
|
|||
language: python
|
||||
# Enable python 3.7 without globally enabling sudo and dist: xenial for other build jobs
|
||||
matrix:
|
||||
include:
|
||||
- python: 3.7
|
||||
dist: xenial
|
||||
sudo: true
|
||||
install:
|
||||
- pip install pipenv
|
||||
- pipenv install --dev
|
||||
|
||||
script: nosetests
|
||||
1
Pipfile
1
Pipfile
|
|
@ -13,6 +13,7 @@ pyyaml = ">=4.2b1"
|
|||
black = "*"
|
||||
nose = "*"
|
||||
coverage = "*"
|
||||
mock = "*"
|
||||
|
||||
[pipenv]
|
||||
allow_prereleases = true
|
||||
|
|
|
|||
24
Pipfile.lock
generated
24
Pipfile.lock
generated
|
|
@ -1,7 +1,7 @@
|
|||
{
|
||||
"_meta": {
|
||||
"hash": {
|
||||
"sha256": "e23347a7d680cf62879fd5c1acfd6431c4d56ff46c30533478629e220623e585"
|
||||
"sha256": "4119ce02f0331e2eae79e97097a00614ec5fb641345868b4cac63f7dd6127a32"
|
||||
},
|
||||
"pipfile-spec": 6,
|
||||
"requires": {},
|
||||
|
|
@ -272,6 +272,14 @@
|
|||
"index": "pypi",
|
||||
"version": "==5.0a4"
|
||||
},
|
||||
"mock": {
|
||||
"hashes": [
|
||||
"sha256:5ce3c71c5545b472da17b72268978914d0252980348636840bd34a00b5cc96c1",
|
||||
"sha256:b158b6df76edd239b8208d481dc46b6afd45a846b7812ff0ce58971cf5bc8bba"
|
||||
],
|
||||
"index": "pypi",
|
||||
"version": "==2.0.0"
|
||||
},
|
||||
"nose": {
|
||||
"hashes": [
|
||||
"sha256:9ff7c6cc443f8c51994b34a667bbcf45afd6d945be7477b52e97516fd17c53ac",
|
||||
|
|
@ -281,6 +289,20 @@
|
|||
"index": "pypi",
|
||||
"version": "==1.3.7"
|
||||
},
|
||||
"pbr": {
|
||||
"hashes": [
|
||||
"sha256:8257baf496c8522437e8a6cfe0f15e00aedc6c0e0e7c9d55eeeeab31e0853843",
|
||||
"sha256:8c361cc353d988e4f5b998555c88098b9d5964c2e11acf7b0d21925a66bb5824"
|
||||
],
|
||||
"version": "==5.1.3"
|
||||
},
|
||||
"six": {
|
||||
"hashes": [
|
||||
"sha256:3350809f0555b11f552448330d0b52d5f24c91a322ea4a15ef22629740f3761c",
|
||||
"sha256:d16a0141ec1a18405cd4ce8b4613101da75da0e9a7aec5bdd4fa804d0e0eba73"
|
||||
],
|
||||
"version": "==1.12.0"
|
||||
},
|
||||
"toml": {
|
||||
"hashes": [
|
||||
"sha256:229f81c57791a41d65e399fc06bf0848bab550a9dfd5ed66df18ce5f05e73d5c",
|
||||
|
|
|
|||
22
appveyor.yml
Normal file
22
appveyor.yml
Normal file
|
|
@ -0,0 +1,22 @@
|
|||
environment:
|
||||
|
||||
matrix:
|
||||
|
||||
# For Python versions available on Appveyor, see
|
||||
# http://www.appveyor.com/docs/installed-software#python
|
||||
|
||||
- PYTHON: "C:\\Python37"
|
||||
|
||||
install:
|
||||
- pip install pipenv
|
||||
- pipenv --python 3
|
||||
- pipenv install --dev
|
||||
|
||||
build: off
|
||||
|
||||
test_script:
|
||||
# Put your test command here.
|
||||
# Note that you must use the environment variable %PYTHON% to refer to
|
||||
# the interpreter you're using - Appveyor does not do anything special
|
||||
# to put the Python version you want to use on PATH.
|
||||
- "pipenv run nosetests"
|
||||
|
|
@ -43,11 +43,6 @@ xmppserverlog = logging.getLogger("xmppserver")
|
|||
# xmppserverlog.setLevel(logging.INFO)
|
||||
|
||||
|
||||
def __init__(self, db=None):
|
||||
if db:
|
||||
self.db = db
|
||||
|
||||
|
||||
def get_milli_time(timetoconvert):
|
||||
return int(round(timetoconvert * 1000))
|
||||
|
||||
|
|
@ -56,6 +51,10 @@ def db_file():
|
|||
if db:
|
||||
return db
|
||||
|
||||
return os_db_path()
|
||||
|
||||
|
||||
def os_db_path():
|
||||
if platform.system() == "Windows":
|
||||
return os.path.join(os.getenv("APPDATA"), "bumper.db")
|
||||
else:
|
||||
|
|
@ -67,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
|
||||
|
||||
|
|
@ -170,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,
|
||||
|
|
|
|||
|
|
@ -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(time.time()),
|
||||
"time": bumper.get_milli_time(
|
||||
datetime.utcnow().timestamp()
|
||||
),
|
||||
}
|
||||
return web.json_response(body)
|
||||
|
||||
|
|
@ -208,7 +210,7 @@ class ConfServer:
|
|||
"code": bumper.ERR_USER_NOT_ACTIVATED,
|
||||
"data": None,
|
||||
"msg": "当前密码错误",
|
||||
"time": bumper.get_milli_time(time.time()),
|
||||
"time": bumper.get_milli_time(datetime.utcnow().timestamp()),
|
||||
}
|
||||
|
||||
return web.json_response(body)
|
||||
|
|
@ -233,7 +235,7 @@ class ConfServer:
|
|||
"username": "fusername_{}".format(user["userid"]),
|
||||
},
|
||||
"msg": "操作成功",
|
||||
"time": bumper.get_milli_time(time.time()),
|
||||
"time": bumper.get_milli_time(datetime.utcnow().timestamp()),
|
||||
}
|
||||
return web.json_response(body)
|
||||
|
||||
|
|
@ -242,7 +244,7 @@ class ConfServer:
|
|||
"code": bumper.ERR_TOKEN_INVALID,
|
||||
"data": None,
|
||||
"msg": "当前密码错误",
|
||||
"time": bumper.get_milli_time(time.time()),
|
||||
"time": bumper.get_milli_time(datetime.utcnow().timestamp()),
|
||||
}
|
||||
return web.json_response(body)
|
||||
|
||||
|
|
@ -276,7 +278,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 +297,7 @@ class ConfServer:
|
|||
"username": "fusername_{}".format(tmpuser["userid"]),
|
||||
},
|
||||
"msg": "操作成功",
|
||||
"time": bumper.get_milli_time(time.time()),
|
||||
"time": bumper.get_milli_time(datetime.utcnow().timestamp()),
|
||||
}
|
||||
|
||||
return body
|
||||
|
|
@ -319,7 +321,7 @@ class ConfServer:
|
|||
"code": bumper.RETURN_API_SUCCESS,
|
||||
"data": None,
|
||||
"msg": "操作成功",
|
||||
"time": bumper.get_milli_time(time.time()),
|
||||
"time": bumper.get_milli_time(datetime.utcnow().timestamp()),
|
||||
}
|
||||
|
||||
return web.json_response(body)
|
||||
|
|
@ -355,7 +357,9 @@ class ConfServer:
|
|||
"ecovacsUid": request.query["uid"],
|
||||
},
|
||||
"msg": "操作成功",
|
||||
"time": bumper.get_milli_time(time.time()),
|
||||
"time": bumper.get_milli_time(
|
||||
datetime.utcnow().timestamp()
|
||||
),
|
||||
}
|
||||
return web.json_response(body)
|
||||
|
||||
|
|
@ -363,7 +367,7 @@ class ConfServer:
|
|||
"code": bumper.ERR_TOKEN_INVALID,
|
||||
"data": None,
|
||||
"msg": "当前密码错误",
|
||||
"time": bumper.get_milli_time(time.time()),
|
||||
"time": bumper.get_milli_time(datetime.utcnow().timestamp()),
|
||||
}
|
||||
|
||||
return web.json_response(body)
|
||||
|
|
@ -385,7 +389,7 @@ class ConfServer:
|
|||
"v": None,
|
||||
},
|
||||
"msg": "操作成功",
|
||||
"time": bumper.get_milli_time(time.time()),
|
||||
"time": bumper.get_milli_time(datetime.utcnow().timestamp()),
|
||||
}
|
||||
|
||||
return web.json_response(body)
|
||||
|
|
@ -399,7 +403,7 @@ class ConfServer:
|
|||
"code": bumper.RETURN_API_SUCCESS,
|
||||
"data": [],
|
||||
"msg": "操作成功",
|
||||
"time": bumper.get_milli_time(time.time()),
|
||||
"time": bumper.get_milli_time(datetime.utcnow().timestamp()),
|
||||
}
|
||||
|
||||
return web.json_response(body)
|
||||
|
|
@ -421,10 +425,10 @@ class ConfServer:
|
|||
"hasCampaign": "N",
|
||||
"imageUrl": None,
|
||||
"nextAlertTime": nextAlert,
|
||||
"serverTime": bumper.get_milli_time(time.time()),
|
||||
"serverTime": bumper.get_milli_time(datetime.utcnow().timestamp()),
|
||||
},
|
||||
"msg": "操作成功",
|
||||
"time": bumper.get_milli_time(time.time()),
|
||||
"time": bumper.get_milli_time(datetime.utcnow().timestamp()),
|
||||
}
|
||||
|
||||
return web.json_response(body)
|
||||
|
|
@ -511,65 +515,70 @@ class ConfServer:
|
|||
confserverlog.exception("{}".format(e))
|
||||
|
||||
async def handle_usersapi(self, request):
|
||||
try:
|
||||
if not request.method == "GET": # Skip GET for now
|
||||
try:
|
||||
|
||||
body = {}
|
||||
postbody = {}
|
||||
if request.content_type == "application/x-www-form-urlencoded":
|
||||
postbody = await request.post()
|
||||
body = {}
|
||||
postbody = {}
|
||||
if request.content_type == "application/x-www-form-urlencoded":
|
||||
postbody = await request.post()
|
||||
|
||||
else:
|
||||
postbody = json.loads(await request.text())
|
||||
else:
|
||||
postbody = json.loads(await request.text())
|
||||
|
||||
todo = postbody["todo"]
|
||||
if todo == "FindBest":
|
||||
service = postbody["service"]
|
||||
if service == "EcoMsgNew":
|
||||
todo = postbody["todo"]
|
||||
if todo == "FindBest":
|
||||
service = postbody["service"]
|
||||
if service == "EcoMsgNew":
|
||||
body = {
|
||||
"result": "ok",
|
||||
"ip": socket.gethostbyname(socket.gethostname()),
|
||||
"port": 5223,
|
||||
}
|
||||
elif service == "EcoUpdate":
|
||||
body = {"result": "ok", "ip": "47.88.66.164", "port": 8005}
|
||||
|
||||
elif todo == "loginByItToken":
|
||||
if bumper.check_authcode(postbody["userId"], postbody["token"]):
|
||||
body = {
|
||||
"resource": postbody["resource"],
|
||||
"result": "ok",
|
||||
"todo": "result",
|
||||
"token": postbody["token"],
|
||||
"userId": postbody["userId"],
|
||||
}
|
||||
|
||||
elif todo == "GetDeviceList":
|
||||
body = {
|
||||
"result": "ok",
|
||||
"ip": socket.gethostbyname(socket.gethostname()),
|
||||
"port": 5223,
|
||||
}
|
||||
elif service == "EcoUpdate":
|
||||
body = {"result": "ok", "ip": "47.88.66.164", "port": 8005}
|
||||
|
||||
elif todo == "loginByItToken":
|
||||
if bumper.check_authcode(postbody["userId"], postbody["token"]):
|
||||
body = {
|
||||
"resource": postbody["resource"],
|
||||
"devices": bumper.db_get().table("bots").all(),
|
||||
"result": "ok",
|
||||
"todo": "result",
|
||||
"token": postbody["token"],
|
||||
"userId": postbody["userId"],
|
||||
}
|
||||
|
||||
elif todo == "GetDeviceList":
|
||||
body = {
|
||||
"devices": bumper.db_get().table("bots").all(),
|
||||
"result": "ok",
|
||||
"todo": "result",
|
||||
}
|
||||
elif todo == "SetDeviceNick":
|
||||
bumper.bot_set_nick(postbody["did"], postbody["nick"])
|
||||
body = {"result": "ok", "todo": "result"}
|
||||
|
||||
elif todo == "SetDeviceNick":
|
||||
bumper.bot_set_nick(postbody["did"], postbody["nick"])
|
||||
body = {"result": "ok", "todo": "result"}
|
||||
elif todo == "AddOneDevice":
|
||||
bumper.bot_set_nick(postbody["did"], postbody["nick"])
|
||||
body = {"result": "ok", "todo": "result"}
|
||||
|
||||
elif todo == "AddOneDevice":
|
||||
bumper.bot_set_nick(postbody["did"], postbody["nick"])
|
||||
body = {"result": "ok", "todo": "result"}
|
||||
elif todo == "DeleteOneDevice":
|
||||
bumper.bot_remove(postbody["did"])
|
||||
body = {"result": "ok", "todo": "result"}
|
||||
|
||||
elif todo == "DeleteOneDevice":
|
||||
bumper.bot_remove(postbody["did"])
|
||||
body = {"result": "ok", "todo": "result"}
|
||||
confserverlog.debug(
|
||||
"\r\n POST: {} \r\n Response: {}".format(postbody, body)
|
||||
)
|
||||
|
||||
confserverlog.debug(
|
||||
"\r\n POST: {} \r\n Response: {}".format(postbody, body)
|
||||
)
|
||||
return web.json_response(body)
|
||||
|
||||
return web.json_response(body)
|
||||
except Exception as e:
|
||||
confserverlog.exception("{}".format(e))
|
||||
|
||||
except Exception as e:
|
||||
confserverlog.exception("{}".format(e))
|
||||
# Return fail for GET
|
||||
body = {"result": "fail", "todo": "result"}
|
||||
return web.json_response(body)
|
||||
|
||||
async def handle_lookup(self, request):
|
||||
try:
|
||||
|
|
|
|||
|
|
@ -339,6 +339,7 @@ class BumperMQTTServer_Plugin:
|
|||
tmpbotdetail[1],
|
||||
"eco-ng",
|
||||
)
|
||||
|
||||
mqttserverlog.debug(
|
||||
"new bot authenticated SN: {} DID: {}".format(
|
||||
username, didsplit[0]
|
||||
|
|
@ -372,7 +373,7 @@ class BumperMQTTServer_Plugin:
|
|||
authenticated = False
|
||||
|
||||
except Exception as e:
|
||||
mqttserverlog.exception("{}".format(e))
|
||||
mqttserverlog.exception("Session: {} - {}".format((kwargs.get("session", None)),e))
|
||||
authenticated = False
|
||||
|
||||
return authenticated
|
||||
|
|
|
|||
|
|
@ -1,14 +1,184 @@
|
|||
from nose.tools import *
|
||||
import nose
|
||||
import mock
|
||||
from tinydb.storages import MemoryStorage
|
||||
from tinydb import TinyDB, Query
|
||||
import bumper
|
||||
import os
|
||||
import datetime, time
|
||||
import platform
|
||||
|
||||
|
||||
def test_useradd():
|
||||
btest = bumper
|
||||
btest.db = "tests/tmp.db"
|
||||
btest.user_add("testuser")
|
||||
def test_get_milli_time():
|
||||
assert_equals(
|
||||
bumper.get_milli_time(
|
||||
datetime.datetime(
|
||||
2018, 1, 1, 1, 0, 0, 0, tzinfo=datetime.timezone.utc
|
||||
).timestamp()
|
||||
),
|
||||
1514768400000,
|
||||
)
|
||||
|
||||
|
||||
def test_user_db():
|
||||
if os.path.exists("tests/tmp.db"):
|
||||
os.remove("tests/tmp.db") # Remove existing db
|
||||
|
||||
# Test os_db_path
|
||||
platform.system = mock.MagicMock(return_value="Windows")
|
||||
p = platform.system()
|
||||
os.getenv = mock.MagicMock(return_value="C:\AppData")
|
||||
o = os.getenv("APPDATA")
|
||||
assert_equals(bumper.os_db_path(), os.path.join(os.getenv("APPDATA"), "bumper.db"))
|
||||
|
||||
platform.system = mock.MagicMock(return_value="Linux")
|
||||
assert_equals(bumper.os_db_path(), os.path.expanduser("~/.config/bumper.db"))
|
||||
|
||||
bumper.db = "tests/tmp.db" # Set db location for testing
|
||||
bumper.user_add("testuser") # Add testuser
|
||||
|
||||
assert_equals(
|
||||
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_equals(
|
||||
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_true(
|
||||
"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_true(
|
||||
"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_true(
|
||||
"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_true(
|
||||
bumper.check_token("testuser", "token_1234")
|
||||
) # Test that token was found for testuser
|
||||
|
||||
assert_true(
|
||||
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_true(
|
||||
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_false(
|
||||
bumper.check_authcode("testuser", "auth_1234")
|
||||
) # Test that authcode was not found for testuser
|
||||
bumper.user_revoke_token("testuser", "token_1234") # Remove token from testuser
|
||||
assert_false(
|
||||
bumper.check_token("testuser", "token_1234")
|
||||
) # 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_equals(
|
||||
len(bumper.user_get_tokens("testuser")), 2
|
||||
) # Test 2 tokens are available
|
||||
bumper.user_revoke_all_tokens("testuser") # Revoke all tokens
|
||||
assert_equals(
|
||||
len(bumper.user_get_tokens("testuser")), 0
|
||||
) # Test 0 tokens are available
|
||||
|
||||
db = TinyDB("tests/tmp.db")
|
||||
users = db.table("users").search(Query().userid == "testuser")
|
||||
assert_equals(len(users), 1)
|
||||
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_equals(
|
||||
len(bumper.user_get_tokens("testuser")), 1
|
||||
) # Test 1 tokens are available
|
||||
bumper.user_revoke_expired_tokens("testuser") # Revoke expired tokens
|
||||
assert_equals(
|
||||
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_equals(
|
||||
len(bumper.user_get_tokens("testuser")), 1
|
||||
) # Test 1 tokens are available
|
||||
bumper.revoke_expired_tokens() # Revoke expired tokens
|
||||
assert_equals(
|
||||
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_true(bumper.bot_get("did_123")) # Test that bot was added to db
|
||||
|
||||
bumper.bot_set_nick("did_123", "nick_123")
|
||||
assert_equals(
|
||||
bumper.bot_get("did_123")["nick"], "nick_123"
|
||||
) # Test that nick was added to bot
|
||||
|
||||
bumper.bot_set_mqtt("did_123", True)
|
||||
assert_true(
|
||||
bumper.bot_get("did_123")["mqtt_connection"]
|
||||
) # Test that mqtt was set True for bot
|
||||
|
||||
bumper.bot_set_xmpp("did_123", True)
|
||||
assert_true(
|
||||
bumper.bot_get("did_123")["xmpp_connection"]
|
||||
) # Test that xmpp was set True for bot
|
||||
|
||||
bumper.bot_remove("did_123")
|
||||
assert_false(bumper.bot_get("did_123")) # 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_true(bumper.client_get("resource_123")) # Test client was added
|
||||
|
||||
bumper.client_set_mqtt("resource_123", True)
|
||||
assert_true(
|
||||
bumper.client_get("resource_123")["mqtt_connection"]
|
||||
) # Test that mqtt was set True for client
|
||||
|
||||
bumper.client_set_xmpp("resource_123", False)
|
||||
assert_false(
|
||||
bumper.client_get("resource_123")["xmpp_connection"]
|
||||
) # Test that xmpp was set False for client
|
||||
assert_equals(len(bumper.get_disconnected_xmpp_clients()), 1) # Test len of connected xmpp clients is 1
|
||||
566
tests/test_confserver.py
Normal file
566
tests/test_confserver.py
Normal file
|
|
@ -0,0 +1,566 @@
|
|||
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, mock.MagicMock)
|
||||
confserver.confserver_app()
|
||||
app = confserver.app
|
||||
|
||||
def async_return(result):
|
||||
f = asyncio.Future()
|
||||
f.set_result(result)
|
||||
return f
|
||||
|
||||
def test_disconnect():
|
||||
confserver.disconnect()
|
||||
|
||||
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
|
||||
|
||||
# Test
|
||||
loop.run_until_complete(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()
|
||||
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())
|
||||
|
||||
# Add a user to db and test with existing users
|
||||
bumper.user_add("testuser")
|
||||
# Test
|
||||
loop.run_until_complete(test_handle_login())
|
||||
|
||||
# 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
|
||||
|
||||
|
||||
def test_logout():
|
||||
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
|
||||
|
||||
# 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"))
|
||||
|
||||
loop.run_until_complete(
|
||||
client.close()
|
||||
) # Close test server after all tests are done
|
||||
|
||||
|
||||
def test_checkLogin():
|
||||
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(token=None):
|
||||
resp = await client.get(
|
||||
"/1/private/us/en/dev_1234/ios/1/0/0/user/checkLogin?accessToken={}".format(
|
||||
token
|
||||
)
|
||||
)
|
||||
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())
|
||||
|
||||
# Add a user to db and test with existing users
|
||||
bumper.user_add("testuser")
|
||||
# Test
|
||||
loop.run_until_complete(test_handle_checkLogin())
|
||||
|
||||
# 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")
|
||||
# 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
|
||||
|
||||
|
||||
def test_getAuthCode():
|
||||
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
|
||||
|
||||
# Test without user or token
|
||||
loop.run_until_complete(test_handle_getAuthCode())
|
||||
|
||||
# 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"))
|
||||
|
||||
# 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"
|
||||
)
|
||||
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
|
||||
|
||||
|
||||
def test_homePageAlert():
|
||||
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_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
|
||||
|
||||
|
||||
def test_checkVersion():
|
||||
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_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
|
||||
|
||||
|
||||
def test_getProductIotMap():
|
||||
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_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
|
||||
|
||||
|
||||
def test_getUsersAPI():
|
||||
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_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
|
||||
|
||||
|
||||
def test_postUsersAPI():
|
||||
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_postUsersApi(postbody=None):
|
||||
resp = await client.post("/api/users/user.do", json=postbody)
|
||||
|
||||
assert resp.status == 200
|
||||
text = await resp.text()
|
||||
jsonresp = json.loads(text)
|
||||
if jsonresp:
|
||||
assert jsonresp["result"] == "ok"
|
||||
else:
|
||||
assert jsonresp
|
||||
|
||||
# Test FindBest
|
||||
postbody = {"todo": "FindBest", "service": "EcoMsgNew"}
|
||||
# Test
|
||||
loop.run_until_complete(test_handle_postUsersApi(postbody))
|
||||
|
||||
# Test EcoUpdate
|
||||
postbody = {"todo": "FindBest", "service": "EcoUpdate"}
|
||||
# Test
|
||||
loop.run_until_complete(test_handle_postUsersApi(postbody))
|
||||
|
||||
# Test loginByItToken - Uses the authcode
|
||||
bumper.user_add("testuser")
|
||||
bumper.user_add_device("testuser", "dev_1234")
|
||||
bumper.user_add_token("testuser", "token_1234")
|
||||
bumper.user_add_authcode("testuser", "token_1234", "auth_1234")
|
||||
bumper.user_add_bot("testuser", "did_1234")
|
||||
bumper.bot_add("sn_1234", "did_1234", "class_1234", "res_1234", "com_1234")
|
||||
# Test
|
||||
postbody = {
|
||||
"country": "US",
|
||||
"last": "",
|
||||
"realm": "ecouser.net",
|
||||
"resource": "dev_1234",
|
||||
"todo": "loginByItToken",
|
||||
"token": "auth_1234",
|
||||
"userId": "testuser",
|
||||
}
|
||||
loop.run_until_complete(test_handle_postUsersApi(postbody))
|
||||
|
||||
# Test GetDeviceList
|
||||
postbody = {
|
||||
"auth": {
|
||||
"realm": "ecouser.net",
|
||||
"resource": "dev_1234",
|
||||
"token": "token_1234",
|
||||
"userid": "testuser",
|
||||
"with": "users",
|
||||
},
|
||||
"todo": "GetDeviceList",
|
||||
"userid": "testuser",
|
||||
}
|
||||
loop.run_until_complete(test_handle_postUsersApi(postbody))
|
||||
|
||||
# Test SetDeviceNick
|
||||
postbody = {
|
||||
"auth": {
|
||||
"realm": "ecouser.net",
|
||||
"resource": "dev_1234",
|
||||
"token": "token_1234",
|
||||
"userid": "testuser",
|
||||
"with": "users",
|
||||
},
|
||||
"todo": "SetDeviceNick",
|
||||
"nick": "botnick",
|
||||
"did": "did_1234",
|
||||
}
|
||||
loop.run_until_complete(test_handle_postUsersApi(postbody))
|
||||
|
||||
# Test AddOneDevice - Same as set nick for some bots
|
||||
postbody = {
|
||||
"auth": {
|
||||
"realm": "ecouser.net",
|
||||
"resource": "dev_1234",
|
||||
"token": "token_1234",
|
||||
"userid": "testuser",
|
||||
"with": "users",
|
||||
},
|
||||
"todo": "AddOneDevice",
|
||||
"nick": "botnick",
|
||||
"did": "did_1234",
|
||||
}
|
||||
loop.run_until_complete(test_handle_postUsersApi(postbody))
|
||||
|
||||
# Test DeleteOneDevice - remove bot
|
||||
postbody = {
|
||||
"auth": {
|
||||
"realm": "ecouser.net",
|
||||
"resource": "dev_1234",
|
||||
"token": "token_1234",
|
||||
"userid": "testuser",
|
||||
"with": "users",
|
||||
},
|
||||
"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
|
||||
|
||||
|
||||
def test_postLookup():
|
||||
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
|
||||
|
||||
# Test FindBest
|
||||
postbody = {"todo": "FindBest", "service": "EcoMsgNew"}
|
||||
# Test
|
||||
loop.run_until_complete(test_handle_lookup(postbody))
|
||||
|
||||
# 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
|
||||
|
||||
def test_devmgr():
|
||||
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
|
||||
|
||||
# Test PollSCResult
|
||||
postbody = {"td": "PollSCResult"}
|
||||
# Test
|
||||
loop.run_until_complete(test_devmanager(postbody, command=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))
|
||||
# Test
|
||||
loop.run_until_complete(test_devmanager(postbody, command=True))
|
||||
|
||||
# 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))
|
||||
|
||||
# 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))
|
||||
|
||||
loop.run_until_complete(
|
||||
client.close()
|
||||
) # Close test server after all tests are done
|
||||
16
tests/tests.md
Normal file
16
tests/tests.md
Normal file
|
|
@ -0,0 +1,16 @@
|
|||
# Bumper tests
|
||||
Bumper uses nosetests for the majority of test cases. Install requirements using `pipenv install --dev`
|
||||
|
||||
## Testing
|
||||
Enter pipenv shell `pipenv shell`
|
||||
|
||||
### Run tests
|
||||
`nosetests`
|
||||
|
||||
### Run tests with coverage
|
||||
`nosetests --cover-package bumper --with-coverage`
|
||||
|
||||
### Run tests with coverage html report
|
||||
`nosetests --cover-package bumper --with-coverage --cover-html-dir="tests/report" --cover-html`
|
||||
|
||||
The report will be output into tests/report/index.html for further analysis.
|
||||
Loading…
Add table
Add a link
Reference in a new issue