summaryrefslogtreecommitdiff
path: root/openstackclient/tests/functional/compute/v2/common.py
blob: 885dc1d915db84822cb8fa38c810a286e0b99774 (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
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
#   Licensed under the Apache License, Version 2.0 (the "License"); you may
#   not use this file except in compliance with the License. You may obtain
#   a copy of the License at
#
#        http://www.apache.org/licenses/LICENSE-2.0
#
#   Unless required by applicable law or agreed to in writing, software
#   distributed under the License is distributed on an "AS IS" BASIS, 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 time
import uuid

from tempest.lib import exceptions

from openstackclient.tests.functional import base


class ComputeTestCase(base.TestCase):
    """Common functional test bits for Compute commands"""

    flavor_name = None
    image_name = None
    network_arg = None

    def setUp(self):
        """Select common resources"""
        super(ComputeTestCase, self).setUp()
        self.flavor_name = self.get_flavor()
        self.image_name = self.get_image()
        self.network_arg = self.get_network()

    @classmethod
    def get_flavor(cls):
        # NOTE(rtheis): Get cirros256 or m1.tiny flavors since functional
        #               tests may create other flavors.
        flavors = cls.openstack("flavor list", parse_output=True)
        server_flavor = None
        for flavor in flavors:
            if flavor['Name'] in ['m1.tiny', 'cirros256']:
                server_flavor = flavor['Name']
                break
        return server_flavor

    @classmethod
    def get_image(cls):
        # NOTE(rtheis): Get first Cirros image since functional tests may
        #               create other images.  Image may be named '-uec' or
        #               '-disk'.
        images = cls.openstack("image list", parse_output=True)
        server_image = None
        for image in images:
            if image['Name'].startswith('cirros-') and (
                image['Name'].endswith('-uec')
                or image['Name'].endswith('-disk')
            ):
                server_image = image['Name']
                break
        return server_image

    @classmethod
    def get_network(cls):
        try:
            # NOTE(rtheis): Get private network since functional tests may
            #               create other networks.
            cmd_output = cls.openstack(
                'network show private',
                parse_output=True,
            )
        except exceptions.CommandFailed:
            return ''
        return '--nic net-id=' + cmd_output['id']

    def server_create(self, name=None, cleanup=True):
        """Create server, with cleanup"""
        if not self.flavor_name:
            self.flavor_name = self.get_flavor()
        if not self.image_name:
            self.image_name = self.get_image()
        if not self.network_arg:
            self.network_arg = self.get_network()
        name = name or uuid.uuid4().hex
        cmd_output = self.openstack(
            'server create '
            + '--flavor '
            + self.flavor_name
            + ' '
            + '--image '
            + self.image_name
            + ' '
            + self.network_arg
            + ' '
            + '--wait '
            + name,
            parse_output=True,
        )
        self.assertIsNotNone(cmd_output["id"])
        self.assertEqual(
            name,
            cmd_output["name"],
        )
        if cleanup:
            self.addCleanup(self.server_delete, name)
        return cmd_output

    def server_delete(self, name):
        """Delete server by name"""
        raw_output = self.openstack('server delete ' + name)
        self.assertOutput('', raw_output)

    def wait_for_status(
        self,
        name,
        expected_status='ACTIVE',
        wait=900,
        interval=10,
    ):
        """Wait until server reaches expected status"""
        # TODO(thowe): Add a server wait command to osc
        failures = ['ERROR']
        total_sleep = 0
        while total_sleep < wait:
            cmd_output = self.openstack(
                'server show ' + name,
                parse_output=True,
            )
            status = cmd_output['status']
            if status == expected_status:
                print('Server {} now has status {}'.format(name, status))
                break
            print(
                'Server {}: Waiting for {}, current status: {}'.format(
                    name, expected_status, status
                )
            )
            self.assertNotIn(status, failures)
            time.sleep(interval)
            total_sleep += interval

        cmd_output = self.openstack(
            'server show ' + name,
            parse_output=True,
        )
        status = cmd_output['status']
        self.assertEqual(status, expected_status)
        # give it a little bit more time
        time.sleep(5)