diff --git a/bumper/__init__.py b/bumper/__init__.py index 649f580..d8f78dd 100644 --- a/bumper/__init__.py +++ b/bumper/__init__.py @@ -50,7 +50,7 @@ def get_milli_time(timetoconvert): def db_file(): if db: return db - + return os_db_path() @@ -58,7 +58,7 @@ def os_db_path(): if platform.system() == "Windows": return os.path.join(os.getenv("APPDATA"), "bumper.db") else: - return os.path.expanduser("~/.config/bumper.db") + return os.path.expanduser("~/.config/bumper.db") def db_get(): @@ -95,7 +95,7 @@ def user_add(userid): def user_get(userid): users = db_get().table("users") - User = Query() + User = Query() return users.get(User.userid == userid) diff --git a/bumper/confserver.py b/bumper/confserver.py index 16d9a1e..f826a54 100644 --- a/bumper/confserver.py +++ b/bumper/confserver.py @@ -210,9 +210,7 @@ class ConfServer: "code": bumper.ERR_USER_NOT_ACTIVATED, "data": None, "msg": "当前密码错误", - "time": bumper.get_milli_time( - datetime.utcnow().timestamp() - ), + "time": bumper.get_milli_time(datetime.utcnow().timestamp()), } return web.json_response(body) @@ -427,9 +425,7 @@ class ConfServer: "hasCampaign": "N", "imageUrl": None, "nextAlertTime": nextAlert, - "serverTime": bumper.get_milli_time( - datetime.utcnow().timestamp() - ), + "serverTime": bumper.get_milli_time(datetime.utcnow().timestamp()), }, "msg": "操作成功", "time": bumper.get_milli_time(datetime.utcnow().timestamp()), @@ -519,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: diff --git a/tests/test_confserver.py b/tests/test_confserver.py index dcf6030..ef6cdb0 100644 --- a/tests/test_confserver.py +++ b/tests/test_confserver.py @@ -28,7 +28,8 @@ def test_base(): text = await resp.text() assert "Bumper!" in text - loop.run_until_complete(test_handle_base()) # Test handle_base + # Test + loop.run_until_complete(test_handle_base()) loop.run_until_complete( client.close() @@ -48,30 +49,34 @@ def test_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"] + 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 loginresp + assert jsonresp - loop.run_until_complete(test_handle_login()) # Test handle_login + # Test + loop.run_until_complete(test_handle_login()) - #Add a user to db and test with existing users + # 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 + # Test + loop.run_until_complete(test_handle_login()) - #Add a bot to db that will be added to user + # 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 + # Test + loop.run_until_complete(test_handle_login()) loop.run_until_complete( client.close() ) # Close test server after all tests are done -def test_check_login(): + +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 @@ -80,49 +85,417 @@ def test_check_login(): 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") + 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() - 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"] + jsonresp = json.loads(text) + if jsonresp: + assert jsonresp["code"] == bumper.RETURN_API_SUCCESS else: - assert loginresp + assert jsonresp - 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 + # Add a token to user and test 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_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 + # 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 + diff --git a/tests/tests.md b/tests/tests.md new file mode 100644 index 0000000..d4756ce --- /dev/null +++ b/tests/tests.md @@ -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. \ No newline at end of file