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/7385/head
Wong Hoi Sing Edison 2021-06-17 23:43:13 +08:00 committed by Guillaume Abrioux
parent eadef1cec7
commit bada32b4e0
2 changed files with 57 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

90
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,35 +256,7 @@ def generate_caps(_type, caps):
return caps_cli return caps_cli
def generate_ceph_cmd(cluster, args, user, user_key_path, container_image=None): def generate_ceph_authtool_cmd(cluster, name, secret, caps, dest, container_image=None): # noqa E501
'''
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
''' '''
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

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,7 +63,11 @@ 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):
@ -71,7 +75,7 @@ class TestCephKeyModule(object):
fake_args = ['arg'] fake_args = ['arg']
fake_user = "fake-user" fake_user = "fake-user"
fake_user_key = "/tmp/my-key" fake_user_key = "/tmp/my-key"
fake_container_image = "quay.io/ceph/daemon:latest-luminous" fake_container_image = "quay.ceph.io/ceph-ci/daemon:latest-luminous"
expected_command_list = ['docker', expected_command_list = ['docker',
'run', 'run',
'--rm', '--rm',
@ -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_user_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):