Add unit tests

.travis.yml

@@ -0,0 +1,19 @@
+# -*- coding: utf-8 -*-
+#
+# This software may be modified and distributed under the terms
+# of the Apache License, Version 2.0 license.  See the LICENSE file for details.
+
+os: linux
+dist: bionic
+sudo: false
+language: python
+cache: pip
+
+python:
+  - "3.6"
+  - "3.7"
+  - "3.8"
+  - "pypy3"
+
+install: pip install tox-travis
+script: tox

tests/full_output_test.py

@@ -0,0 +1,66 @@
+# -*- coding: utf-8 -*-
+#
+# This software may be modified and distributed under the terms
+# of the Apache License, Version 2.0 license.  See the LICENSE file for details.
+
+from os.path import join
+from unittest import mock
+import sys
+import unittest
+
+from ddt import data, ddt, unpack
+
+from kubecargoload import KubernetesCargoLoadOverviewProvider
+from kubecargoload import main as kubecargoload_main
+
+
+# pylint: disable=line-too-long
+
+
+TEST_VARIATIONS = (
+    # output name, sys.argv, pods.json, pods.top
+    ('output_cpu', ['--cpu', '--all-namespaces'], 'pods.json', 'pods.top'),
+    ('output_cpu_default', ['--cpu', '--namespace', 'default'], 'pods_default.json', 'pods_default.top'),
+    ('output_cpu_default_no_header', ['--cpu', '--no-headers', '--namespace', 'default'], 'pods_default.json', 'pods_default.top'),
+    ('output_cpu_no_header', ['--cpu', '--no-headers'], 'pods.json', 'pods.top'),
+    ('output_memory', ['--all-namespaces'], 'pods.json', 'pods.top'),
+    ('output_memory_default', ['--namespace', 'default'], 'pods_default.json', 'pods_default.top'),
+    ('output_memory_default_no_header', ['--no-headers', '--namespace', 'default'], 'pods_default.json', 'pods_default.top'),
+    ('output_memory_default_no_header_sort_by_requests_limits', ['--no-headers', '--namespace', 'default', '--sort', 'requests,limits'], 'pods_default.json', 'pods_default.top'),
+    ('output_memory_default_sort_by_requests_limits', ['--namespace', 'default', '--sort', 'requests,limits'], 'pods_default.json', 'pods_default.top'),
+    ('output_memory_no_header', ['--all-namespaces', '--no-headers'], 'pods.json', 'pods.top'),
+    ('output_memory_no_header_sort_by_requests_limits', ['--all-namespaces', '--no-headers', '--sort', 'requests,limits'], 'pods.json', 'pods.top'),
+    ('output_memory_sort_by_requests_limits', ['--all-namespaces', '--sort', 'requests,limits'], 'pods.json', 'pods.top'),
+)
+
+
+@ddt
+class FullOutputTest(unittest.TestCase):
+
+    def setUp(self):
+        super().setUp()
+        self.maxDiff = None  # pylint: disable=invalid-name
+
+    @data(*TEST_VARIATIONS)
+    @unpack
+    @mock.patch.object(KubernetesCargoLoadOverviewProvider, '_execute_kubectl_get_pods')
+    @mock.patch.object(KubernetesCargoLoadOverviewProvider, '_execute_kubectl_top_pods')
+    def test_full_output(self, output_name, argv, pods_json, pods_top, mocked_top_pods, mocked_get_pods):
+        pods_json_content = self._read_file_contents(pods_json)
+        pods_top_content = self._read_file_contents(pods_top)
+        expected_output = self._read_file_contents(output_name)
+
+        mocked_top_pods.return_value = pods_top_content
+        mocked_get_pods.return_value = pods_json_content
+
+        # mock sys.argv command line arguments
+        with mock.patch.object(sys, 'argv', ['kubecargoload.py'] + argv):
+            kubecargoload_main()
+        # check
+        output = sys.stdout.getvalue()  # pylint: disable=no-member
+        self.assertEqual(output, expected_output)
+
+    def _read_file_contents(self, filename):  # pylint: disable=no-self-use
+        path = join('tests/test_data', filename)
+        with open(path) as file_h:
+            return file_h.read()

