File: test_service_status.py

package info (click to toggle)
python-openstacksdk 4.4.0-5
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 13,352 kB
  • sloc: python: 122,960; sh: 153; makefile: 23
file content (54 lines) | stat: -rw-r--r-- 1,910 bytes parent folder | download
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
# 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.

from openstack import connection
from openstack.tests.functional import base


class TestServiceStatus(base.BaseFunctionalTest):
    def setUp(self):
        super().setUp()
        self.require_service('dns')

        self.conn = connection.from_config(cloud_name=base.TEST_CLOUD_NAME)

        self.service_names = [
            "api",
            "backend",
            "central",
            "mdns",
            "producer",
            "sink",
            "storage",
            "worker",
        ]
        self.service_status = ["UP", "DOWN"]

    def test_service_status(self):
        service_statuses = list(self.conn.dns.service_statuses())
        if not service_statuses:
            self.skipTest(
                "The Service in Designate System is required for this test"
            )

        names = [f.service_name for f in service_statuses]
        statuses = [f.status for f in service_statuses]

        self.assertTrue(
            all(status in self.service_status for status in statuses)
        )
        self.assertTrue(all(name in self.service_names for name in names))

        # Test that we can fetch a service status
        service_status = self.conn.dns.get_service_status(service_statuses[0])
        self.assertIn(service_status.service_name, self.service_names)
        self.assertIn(service_status.status, self.service_status)