Merge "Python 3.13: do not use removed module "pipes""

This commit is contained in:
Zuul 2024-10-01 16:45:19 +00:00 committed by Gerrit Code Review
commit b30ce4a456
6 changed files with 14 additions and 15 deletions

View File

@ -14,7 +14,7 @@
# under the License.
from http import cookiejar as http_cookiejar
import pipes
import shlex
from urllib import error as url_error
from urllib import request as url_request
@ -147,7 +147,7 @@ class SSHConnector(object):
password=self.password)
def run_ssh(self, cmd_list, check_exit_code=False):
command = ' '.join(pipes.quote(cmd_arg) for cmd_arg in cmd_list)
command = ' '.join(shlex.quote(cmd_arg) for cmd_arg in cmd_list)
with self.sshpool.item() as ssh:
try:

View File

@ -15,8 +15,8 @@
import io
import os
import pipes
import re
import shlex
import sys
from oslo_log import log
@ -294,7 +294,7 @@ class GaneshaManager(object):
fname + ".XXXXXX")[0][:-1]
self.execute(
'sh', '-c',
'echo %s > %s' % (pipes.quote(data), pipes.quote(tmpf)),
'echo %s > %s' % (shlex.quote(data), shlex.quote(tmpf)),
message='writing ' + tmpf)
return tmpf
@ -640,5 +640,5 @@ class GaneshaManager(object):
def reset_exports(self):
"""Delete all export files."""
self.execute('sh', '-c',
'rm -f %s/*.conf' % pipes.quote(self.ganesha_export_dir))
'rm -f %s/*.conf' % shlex.quote(self.ganesha_export_dir))
self._mkindex()

View File

@ -14,7 +14,7 @@
# under the License.
import os
import pipes
import shlex
from oslo_concurrency import processutils
from oslo_log import log
@ -72,7 +72,7 @@ class SSHExecutor(object):
# method. So implement workaround to enable or disable 'run as root'
# behavior.
run_as_root = kwargs.pop('run_as_root', False)
cmd = ' '.join(pipes.quote(a) for a in args)
cmd = ' '.join(shlex.quote(a) for a in args)
if run_as_root:
cmd = ' '.join(['sudo', cmd])
ssh = self.pool.get()

View File

@ -27,7 +27,7 @@ Configuration Requirements:
import math
import os
import pipes
import shlex
import socket
from oslo_concurrency import processutils
@ -115,7 +115,7 @@ class HDFSNativeShareDriver(driver.ExecuteMixin, driver.ShareDriver):
return self._run_ssh(host, cmd, check_exit_code)
def _run_ssh(self, host, cmd_list, check_exit_code=False):
command = ' '.join(pipes.quote(cmd_arg) for cmd_arg in cmd_list)
command = ' '.join(shlex.quote(cmd_arg) for cmd_arg in cmd_list)
connection = self.ssh_connections.get(host)
if not connection:
hdfs_ssh_name = self.configuration.hdfs_ssh_name

View File

@ -17,7 +17,7 @@ Utility for processing MapR cluster operations
"""
import json
import pipes
import shlex
import socket
from oslo_concurrency import processutils
@ -82,7 +82,7 @@ class BaseDriverUtil(object):
raise exception.ProcessExecutionError(str(e))
def _run_ssh(self, host, cmd_list, check_exit_code=False):
command = ' '.join(pipes.quote(cmd_arg) for cmd_arg in cmd_list)
command = ' '.join(shlex.quote(cmd_arg) for cmd_arg in cmd_list)
connection = self.ssh_connections.get(host)
if connection is None:
ssh_name = self.configuration.maprfs_ssh_name
@ -123,7 +123,7 @@ class BaseDriverUtil(object):
@staticmethod
def _as_user(cmd, user):
return ['sudo', 'su', '-', user, '-c',
' '.join(pipes.quote(cmd_arg) for cmd_arg in cmd)]
' '.join(shlex.quote(cmd_arg) for cmd_arg in cmd)]
@staticmethod
def _add_params(cmd, **kwargs):

View File

@ -26,7 +26,6 @@ from manila.share.drivers.ganesha import manager
from manila import test
from manila import utils
test_export_id = 101
test_name = 'fakefile'
test_path = '/fakedir0/export.d/fakefile.conf'
@ -349,7 +348,7 @@ class GaneshaManagerTestCase(test.TestCase):
self._manager._get_export_rados_object_name('fakeobj'))
def test_write_tmp_conf_file(self):
self.mock_object(manager.pipes, 'quote',
self.mock_object(manager.shlex, 'quote',
mock.Mock(side_effect=['fakedata',
test_tmp_path]))
test_args = [
@ -371,7 +370,7 @@ class GaneshaManagerTestCase(test.TestCase):
self._manager.execute.assert_has_calls([
mock.call(*test_args[0]),
mock.call(*test_args[1], **test_kwargs)])
manager.pipes.quote.assert_has_calls([
manager.shlex.quote.assert_has_calls([
mock.call('fakedata'),
mock.call(test_tmp_path)])
self.assertEqual(test_tmp_path, ret)