Add func-tests for ceph-osd 'service' action
This commit is contained in:
Martin Kalcok
@@ -16,6 +16,10 @@
|
||||
|
||||
import logging
|
||||
import unittest
|
||||
import re
|
||||
|
||||
from copy import deepcopy
|
||||
from typing import List
|
||||
|
||||
import zaza.openstack.charm_tests.test_utils as test_utils
|
||||
import zaza.model as zaza_model
|
||||
@@ -47,3 +51,176 @@ class SecurityTest(unittest.TestCase):
|
||||
expected_passes,
|
||||
expected_failures,
|
||||
expected_to_pass=True)
|
||||
|
||||
|
||||
class OsdService:
|
||||
"""Simple representation of ceph-osd systemd service."""
|
||||
|
||||
def __init__(self, id_: int):
|
||||
"""
|
||||
Init service using its ID.
|
||||
|
||||
e.g.: id_=1 -> ceph.-osd@1.service
|
||||
"""
|
||||
self.id = id_
|
||||
self.name = 'ceph-osd@{}'.format(id_)
|
||||
|
||||
|
||||
class ServiceTest(unittest.TestCase):
|
||||
"""ceph-osd systemd service tests."""
|
||||
|
||||
TESTED_UNIT = 'ceph-osd/0'
|
||||
SERVICE_PATTERN = re.compile(r'ceph-osd@(?P<service_id>\d+)\.service')
|
||||
|
||||
def __init__(self, methodName='runTest'):
|
||||
"""Initialize Test Case."""
|
||||
super(ServiceTest, self).__init__(methodName)
|
||||
self._available_services = None
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
"""Run class setup for running ceph service tests."""
|
||||
super(ServiceTest, cls).setUpClass()
|
||||
|
||||
def setUp(self):
|
||||
"""Run test setup."""
|
||||
# Note: This counter reset is needed because ceph-osd service is
|
||||
# limited to 3 restarts per 30 mins which is insufficient
|
||||
# when running functional tests for 'service' action. This
|
||||
# limitation is defined in /lib/systemd/system/ceph-osd@.service
|
||||
# in section [Service] with options 'StartLimitInterval' and
|
||||
# 'StartLimitBurst'
|
||||
reset_counter = 'systemctl reset-failed'
|
||||
zaza_model.run_on_unit(self.TESTED_UNIT, reset_counter)
|
||||
|
||||
def tearDown(self):
|
||||
"""Start ceph-osd services after each test.
|
||||
|
||||
This ensures that the environment is ready for the next tests.
|
||||
"""
|
||||
zaza_model.run_action_on_units([self.TESTED_UNIT, ], 'service',
|
||||
action_params={'start': 'all'},
|
||||
raise_on_failure=True)
|
||||
|
||||
@property
|
||||
def available_services(self) -> List[OsdService]:
|
||||
"""Return list of all ceph-osd services present on the TESTED_UNIT."""
|
||||
if self._available_services is None:
|
||||
self._available_services = self._fetch_osd_services()
|
||||
return self._available_services
|
||||
|
||||
def _fetch_osd_services(self) -> List[OsdService]:
|
||||
"""Fetch all ceph-osd services present on the TESTED_UNIT."""
|
||||
service_list = []
|
||||
service_list_cmd = 'systemctl list-units --full --all ' \
|
||||
'--no-pager -t service'
|
||||
result = zaza_model.run_on_unit(self.TESTED_UNIT, service_list_cmd)
|
||||
for line in result['Stdout'].split('\n'):
|
||||
service_name = self.SERVICE_PATTERN.search(line)
|
||||
if service_name:
|
||||
service_id = int(service_name.group('service_id'))
|
||||
service_list.append(OsdService(service_id))
|
||||
return service_list
|
||||
|
||||
def test_start_stop_all_by_keyword(self):
|
||||
"""Start and Stop all ceph-osd services using keyword 'all'."""
|
||||
service_list = [service.name for service in self.available_services]
|
||||
|
||||
logging.info("Running 'service stop=all' action on {} "
|
||||
"unit".format(self.TESTED_UNIT))
|
||||
zaza_model.run_action_on_units([self.TESTED_UNIT, ], 'service',
|
||||
action_params={'stop': 'all'})
|
||||
zaza_model.block_until_service_status(unit_name=self.TESTED_UNIT,
|
||||
services=service_list,
|
||||
target_status='stopped')
|
||||
|
||||
logging.info("Running 'service start=all' action on {} "
|
||||
"unit".format(self.TESTED_UNIT))
|
||||
zaza_model.run_action_on_units([self.TESTED_UNIT, ], 'service',
|
||||
action_params={'start': 'all'})
|
||||
zaza_model.block_until_service_status(unit_name=self.TESTED_UNIT,
|
||||
services=service_list,
|
||||
target_status='running',
|
||||
pgrep_full=True)
|
||||
|
||||
def test_start_stop_all_by_list(self):
|
||||
"""Start and Stop all ceph-osd services using explicit list."""
|
||||
service_list = [service.name for service in self.available_services]
|
||||
service_ids = [str(service.id) for service in self.available_services]
|
||||
action_params = ','.join(service_ids)
|
||||
|
||||
logging.info("Running 'service stop={}' action on {} "
|
||||
"unit".format(action_params, self.TESTED_UNIT))
|
||||
zaza_model.run_action_on_units([self.TESTED_UNIT, ], 'service',
|
||||
action_params={'stop': action_params})
|
||||
zaza_model.block_until_service_status(unit_name=self.TESTED_UNIT,
|
||||
services=service_list,
|
||||
target_status='stopped')
|
||||
|
||||
logging.info("Running 'service start={}' action on {} "
|
||||
"unit".format(action_params, self.TESTED_UNIT))
|
||||
zaza_model.run_action_on_units([self.TESTED_UNIT, ], 'service',
|
||||
action_params={'start': action_params})
|
||||
zaza_model.block_until_service_status(unit_name=self.TESTED_UNIT,
|
||||
services=service_list,
|
||||
target_status='running',
|
||||
pgrep_full=True)
|
||||
|
||||
def test_stop_specific(self):
|
||||
"""Stop only specified ceph-osd service."""
|
||||
if len(self.available_services) < 2:
|
||||
raise unittest.SkipTest('This test can be performed only if '
|
||||
'there\'s more than one ceph-osd service '
|
||||
'present on the tested unit')
|
||||
|
||||
should_run = deepcopy(self.available_services)
|
||||
to_stop = should_run.pop()
|
||||
should_run = [service.name for service in should_run]
|
||||
|
||||
logging.info("Running 'service stop={} on {} "
|
||||
"unit".format(to_stop.id, self.TESTED_UNIT))
|
||||
|
||||
zaza_model.run_action_on_units([self.TESTED_UNIT, ], 'service',
|
||||
action_params={'stop': to_stop.id})
|
||||
|
||||
zaza_model.block_until_service_status(unit_name=self.TESTED_UNIT,
|
||||
services=[to_stop.name, ],
|
||||
target_status='stopped')
|
||||
zaza_model.block_until_service_status(unit_name=self.TESTED_UNIT,
|
||||
services=should_run,
|
||||
target_status='running',
|
||||
pgrep_full=True)
|
||||
|
||||
def test_start_specific(self):
|
||||
"""Start only specified ceph-osd service."""
|
||||
if len(self.available_services) < 2:
|
||||
raise unittest.SkipTest('This test can be performed only if '
|
||||
'there\'s more than one ceph-osd service '
|
||||
'present on the tested unit')
|
||||
|
||||
service_names = [service.name for service in self.available_services]
|
||||
should_stop = deepcopy(self.available_services)
|
||||
to_start = should_stop.pop()
|
||||
should_stop = [service.name for service in should_stop]
|
||||
|
||||
logging.info("Stopping all running ceph-osd services")
|
||||
service_stop_cmd = 'systemctl stop ceph-osd.target'
|
||||
zaza_model.run_on_unit(self.TESTED_UNIT, service_stop_cmd)
|
||||
|
||||
zaza_model.block_until_service_status(unit_name=self.TESTED_UNIT,
|
||||
services=service_names,
|
||||
target_status='stopped')
|
||||
|
||||
logging.info("Running 'service start={} on {} "
|
||||
"unit".format(to_start.id, self.TESTED_UNIT))
|
||||
|
||||
zaza_model.run_action_on_units([self.TESTED_UNIT, ], 'service',
|
||||
action_params={'start': to_start.id})
|
||||
|
||||
zaza_model.block_until_service_status(unit_name=self.TESTED_UNIT,
|
||||
services=[to_start.name, ],
|
||||
target_status='running',
|
||||
pgrep_full=True)
|
||||
zaza_model.block_until_service_status(unit_name=self.TESTED_UNIT,
|
||||
services=should_stop,
|
||||
target_status='stopped')
|
||||
|
||||
Reference in New Issue
Block a user