Adding upstream version 2.2.0.upstream/2.2.0upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 199 insertions, 0 deletions

diff --git a/src/bin/shell/tests/shell_unittest.py.in b/src/bin/shell/tests/shell_unittest.py.in
new file mode 100644
index 0000000..08a05f5
--- /dev/null
+++ b/src/bin/shell/tests/shell_unittest.py.in
@@ -0,0 +1,199 @@
+#!@PYTHON@
+
+# Copyright (C) 2017-2020 Internet Systems Consortium, Inc. ("ISC")
+#
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+"""
+Kea shell unittest (python part)
+"""
+
+import unittest
+import sys
+from base64 import b64encode
+
+from kea_conn import CARequest
+
+class CARequestUnitTest(unittest.TestCase):
+    """
+    This class is dedicated to testing CARequest class. That class
+    is responsible for generation of the body and headers.
+    """
+
+    def setUp(self):
+        """
+        This method is called before each test. Currently it does nothing.
+        """
+
+    def test_body_with_service(self):
+        """
+        This test verifies if the CARequest object generates the request
+        content properly when there is one target service.
+        """
+        request = CARequest()
+        request.command = "foo"
+        request.service = ["service1"]
+        request.generate_body()
+        self.assertEqual(request.content,
+                         '{ "command": "foo", "service": ["service1"] }')
+
+    def test_body_with_multiple_service(self):
+        """
+        This test verifies if the CARequest object generates the request
+        content properly when there are two target service.
+        """
+        request = CARequest()
+        request.command = "foo"
+        request.service = ["service1", "service2/2"]
+        request.generate_body()
+        self.assertEqual(request.content,
+                         '{ "command": "foo", "service": ["service1","service2/2"] }')
+
+    def test_body_with_malformed_service(self):
+        """
+        This test verifies if the CARequest object generates the request
+        content properly when there are two target service, one is empty
+        """
+        request = CARequest()
+        request.command = "foo"
+        request.service = ["service1", ""]
+        request.generate_body()
+        self.assertEqual(request.content,
+                         '{ "command": "foo", "service": ["service1"] }')
+
+    def test_body_without_args(self):
+        """
+        This test verifies if the CARequest object generates the request
+        content properly when there are no arguments.
+        """
+        request = CARequest()
+        request.command = "foo"
+        request.generate_body()
+        self.assertEqual(request.content, '{ "command": "foo" }')
+
+    def test_body_with_args(self):
+        """
+        This test verifies if the CARequest object generates the request
+        content properly when there are arguments.
+        """
+        request = CARequest()
+        request.command = "foo"
+        request.args = '"bar": "baz"'
+        request.generate_body()
+        self.assertEqual(request.content,
+                         '{ "command": "foo", "arguments": { "bar": "baz" } }')
+
+    @staticmethod
+    def check_header(headers, header_name, value):
+        """
+        Checks if headers array contains an entry specified by
+        header_name and that its value matches specified value
+        """
+        if header_name in headers:
+            if headers[header_name] == value:
+                return True
+            print("Expected value: " + value +
+                  " does not match actual value: " +
+                  headers[header_name])
+            return False
+        print("Expected header: " + header_name + " missing")
+        return False
+
+    def test_headers(self):
+        """
+        This test checks if the headers are generated properly. Note that since
+        the content is not specified, it is 0. Therefore Content-Length is 0.
+        """
+        request = CARequest()
+        request.generate_headers()
+
+        self.assertTrue(self.check_header(request.headers,
+                                          'Content-Type', 'application/json'))
+        self.assertTrue(self.check_header(request.headers,
+                                          'Accept', '*/*'))
+        self.assertTrue(self.check_header(request.headers,
+                                          'Content-Length', '0'))
+
+    def test_header_length(self):
+        """
+        This test checks if the headers are generated properly. In
+        this test there is specific content of non-zero length, and
+        its size should be reflected in the header.
+        """
+        request = CARequest()
+        request.content = '{ "command": "foo" }'
+        request.generate_headers()
+
+        self.assertTrue(self.check_header(request.headers, 'Content-Length',
+                                          str(len(request.content))))
+
+    def test_header_version(self):
+        """
+        This test checks if the version reported in HTTP headers is
+        generated properly.
+        """
+        request = CARequest()
+        request.version = "1.2.3"
+        request.generate_headers()
+        self.assertTrue(self.check_header(request.headers, 'User-Agent',
+                                          'Kea-shell/1.2.3'))
+
+    def test_basic_http_auth(self):
+        """
+        This test check if the basic HTTP authentication credential
+        generated properly.
+        """
+        user = 'foo'
+        password = 'bar'
+        buser = user.encode('utf-8')
+        bpassword = password.encode('utf-8')
+        secret = b':'.join((buser, bpassword))
+        self.assertEqual(b'foo:bar', secret)
+        if sys.version_info[0] == 3:
+            auth = b64encode(secret).strip().decode('ascii')
+        else:
+            auth = b64encode(secret).strip().encode('ascii')
+        self.assertEqual('Zm9vOmJhcg==', auth)
+
+    def test_auth_unicode(self):
+        """
+        This test check if the basic HTTP authentication uses UTF-8 encoding.
+        """
+        if sys.version_info[0] == 3:
+            user = 'libert\xe9'
+            password = '\xe9galit\xe9'
+        else:
+            user = u'libert\xe9'
+            password = u'\xe9galit\xe9'
+        buser = user.encode('utf-8')
+        bpassword = password.encode('utf-8')
+        secret = b':'.join((buser, bpassword))
+        self.assertEqual(b'libert\xc3\xa9:\xc3\xa9galit\xc3\xa9', secret)
+        if sys.version_info[0] == 3:
+            auth = b64encode(secret).strip().decode('ascii')
+        else:
+            auth = b64encode(secret).strip().encode('ascii')
+        self.assertEqual('bGliZXJ0w6k6w6lnYWxpdMOp', auth)
+
+    def test_header_auth(self):
+        """
+        This test checks if the basic HTTP authentication header is
+        generated properly.
+        """
+        request = CARequest()
+        request.auth = "Zm9vOmJhcg=="
+        request.generate_headers()
+
+        self.assertTrue(self.check_header(request.headers,
+                                          'Authorization',
+                                          'Basic Zm9vOmJhcg=='))
+
+    def tearDown(self):
+        """
+        This method is called after each test. Currently it does nothing.
+        """
+
+if __name__ == '__main__':
+    unittest.main()