Add unit tests

kubemarine/procedures/install.py

@@ -606,16 +606,23 @@ def overview(cluster: KubernetesCluster) -> None:
 
 
 def run_tasks(res: DynamicResources, tasks_filter: List[str] = None) -> None:
-    flow.run_tasks(res, tasks, cumulative_points=cumulative_points, tasks_filter=tasks_filter)
+    _run_tasks(res, tasks, tasks_filter)
+
+
+def _run_tasks(res: DynamicResources, tasks_: OrderedDict, tasks_filter: List[str] = None) -> None:
+    flow.run_tasks(res, tasks_, cumulative_points=cumulative_points, tasks_filter=tasks_filter)
 
 
 class InstallAction(Action):
-    def __init__(self) -> None:
+    def __init__(self, tasks_: OrderedDict = None) -> None:
         super().__init__('install')
+        if tasks_ is None:
+            tasks_ = tasks
+        self.action_tasks = tasks_
         self.target_version = "not supported"
 
     def run(self, res: DynamicResources) -> None:
-        run_tasks(res)
+        _run_tasks(res, self.action_tasks)
         self.target_version = kubernetes.get_kubernetes_version(res.cluster().inventory)
 
 

kubemarine/procedures/upgrade.py

@@ -149,21 +149,22 @@ def _run(self, resources: DynamicResources) -> None:
 
 
 class UpgradeAction(Action):
-    def __init__(self, upgrade_step: int) -> None:
+    def __init__(self, upgrade_step: int, tasks_: OrderedDict = None) -> None:
         super().__init__(f'upgrade step {upgrade_step + 1}', recreate_inventory=True)
+        if tasks_ is None:
+            tasks_ = tasks
+        self.action_tasks = copy.deepcopy(tasks_)
+        if upgrade_step > 0 and 'cleanup_tmp_dir' in self.action_tasks:
+            del self.action_tasks['cleanup_tmp_dir']
         self.upgrade_step = upgrade_step
         self.upgrade_version = 'not supported'
 
     def run(self, res: DynamicResources) -> None:
-        action_tasks = copy.deepcopy(tasks)
-        if self.upgrade_step > 0:
-            del action_tasks['cleanup_tmp_dir']
-
         context = res.context
         context['upgrade_step'] = self.upgrade_step
         res.reset_context(EnrichmentStage.DEFAULT)
         try:
-            flow.run_tasks(res, action_tasks)
+            flow.run_tasks(res, self.action_tasks)
         finally:
             upgrade_version_verified = (kubernetes.verify_version.name
                                         in res.working_context.get('proceeded_functions', []))

test/unit/core/test_cluster.py

@@ -12,12 +12,22 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-
-
+import collections
+import logging
+import os.path
+import tarfile
+import tempfile
 import unittest
+from typing import List, Optional, Set
+
+import yaml
 
 from kubemarine import demo
+from kubemarine.core import utils, log, action, errors
+from kubemarine.core.cluster import EnrichmentStage
 from kubemarine.demo import FakeKubernetesCluster
+from kubemarine.procedures import install, upgrade
+from test.unit import utils as test_utils
 
 
 def get_os_family(cluster: FakeKubernetesCluster):
@@ -102,5 +112,119 @@ def _nodes_context_one_different_os(self, inventory, host_different_os):
         return nodes_context
 
 
