Additional tests for confserver

Additional tests for confserver

bumper/confserver.py

@@ -210,9 +210,7 @@ class ConfServer:
                     "code": bumper.ERR_USER_NOT_ACTIVATED,
                     "data": None,
                     "msg": "当前密码错误",
-                    "time": bumper.get_milli_time(
+                    "time": bumper.get_milli_time(datetime.utcnow().timestamp()),
-                        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(
+                    "serverTime": bumper.get_milli_time(datetime.utcnow().timestamp()),
-                        datetime.utcnow().timestamp()
-                    ),
                 },
                 "msg": "操作成功",
                 "time": bumper.get_milli_time(datetime.utcnow().timestamp()),
@@ -519,6 +515,7 @@ class ConfServer:
             confserverlog.exception("{}".format(e))
 
     async def handle_usersapi(self, request):
+        if not request.method == "GET":  # Skip GET for now
             try:
 
                 body = {}
@@ -579,6 +576,10 @@ class ConfServer:
             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:
 

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)
+        jsonresp = json.loads(text)
-        if loginresp:
+        if jsonresp:
-            assert loginresp["code"] == "0000"
+            assert jsonresp["code"] == bumper.RETURN_API_SUCCESS
-            assert "accessToken" in loginresp["data"]
+            assert "accessToken" in jsonresp["data"]
-            assert "uid" in loginresp["data"]
+            assert "uid" in jsonresp["data"]
-            assert "username" in loginresp["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
     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
     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,39 +85,70 @@ def test_check_login():
     loop.run_until_complete(client.start_server())
     root = "http://{}".format(confserver.address)
 
-    async def test_handle_checkLogin_nouser():
+    async def test_handle_logout(token=None):
-        resp = await client.get("/1/private/us/en/dev_1234/ios/1/0/0/user/checkLogin?accessToken=token_1234")
+        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)
+        jsonresp = json.loads(text)
-        if loginresp:
+        if jsonresp:
-            assert loginresp["code"] == "0000"
+            assert jsonresp["code"] == bumper.RETURN_API_SUCCESS
-            assert "accessToken" in loginresp["data"]
-            assert loginresp["data"]["accessToken"] != "token_1234"
-            assert "uid" in loginresp["data"]
-            assert "username" in loginresp["data"]
         else:
-            assert loginresp
+            assert jsonresp
 
-    async def test_handle_checkLogin_withuser():
+    # Add a token to user and test
-        resp = await client.get("/1/private/us/en/dev_1234/ios/1/0/0/user/checkLogin?accessToken=token_1234")
+    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()
-        loginresp = json.loads(text)
+        jsonresp = json.loads(text)
-        if loginresp:
+        if jsonresp:
-            assert loginresp["code"] == "0000"
+            assert jsonresp["code"] == bumper.RETURN_API_SUCCESS
-            assert "accessToken" in loginresp["data"]
+            assert "accessToken" in jsonresp["data"]
-            assert loginresp["data"]["accessToken"] == "token_1234"
+            if not token:
-            assert "uid" in loginresp["data"]
+                assert jsonresp["data"]["accessToken"] != "token_1234"
-            assert "username" in loginresp["data"]
             else:
-            assert loginresp            
+                assert jsonresp["data"]["accessToken"] == "token_1234"
 
-    loop.run_until_complete(test_handle_checkLogin_nouser())  # Test handle_login no user
+            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")
-    loop.run_until_complete(test_handle_checkLogin_nouser())  # Test handle_login with user in db
+    # Test
+    loop.run_until_complete(test_handle_checkLogin())
 
     # Remove dev from tmpuser
     bumper.user_remove_device("tmpuser", "dev_1234")
@@ -121,8 +157,345 @@ def test_check_login():
     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
+    # 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
+

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.