diff --git a/.gitignore b/.gitignore index ebef324..e470b09 100644 --- a/.gitignore +++ b/.gitignore @@ -3,6 +3,7 @@ __pycache__ .vscode/ .coverage .noseids +.pytest_cache nosetests.xml tests/report tests/tmp.db diff --git a/Pipfile b/Pipfile index b7df90c..aafbc10 100644 --- a/Pipfile +++ b/Pipfile @@ -11,11 +11,13 @@ pyyaml = ">=4.2b1" [dev-packages] black = "*" -nose = "*" coverage = "*" mock = "*" pylint = "*" pbr = "*" +pytest-asyncio = "*" +pytest-aiohttp = "*" +pytest-cov = "*" [pipenv] allow_prereleases = true diff --git a/Pipfile.lock b/Pipfile.lock index 1f64b3a..b7c6c8b 100644 --- a/Pipfile.lock +++ b/Pipfile.lock @@ -1,7 +1,7 @@ { "_meta": { "hash": { - "sha256": "6e5b4ccc879dfdf82f53f7966eac54af90f9b9a2ffd1da6f1ce126d0ca125fc9" + "sha256": "89b1dc5fde60f217d24d1b66c94fa5640ca23925420992a2c14718f61866dc61" }, "pipfile-spec": 6, "requires": {}, @@ -208,6 +208,34 @@ } }, "develop": { + "aiohttp": { + "hashes": [ + "sha256:02396865118790ebb0bd14d6935e9e4fa80d87683240618894e585048a8b83ec", + "sha256:0f448b5d5cd45e642b9fcf4798e48cca4149ae479022f59bd67a47caed44657e", + "sha256:1af72f53ccead6d161296e576aa2b5d0c6c7403a389601dd7aa7a7a30a9f41c3", + "sha256:32eeef64a5bcf6dc652f19e020b16f36ec66d291957f62ffad18ecfc58695966", + "sha256:36c4e234a85a81e325d8b1d7fdeb696a3d51bb8eb09a8c8e69c7532b421e8a29", + "sha256:371a5f1bb604fc262ba32893931f5aad3bb3e797dbebcf2b5b2c36f97603e4c7", + "sha256:3f5a30f67e4152d4075063ae9f2286313af2e39b950869edba76232569e08793", + "sha256:48c65d65d0de79f1a4391bc29c8c8ec7655671a03ad7902d5e66fd735531893e", + "sha256:4ede22808195126d55879f1a14aaa3d1004f188c341f92ec3b9ca5ea13e695b9", + "sha256:53228028648b40f59fc941c0cd67b7899bce52f35eb66aeb9e33b75bac0aac0a", + "sha256:67cb8e71c043686f806cfdb4189774c57ef929d32890024c3775d72b54381797", + "sha256:6b1e99dc838c28f14e45bffef6fb77bf5b54bd2bfb1a475c776da5ba8a9fec2d", + "sha256:6c30eca95e7d60fbf13e6ebcaf62b7c0f3f93ec5de8862919aa5ede3720b1c86", + "sha256:6f2c905ab82aa0ee8a06210d5a6c1359c79f3f2088cc1b53c8758cae1dbc0d55", + "sha256:746b73eff86a1618025093b9c86ff4d642cff71a7ac7244f7fd8a186967cee22", + "sha256:9ebf518c7bc08e65b5396f80e8155c7bb1380b921b84b510570e62196c9fca56", + "sha256:9efc80ced0936f3fb2774e1bc981006324c74d1e58fc7f7164ecd98ede727c7d", + "sha256:abc43651f2c6d70b812cb874fc8caa00eb69fac11930cd2640d32a11ee7beef0", + "sha256:b4c6e1f5a591511537d4ccd4c807f6c8bb8d1b8a5043395950ba339086380904", + "sha256:bee80820a8ef5f6bddd58d762fc9f981ef1041a464d382d7d9a992bb94cd23c6", + "sha256:f0c304dcc1494dbc3fc492ebc2e61d0db323be9756a1bb14f407097d8adf82ec", + "sha256:fb9c2f27d2db6d709a02421733bb6ad22e16104beb2cf403e1120c24f3cc8668" + ], + "index": "pypi", + "version": "==4.0.0a0" + }, "appdirs": { "hashes": [ "sha256:9e5896d1372858f8dd3344faf4e5014d21849c756c8d5701f78f8a103b372d92", @@ -222,6 +250,20 @@ ], "version": "==2.2.5" }, + "async-timeout": { + "hashes": [ + "sha256:0c3c816a028d47f659d6ff5c745cb2acf1f966da1fe5c19c77a70282b25f4c5f", + "sha256:4291ca197d287d274d0b6cb5d6f8f8f82d434ed288f962539ff18cc9012f9ea3" + ], + "version": "==3.0.1" + }, + "atomicwrites": { + "hashes": [ + "sha256:03472c30eb2c5d1ba9227e4c2ca66ab8287fbfbbda3888aa93dc2e28fc6811b4", + "sha256:75a9445bac02d8d058d5e1fe689654ba5a6556a1dfd8ce6ec55a0ed79866cfa6" + ], + "version": "==1.3.0" + }, "attrs": { "hashes": [ "sha256:69c0dbf2ed392de1cb5ec704444b08a5ef81680a61cb899dc08127123af36a79", @@ -237,6 +279,13 @@ "index": "pypi", "version": "==19.3b0" }, + "chardet": { + "hashes": [ + "sha256:84ab92ed1c4d4f16916e05906b6b75a6c0fb5db821cc65e70cbd64a3e2a5eaae", + "sha256:fc323ffcaeaed0e0a02bf4d117757b98aed530d9ed4531e3e15460124c106691" + ], + "version": "==3.0.4" + }, "click": { "hashes": [ "sha256:2335065e6395b9e67ca716de5f7526736bfa6ceead690adf616d925bdc622b13", @@ -277,6 +326,20 @@ "index": "pypi", "version": "==5.0a5" }, + "idna": { + "hashes": [ + "sha256:c357b3f628cf53ae2c4c05627ecc484553142ca23264e593d327bcde5e9c3407", + "sha256:ea8b7f6188e6fa117537c3df7da9fc686d485087abf6ac197f9c46432f7e4a3c" + ], + "version": "==2.8" + }, + "importlib-metadata": { + "hashes": [ + "sha256:a9f185022cfa69e9ca5f7eabfd5a58b689894cb78a11e3c8c89398a8ccbb8e7f", + "sha256:df1403cd3aebeb2b1dcd3515ca062eecb5bd3ea7611f18cba81130c68707e879" + ], + "version": "==0.17" + }, "isort": { "hashes": [ "sha256:c40744b6bc5162bbb39c1257fe298b7a393861d50978b565f3ccd9cb9de0182a", @@ -322,14 +385,47 @@ "index": "pypi", "version": "==3.0.5" }, - "nose": { + "more-itertools": { "hashes": [ - "sha256:9ff7c6cc443f8c51994b34a667bbcf45afd6d945be7477b52e97516fd17c53ac", - "sha256:dadcddc0aefbf99eea214e0f1232b94f2fa9bd98fa8353711dacb112bfcbbb2a", - "sha256:f1bffef9cbc82628f6e7d7b40d7e255aefaa1adb6a1b1d26c69a8b79e6208a98" + "sha256:2112d2ca570bb7c3e53ea1a35cd5df42bb0fd10c45f0fb97178679c3c03d64c7", + "sha256:c3e4748ba1aad8dba30a4886b0b1a2004f9a863837b8654e7059eebf727afa5a" ], - "index": "pypi", - "version": "==1.3.7" + "markers": "python_version > '2.7'", + "version": "==7.0.0" + }, + "multidict": { + "hashes": [ + "sha256:024b8129695a952ebd93373e45b5d341dbb87c17ce49637b34000093f243dd4f", + "sha256:041e9442b11409be5e4fc8b6a97e4bcead758ab1e11768d1e69160bdde18acc3", + "sha256:045b4dd0e5f6121e6f314d81759abd2c257db4634260abcfe0d3f7083c4908ef", + "sha256:047c0a04e382ef8bd74b0de01407e8d8632d7d1b4db6f2561106af812a68741b", + "sha256:068167c2d7bbeebd359665ac4fff756be5ffac9cda02375b5c5a7c4777038e73", + "sha256:148ff60e0fffa2f5fad2eb25aae7bef23d8f3b8bdaf947a65cdbe84a978092bc", + "sha256:1d1c77013a259971a72ddaa83b9f42c80a93ff12df6a4723be99d858fa30bee3", + "sha256:1d48bc124a6b7a55006d97917f695effa9725d05abe8ee78fd60d6588b8344cd", + "sha256:31dfa2fc323097f8ad7acd41aa38d7c614dd1960ac6681745b6da124093dc351", + "sha256:34f82db7f80c49f38b032c5abb605c458bac997a6c3142e0d6c130be6fb2b941", + "sha256:3d5dd8e5998fb4ace04789d1d008e2bb532de501218519d70bb672c4c5a2fc5d", + "sha256:4a6ae52bd3ee41ee0f3acf4c60ceb3f44e0e3bc52ab7da1c2b2aa6703363a3d1", + "sha256:4b02a3b2a2f01d0490dd39321c74273fed0568568ea0e7ea23e02bd1fb10a10b", + "sha256:4b843f8e1dd6a3195679d9838eb4670222e8b8d01bc36c9894d6c3538316fa0a", + "sha256:5de53a28f40ef3c4fd57aeab6b590c2c663de87a5af76136ced519923d3efbb3", + "sha256:61b2b33ede821b94fa99ce0b09c9ece049c7067a33b279f343adfe35108a4ea7", + "sha256:6a3a9b0f45fd75dc05d8e93dc21b18fc1670135ec9544d1ad4acbcf6b86781d0", + "sha256:76ad8e4c69dadbb31bad17c16baee61c0d1a4a73bed2590b741b2e1a46d3edd0", + "sha256:7ba19b777dc00194d1b473180d4ca89a054dd18de27d0ee2e42a103ec9b7d014", + "sha256:7c1b7eab7a49aa96f3db1f716f0113a8a2e93c7375dd3d5d21c4941f1405c9c5", + "sha256:7fc0eee3046041387cbace9314926aa48b681202f8897f8bff3809967a049036", + "sha256:8ccd1c5fff1aa1427100ce188557fc31f1e0a383ad8ec42c559aabd4ff08802d", + "sha256:8e08dd76de80539d613654915a2f5196dbccc67448df291e69a88712ea21e24a", + "sha256:c18498c50c59263841862ea0501da9f2b3659c00db54abfbf823a80787fde8ce", + "sha256:c49db89d602c24928e68c0d510f4fcf8989d77defd01c973d6cbe27e684833b1", + "sha256:ce20044d0317649ddbb4e54dab3c1bcc7483c78c27d3f58ab3d0c7e6bc60d26a", + "sha256:d1071414dd06ca2eafa90c85a079169bfeb0e5f57fd0b45d44c092546fcd6fd9", + "sha256:d3be11ac43ab1a3e979dac80843b42226d5d3cccd3986f2e03152720a4297cd7", + "sha256:db603a1c235d110c860d5f39988ebc8218ee028f07a7cbc056ba6424372ca31b" + ], + "version": "==4.5.2" }, "pbr": { "hashes": [ @@ -339,6 +435,20 @@ "index": "pypi", "version": "==5.2.1" }, + "pluggy": { + "hashes": [ + "sha256:0825a152ac059776623854c1543d65a4ad408eb3d33ee114dff91e57ec6ae6fc", + "sha256:b9817417e95936bf75d85d3f8767f7df6cdde751fc40aed3bb3074cbcb77757c" + ], + "version": "==0.12.0" + }, + "py": { + "hashes": [ + "sha256:64f65755aee5b381cea27766a3a147c3f15b9b6b9ac88676de66ba2ae36793fa", + "sha256:dc639b046a6e2cff5bbe40194ad65936d6ba360b52b3c3fe1d08a82dd50b5e53" + ], + "version": "==1.8.0" + }, "pylint": { "hashes": [ "sha256:5d77031694a5fb97ea95e828c8d10fc770a1df6eb3906067aaed42201a8a6a09", @@ -347,6 +457,37 @@ "index": "pypi", "version": "==2.3.1" }, + "pytest": { + "hashes": [ + "sha256:1a8aa4fa958f8f451ac5441f3ac130d9fc86ea38780dd2715e6d5c5882700b24", + "sha256:b8bf138592384bd4e87338cb0f256bf5f615398a649d4bd83915f0e4047a5ca6" + ], + "version": "==4.5.0" + }, + "pytest-aiohttp": { + "hashes": [ + "sha256:0b9b660b146a65e1313e2083d0d2e1f63047797354af9a28d6b7c9f0726fa33d", + "sha256:c929854339637977375838703b62fef63528598bc0a9d451639eba95f4aaa44f" + ], + "index": "pypi", + "version": "==0.3.0" + }, + "pytest-asyncio": { + "hashes": [ + "sha256:9fac5100fd716cbecf6ef89233e8590a4ad61d729d1732e0a96b84182df1daaf", + "sha256:d734718e25cfc32d2bf78d346e99d33724deeba774cc4afdf491530c6184b63b" + ], + "index": "pypi", + "version": "==0.10.0" + }, + "pytest-cov": { + "hashes": [ + "sha256:2b097cde81a302e1047331b48cadacf23577e431b61e9c6f49a1170bbe3d3da6", + "sha256:e00ea4fdde970725482f1f35630d12f074e121a23801aabf2ae154ec6bdd343a" + ], + "index": "pypi", + "version": "==2.7.1" + }, "six": { "hashes": [ "sha256:3350809f0555b11f552448330d0b52d5f24c91a322ea4a15ef22629740f3761c", @@ -386,11 +527,41 @@ "markers": "implementation_name == 'cpython'", "version": "==1.3.5" }, + "wcwidth": { + "hashes": [ + "sha256:3df37372226d6e63e1b1e1eda15c594bca98a22d33a23832a90998faa96bc65e", + "sha256:f4ebe71925af7b40a864553f761ed559b43544f8f71746c2d756c7fe788ade7c" + ], + "version": "==0.1.7" + }, "wrapt": { "hashes": [ "sha256:4aea003270831cceb8a90ff27c4031da6ead7ec1886023b80ce0dfe0adf61533" ], "version": "==1.11.1" + }, + "yarl": { + "hashes": [ + "sha256:024ecdc12bc02b321bc66b41327f930d1c2c543fa9a561b39861da9388ba7aa9", + "sha256:2f3010703295fbe1aec51023740871e64bb9664c789cba5a6bdf404e93f7568f", + "sha256:3890ab952d508523ef4881457c4099056546593fa05e93da84c7250516e632eb", + "sha256:3e2724eb9af5dc41648e5bb304fcf4891adc33258c6e14e2a7414ea32541e320", + "sha256:5badb97dd0abf26623a9982cd448ff12cb39b8e4c94032ccdedf22ce01a64842", + "sha256:73f447d11b530d860ca1e6b582f947688286ad16ca42256413083d13f260b7a0", + "sha256:7ab825726f2940c16d92aaec7d204cfc34ac26c0040da727cf8ba87255a33829", + "sha256:b25de84a8c20540531526dfbb0e2d2b648c13fd5dd126728c496d7c3fea33310", + "sha256:c6e341f5a6562af74ba55205dbd56d248daf1b5748ec48a0200ba227bb9e33f4", + "sha256:c9bb7c249c4432cd47e75af3864bc02d26c9594f49c82e2a28624417f0ae63b8", + "sha256:e060906c0c585565c718d1c3841747b61c5439af2211e185f6739a9412dfbde1" + ], + "version": "==1.3.0" + }, + "zipp": { + "hashes": [ + "sha256:8c1019c6aad13642199fbe458275ad6a84907634cc9f0989877ccc4a2840139d", + "sha256:ca943a7e809cc12257001ccfb99e3563da9af99d52f261725e96dfe0f9275bc3" + ], + "version": "==0.5.1" } } } diff --git a/bumper/confserver.py b/bumper/confserver.py index 91895aa..dc99543 100644 --- a/bumper/confserver.py +++ b/bumper/confserver.py @@ -64,8 +64,11 @@ class ConfServer: self.run_async = False self.app = None - def confserver_app(self): - self.app = web.Application() + def confserver_app(self, loop=None): + if loop: + self.app = web.Application(loop=loop) + else: + self.app = web.Application() self.app.add_routes( [ diff --git a/tests/test_bumper.py b/tests/test_bumper.py index 2e43d51..d1e9060 100644 --- a/tests/test_bumper.py +++ b/tests/test_bumper.py @@ -1,6 +1,5 @@ -from nose.tools import * -import nose import mock +import pytest from tinydb.storages import MemoryStorage from tinydb import TinyDB, Query import bumper @@ -10,13 +9,13 @@ import platform def test_get_milli_time(): - assert_equals( + assert ( bumper.get_milli_time( datetime.datetime( 2018, 1, 1, 1, 0, 0, 0, tzinfo=datetime.timezone.utc ).timestamp() - ), - 1514768400000, + ) + == 1514768400000 ) @@ -27,80 +26,73 @@ def test_user_db(): # Test os_db_path platform.system = mock.MagicMock(return_value="Windows") p = platform.system() - os.getenv = mock.MagicMock(return_value="C:\AppData") + os.getenv = mock.MagicMock(return_value="C:\\AppData") o = os.getenv("APPDATA") - assert_equals(bumper.os_db_path(createdir=False), os.path.join(os.getenv("APPDATA"), "bumper.db")) + assert bumper.os_db_path(createdir=False) == os.path.join( + os.getenv("APPDATA"), "bumper.db" + ) platform.system = mock.MagicMock(return_value="Linux") - assert_equals(bumper.os_db_path(createdir=False), os.path.expanduser("~/.config/bumper.db")) + assert bumper.os_db_path(createdir=False) == 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" + assert ( + 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" + assert ( + 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 + assert "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 + assert "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 + assert "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 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 + assert 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 + assert 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 + assert bumper.check_authcode("testuser", "auth_1234") == False + # 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") + assert ( + bumper.check_token("testuser", "token_1234") == False ) # 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 + assert 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 + assert len(bumper.user_get_tokens("testuser")) == 0 # Test 0 tokens are available db = TinyDB("tests/tmp.db") tokens = db.table("tokens") @@ -114,13 +106,9 @@ def test_user_db(): } ) # Add expired token db.close() - assert_equals( - len(bumper.user_get_tokens("testuser")), 1 - ) # Test 1 tokens are available + assert 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 + assert len(bumper.user_get_tokens("testuser")) == 0 # Test 0 tokens are available db = TinyDB("tests/tmp.db") tokens = db.table("tokens") @@ -134,53 +122,49 @@ def test_user_db(): } ) # Add expired token db.close() - assert_equals( - len(bumper.user_get_tokens("testuser")), 1 - ) # Test 1 tokens are available + assert 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 + assert 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 + assert 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" + assert ( + 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 + assert 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 + assert 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 + assert bumper.bot_get("did_123") == None # 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 + assert 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 + assert 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"] + assert ( + bumper.client_get("resource_123")["xmpp_connection"] == False ) # Test that xmpp was set False for client - assert_equals( - len(bumper.get_disconnected_xmpp_clients()), 1 + assert ( + len(bumper.get_disconnected_xmpp_clients()) == 1 ) # Test len of connected xmpp clients is 1 diff --git a/tests/test_confserver.py b/tests/test_confserver.py index 4fd9a2a..eae8cb1 100644 --- a/tests/test_confserver.py +++ b/tests/test_confserver.py @@ -1,16 +1,22 @@ -from nose.tools import * import mock import bumper import asyncio +import pytest import os import json import tinydb -from aiohttp.test_utils import TestClient, TestServer, loop_context -from aiohttp import request +import pytest_aiohttp +from aiohttp import web -confserver = bumper.ConfServer("127.0.0.1:11111", False, mock.MagicMock) -confserver.confserver_app() -app = confserver.app + +def create_confserver(): + return bumper.ConfServer("127.0.0.1:11111", False, mock.MagicMock) + + +def create_app(loop): + confserver = bumper.ConfServer("127.0.0.1:11111", False, mock.MagicMock) + confserver.confserver_app(loop=loop) + return confserver.app def async_return(result): @@ -19,151 +25,115 @@ def async_return(result): return f -def test_disconnect(): - async def test_disconnect_async(): - await confserver.disconnect() - - loop = asyncio.get_event_loop() - # Test - loop.run_until_complete(test_disconnect_async()) - - -def test_base(): +async def test_base(aiohttp_client): 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) + client = await aiohttp_client(create_app) - 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 + resp = await client.get("/") + assert resp.status == 200 + text = await resp.text() + assert "Bumper!" in text -def test_login(): +async def test_login(aiohttp_client): 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) + client = await aiohttp_client(create_app) - 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()) + # Test without user + 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) + assert jsonresp["code"] == bumper.RETURN_API_SUCCESS + assert "accessToken" in jsonresp["data"] + assert "uid" in jsonresp["data"] + assert "username" in jsonresp["data"] # Add a user to db and test with existing users bumper.user_add("testuser") - # Test - loop.run_until_complete(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) + assert jsonresp["code"] == bumper.RETURN_API_SUCCESS + assert "accessToken" in jsonresp["data"] + assert "uid" in jsonresp["data"] + assert "username" in jsonresp["data"] # 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 + 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) + assert jsonresp["code"] == bumper.RETURN_API_SUCCESS + assert "accessToken" in jsonresp["data"] + assert "uid" in jsonresp["data"] + assert "username" in jsonresp["data"] -def test_logout(): +async def test_logout(aiohttp_client): 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 + client = await aiohttp_client(create_app) # 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")) + resp = await client.get( + "/1/private/us/en/dev_1234/ios/1/0/0/user/logout?accessToken={}".format( + "token_1234" + ) + ) - loop.run_until_complete( - client.close() - ) # Close test server after all tests are done + assert resp.status == 200 + text = await resp.text() + jsonresp = json.loads(text) + assert jsonresp["code"] == bumper.RETURN_API_SUCCESS -def test_checkLogin(): +async def test_checkLogin(aiohttp_client): 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) + client = await aiohttp_client(create_app) - 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 - ) + # Test without token + resp = await client.get( + "/1/private/us/en/dev_1234/ios/1/0/0/user/checkLogin?accessToken={}".format( + None ) - 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()) + ) + assert resp.status == 200 + text = await resp.text() + jsonresp = json.loads(text) + assert jsonresp["code"] == bumper.RETURN_API_SUCCESS + assert "accessToken" in jsonresp["data"] + assert jsonresp["data"]["accessToken"] != "token_1234" + assert "uid" in jsonresp["data"] + assert "username" in jsonresp["data"] # Add a user to db and test with existing users bumper.user_add("testuser") - # Test - loop.run_until_complete(test_handle_checkLogin()) + resp = await client.get( + "/1/private/us/en/dev_1234/ios/1/0/0/user/checkLogin?accessToken={}".format( + None + ) + ) + assert resp.status == 200 + text = await resp.text() + jsonresp = json.loads(text) + assert jsonresp["code"] == bumper.RETURN_API_SUCCESS + assert "accessToken" in jsonresp["data"] + assert jsonresp["data"]["accessToken"] != "token_1234" + assert "uid" in jsonresp["data"] + assert "username" in jsonresp["data"] # Remove dev from tmpuser bumper.user_remove_device("tmpuser", "dev_1234") @@ -172,231 +142,156 @@ def test_checkLogin(): 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 + resp = await client.get( + "/1/private/us/en/dev_1234/ios/1/0/0/user/checkLogin?accessToken={}".format( + "token_1234" + ) + ) + assert resp.status == 200 + text = await resp.text() + jsonresp = json.loads(text) + assert jsonresp["code"] == bumper.RETURN_API_SUCCESS + assert "accessToken" in jsonresp["data"] + assert jsonresp["data"]["accessToken"] == "token_1234" + assert "uid" in jsonresp["data"] + assert "username" in jsonresp["data"] -def test_getAuthCode(): +async def test_getAuthCode(aiohttp_client): 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 + client = await aiohttp_client(create_app) # Test without user or token - loop.run_until_complete(test_handle_getAuthCode()) + resp = await client.get( + "/1/private/us/en/dev_1234/ios/1/0/0/user/getAuthCode?uid={}&accessToken={}".format( + None, None + ) + ) + assert resp.status == 200 + text = await resp.text() + jsonresp = json.loads(text) + assert jsonresp["code"] == bumper.ERR_TOKEN_INVALID # 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")) + resp = await client.get( + "/1/private/us/en/dev_1234/ios/1/0/0/user/getAuthCode?uid={}&accessToken={}".format( + "testuser", "token_1234" + ) + ) + assert resp.status == 200 + text = await resp.text() + jsonresp = json.loads(text) + assert jsonresp["code"] == bumper.RETURN_API_SUCCESS + assert "authCode" in jsonresp["data"] + assert "ecovacsUid" in jsonresp["data"] # 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" + resp = await client.get( + "/1/private/us/en/dev_1234/ios/1/0/0/user/getAuthCode?uid={}&accessToken={}".format( + "testuser", "token_1234" ) - 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 + ) + assert resp.status == 200 + text = await resp.text() + jsonresp = json.loads(text) + assert jsonresp["code"] == bumper.RETURN_API_SUCCESS + assert "authCode" in jsonresp["data"] + assert "ecovacsUid" in jsonresp["data"] -def test_homePageAlert(): +async def test_checkAgreement(aiohttp_client): 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) + client = await aiohttp_client(create_app) - 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 + 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) + assert jsonresp["code"] == bumper.RETURN_API_SUCCESS -def test_checkVersion(): +async def test_homePageAlert(aiohttp_client): 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) + client = await aiohttp_client(create_app) - 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 + 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) + assert jsonresp["code"] == bumper.RETURN_API_SUCCESS -def test_getProductIotMap(): +async def test_checkVersion(aiohttp_client): 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) + client = await aiohttp_client(create_app) - 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 + 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) + assert jsonresp["code"] == bumper.RETURN_API_SUCCESS -def test_getUsersAPI(): +async def test_getProductIotMap(aiohttp_client): 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) + client = await aiohttp_client(create_app) - 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 + resp = await client.post("/api/pim/product/getProductIotMap") + assert resp.status == 200 + text = await resp.text() + jsonresp = json.loads(text) + assert jsonresp["code"] == bumper.RETURN_API_SUCCESS -def test_postUsersAPI(): +async def test_getUsersAPI(aiohttp_client): 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) + client = await aiohttp_client(create_app) - async def test_handle_postUsersApi(postbody=None): - resp = await client.post("/api/users/user.do", json=postbody) + resp = await client.get("/api/users/user.do") + assert resp.status == 200 + text = await resp.text() + jsonresp = json.loads(text) + assert jsonresp["result"] == "fail" - assert resp.status == 200 - text = await resp.text() - jsonresp = json.loads(text) - if jsonresp: - assert jsonresp["result"] == "ok" - else: - assert jsonresp + +async def test_postUsersAPI(aiohttp_client): + 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 + client = await aiohttp_client(create_app) # Test FindBest postbody = {"todo": "FindBest", "service": "EcoMsgNew"} - # Test - loop.run_until_complete(test_handle_postUsersApi(postbody)) + resp = await client.post("/api/users/user.do", json=postbody) + assert resp.status == 200 + text = await resp.text() + jsonresp = json.loads(text) + assert jsonresp["result"] == "ok" # Test EcoUpdate postbody = {"todo": "FindBest", "service": "EcoUpdate"} - # Test - loop.run_until_complete(test_handle_postUsersApi(postbody)) + resp = await client.post("/api/users/user.do", json=postbody) + assert resp.status == 200 + text = await resp.text() + jsonresp = json.loads(text) + assert jsonresp["result"] == "ok" # Test loginByItToken - Uses the authcode bumper.user_add("testuser") @@ -415,7 +310,11 @@ def test_postUsersAPI(): "token": "auth_1234", "userId": "testuser", } - loop.run_until_complete(test_handle_postUsersApi(postbody)) + resp = await client.post("/api/users/user.do", json=postbody) + assert resp.status == 200 + text = await resp.text() + jsonresp = json.loads(text) + assert jsonresp["result"] == "ok" # Test GetDeviceList postbody = { @@ -429,7 +328,11 @@ def test_postUsersAPI(): "todo": "GetDeviceList", "userid": "testuser", } - loop.run_until_complete(test_handle_postUsersApi(postbody)) + resp = await client.post("/api/users/user.do", json=postbody) + assert resp.status == 200 + text = await resp.text() + jsonresp = json.loads(text) + assert jsonresp["result"] == "ok" # Test SetDeviceNick postbody = { @@ -444,7 +347,11 @@ def test_postUsersAPI(): "nick": "botnick", "did": "did_1234", } - loop.run_until_complete(test_handle_postUsersApi(postbody)) + resp = await client.post("/api/users/user.do", json=postbody) + assert resp.status == 200 + text = await resp.text() + jsonresp = json.loads(text) + assert jsonresp["result"] == "ok" # Test AddOneDevice - Same as set nick for some bots postbody = { @@ -459,7 +366,11 @@ def test_postUsersAPI(): "nick": "botnick", "did": "did_1234", } - loop.run_until_complete(test_handle_postUsersApi(postbody)) + resp = await client.post("/api/users/user.do", json=postbody) + assert resp.status == 200 + text = await resp.text() + jsonresp = json.loads(text) + assert jsonresp["result"] == "ok" # Test DeleteOneDevice - remove bot postbody = { @@ -473,79 +384,50 @@ def test_postUsersAPI(): "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 + resp = await client.post("/api/users/user.do", json=postbody) + assert resp.status == 200 + text = await resp.text() + jsonresp = json.loads(text) + assert jsonresp["result"] == "ok" -def test_postLookup(): +async def test_postLookup(aiohttp_client): 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 + client = await aiohttp_client(create_app) # Test FindBest postbody = {"todo": "FindBest", "service": "EcoMsgNew"} - # Test - loop.run_until_complete(test_handle_lookup(postbody)) + resp = await client.post("/lookup.do", json=postbody) + assert resp.status == 200 + text = await resp.text() + test_resp = json.loads(text) + assert test_resp["result"] == "ok" # 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 + resp = await client.post("/lookup.do", json=postbody) + assert resp.status == 200 + text = await resp.text() + test_resp = json.loads(text) + assert test_resp["result"] == "ok" -def test_devmgr(): +async def test_devmgr(aiohttp_client): 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 + confserver = create_confserver() + client = await aiohttp_client(create_app) # Test PollSCResult postbody = {"td": "PollSCResult"} - # Test - loop.run_until_complete(test_devmanager(postbody, command=False)) + resp = await client.post("/api/dim/devmanager.do", json=postbody) + assert resp.status == 200 + text = await resp.text() + test_resp = json.loads(text) + assert test_resp["ret"] == "ok" # Test BotCommand bumper.bot_add("sn_1234", "did_1234", "dev_1234", "res_1234", "eco-ng") @@ -561,25 +443,99 @@ def test_devmgr(): confserver.helperbot.send_command = mock.MagicMock( return_value=async_return(command_getstatus_resp) ) - # Test - loop.run_until_complete(test_devmanager(postbody, command=True)) + resp = await client.post("/api/dim/devmanager.do", json=postbody) + assert resp.status == 200 + text = await resp.text() + test_resp = json.loads(text) + assert test_resp["ret"] == "ok" # 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)) + resp = await client.post("/api/dim/devmanager.do", json=postbody) + assert resp.status == 200 + text = await resp.text() + test_resp = json.loads(text) + assert test_resp["ret"] == "fail" # 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)) + resp = await client.post("/api/dim/devmanager.do", json=postbody) + assert resp.status == 200 + text = await resp.text() + test_resp = json.loads(text) + assert test_resp["ret"] == "fail" + + +async def test_dim_devmanager(aiohttp_client): + 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 + confserver = create_confserver() + client = await aiohttp_client(create_app) + + # Test PollSCResult + postbody = {"td": "PollSCResult"} + resp = await client.post("/api/dim/devmanager.do", json=postbody) + assert resp.status == 200 + text = await resp.text() + test_resp = json.loads(text) + assert test_resp["ret"] == "ok" + + # Test HasUnreadMsg + postbody = {"td": "HasUnreadMsg"} + resp = await client.post("/api/dim/devmanager.do", json=postbody) + assert resp.status == 200 + text = await resp.text() + test_resp = json.loads(text) + assert test_resp["ret"] == "ok" + assert test_resp["unRead"] == 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": "", + "ret": "ok", + } + confserver.helperbot.send_command = mock.MagicMock( + return_value=async_return(command_getstatus_resp) + ) + resp = await client.post("/api/dim/devmanager.do", json=postbody) + assert resp.status == 200 + text = await resp.text() + test_resp = json.loads(text) + assert test_resp["ret"] == "ok" + + # 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) + ) + resp = await client.post("/api/dim/devmanager.do", json=postbody) + assert resp.status == 200 + text = await resp.text() + test_resp = json.loads(text) + assert test_resp["ret"] == "fail" + assert test_resp["errno"] == "timeout" + + # 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) + ) + resp = await client.post("/api/dim/devmanager.do", json=postbody) + assert resp.status == 200 + text = await resp.text() + test_resp = json.loads(text) + assert test_resp["ret"] == "fail" - loop.run_until_complete( - client.close() - ) # Close test server after all tests are done diff --git a/tests/tests.md b/tests/tests.md index d4756ce..c485135 100644 --- a/tests/tests.md +++ b/tests/tests.md @@ -1,16 +1,16 @@ # Bumper tests -Bumper uses nosetests for the majority of test cases. Install requirements using `pipenv install --dev` +Bumper uses pytest for the majority of test cases. Install requirements using `pipenv install --dev` ## Testing Enter pipenv shell `pipenv shell` ### Run tests -`nosetests` +`python -m pytest` ### Run tests with coverage -`nosetests --cover-package bumper --with-coverage` +`python -m pytest --cov=bumper tests/` ### Run tests with coverage html report -`nosetests --cover-package bumper --with-coverage --cover-html-dir="tests/report" --cover-html` +`python -m pytest --cov=bumper tests/ --cov-report html:tests/report` The report will be output into tests/report/index.html for further analysis. \ No newline at end of file