full coverage for init

full coverage for init
- 100% coverage of db operations

Pipfile

@@ -13,6 +13,7 @@ pyyaml = ">=4.2b1"
 black = "*"
 nose = "*"
 coverage = "*"
+mock = "*"
 
 [pipenv]
 allow_prereleases = true

Pipfile.lock

@@ -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",

bumper/__init__.py

@@ -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))
 
@@ -55,11 +50,15 @@ def get_milli_time(timetoconvert):
 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:
-        return os.path.expanduser("~/.config/bumper.db")
+        return os.path.expanduser("~/.config/bumper.db")    
 
 
 def db_get():

bumper/confserver.py

@@ -200,7 +200,7 @@ class ConfServer:
                                 "username": "fusername_{}".format(user["userid"]),
                             },
                             "msg": "操作成功",
-                            "time": bumper.get_milli_time(time.time()),
+                            "time": bumper.get_milli_time(datetime.datetime.utcnow().timestamp()),
                         }
                         return web.json_response(body)
 
@@ -208,7 +208,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.datetime.utcnow().timestamp()),
                 }
 
                 return web.json_response(body)
@@ -233,7 +233,7 @@ class ConfServer:
                     "username": "fusername_{}".format(user["userid"]),
                 },
                 "msg": "操作成功",
-                "time": bumper.get_milli_time(time.time()),
+                "time": bumper.get_milli_time(datetime.datetime.utcnow().timestamp()),
             }
             return web.json_response(body)
 
@@ -242,7 +242,7 @@ class ConfServer:
                 "code": bumper.ERR_TOKEN_INVALID,
                 "data": None,
                 "msg": "当前密码错误",
-                "time": bumper.get_milli_time(time.time()),
+                "time": bumper.get_milli_time(datetime.datetime.utcnow().timestamp()),
             }
             return web.json_response(body)
 
@@ -295,7 +295,7 @@ class ConfServer:
                     "username": "fusername_{}".format(tmpuser["userid"]),
                 },
                 "msg": "操作成功",
-                "time": bumper.get_milli_time(time.time()),
+                "time": bumper.get_milli_time(datetime.datetime.utcnow().timestamp()),
             }
 
             return body
@@ -319,7 +319,7 @@ class ConfServer:
                 "code": bumper.RETURN_API_SUCCESS,
                 "data": None,
                 "msg": "操作成功",
-                "time": bumper.get_milli_time(time.time()),
+                "time": bumper.get_milli_time(datetime.datetime.utcnow().timestamp()),
             }
 
             return web.json_response(body)
@@ -355,7 +355,7 @@ class ConfServer:
                                 "ecovacsUid": request.query["uid"],
                             },
                             "msg": "操作成功",
-                            "time": bumper.get_milli_time(time.time()),
+                            "time": bumper.get_milli_time(datetime.datetime.utcnow().timestamp()),
                         }
                         return web.json_response(body)
 
@@ -363,7 +363,7 @@ class ConfServer:
                 "code": bumper.ERR_TOKEN_INVALID,
                 "data": None,
                 "msg": "当前密码错误",
-                "time": bumper.get_milli_time(time.time()),
+                "time": bumper.get_milli_time(datetime.datetime.utcnow().timestamp()),
             }
 
             return web.json_response(body)
@@ -385,7 +385,7 @@ class ConfServer:
                     "v": None,
                 },
                 "msg": "操作成功",
-                "time": bumper.get_milli_time(time.time()),
+                "time": bumper.get_milli_time(datetime.datetime.utcnow().timestamp()),
             }
 
             return web.json_response(body)
@@ -399,7 +399,7 @@ class ConfServer:
                 "code": bumper.RETURN_API_SUCCESS,
                 "data": [],
                 "msg": "操作成功",
-                "time": bumper.get_milli_time(time.time()),
+                "time": bumper.get_milli_time(datetime.datetime.utcnow().timestamp()),
             }
 
             return web.json_response(body)
@@ -421,10 +421,10 @@ class ConfServer:
                     "hasCampaign": "N",
                     "imageUrl": None,
                     "nextAlertTime": nextAlert,
-                    "serverTime": bumper.get_milli_time(time.time()),
+                    "serverTime": bumper.get_milli_time(datetime.datetime.utcnow().timestamp()),
                 },
                 "msg": "操作成功",
-                "time": bumper.get_milli_time(time.time()),
+                "time": bumper.get_milli_time(datetime.datetime.utcnow().timestamp()),
             }
 
             return web.json_response(body)

tests/test_bumper.py

@@ -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():
+def test_get_milli_time():
-    btest = bumper
+    assert_equals(
-    btest.db = "tests/tmp.db"
+        bumper.get_milli_time(
-    btest.user_add("testuser")
+            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")
+    tokens = db.table("tokens")
-    assert_equals(len(users), 1)
+    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