+class FakeResources(demo.FakeResources):
+    def __init__(self, context: dict, inventory: dict, procedure_inventory: dict = None):
+        super().__init__(context, inventory, procedure_inventory, demo.generate_nodes_context(inventory))
+        args: dict = context['execution_arguments']
+        self.inventory_filepath = args['config']
+        self.procedure_inventory_filepath: Optional[str] = args.get('procedure_config')
+
+    def _store_finalized_inventory(self, finalized_inventory: dict) -> None:
+        super()._store_finalized_inventory(finalized_inventory)
+        utils.dump_file(self, yaml.dump(finalized_inventory), "cluster_finalized.yaml")
+
+
+class ClusterStorageTest(unittest.TestCase):
+    def setUp(self):
+        self.inventory = demo.generate_inventory(**demo.ALLINONE)
+        self.procedure_inventory = None
+
+    def prepare_context(self, args: list = None, procedure: str = 'install'):
+        self.tmpdir = tempfile.TemporaryDirectory()
+
+        self.context = demo.create_silent_context(args, procedure)
+        self.context['preserve_inventory'] = True
+        self.args = self.context['execution_arguments']
+        self.args['disable_dump'] = False
+        self.args['dump_location'] = self.tmpdir.name
+        self.args['config'] = os.path.join(self.tmpdir.name, 'cluster.yaml')
+
+        utils.prepare_dump_directory(self.args['dump_location'])
+        utils.dump_file(self.context, yaml.dump(self.inventory), self.args['config'], dump_location=False)
+        if self.procedure_inventory is not None:
+            self.args['procedure_config'] = os.path.join(self.tmpdir.name, 'procedure.yaml')
+            utils.dump_file(self.context, yaml.dump(self.procedure_inventory), self.args['procedure_config'], dump_location=False)
+
+    def tearDown(self):
+        logger = logging.getLogger("k8s.fake.local")
+        for h in logger.handlers:
+            if isinstance(h, log.FileHandlerWithHeader):
+                h.close()
+        self.tmpdir.cleanup()
+
+    def _run_actions(self, actions: List[action.Action]) -> demo.FakeClusterStorage:
+        resources = FakeResources(self.context, self.inventory,
+                                  procedure_inventory=self.procedure_inventory)
+        try:
+            test_utils.run_actions(resources, actions)
+        except errors.FailException:
+            pass
+
+        return resources.cluster(EnrichmentStage.LIGHT).cluster_storage
+
+    def _check_local_archive(self, local_archive_path: str, files: Set[str]):
+        with tarfile.open(local_archive_path, "r:gz") as tar:
+            self.assertEqual(files, set(tar.getnames()))
+
+    def test_local_archive_simple(self):
+        self.prepare_context()
+        cluster_storage = self._run_actions([install.InstallAction(collections.OrderedDict())])
+        self._check_local_archive(
+            cluster_storage.uploaded_archives[0],
+            {'dump/procedure_parameters',
+             'dump/cluster_initial.yaml', 'dump/cluster_light.yaml', 'dump/cluster.yaml', 'dump/cluster_finalized.yaml',
+             'cluster.yaml', 'version'})
+
+    def test_local_archives_upgrade(self):
+        self.inventory['services']['kubeadm'] = {
+            'kubernetesVersion': 'v1.26.11'
+        }
+        self.procedure_inventory = demo.generate_procedure_inventory('upgrade')
+        self.procedure_inventory['upgrade_plan'] = ['v1.27.8', 'v1.28.4']
+
+        self.prepare_context(['fake.yaml'], procedure='upgrade')
+        cluster_storage = self._run_actions([upgrade.UpgradeAction(i, collections.OrderedDict()) for i in range(2)])
+        self.assertEqual(2, len(cluster_storage.uploaded_archives))
+        self._check_local_archive(
+            cluster_storage.uploaded_archives[0],
+            {'dump/procedure_parameters', 'dump/procedure.yaml',
+             'dump/cluster_initial.yaml', 'dump/cluster_light.yaml', 'dump/cluster_default.yaml', 'dump/cluster.yaml', 'dump/cluster_finalized.yaml',
+             'cluster.yaml', 'version'})
+        self._check_local_archive(
+            cluster_storage.uploaded_archives[1],
+            {'dump/procedure_parameters', 'dump/procedure.yaml',
+             'dump/cluster_initial.yaml', 'dump/cluster.yaml', 'dump/cluster_finalized.yaml',
+             'cluster.yaml', 'version'})
+
+        self.assertEqual({'debug.log', 'v1.27.8', 'v1.28.4'},
+                         set(os.listdir(os.path.join(self.tmpdir.name, 'dump'))))
+
+    def test_local_archive_upgrade_failed(self):
+        self.inventory['services']['kubeadm'] = {
+            'kubernetesVersion': 'v1.26.11'
+        }
+        self.procedure_inventory = demo.generate_procedure_inventory('upgrade')
+        self.procedure_inventory['upgrade_plan'] = ['v1.27.8', 'v1.28.4']
+
+        self.prepare_context(['fake.yaml'], procedure='upgrade')
+
+        def test_func(_: FakeKubernetesCluster):
+            raise Exception("test")
+
+        cluster_storage = self._run_actions([
+            upgrade.UpgradeAction(0, collections.OrderedDict()),
+            upgrade.UpgradeAction(1, collections.OrderedDict({"test": test_func})),
+        ])
+        self.assertEqual(1, len(cluster_storage.uploaded_archives))
+        self._check_local_archive(
+            cluster_storage.uploaded_archives[0],
+            {'dump/procedure_parameters', 'dump/procedure.yaml',
+             'dump/cluster_initial.yaml', 'dump/cluster_light.yaml', 'dump/cluster_default.yaml', 'dump/cluster.yaml', 'dump/cluster_finalized.yaml',
+             'cluster.yaml', 'version'})
+
+        self.assertEqual({'debug.log', 'v1.27.8', 'v1.28.4'},
+                         set(os.listdir(os.path.join(self.tmpdir.name, 'dump'))))
+
+
 if __name__ == '__main__':
     unittest.main()

