openstack
/
python-karborclient
Code Issues Proposed changes

Add commands for service management API 

Change-Id: I3e9a62327653ea1dc9b5807f50f250c739c1566d
Implements: blueprint karbor-service-management
This commit is contained in:
Jiao Pengju 2017-10-25 18:09:54 +08:00
parent f13c3faef4
commit aba2875905
4 changed files with 212 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

104
karborclient/tests/unit/v1/test_services.py Normal file
View File

@ -0,0 +1,104 @@
# 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 mock
from karborclient.tests.unit import base
from karborclient.tests.unit.v1 import fakes
cs = fakes.FakeClient()
mock_request_return = ({}, {'service': {}})
class ServicesTest(base.TestCaseShell):
@mock.patch('karborclient.common.http.HTTPClient.json_request')
def test_list_services(self, mock_request):
mock_request.return_value = mock_request_return
cs.services.list()
mock_request.assert_called_with(
'GET',
'/os-services',
headers={}
)
@mock.patch('karborclient.common.http.HTTPClient.json_request')
def test_list_services_with_host(self, mock_request):
mock_request.return_value = mock_request_return
cs.services.list(host='fake_host')
mock_request.assert_called_with(
'GET',
'/os-services?host=fake_host',
headers={}
)
@mock.patch('karborclient.common.http.HTTPClient.json_request')
def test_list_services_with_binary(self, mock_request):
mock_request.return_value = mock_request_return
cs.services.list(binary='fake_binary')
mock_request.assert_called_with(
'GET',
'/os-services?binary=fake_binary',
headers={}
)
@mock.patch('karborclient.common.http.HTTPClient.json_request')
def test_list_services_with_host_and_binary(self, mock_request):
mock_request.return_value = mock_request_return
cs.services.list(host='fake_host', binary='fake_binary')
mock_request.assert_called_with(
'GET',
'/os-services?binary=fake_binary&host=fake_host',
headers={}
)
@mock.patch('karborclient.common.http.HTTPClient.json_request')
def test_enable_service(self, mock_request):
mock_request.return_value = mock_request_return
body = {
'status': 'enabled'
}
cs.services.enable('1')
mock_request.assert_called_with(
'PUT',
'/os-services/1',
data=body,
headers={}
)
@mock.patch('karborclient.common.http.HTTPClient.json_request')
def test_disable_service(self, mock_request):
mock_request.return_value = mock_request_return
body = {
'status': 'disabled'
}
cs.services.disable('1')
mock_request.assert_called_with(
'PUT',
'/os-services/1',
data=body,
headers={}
)
@mock.patch('karborclient.common.http.HTTPClient.json_request')
def test_disable_service_with_reason(self, mock_request):
mock_request.return_value = mock_request_return
body = {
'status': 'disabled',
'disabled_reason': 'fake_reason'
}
cs.services.disable_log_reason('1', 'fake_reason')
mock_request.assert_called_with(
'PUT',
'/os-services/1',
data=body,
headers={}
)

2
karborclient/v1/client.py
View File

@ -20,6 +20,7 @@ from karborclient.v1 import protectables
from karborclient.v1 import providers
from karborclient.v1 import restores
from karborclient.v1 import scheduled_operations
from karborclient.v1 import services
from karborclient.v1 import triggers
from karborclient.v1 import verifications
@ -48,3 +49,4 @@ class Client(object):
operation_logs.OperationLogManager(self.http_client)
self.verifications = verifications.VerificationManager(
self.http_client)
self.services = services.ServiceManager(self.http_client)

64
karborclient/v1/services.py Normal file
View File

@ -0,0 +1,64 @@
# 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 karborclient.common import base
class Service(base.Resource):
def __repr__(self):
return "<Service %s>" % self._info
class ServiceManager(base.ManagerWithFind):
resource_class = Service
def enable(self, service_id):
"""Enable the service specified by the service ID
:param service_id: The ID of the service to enable.
"""
body = {
'status': 'enabled'
}
return self._update('/os-services/%s' % service_id, body, "service")
def disable(self, service_id):
"""Disable the service specified by the service ID.
:param service_id: The ID of the service to disable.
"""
body = {
'status': 'disabled'
}
return self._update('/os-services/%s' % service_id, body, "service")
def disable_log_reason(self, service_id, reason):
"""Disable the service with a reason.
:param service_id: The ID of the service to disable.
:param reason: The reason for disabling a service.
"""
body = {
'status': 'disabled',
'disabled_reason': reason
}
return self._update("/os-services/%s" % service_id, body, "service")
def list(self, host=None, binary=None):
"""Lists all services."""
search_opts = {
'host': host,
'binary': binary
}
resource_type = "os-services"
url = self._build_list_url(resource_type, search_opts=search_opts)
return self._list(url, 'services')

42
karborclient/v1/shell.py
View File

@ -1245,3 +1245,45 @@ def do_operationlog_show(cs, args):
"""Shows operation_log details."""
operation_log = cs.operation_logs.get(args.operation_log)
utils.print_dict(operation_log.to_dict())
@utils.arg('--host',
metavar='<hostname>',
default=None,
help='Name of host.')
@utils.arg('--binary',
metavar='<binary>',
default=None,
help='Service binary.')
def do_service_list(cs, args):
"""Show a list of all running services. Filter by host & binary."""
result = cs.services.list(host=args.host, binary=args.binary)
columns = ["Id", "Binary", "Host", "Status", "State",
"Updated_at", "Disabled Reason"]
utils.print_list(result, columns)
@utils.arg('service_id',
metavar='<service_id>',
help='ID of the service.')
def do_service_enable(cs, args):
"""Enable the service."""
result = cs.services.enable(args.service_id)
utils.print_list([result], ["Id", "Binary", "Host", "Status", "State",
"Updated_at", "Disabled Reason"])
@utils.arg('service_id',
metavar='<service_id>',
help='ID of the service.')
@utils.arg('--reason',
metavar='<reason>',
help='Reason for disabling the service.')
def do_service_disable(cs, args):
"""Disable the service"""
if args.reason:
result = cs.services.disable_log_reason(args.service_id, args.reason)
else:
result = cs.services.disable(args.service_id)
utils.print_list([result], ["Id", "Binary", "Host", "Status", "State",
"Updated_at", "Disabled Reason"])