Add unit tests

kubemarine/procedures/install.py

@@ -611,16 +611,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:
             cluster = res.cluster_unsafe()
             procedure_context = ({} if cluster is None or cluster.procedure_context is None

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, resources as res, flow
+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):
@@ -26,8 +36,6 @@ def get_os_family(cluster: FakeKubernetesCluster):
 
 class KubernetesClusterTest(unittest.TestCase):
 
-    # TODO: add more tests
-
     def setUp(self):
         self.cluster = demo.new_cluster(demo.generate_inventory(**demo.FULLHA))
 
@@ -90,7 +98,44 @@ def test_remove_node_different_os_get_os_family_single(self):
         cluster = demo.new_cluster(inventory, procedure_inventory=remove_node, context=context,
                                    nodes_context=nodes_context)
         self.assertEqual('debian', get_os_family(cluster),
-                         msg="One node has different OS family and thus global OS family should be 'multiple'")
+                         msg="The only node with different OS family is removed, "
+                             "and global OS family should the specific remained")
+
+    def test_remove_node_different_os_get_package_associations(self):
+        inventory = demo.generate_inventory(**demo.MINIHA_KEEPALIVED)
+        context = demo.create_silent_context(['fake.yaml'], procedure='remove_node')
+        host_different_os = inventory['nodes'][0]['address']
+        remained_host = inventory['nodes'][1]['address']
+        nodes_context = self._nodes_context_one_different_os(inventory, host_different_os)
+        remove_node = demo.generate_procedure_inventory('remove_node')
+        remove_node['nodes'] = [{"name": inventory["nodes"][0]["name"]}]
+        cluster = demo.new_cluster(inventory, procedure_inventory=remove_node, context=context,
+                                   nodes_context=nodes_context)
+        self.assertEqual('conntrack-tools',
+                         cluster.get_package_association_for_node(host_different_os, 'conntrack', 'package_name'),
+                         msg="Unexpected package associations of node to remove")
+        self.assertEqual('conntrack',
+                         cluster.get_package_association_for_node(remained_host, 'conntrack', 'package_name'),
+                         msg="Unexpected package associations of remained node")
+
+    def test_upgrade_get_redefined_package_associations(self):
+        context = demo.create_silent_context(['fake.yaml'], procedure='upgrade')
+        context['upgrade_step'] = 0
+
+        inventory = demo.generate_inventory(**demo.MINIHA_KEEPALIVED)
+        inventory['services']['kubeadm'] = {
+            'kubernetesVersion': 'v1.27.8'
+        }
+        upgrade = demo.generate_procedure_inventory('upgrade')
+        upgrade['upgrade_plan'] = ['v1.28.4']
+        upgrade.setdefault('v1.28.4', {})['packages'] = {
+            'associations': {'containerd': {'package_name': 'containerd_new'}}
+        }
+        cluster = demo.new_cluster(inventory, procedure_inventory=upgrade, context=context)
+        self.assertEqual('containerd_new',
+                         cluster.get_package_association_for_node(cluster.nodes['all'].get_any_member().get_host(),
+                                                                  'containerd', 'package_name'),
+                         msg="Package associations are not redefined")
 
     def _nodes_context_one_different_os(self, inventory, host_different_os):
         nodes_context = demo.generate_nodes_context(inventory, os_name='ubuntu', os_version='20.04')
@@ -102,5 +147,142 @@ def _nodes_context_one_different_os(self, inventory, host_different_os):
         return nodes_context
 
 
+class FakeResources(test_utils.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:
+            flow.run_actions(resources, actions)
+        except Exception:
+            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_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_upgrade_two_versions(self):
+        self.inventory['values'] = {
+            'before': 'v1.26.11', 'through': 'v1.27.8', 'after': 'v1.28.4',
+        }
+        self.inventory['services']['kubeadm'] = {
+            'kubernetesVersion': '{{ values.before }}'
+        }
+        self.procedure_inventory = demo.generate_procedure_inventory('upgrade')
+        self.procedure_inventory['upgrade_plan'] = ['{{ values.through }}', '{{ values.after }}']
+
+        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_upgrade_second_version_failed(self):
+        self.inventory['values'] = {
+            'before': 'v1.26.11', 'through': 'v1.27.8', 'after': 'v1.28.4',
+        }
+        self.inventory['services']['kubeadm'] = {
+            'kubernetesVersion': '{{ values.before }}'
+        }
+        self.procedure_inventory = demo.generate_procedure_inventory('upgrade')
+        self.procedure_inventory['upgrade_plan'] = ['{{ values.through }}', '{{ values.after }}']
+
+        self.prepare_context(['fake.yaml'], procedure='upgrade')
+
+        def failed_task(_: FakeKubernetesCluster):
+            raise Exception("test")
+
+        cluster_storage = self._run_actions([
+            upgrade.UpgradeAction(0, collections.OrderedDict()),
+            upgrade.UpgradeAction(1, collections.OrderedDict({"test": failed_task})),
+        ])
+        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'))))
+
+    def test_run_two_actions_second_failed(self):
+        self.prepare_context()
+
+        def failed_action(_: res.DynamicResources):
+            raise Exception("test")
+
+        cluster_storage = self._run_actions([
+            test_utils.new_action("test_cluster1", action_=lambda resources: resources.cluster()),
+            test_utils.new_action("test_cluster2", action_=failed_action),
+        ])
+        self.assertEqual(1, len(cluster_storage.uploaded_archives))
+        self._check_local_archive(
+            cluster_storage.uploaded_archives[0],
+            {'dump/procedure_parameters',
+             'dump/cluster_initial.yaml', 'dump/cluster_light.yaml', 'dump/cluster.yaml',
+             'cluster.yaml', 'version'})
+
+
 if __name__ == '__main__':
     unittest.main()