ceph
/
ceph-ansible
mirror of https://github.com/ceph/ceph-ansible.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

library/ceph_key.py: rewrite for generate_ceph_cmd() 

Also code lint with flake8

Signed-off-by: Wong Hoi Sing Edison <hswong3i@pantarei-design.com>
pull/6615/head
Wong Hoi Sing Edison 2021-06-17 23:43:13 +08:00 committed by Guillaume Abrioux
parent 2491d4e004
commit 793d529302
2 changed files with 60 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

96
library/ceph_key.py
View File

@ -19,9 +19,15 @@ __metaclass__ = type
from ansible.module_utils.basic import AnsibleModule from ansible.module_utils.basic import AnsibleModule
try: try:
from ansible.module_utils.ca_common import is_containerized, container_exec, fatal from ansible.module_utils.ca_common import generate_ceph_cmd, \
is_containerized, \
container_exec, \
fatal
except ImportError: except ImportError:
from module_utils.ca_common import is_containerized, container_exec, fatal from module_utils.ca_common import generate_ceph_cmd, \
is_containerized, \
container_exec, \
fatal
import datetime import datetime
import json import json
import os import os
@ -250,34 +256,6 @@ def generate_caps(_type, caps):
return caps_cli return caps_cli
def generate_ceph_cmd(cluster, args, user, user_key_path, container_image=None):
'''
Generate 'ceph' command line to execute
'''
if container_image:
binary = 'ceph'
cmd = container_exec(
binary, container_image)
else:
binary = ['ceph']
cmd = binary
base_cmd = [
'-n',
user,
'-k',
user_key_path,
'--cluster',
cluster,
'auth',
]
cmd.extend(base_cmd + args)
return cmd
def generate_ceph_authtool_cmd(cluster, name, secret, caps, dest, container_image=None): # noqa E501 def generate_ceph_authtool_cmd(cluster, name, secret, caps, dest, container_image=None): # noqa E501
''' '''
Generate 'ceph-authtool' command line to execute Generate 'ceph-authtool' command line to execute
@ -306,7 +284,7 @@ def generate_ceph_authtool_cmd(cluster, name, secret, caps, dest, container_imag
return cmd return cmd
def create_key(module, result, cluster, user, user_key_path, name, secret, caps, import_key, dest, container_image=None): # noqa E501 def create_key(module, result, cluster, user, user_key, name, secret, caps, import_key, dest, container_image=None): # noqa E501
''' '''
Create a CephX key Create a CephX key
''' '''
@ -326,13 +304,17 @@ def create_key(module, result, cluster, user, user_key_path, name, secret, caps,
cluster, name, secret, caps, dest, container_image)) cluster, name, secret, caps, dest, container_image))
if import_key or user != 'client.admin': if import_key or user != 'client.admin':
cmd_list.append(generate_ceph_cmd( cmd_list.append(generate_ceph_cmd(sub_cmd=['auth'],
cluster, args, user, user_key_path, container_image)) args=args,
cluster=cluster,
user=user,
user_key=user_key,
container_image=container_image))
return cmd_list return cmd_list
def delete_key(cluster, user, user_key_path, name, container_image=None): def delete_key(cluster, user, user_key, name, container_image=None):
''' '''
Delete a CephX key Delete a CephX key
''' '''
@ -344,13 +326,17 @@ def delete_key(cluster, user, user_key_path, name, container_image=None):
name, name,
] ]
cmd_list.append(generate_ceph_cmd( cmd_list.append(generate_ceph_cmd(sub_cmd=['auth'],
cluster, args, user, user_key_path, container_image)) args=args,
cluster=cluster,
user=user,
user_key=user_key,
container_image=container_image))
return cmd_list return cmd_list
def get_key(cluster, user, user_key_path, name, dest, container_image=None): def get_key(cluster, user, user_key, name, dest, container_image=None):
''' '''
Get a CephX key (write on the filesystem) Get a CephX key (write on the filesystem)
''' '''
@ -364,13 +350,17 @@ def get_key(cluster, user, user_key_path, name, dest, container_image=None):
dest, dest,
] ]
cmd_list.append(generate_ceph_cmd( cmd_list.append(generate_ceph_cmd(sub_cmd=['auth'],
cluster, args, user, user_key_path, container_image)) args=args,
cluster=cluster,
user=user,
user_key=user_key,
container_image=container_image))
return cmd_list return cmd_list
def info_key(cluster, name, user, user_key_path, output_format, container_image=None): # noqa E501 def info_key(cluster, name, user, user_key, output_format, container_image=None): # noqa E501
''' '''
Get information about a CephX key Get information about a CephX key
''' '''
@ -384,13 +374,17 @@ def info_key(cluster, name, user, user_key_path, output_format, container_image=
output_format, output_format,
] ]
cmd_list.append(generate_ceph_cmd( cmd_list.append(generate_ceph_cmd(sub_cmd=['auth'],
cluster, args, user, user_key_path, container_image)) args=args,
cluster=cluster,
user=user,
user_key=user_key,
container_image=container_image))
return cmd_list return cmd_list
def list_keys(cluster, user, user_key_path, container_image=None): def list_keys(cluster, user, user_key, container_image=None):
''' '''
List all CephX keys List all CephX keys
''' '''
@ -403,8 +397,12 @@ def list_keys(cluster, user, user_key_path, container_image=None):
'json', 'json',
] ]
cmd_list.append(generate_ceph_cmd( cmd_list.append(generate_ceph_cmd(sub_cmd=['auth'],
cluster, args, user, user_key_path, container_image)) args=args,
cluster=cluster,
user=user,
user_key=user_key,
container_image=container_image))
return cmd_list return cmd_list
@ -481,15 +479,15 @@ def run_module():
module_args = dict( module_args = dict(
cluster=dict(type='str', required=False, default='ceph'), cluster=dict(type='str', required=False, default='ceph'),
name=dict(type='str', required=False), name=dict(type='str', required=False),
state=dict(type='str', required=False, default='present', choices=['present', 'update', 'absent', state=dict(type='str', required=False, default='present', choices=['present', 'update', 'absent', # noqa E501
'list', 'info', 'fetch_initial_keys', 'generate_secret']), 'list', 'info', 'fetch_initial_keys', 'generate_secret']), # noqa E501
caps=dict(type='dict', required=False, default=None), caps=dict(type='dict', required=False, default=None),
secret=dict(type='str', required=False, default=None, no_log=True), secret=dict(type='str', required=False, default=None, no_log=True),
import_key=dict(type='bool', required=False, default=True), import_key=dict(type='bool', required=False, default=True),
dest=dict(type='str', required=False, default='/etc/ceph/'), dest=dict(type='str', required=False, default='/etc/ceph/'),
user=dict(type='str', required=False, default='client.admin'), user=dict(type='str', required=False, default='client.admin'),
user_key=dict(type='str', required=False, default=None), user_key=dict(type='str', required=False, default=None),
output_format=dict(type='str', required=False, default='json', choices=['json', 'plain', 'xml', 'yaml']) output_format=dict(type='str', required=False, default='json', choices=['json', 'plain', 'xml', 'yaml']) # noqa E501
) )
module = AnsibleModule( module = AnsibleModule(
@ -610,7 +608,7 @@ def run_module():
elif state == "absent": elif state == "absent":
if key_exist == 0: if key_exist == 0:
rc, cmd, out, err = exec_commands( rc, cmd, out, err = exec_commands(
module, delete_key(cluster, user, user_key_path, name, container_image)) module, delete_key(cluster, user, user_key_path, name, container_image)) # noqa E501
if rc == 0: if rc == 0:
changed = True changed = True
else: else:

17
tests/library/test_ceph_key.py
View File

@ -50,7 +50,7 @@ class TestCephKeyModule(object):
fake_cluster = "fake" fake_cluster = "fake"
fake_args = ['arg'] fake_args = ['arg']
fake_user = "fake-user" fake_user = "fake-user"
fake_key = "/tmp/my-key" fake_user_key = "/tmp/my-key"
expected_command_list = [ expected_command_list = [
'ceph', 'ceph',
'-n', '-n',
@ -63,14 +63,18 @@ class TestCephKeyModule(object):
'arg' 'arg'
] ]
result = ceph_key.generate_ceph_cmd( result = ceph_key.generate_ceph_cmd(
fake_cluster, fake_args, fake_user, fake_key) sub_cmd=['auth'],
args=fake_args,
cluster=fake_cluster,
user=fake_user,
user_key=fake_user_key)
assert result == expected_command_list assert result == expected_command_list
def test_generate_ceph_cmd_list_container(self): def test_generate_ceph_cmd_list_container(self):
fake_cluster = "fake" fake_cluster = "fake"
fake_args = ['arg'] fake_args = ['arg']
fake_user = "fake-user" fake_user = "fake-user"
fake_key = "/tmp/my-key" fake_user_key = "/tmp/my-key"
fake_container_image = "quay.ceph.io/ceph-ci/daemon:latest-luminous" fake_container_image = "quay.ceph.io/ceph-ci/daemon:latest-luminous"
expected_command_list = ['docker', expected_command_list = ['docker',
'run', 'run',
@ -90,7 +94,12 @@ class TestCephKeyModule(object):
'auth', 'auth',
'arg'] 'arg']
result = ceph_key.generate_ceph_cmd( result = ceph_key.generate_ceph_cmd(
fake_cluster, fake_args, fake_user, fake_key, fake_container_image) sub_cmd=['auth'],
args=fake_args,
cluster=fake_cluster,
user=fake_user,
user_key=fake_user_key,
container_image=fake_container_image)
assert result == expected_command_list assert result == expected_command_list
def test_generate_ceph_authtool_cmd_non_container_no_auid(self): def test_generate_ceph_authtool_cmd_non_container_no_auid(self):