summaryrefslogtreecommitdiffstats
path: root/tests/functional/heketi/test_heketi_cluster_operations.py
blob: b8ce8822aa74491960c737c10bfa95d200a7b93f (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
import ddt

from openshiftstoragelibs import baseclass
from openshiftstoragelibs import heketi_ops


@ddt.ddt
class TestClusterOperationsTestCases(baseclass.BaseClass):
    """Class for heketi cluster creation related test cases"""

    @ddt.data("", "block", "file")
    def test_heketi_cluster_create(self, disable_volume_type):
        """Test heketi cluster creation"""
        kwargs = {"json": True}
        if disable_volume_type:
            kwargs.update({disable_volume_type: False})

        # Create heketi cluster
        cluster_info = heketi_ops.heketi_cluster_create(
            self.heketi_client_node, self.heketi_server_url, **kwargs)
        self.addCleanup(
            heketi_ops.heketi_cluster_delete, self.heketi_client_node,
            self.heketi_server_url, cluster_info["id"])

        # Validate block and file options
        err_msg = "Cluster with %s option, unexpectedly set to %s"
        if disable_volume_type:
            self.assertFalse(
                cluster_info[disable_volume_type],
                err_msg % (disable_volume_type, "True"))
        else:
            self.assertTrue(
                cluster_info["block"], err_msg % ("block", "False"))
            self.assertTrue(cluster_info["file"], err_msg % ("file", "False"))

    def test_heketi_cluster_list(self):
        """Test and validateheketi cluster list operation"""
        # Create heketi cluster
        cluster_info = heketi_ops.heketi_cluster_create(
            self.heketi_client_node, self.heketi_server_url, json=True)
        self.addCleanup(
            heketi_ops.heketi_cluster_delete, self.heketi_client_node,
            self.heketi_server_url, cluster_info["id"])

        # Get heketi cluster list and validate presence of newly
        # created cluster
        cluster_list = heketi_ops.heketi_cluster_list(
            self.heketi_client_node, self.heketi_server_url, json=True)
        err_msg = (
            "Cluster id %s not found in cluster list %s"
            % (cluster_info["id"], cluster_list["clusters"]))
        self.assertIn(cluster_info["id"], cluster_list["clusters"], err_msg)

    def test_heketi_cluster_info(self):
        """Test and validateheketi cluster info operation"""
        # Create heketi cluster
        cluster_info = heketi_ops.heketi_cluster_create(
            self.heketi_client_node, self.heketi_server_url, json=True)
        self.addCleanup(
            heketi_ops.heketi_cluster_delete, self.heketi_client_node,
            self.heketi_server_url, cluster_info["id"])

        # Get newly created heketi cluster info
        get_cluster_info = heketi_ops.heketi_cluster_info(
            self.heketi_client_node, self.heketi_server_url,
            cluster_info["id"], json=True)

        # Validate newly created heketi cluster info
        params = (
            ("id", cluster_info["id"]),
            ("block", True),
            ("file", True),
            ("blockvolumes", []),
            ("volumes", []),
            ("nodes", []))
        for param, value in params:
            self.assertEqual(get_cluster_info[param], value)

    def test_heketi_cluster_delete(self):
        """Test and validateheketi cluster delete operation"""
        # Create heketi cluster
        cluster_info = heketi_ops.heketi_cluster_create(
            self.heketi_client_node, self.heketi_server_url, json=True)

        # Delete newly created cluster
        heketi_ops.heketi_cluster_delete(
            self.heketi_client_node, self.heketi_server_url,
            cluster_info["id"])

        # Get heketi cluster list and check for absence of deleted cluster
        cluster_list = heketi_ops.heketi_cluster_list(
            self.heketi_client_node, self.heketi_server_url, json=True)
        err_msg = (
            "Cluster id %s was not expected in cluster list %s"
            % (cluster_info["id"], cluster_list["clusters"]))
        self.assertNotIn(cluster_info["id"], cluster_list["clusters"], err_msg)