tests/get_nested_pod_data_attribute_test.py

@@ -0,0 +1,114 @@
+# -*- coding: utf-8 -*-
+#
+# This software may be modified and distributed under the terms
+# of the Apache License, Version 2.0 license.  See the LICENSE file for details.
+
+from unittest import mock
+import json
+import unittest
+
+from kubecargoload import KubernetesCargoLoadOverviewProvider
+
+
+# pylint: disable=protected-access
+
+
+class GetNestedPodDataAttributeTest(unittest.TestCase):
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._pod_data = None
+
+    def setUp(self):
+        super().setUp()
+
+        filename = 'tests/test_data/pods.json'
+        with open(filename) as all_pod_data_f:
+            all_pod_data = json.load(all_pod_data_f)
+
+        self._pod_data = all_pod_data['items'][0]
+
+    def test_get_nested_pod_data_attribute_flat(self):
+        provider = self._factor_provider()
+        # test
+        result = provider._get_nested_pod_data_attribute('kind', pod_data=self._pod_data)
+        # check
+        expected_result = 'Pod'
+        self.assertEqual(result, expected_result)
+
+    def _factor_provider(self):  # pylint: disable=no-self-use
+        return KubernetesCargoLoadOverviewProvider(
+            namespace=None,
+            context=None,
+            show_cpu_usage=None)
+
+    def test_get_nested_pod_data_attribute_nested(self):
+        provider = self._factor_provider()
+        # test
+        result = provider._get_nested_pod_data_attribute(
+            'spec', 'schedulerName',
+            pod_data=self._pod_data)
+        # check
+        expected_result = 'default-scheduler'
+        self.assertEqual(result, expected_result)
+
+    def test_get_nested_pod_data_attribute_flat_default(self):
+        provider = self._factor_provider()
+        default = 'default-for-nonexistent-key'
+        # test
+        result = provider._get_nested_pod_data_attribute(
+            'non-nonexistent-key',
+            default=default,
+            pod_data=self._pod_data)
+        # check
+        expected_result = default
+        self.assertEqual(result, expected_result)
+
+    def test_get_nested_pod_data_attribute_nested_default(self):
+        provider = self._factor_provider()
+        default = 'default-for-nonexistent-key'
+        # test
+        result = provider._get_nested_pod_data_attribute(
+            'spec', 'nested-non-nonexistent-key',
+            default=default,
+            pod_data=self._pod_data)
+        # check
+        expected_result = default
+        self.assertEqual(result, expected_result)
+
+    def test_get_nested_pod_data_attribute_nested_default_ex(self):
+        provider = self._factor_provider()
+        default = 'default-for-nonexistent-key'
+        # test
+        result = provider._get_nested_pod_data_attribute(
+            'non-nonexistent-key', 'nested-non-nonexistent-key',
+            default=default,
+            pod_data=self._pod_data)
+        # check
+        expected_result = default
+        self.assertEqual(result, expected_result)
+
+    def test_get_nested_pod_data_attribute_empty_pod_data(self):
+        provider = self._factor_provider()
+        default = 'default-for-nonexistent-key'
+        # test
+        with mock.patch.object(provider, '_pod_data', dict()):
+            result = provider._get_nested_pod_data_attribute(
+                'non-nonexistent-key', 'nested-non-nonexistent-key',
+                default=default)
+        # check
+        expected_result = default
+        self.assertEqual(result, expected_result)
+
+    def test_get_nested_pod_data_attribute_empty_pod_data_passed(self):
+        provider = self._factor_provider()
+        default = 'default-for-nonexistent-key'
+        # test
+        with mock.patch.object(provider, '_pod_data', dict()):
+            result = provider._get_nested_pod_data_attribute(
+                'non-nonexistent-key', 'nested-non-nonexistent-key',
+                default=default,
+                pod_data=dict())
+        # check
+        expected_result = default
+        self.assertEqual(result, expected_result)