test/unit/core/test_flow.py

@@ -291,6 +291,7 @@ def test_detect_nodes_context(self):
                          "Here should be all 4 calls of test_func")
 
         self.assertEqual("rhel", cluster.get_os_family())
+        self.assertEqual(len(hosts), len(cluster.nodes_context))
         for host, node_context in cluster.nodes_context.items():
             self.assertEqual({'online': True, 'accessible': True, 'sudo': 'Root'}, node_context["access"])
             self.assertEqual({'name': 'centos', 'version': '7.6', 'family': 'rhel'}, node_context["os"])
@@ -332,6 +333,22 @@ def test_any_offline_node_interrupts(self):
         self.assertIsInstance(exc, errors.FailException, msg="Exception should be raised")
         self.assertTrue(f"['{offline}'] are not reachable." in str(exc.reason))
 
+    def test_all_nodes_offline_check_iaas(self):
+        inventory = demo.generate_inventory(**demo.FULLHA_KEEPALIVED)
+        self._stub_detect_nodes_context(inventory, [], [])
+        context = demo.create_silent_context(procedure='check_iaas')
+        res = demo.FakeResources(context, inventory, fake_shell=self.light_fake_shell)
+        flow.run_tasks(res, tasks)
+        cluster = res.cluster()
+        self.assertEqual(4, cluster.context["test_info"],
+                         "Here should be all 4 calls of test_func")
+
+        self.assertEqual("<undefined>", cluster.get_os_family())
+        for host, node_context in cluster.nodes_context.items():
+            self.assertEqual({'online': False, 'accessible': False, 'sudo': "No"}, node_context["access"])
+            self.assertEqual({'name': "<undefined>", 'version': "<undefined>", 'family': "<undefined>"}, node_context["os"])
+            self.assertEqual("<undefined>", node_context["active_interface"])
+
     def test_any_removed_node_can_be_offline(self):
         inventory = demo.generate_inventory(**demo.FULLHA_KEEPALIVED)
         online_hosts = [node["address"] for node in inventory["nodes"]]
@@ -349,6 +366,31 @@ def test_any_removed_node_can_be_offline(self):
         # no exception should occur
         flow.run_tasks(res, tasks)
 
