Merge branch 'token_redis_improvements' of https://github.com/javicacheiro/websockifyHEADmaster

1 files changed, 150 insertions, 0 deletions

diff --git a/tests/test_token_plugins.py b/tests/test_token_plugins.py
index 3e1fd19..d13d944 100644
--- a/tests/test_token_plugins.py
+++ b/tests/test_token_plugins.py
@@ -203,3 +203,153 @@ class TokenRedisTestCase(unittest.TestCase):
         self.assertIsNotNone(result)
         self.assertEqual(result[0], 'remote_host')
         self.assertEqual(result[1], 'remote_port')
+
+    @patch('redis.Redis')
+    def test_json_token_with_spaces(self, mock_redis):
+        plugin = TokenRedis('127.0.0.1:1234')
+
+        instance = mock_redis.return_value
+        instance.get.return_value = b' {"host": "remote_host:remote_port"} '
+
+        result = plugin.lookup('testhost')
+
+        instance.get.assert_called_once_with('testhost')
+        self.assertIsNotNone(result)
+        self.assertEqual(result[0], 'remote_host')
+        self.assertEqual(result[1], 'remote_port')
+
+    @patch('redis.Redis')
+    def test_text_token(self, mock_redis):
+        plugin = TokenRedis('127.0.0.1:1234')
+
+        instance = mock_redis.return_value
+        instance.get.return_value = b'remote_host:remote_port'
+
+        result = plugin.lookup('testhost')
+
+        instance.get.assert_called_once_with('testhost')
+        self.assertIsNotNone(result)
+        self.assertEqual(result[0], 'remote_host')
+        self.assertEqual(result[1], 'remote_port')
+
+    @patch('redis.Redis')
+    def test_text_token_with_spaces(self, mock_redis):
+        plugin = TokenRedis('127.0.0.1:1234')
+
+        instance = mock_redis.return_value
+        instance.get.return_value = b' remote_host:remote_port '
+
+        result = plugin.lookup('testhost')
+
+        instance.get.assert_called_once_with('testhost')
+        self.assertIsNotNone(result)
+        self.assertEqual(result[0], 'remote_host')
+        self.assertEqual(result[1], 'remote_port')
+
+    @patch('redis.Redis')
+    def test_invalid_token(self, mock_redis):
+        plugin = TokenRedis('127.0.0.1:1234')
+
+        instance = mock_redis.return_value
+        instance.get.return_value = b'{"host": "remote_host:remote_port"   '
+
+        result = plugin.lookup('testhost')
+
+        instance.get.assert_called_once_with('testhost')
+        self.assertIsNone(result)
+
+    def test_src_only_host(self):
+        plugin = TokenRedis('127.0.0.1')
+
+        self.assertEqual(plugin._server, '127.0.0.1')
+        self.assertEqual(plugin._port, 6379)
+        self.assertEqual(plugin._db, 0)
+        self.assertEqual(plugin._password, None)
+
+    def test_src_with_host_port(self):
+        plugin = TokenRedis('127.0.0.1:1234')
+
+        self.assertEqual(plugin._server, '127.0.0.1')
+        self.assertEqual(plugin._port, 1234)
+        self.assertEqual(plugin._db, 0)
+        self.assertEqual(plugin._password, None)
+
+    def test_src_with_host_port_db(self):
+        plugin = TokenRedis('127.0.0.1:1234:2')
+
+        self.assertEqual(plugin._server, '127.0.0.1')
+        self.assertEqual(plugin._port, 1234)
+        self.assertEqual(plugin._db, 2)
+        self.assertEqual(plugin._password, None)
+
+    def test_src_with_host_port_db_pass(self):
+        plugin = TokenRedis('127.0.0.1:1234:2:verysecret')
+
+        self.assertEqual(plugin._server, '127.0.0.1')
+        self.assertEqual(plugin._port, 1234)
+        self.assertEqual(plugin._db, 2)
+        self.assertEqual(plugin._password, 'verysecret')
+
+    def test_src_with_host_empty_port_empty_db_pass(self):
+        plugin = TokenRedis('127.0.0.1:::verysecret')
+
+        self.assertEqual(plugin._server, '127.0.0.1')
+        self.assertEqual(plugin._port, 6379)
+        self.assertEqual(plugin._db, 0)
+        self.assertEqual(plugin._password, 'verysecret')
+
+    def test_src_with_host_empty_port_empty_db_empty_pass(self):
+        plugin = TokenRedis('127.0.0.1:::')
+
+        self.assertEqual(plugin._server, '127.0.0.1')
+        self.assertEqual(plugin._port, 6379)
+        self.assertEqual(plugin._db, 0)
+        self.assertEqual(plugin._password, None)
+
+    def test_src_with_host_empty_port_empty_db_no_pass(self):
+        plugin = TokenRedis('127.0.0.1::')
+
+        self.assertEqual(plugin._server, '127.0.0.1')
+        self.assertEqual(plugin._port, 6379)
+        self.assertEqual(plugin._db, 0)
+        self.assertEqual(plugin._password, None)
+
+    def test_src_with_host_empty_port_no_db_no_pass(self):
+        plugin = TokenRedis('127.0.0.1:')
+
+        self.assertEqual(plugin._server, '127.0.0.1')
+        self.assertEqual(plugin._port, 6379)
+        self.assertEqual(plugin._db, 0)
+        self.assertEqual(plugin._password, None)
+
+    def test_src_with_host_empty_port_db_no_pass(self):
+        plugin = TokenRedis('127.0.0.1::2')
+
+        self.assertEqual(plugin._server, '127.0.0.1')
+        self.assertEqual(plugin._port, 6379)
+        self.assertEqual(plugin._db, 2)
+        self.assertEqual(plugin._password, None)
+
+    def test_src_with_host_port_empty_db_pass(self):
+        plugin = TokenRedis('127.0.0.1:1234::verysecret')
+
+        self.assertEqual(plugin._server, '127.0.0.1')
+        self.assertEqual(plugin._port, 1234)
+        self.assertEqual(plugin._db, 0)
+        self.assertEqual(plugin._password, 'verysecret')
+
+    def test_src_with_host_empty_port_db_pass(self):
+        plugin = TokenRedis('127.0.0.1::2:verysecret')
+
+        self.assertEqual(plugin._server, '127.0.0.1')
+        self.assertEqual(plugin._port, 6379)
+        self.assertEqual(plugin._db, 2)
+        self.assertEqual(plugin._password, 'verysecret')
+
+    def test_src_with_host_empty_port_db_empty_pass(self):
+        plugin = TokenRedis('127.0.0.1::2:')
+
+        self.assertEqual(plugin._server, '127.0.0.1')
+        self.assertEqual(plugin._port, 6379)
+        self.assertEqual(plugin._db, 2)
+        self.assertEqual(plugin._password, None)