tests/humanize_bytes_test.py

@@ -0,0 +1,90 @@
+# -*- coding: utf-8 -*-
+#
+# This software may be modified and distributed under the terms
+# of the Apache License, Version 2.0 license.  See the LICENSE file for details.
+
+from decimal import Decimal
+import unittest
+
+from ddt import data, ddt, unpack
+
+from kubecargoload import KubernetesCargoLoadOverviewPrinter
+
+
+# pylint: disable=protected-access
+
+
+TEST_VARIATIONS = (
+    # bytes, precision, show_cpu_usage, result
+    (0, 1, False, '0.0  B'),
+    (1, 1, False, '1.0  B'),
+    (10, 1, False, '10.0  B'),
+    (100, 1, False, '100.0  B'),
+    (1000, 1, False, '1000.0  B'),
+    (1023, 1, False, '1023.0  B'),
+
+    (1024, 1, False, '1.0 Ki'),
+    (1025, 1, False, '1.0 Ki'),
+    (1024 * 1.04, 1, False, '1.0 Ki'),
+    (1024 * 1.05, 1, False, '1.1 Ki'),
+
+    (1024 * 1024 - 1024, 1, False, '1023.0 Ki'),
+    (1024 * 1024, 1, False, '1.0 Mi'),
+    (1024 * 1024 + 1, 1, False, '1.0 Mi'),
+    (1024 * 1024 * 1.04, 1, False, '1.0 Mi'),
+    (1024 * 1024 * 1.05, 1, False, '1.1 Mi'),
+    (1024 * 1024 * 1.2 - 1, 1, False, '1.2 Mi'),
+
+    (1024 * 1024 * 1024 - 1024 * 1024, 1, False, '1023.0 Mi'),
+    (1024 * 1024 * 1024, 1, False, '1.0 Gi'),
+    (1024 * 1024 * 1024 + 1, 1, False, '1.0 Gi'),
+    (1024 * 1024 * 1024 * 1.04, 1, False, '1.0 Gi'),
+    (1024 * 1024 * 1024 * 1.05, 1, False, '1.1 Gi'),
+    (1024 * 1024 * 1024 * 1.2 - 1, 1, False, '1.2 Gi'),
+
+    (1024 * 1024 * 1024 * 1024 - 1024 * 1024 * 1024, 1, False, '1023.0 Gi'),
+    (1024 * 1024 * 1024 * 1024, 1, False, '1.0 Ti'),
+    (1024 * 1024 * 1024 * 1024 + 1, 1, False, '1.0 Ti'),
+    (1024 * 1024 * 1024 * 1024 * 1.04, 1, False, '1.0 Ti'),
+    (1024 * 1024 * 1024 * 1024 * 1.05, 1, False, '1.1 Ti'),
+    (1024 * 1024 * 1024 * 1024 * 1.2 - 1, 1, False, '1.2 Ti'),
+
+    # precision 0
+    (0, 0, False, '0  B'),
+    (1024, 0, False, '1 Ki'),
+    (1025, 0, False, '1 Ki'),
+    (1024 * 1.04, 0, False, '1 Ki'),
+    (1024 * 1.05, 0, False, '1 Ki'),
+    (1024 * 1.5, 0, False, '2 Ki'),
+
+    # precision 2
+    (0, 2, False, '0.00  B'),
+    (1024, 2, False, '1.00 Ki'),
+    (1025, 2, False, '1.00 Ki'),
+    (1024 * 1.04, 2, False, '1.04 Ki'),
+    (1024 * 1.05, 2, False, '1.05 Ki'),
+    (1024 * 1.5, 2, False, '1.50 Ki'),
+
+    # cpu
+    (0, 0, True, '0 m'),
+    (1, 0, True, '1000 m'),
+    (24, 0, True, '24000 m'),
+    (1024, 0, True, '1024000 m'),
+)
+
+
+@ddt
+class HumanizeBytesTest(unittest.TestCase):
+
+    @data(*TEST_VARIATIONS)
+    @unpack
+    def test_humanize_bytes(self, bytes_, precision, show_cpu_usage, expected_result):
+        printer = KubernetesCargoLoadOverviewPrinter(
+            None,
+            no_header=False,
+            sort=None,
+            show_cpu_usage=show_cpu_usage)
+        # test
+        result = printer._humanize_bytes(bytes_=Decimal(bytes_), precision=precision)
+        # check
+        self.assertEqual(result, expected_result)

