def mock_remoto_process_check_success(conn, args):
secret = new.generate_auth_key()
out = '[mon.]\nkey = %s\ncaps mon = allow *\n' % secret
- return out.encode('utf-8').split(b'\n'), [], 0
+ return out.split('\n'), [], 0
def mock_remoto_process_check_rc_error(conn, args):
- return [b""], [b"this failed\n"], 1
+ return [""], ["this failed\n"], 1
class TestGatherKeysMissing(object):
keyname = gatherkeys.keytype_path_to(self.args, keytype)
keypath_gen = os.path.join(self.test_dir, keyname)
assert os.path.isfile(keypath_gen)
-
+
@mock.patch('ceph_deploy.lib.remoto.process.check', mock_remoto_process_check_success)
def test_success_osd(self):
keytype = 'osd'
class mock_distro(object):
def __init__(self):
self.conn = mock_conn()
-
+
def mock_hosts_get_file_key_content(host, **kwargs):
output = mock_distro()
mon_keyring = '[mon.]\nkey = %s\ncaps mon = allow *\n' % new.generate_auth_key()
- output.conn.remote_module.get_file_result = mon_keyring.encode('utf-8')
+ output.conn.remote_module.get_file_result = mon_keyring
output.conn.remote_module.longhostname = host
return output
def mock_remoto_process_check_success(conn, args):
out = json.dumps(remoto_process_check_success_output,sort_keys=True, indent=4)
- return out.encode('utf-8').split(b'\n'), [], 0
+ return out.split('\n'), [], 0
def mock_remoto_process_check_rc_error(conn, args):
- return [b""], [b"this failed\n"], 1
+ return [""], ["this failed\n"], 1
def mock_remoto_process_check_out_not_json(conn, args):
- return [b"}bad output{"], [b""], 0
+ return ["}bad output{"], [""], 0
def mock_remoto_process_check_out_missing_quorum(conn, args):
outdata = copy.deepcopy(remoto_process_check_success_output)
del outdata["quorum"]
out = json.dumps(outdata,sort_keys=True, indent=4)
- return out.encode('utf-8').split(b'\n'), [], 0
+ return out.split('\n'), [], 0
def mock_remoto_process_check_out_missing_quorum_1(conn, args):
outdata = copy.deepcopy(remoto_process_check_success_output)
del outdata["quorum"][1]
out = json.dumps(outdata,sort_keys=True, indent=4)
- return out.encode('utf-8').split(b'\n'), [], 0
+ return out.split('\n'), [], 0
def mock_remoto_process_check_out_missing_monmap(conn, args):
outdata = copy.deepcopy(remoto_process_check_success_output)
del outdata["monmap"]
out = json.dumps(outdata,sort_keys=True, indent=4)
- return out.encode('utf-8').split(b'\n'), [], 0
+ return out.split('\n'), [], 0
def mock_remoto_process_check_out_missing_mons(conn, args):
outdata = copy.deepcopy(remoto_process_check_success_output)
del outdata["monmap"]["mons"]
out = json.dumps(outdata,sort_keys=True, indent=4)
- return out.encode('utf-8').split(b'\n'), [], 0
+ return out.split('\n'), [], 0
def mock_remoto_process_check_out_missing_monmap_host1(conn, args):
outdata = copy.deepcopy(remoto_process_check_success_output)
del outdata["monmap"]["mons"][1]
out = json.dumps(outdata,sort_keys=True, indent=4)
- return out.encode('utf-8').split(b'\n'), [], 0
+ return out.split('\n'), [], 0
class TestGatherKeysWithMon(object):