+    def test_detect_nodes_context_removed_node_online(self):
+        inventory = demo.generate_inventory(**demo.FULLHA_KEEPALIVED)
+        hosts = [node["address"] for node in inventory["nodes"]]
+        self._stub_detect_nodes_context(inventory, hosts, hosts)
+
+        i = random.randrange(len(inventory["nodes"]))
+        procedure_inventory = demo.generate_procedure_inventory('remove_node')
+        procedure_inventory["nodes"] = [{"name": inventory["nodes"][i]["name"]}]
+
+        context = demo.create_silent_context(['fake_path.yaml'], procedure='remove_node')
+        res = demo.FakeResources(context, inventory, procedure_inventory=procedure_inventory,
+                                 fake_shell=self.light_fake_shell)
+
+        flow.run_tasks(res, tasks)
+        cluster = res.cluster()
+        self.assertEqual(4, cluster.context["test_info"],
+                         "Here should be all 4 calls of test_func")
+
+        self.assertEqual("rhel", cluster.get_os_family())
+        self.assertEqual(len(hosts), len(cluster.nodes_context))
+        for host, node_context in cluster.nodes_context.items():
+            self.assertEqual({'online': True, 'accessible': True, 'sudo': 'Root'}, node_context["access"])
+            self.assertEqual({'name': 'centos', 'version': '7.6', 'family': 'rhel'}, node_context["os"])
+            self.assertEqual('eth0', node_context["active_interface"])
+
     def test_kubernetes_version_not_allowed(self):
         k8s_versions = list(sorted(static.KUBERNETES_VERSIONS["compatibility_map"], key=utils.version_key))
         k8s_latest = k8s_versions[-1]

test/unit/test_upgrade.py

@@ -16,10 +16,11 @@
 import unittest
 from copy import deepcopy
 from typing import List, Optional, Tuple
+from unittest import mock
 
 from kubemarine import kubernetes
 from kubemarine.core import errors, utils as kutils, static, log
-from kubemarine.procedures import upgrade
+from kubemarine.procedures import upgrade, install
 from kubemarine import demo
 from test.unit import utils
 
@@ -141,7 +142,6 @@ def setUp(self):
         self.old = 'v1.24.2'
         self.new = 'v1.24.11'
         self.inventory, self.context = generate_upgrade_environment(self.old)
-        self.context['upgrade_step'] = 0
         self.nodes_context = demo.generate_nodes_context(self.inventory, os_name='ubuntu', os_version='20.04')
         self.inventory['services'].update({'packages': {'associations': {
             'docker': {},
@@ -160,6 +160,7 @@ def setUp(self):
         }
 
     def _new_cluster(self):
+        self.context['upgrade_step'] = 0
         return demo.new_cluster(deepcopy(self.inventory), procedure_inventory=deepcopy(self.upgrade),
                                 context=self.context, nodes_context=self.nodes_context)
 
@@ -269,6 +270,32 @@ def test_procedure_inventory_upgrade_required_inventory_redefined(self):
                                  'containerd' in cluster.context["upgrade"]["required"]['packages'],
                                  f"CRI was {'not' if expected_upgrade_required else 'unexpectedly'} scheduled for upgrade")
 
+    def test_no_custom_packages_upgrade_not_required(self):
+        self._run_upgrade_packages_and_check(False)
+
+    def test_custom_packages_upgrade_not_required(self):
+        self.inventory['services']['packages']['install'] = ['curl']
+        self._run_upgrade_packages_and_check(False)
+
+    def test_custom_packages_procedure_extended_upgrade_required(self):
+        self.inventory['services']['packages']['install'] = ['curl']
+        self.upgrade[self.new]['packages']['install'] = ['unzip', {'<<': 'merge'}]
+        self._run_upgrade_packages_and_check(True)
+
+    def test_procedure_upgrade_custom_packages_upgrade_required(self):
+        self.upgrade[self.new]['packages']['upgrade'] = ['unzip']
+        self._run_upgrade_packages_and_check(True)
+
+    def _run_upgrade_packages_and_check(self, called: bool):
+        args = self.context['execution_arguments']
+        args['without_act'] = False
+        args['tasks'] = 'packages'
+        resources = demo.FakeResources(self.context, self.inventory,
+                                       procedure_inventory=self.upgrade, nodes_context=self.nodes_context)
+        with mock.patch.object(install, install.manage_custom_packages.__name__) as run:
+            utils.run_actions(resources, [upgrade.UpgradeAction(0)])
+            self.assertEqual(called, run.called, f"Upgrade was {'not' if called else 'unexpectedly'} run")
+
     def test_final_inventory_merge_packages(self):
         self.inventory['services']['packages'].setdefault('install', {})['include'] = ['curl']
         self.upgrade[self.new]['packages']['install'] = ['unzip', {'<<': 'merge'}]