tests/options_test.py

@@ -0,0 +1,90 @@
+# -*- coding: utf-8 -*-
+#
+# This software may be modified and distributed under the terms
+# of the Apache License, Version 2.0 license.  See the LICENSE file for details.
+
+from unittest import mock
+import sys
+import unittest
+
+from kubecargoload import _setup_options
+
+
+class OptionsTest(unittest.TestCase):
+
+    def test_option_unknown(self):
+        test_argv = ['kubecargoload.py', '--unknown-option-which-will-never-exist-abcdefgh']
+        with mock.patch.object(sys, 'argv', test_argv):
+            with self.assertRaises(SystemExit):
+                _setup_options()
+
+    def _test_flag(self, option_short_name, option_long_name, name):
+        # short name
+        if option_short_name is not None:
+            test_argv = ['kubecargoload.py', f'-{option_short_name}']
+            with mock.patch.object(sys, 'argv', test_argv):
+                arguments = _setup_options()
+                self.assertTrue(getattr(arguments, name))
+
+        # long name
+        test_argv = ['kubecargoload.py', f'--{option_long_name}']
+        with mock.patch.object(sys, 'argv', test_argv):
+            arguments = _setup_options()
+            self.assertTrue(getattr(arguments, name))
+
+        # not set
+        test_argv = ['kubecargoload.py']
+        with mock.patch.object(sys, 'argv', test_argv):
+            arguments = _setup_options()
+            self.assertFalse(getattr(arguments, name))
+
+    def test_flag_all_namespaces(self):
+        self._test_flag('A', 'all-namespaces', 'all_namespaces')
+
+    def test_flag_show_cpu_usage(self):
+        self._test_flag('c', 'cpu', 'show_cpu_usage')
+
+    def test_flag_debug(self):
+        self._test_flag('d', 'debug', 'debug')
+
+    def test_flag_no_header(self):
+        self._test_flag('H', 'no-headers', 'no_header')
+
+    def test_flag_version(self):
+        self._test_flag('V', 'version', 'version')
+
+    def _test_option(self, option_short_name, option_long_name, name, value):
+        # short name
+        if option_short_name is not None:
+            test_argv = ['kubecargoload.py', f'-{option_short_name}', value]
+            with mock.patch.object(sys, 'argv', test_argv):
+                arguments = _setup_options()
+                short_name_value = getattr(arguments, name)
+                self.assertEqual(short_name_value, value)
+
+        # long name
+        test_argv = ['kubecargoload.py', f'--{option_long_name}', value]
+        with mock.patch.object(sys, 'argv', test_argv):
+            arguments = _setup_options()
+            long_name_value = getattr(arguments, name)
+            self.assertEqual(long_name_value, value)
+
+    def test_option_context(self):
+        self._test_option(None, 'context', 'context', 'my-k8s-cluster')
+
+    def test_option_namespace(self):
+        self._test_option('n', 'namespace', 'namespace', 'kube-system')
+
+    def test_option_sort(self):
+        self._test_option('s', 'sort', 'sort', 'name,namespace')
+
+    def test_option_exclusive_group_namespace(self):
+        test_argv = ['kubecargoload.py', '--namespace', 'kube-system', '--all-namespaces']
+        with mock.patch.object(sys, 'argv', test_argv):
+            with self.assertRaises(SystemExit):
+                _setup_options()
+
+        test_argv = ['kubecargoload.py', '-n', 'kube-system', '-A']
+        with mock.patch.object(sys, 'argv', test_argv):
+            with self.assertRaises(SystemExit):
+                _setup_options()