x/cloudbase-init
Code Issues Proposed changes

Remove six use from cloudbaseinit

Browse Source 
Change-Id: Ia4b4add954ee5192dbf437b415d8e698c0a271b8
This commit is contained in:
Jakub Kulik 2024-01-25 08:48:39 +00:00 committed by Jakub Kulík
parent 07cae6e8c7
commit a40f5a1508
58 changed files with 106 additions and 181 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
cloudbaseinit/conf/base.py
View File

@ -14,11 +14,8 @@
import abc
import six
@six.add_metaclass(abc.ABCMeta)
class Options(object):
class Options(object, metaclass=abc.ABCMeta):
"""Contact class for all the collections of config options."""

6
cloudbaseinit/metadata/services/azureservice.py
View File

@ -13,13 +13,13 @@
# under the License.
import contextlib
import io
import os
import socket
import time
from xml.etree import ElementTree
from oslo_log import log as oslo_logging
import six
import untangle
from cloudbaseinit import conf as cloudbaseinit_conf
@ -115,13 +115,13 @@ class AzureService(base.BaseHTTPMetadataService):
path, data_xml, headers=all_headers))
if parse_xml:
return untangle.parse(six.StringIO(encoding.get_as_string(data)))
return untangle.parse(io.StringIO(encoding.get_as_string(data)))
else:
return data
@staticmethod
def _encode_xml(xml_root):
bio = six.BytesIO()
bio = io.BytesIO()
ElementTree.ElementTree(xml_root).write(
bio, encoding='utf-8', xml_declaration=True)
return bio.getvalue()

4
cloudbaseinit/metadata/services/base.py
View File

@ -21,7 +21,6 @@ import time
from oslo_log import log as oslo_logging
import requests
import six
from cloudbaseinit import conf as cloudbaseinit_conf
from cloudbaseinit import exception
@ -35,8 +34,7 @@ class NotExistingMetadataException(Exception):
pass
@six.add_metaclass(abc.ABCMeta)
class BaseMetadataService(object):
class BaseMetadataService(object, metaclass=abc.ABCMeta):
_GZIP_MAGIC_NUMBER = b'\x1f\x8b'
def __init__(self):

10
cloudbaseinit/metadata/services/cloudstack.py
View File

@ -13,11 +13,11 @@
# under the License.
import contextlib
import http.client
import posixpath
import urllib
from oslo_log import log as oslo_logging
from six.moves import http_client
from six.moves import urllib
from cloudbaseinit import conf as cloudbaseinit_conf
from cloudbaseinit.metadata.services import base
@ -131,12 +131,12 @@ class CloudStack(base.BaseHTTPMetadataService):
def _password_client(self, body=None, headers=None, decode=True):
"""Client for the Password Server."""
port = CONF.cloudstack.password_server_port
with contextlib.closing(http_client.HTTPConnection(
with contextlib.closing(http.client.HTTPConnection(
self._metadata_host, port, timeout=TIMEOUT)) as connection:
try:
connection.request("GET", "/", body=body, headers=headers)
response = connection.getresponse()
except http_client.HTTPException as exc:
except http.client.HTTPException as exc:
LOG.error("Request failed: %s", exc)
raise
@ -145,7 +145,7 @@ class CloudStack(base.BaseHTTPMetadataService):
content = encoding.get_as_string(content)
if response.status != 200:
raise http_client.HTTPException(
raise http.client.HTTPException(
"%(status)s %(reason)s - %(message)r",
{"status": response.status, "reason": response.reason,
"message": content})

3
cloudbaseinit/metadata/services/httpservice.py
View File

@ -12,8 +12,9 @@
# License for the specific language governing permissions and limitations
# under the License.
from urllib import error
from oslo_log import log as oslo_logging
from six.moves.urllib import error
from cloudbaseinit import conf as cloudbaseinit_conf
from cloudbaseinit.metadata.services import base

4
cloudbaseinit/metadata/services/opennebulaservice.py
View File

@ -21,7 +21,6 @@ import socket
import struct
from oslo_log import log as oslo_logging
import six
from cloudbaseinit.metadata.services import base
from cloudbaseinit.models import network as network_model
@ -108,8 +107,7 @@ class OpenNebulaService(base.BaseMetadataService):
address_chunks = address.split(".")
gateway_chunks = gateway.split(".")
netmask_chunks = []
for achunk, gchunk in six.moves.zip(
address_chunks, gateway_chunks):
for achunk, gchunk in zip(address_chunks, gateway_chunks):
if achunk == gchunk:
nchunk = "255"
else:

5
cloudbaseinit/metadata/services/osconfigdrive/base.py
View File

@ -15,11 +15,8 @@
import abc
import tempfile
import six
@six.add_metaclass(abc.ABCMeta)
class BaseConfigDriveManager(object):
class BaseConfigDriveManager(object, metaclass=abc.ABCMeta):
def __init__(self):
self.target_path = tempfile.mkdtemp()

2
cloudbaseinit/metadata/services/packet.py
View File

@ -15,12 +15,12 @@
import json
import requests
from urllib import error
from cloudbaseinit import conf as cloudbaseinit_conf
from cloudbaseinit import exception
from cloudbaseinit.metadata.services import base
from oslo_log import log as oslo_logging
from six.moves.urllib import error
CONF = cloudbaseinit_conf.CONF
LOG = oslo_logging.getLogger(__name__)

25
cloudbaseinit/osutils/windows.py
View File

@ -20,12 +20,11 @@ import re
import struct
import subprocess
import time
import winreg
import netaddr
from oslo_log import log as oslo_logging
import pywintypes
import six
from six.moves import winreg
from tzlocal import windows_tz
import win32api
from win32com import client
@ -602,7 +601,7 @@ class WindowsUtils(base.BaseOSUtils):
sidNameUse = wintypes.DWORD()
ret_val = advapi32.LookupAccountNameW(
0, six.text_type(username), sid, ctypes.byref(cbSid), domainName,
0, str(username), sid, ctypes.byref(cbSid), domainName,
ctypes.byref(cchReferencedDomainName), ctypes.byref(sidNameUse))
if not ret_val:
raise exception.WindowsCloudbaseInitException(
@ -613,9 +612,9 @@ class WindowsUtils(base.BaseOSUtils):
def add_user_to_local_group(self, username, groupname):
lmi = Win32_LOCALGROUP_MEMBERS_INFO_3()
lmi.lgrmi3_domainandname = six.text_type(username)
lmi.lgrmi3_domainandname = str(username)
ret_val = netapi32.NetLocalGroupAddMembers(0, six.text_type(groupname),
ret_val = netapi32.NetLocalGroupAddMembers(0, str(groupname),
3, ctypes.pointer(lmi), 1)
if ret_val == self.NERR_GroupNotFound:
@ -652,9 +651,9 @@ class WindowsUtils(base.BaseOSUtils):
{"username": username, "domain": domain})
token = wintypes.HANDLE()
ret_val = advapi32.LogonUserW(six.text_type(username),
six.text_type(domain),
six.text_type(password),
ret_val = advapi32.LogonUserW(str(username),
str(domain),
str(password),
logon_type,
self.LOGON32_PROVIDER_DEFAULT,
ctypes.byref(token))
@ -665,7 +664,7 @@ class WindowsUtils(base.BaseOSUtils):
if load_profile:
pi = Win32_PROFILEINFO()
pi.dwSize = ctypes.sizeof(Win32_PROFILEINFO)
pi.lpUserName = six.text_type(username)
pi.lpUserName = str(username)
ret_val = userenv.LoadUserProfileW(token, ctypes.byref(pi))
if not ret_val:
kernel32.CloseHandle(token)
@ -756,8 +755,7 @@ class WindowsUtils(base.BaseOSUtils):
def set_host_name(self, new_host_name):
ret_val = kernel32.SetComputerNameExW(
self.ComputerNamePhysicalDnsHostname,
six.text_type(new_host_name))
self.ComputerNamePhysicalDnsHostname, str(new_host_name))
if not ret_val:
raise exception.WindowsCloudbaseInitException(
"Cannot set host name: %r")
@ -1394,7 +1392,7 @@ class WindowsUtils(base.BaseOSUtils):
def get_volume_label(self, drive):
max_label_size = 261
label = ctypes.create_unicode_buffer(max_label_size)
ret_val = kernel32.GetVolumeInformationW(six.text_type(drive), label,
ret_val = kernel32.GetVolumeInformationW(str(drive), label,
max_label_size, 0, 0, 0, 0, 0)
if ret_val:
return label.value
@ -1404,8 +1402,7 @@ class WindowsUtils(base.BaseOSUtils):
volume_name = ctypes.create_unicode_buffer(max_volume_name_len)
if not kernel32.GetVolumeNameForVolumeMountPointW(
six.text_type(mount_point), volume_name,
max_volume_name_len):
str(mount_point), volume_name, max_volume_name_len):
if kernel32.GetLastError() in [self.ERROR_INVALID_NAME,
self.ERROR_PATH_NOT_FOUND]:
raise exception.ItemNotFoundException(

4
cloudbaseinit/plugins/common/createuser.py
View File

@ -15,7 +15,6 @@
import abc
from oslo_log import log as oslo_logging
import six
from cloudbaseinit import conf as cloudbaseinit_conf
from cloudbaseinit.osutils import factory as osutils_factory
@ -26,8 +25,7 @@ CONF = cloudbaseinit_conf.CONF
LOG = oslo_logging.getLogger(__name__)
@six.add_metaclass(abc.ABCMeta)
class BaseCreateUserPlugin(base.BasePlugin):
class BaseCreateUserPlugin(base.BasePlugin, metaclass=abc.ABCMeta):
"""This is a base class for creating or modifying an user."""
@abc.abstractmethod

5
cloudbaseinit/plugins/common/userdataplugins/base.py
View File

@ -14,11 +14,8 @@
import abc
import six
@six.add_metaclass(abc.ABCMeta)
class BaseUserDataPlugin(object):
class BaseUserDataPlugin(object, metaclass=abc.ABCMeta):
def __init__(self, mime_type):
self._mime_type = mime_type

5
cloudbaseinit/plugins/common/userdataplugins/cloudconfigplugins/base.py
View File

@ -14,11 +14,8 @@
import abc
import six
@six.add_metaclass(abc.ABCMeta)
class BaseCloudConfigPlugin(object):
class BaseCloudConfigPlugin(object, metaclass=abc.ABCMeta):
"""Base plugin class for cloud-config plugins."""
@abc.abstractmethod

4
cloudbaseinit/plugins/common/userdataplugins/cloudconfigplugins/groups.py
View File

@ -12,8 +12,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import six
from oslo_log import log as oslo_logging
from cloudbaseinit import conf as cloudbaseinit_conf
@ -45,7 +43,7 @@ class GroupsPlugin(base.BaseCloudConfigPlugin):
group_name = None
group_users = []
if isinstance(item, six.string_types):
if isinstance(item, str):
group_name = item
elif isinstance(item, dict):
try:

3
cloudbaseinit/plugins/common/userdataplugins/cloudconfigplugins/runcmd.py
View File

@ -13,7 +13,6 @@
# under the License.
import os
import six
from oslo_log import log as oslo_logging
@ -49,7 +48,7 @@ class RunCmdPlugin(base.BaseCloudConfigPlugin):
entries = 0
for command in commands:
if isinstance(command, six.string_types):
if isinstance(command, str):
script_content += command
elif isinstance(command, (list, tuple)):
subcommand_content = []

7
cloudbaseinit/plugins/common/userdataplugins/cloudconfigplugins/users.py
View File

@ -15,7 +15,6 @@
import datetime
import os
import pytz
import six
from oslo_log import log as oslo_logging
@ -41,11 +40,11 @@ class UsersPlugin(base.BaseCloudConfigPlugin):
groups = data.get('groups', None)
primary_group = data.get('primary_group', None)
user_groups = []
if isinstance(groups, six.string_types):
if isinstance(groups, str):
user_groups.extend(groups.split(', '))
elif isinstance(groups, (list, tuple)):
user_groups.extend(groups)
if isinstance(primary_group, six.string_types):
if isinstance(primary_group, str):
user_groups.extend(primary_group.split(', '))
elif isinstance(primary_group, (list, tuple)):
user_groups.extend(primary_group)
@ -68,7 +67,7 @@ class UsersPlugin(base.BaseCloudConfigPlugin):
expiredate = data.get('expiredate', None)
expire_interval = None
if isinstance(expiredate, six.string_types):
if isinstance(expiredate, str):
year, month, day = map(int, expiredate.split('-'))
expiredate = datetime.datetime(year=year, month=month, day=day,
tzinfo=pytz.utc)

3
cloudbaseinit/plugins/common/userdataplugins/cloudconfigplugins/write_files.py
View File

@ -18,7 +18,6 @@ import io
import os
from oslo_log import log as oslo_logging
import six
from cloudbaseinit import exception
from cloudbaseinit.plugins.common.userdataplugins.cloudconfigplugins import (
@ -59,7 +58,7 @@ def _convert_permissions(permissions):
def _process_content(content, encoding):
"""Decode the content taking into consideration the encoding."""
result = content
if six.PY3 and not isinstance(result, six.binary_type):
if not isinstance(result, bytes):
# At this point, content will be string, which is wrong for Python 3.
result = result.encode()

4
cloudbaseinit/plugins/common/userdataplugins/heat.py
View File

@ -15,8 +15,6 @@
import os
import six
from cloudbaseinit import conf as cloudbaseinit_conf
from cloudbaseinit.plugins.common.userdataplugins import base
from cloudbaseinit.plugins.common import userdatautils
@ -47,6 +45,6 @@ class HeatPlugin(base.BaseUserDataPlugin):
# Normalize the payload to bytes, since `execute_user_data_script`
# operates on bytes and `get_payload` returns a string on
# Python 3.
if isinstance(payload, six.text_type):
if isinstance(payload, str):
payload = payload.encode()
return userdatautils.execute_user_data_script(payload)

2
cloudbaseinit/plugins/windows/azureguestagent.py
View File

@ -15,10 +15,10 @@
import datetime
import os
import shutil
import winreg
import zipfile
from oslo_log import log as oslo_logging
from six.moves import winreg
from cloudbaseinit import conf as cloudbaseinit_conf
from cloudbaseinit import exception

4
cloudbaseinit/tests/metadata/services/osconfigdrive/test_windows.py
View File

@ -151,7 +151,7 @@ class TestWindowsConfigDriveManager(unittest.TestCase):
def test_get_iso_file_size(self):
self._test_get_iso_file_size()
@mock.patch("six.moves.builtins.open", new=OPEN)
@mock.patch("builtins.open", new=OPEN)
def test_write_iso_file(self):
file_path = "fake\\path"
file_size = 100 * 512
@ -293,7 +293,7 @@ class TestWindowsConfigDriveManager(unittest.TestCase):
@mock.patch('cloudbaseinit.metadata.services.osconfigdrive.windows.'
'WindowsConfigDriveManager.'
'_extract_iso_from_devices')
@mock.patch("six.moves.builtins.map")
@mock.patch("builtins.map")
def test_get_config_drive_from_raw_hdd(self, mock_map,
mock_extract_iso_from_devices):
Disk = self.conf_module.disk.Disk

2
cloudbaseinit/tests/metadata/services/test_azureservice.py
View File

@ -37,14 +37,12 @@ class AzureServiceTest(unittest.TestCase):
self._mock_untangle = mock.MagicMock()
self._mock_ctypes = mock.MagicMock()
self._mock_wintypes = mock.MagicMock()
self._moves_mock = mock.MagicMock()
self._module_patcher = mock.patch.dict(
'sys.modules',
{'untangle': self._mock_untangle,
'ctypes': self._mock_ctypes,
'ctypes.wintypes': self._mock_wintypes,
'six.moves': self._moves_mock
})
self._module_patcher.start()
self._azureservice_module = importlib.import_module(

2
cloudbaseinit/tests/metadata/services/test_baseconfigdrive.py
View File

@ -117,7 +117,7 @@ class TestBaseConfigDriveService(unittest.TestCase):
@mock.patch('os.path.join')
def test_get_data(self, mock_join, mock_normpath):
fake_path = os.path.join('fake', 'path')
with mock.patch('six.moves.builtins.open',
with mock.patch('builtins.open',
mock.mock_open(read_data='fake data'), create=True):
response = self._config_drive._get_data(fake_path)
self.assertEqual('fake data', response)

2
cloudbaseinit/tests/metadata/services/test_cloudstack.py
View File

@ -15,12 +15,12 @@
import functools
import socket
import unittest
import urllib
try:
import unittest.mock as mock
except ImportError:
import mock
from six.moves import urllib
from cloudbaseinit import conf as cloudbaseinit_conf
from cloudbaseinit.metadata.services import base

2
cloudbaseinit/tests/metadata/services/test_httpservice.py
View File

@ -14,12 +14,12 @@
import os
import unittest
from urllib import error
try:
import unittest.mock as mock
except ImportError:
import mock
from six.moves.urllib import error
from cloudbaseinit import conf as cloudbaseinit_conf
from cloudbaseinit.metadata.services import base

2
cloudbaseinit/tests/metadata/services/test_nocloudservice.py
View File

@ -213,7 +213,7 @@ class TestNoCloudConfigDriveService(unittest.TestCase):
@mock.patch('os.path.join')
def test_get_data(self, mock_join, mock_normpath):
fake_path = os.path.join('fake', 'path')
with mock.patch('six.moves.builtins.open',
with mock.patch('builtins.open',
mock.mock_open(read_data='fake data'), create=True):
response = self._config_drive._get_data(fake_path)
self.assertEqual('fake data', response)

2
cloudbaseinit/tests/metadata/services/test_opennebulaservice.py
View File

@ -126,7 +126,7 @@ class _TestOpenNebulaService(unittest.TestCase):
self._service = opennebulaservice.OpenNebulaService()
@mock.patch("six.moves.builtins.open", new=OPEN)
@mock.patch("builtins.open", new=OPEN)
class TestOpenNebulaService(_TestOpenNebulaService):
@classmethod

2
cloudbaseinit/tests/metadata/services/test_ovfservice.py
View File

@ -37,14 +37,12 @@ class OvfServiceTest(unittest.TestCase):
self._mock_untangle = mock.MagicMock()
self._mock_ctypes = mock.MagicMock()
self._mock_wintypes = mock.MagicMock()
self._moves_mock = mock.MagicMock()
self._module_patcher = mock.patch.dict(
'sys.modules',
{'untangle': self._mock_untangle,
'ctypes': self._mock_ctypes,
'ctypes.wintypes': self._mock_wintypes,
'six.moves': self._moves_mock
})
self._module_patcher.start()
self._ovfservice_module = importlib.import_module(

3
cloudbaseinit/tests/metadata/services/test_packet.py
View File

@ -14,14 +14,13 @@
import importlib
import unittest
from urllib import error
try:
import unittest.mock as mock
except ImportError:
import mock
from six.moves.urllib import error
from cloudbaseinit import conf as cloudbaseinit_conf
from cloudbaseinit import exception
from cloudbaseinit.tests import testutils

19
cloudbaseinit/tests/osutils/test_windows.py
View File

@ -24,7 +24,6 @@ try:
import unittest.mock as mock
except ImportError:
import mock
import six
from cloudbaseinit import conf as cloudbaseinit_conf
from cloudbaseinit import exception
@ -66,7 +65,7 @@ class TestWindowsUtils(testutils.CloudbaseInitTestBase):
self._mi_mock = mock.MagicMock()
self._wmi_mock = mock.MagicMock()
self._wmi_mock.x_wmi = WMIError
self._moves_mock = mock.MagicMock()
self._winreg_mock = mock.MagicMock()
self._xmlrpc_client_mock = mock.MagicMock()
self._ctypes_mock = mock.MagicMock()
self._tzlocal_mock = mock.Mock()
@ -86,12 +85,12 @@ class TestWindowsUtils(testutils.CloudbaseInitTestBase):
'winerror': self._winerror_mock,
'mi': self._mi_mock,
'wmi': self._wmi_mock,
'six.moves': self._moves_mock,
'six.moves.xmlrpc_client': self._xmlrpc_client_mock,
'xmlrpc.client': self._xmlrpc_client_mock,
'ctypes': self._ctypes_mock,
'pywintypes': self._pywintypes_mock,
'tzlocal': self._tzlocal_mock,
'winioctlcon': mock.MagicMock()})
'winioctlcon': mock.MagicMock(),
'winreg': self._winreg_mock})
_module_patcher.start()
self.addCleanup(_module_patcher.stop)
@ -100,7 +99,6 @@ class TestWindowsUtils(testutils.CloudbaseInitTestBase):
with mock.patch("cloudbaseinit.utils.windows.disk.GUID"):
self.windows_utils = importlib.import_module(module_path)
self._winreg_mock = self._moves_mock.winreg
self._windll_mock = self._ctypes_mock.windll
self._wintypes_mock = self._ctypes_mock.wintypes
self._client_mock = self._win32com_mock.client
@ -322,7 +320,7 @@ class TestWindowsUtils(testutils.CloudbaseInitTestBase):
response = self._winutils._get_user_sid_and_domain(self._USERNAME)
advapi32.LookupAccountNameW.assert_called_with(
0, six.text_type(self._USERNAME), sid,
0, str(self._USERNAME), sid,
self._ctypes_mock.byref(cbSid), domainName,
self._ctypes_mock.byref(cchReferencedDomainName),
self._ctypes_mock.byref(sidNameUse))
@ -369,12 +367,11 @@ class TestWindowsUtils(testutils.CloudbaseInitTestBase):
group_name)
netapi32.NetLocalGroupAddMembers.assert_called_with(
0, six.text_type(group_name), 3,
0, str(group_name), 3,
self._ctypes_mock.pointer.return_value, 1)
self._ctypes_mock.pointer.assert_called_once_with(lmi)
self.assertEqual(lmi.lgrmi3_domainandname,
six.text_type(self._USERNAME))
self.assertEqual(lmi.lgrmi3_domainandname, str(self._USERNAME))
def test_add_user_to_local_group_no_error(self):
self._test_add_user_to_local_group(ret_value=0)
@ -567,7 +564,7 @@ class TestWindowsUtils(testutils.CloudbaseInitTestBase):
mock_SetComputerNameExW.assert_called_with(
self._winutils.ComputerNamePhysicalDnsHostname,
six.text_type('fake name'))
str('fake name'))
def test_set_host_name(self):
self._test_set_host_name(ret_value='fake response')

3
cloudbaseinit/tests/plugins/common/test_mtu.py
View File

@ -18,7 +18,6 @@ try:
import unittest.mock as mock
except ImportError:
import mock
import six
from cloudbaseinit.plugins.common import base
from cloudbaseinit.plugins.common import mtu
@ -79,7 +78,7 @@ class MTUPluginTests(unittest.TestCase):
self.assertFalse(mock_get_os_utils().set_network_mtu.called)
def test_execute_success(self, mock_get_os_utils):
dhcp_options = {dhcp.OPTION_MTU: six.b("\x00\x04")}
dhcp_options = {dhcp.OPTION_MTU: "\x00\x04".encode("latin-1")}
self._test_execute(mock_get_os_utils,
dhcp_options=dhcp_options)

5
cloudbaseinit/tests/plugins/windows/test_azureguestagent.py
View File

@ -35,17 +35,16 @@ class AzureGuestAgentPluginTest(unittest.TestCase):
def setUp(self):
self.mock_wmi = mock.MagicMock()
self._moves_mock = mock.MagicMock()
self._winreg_mock = mock.MagicMock()
patcher = mock.patch.dict(
"sys.modules",
{
"wmi": self.mock_wmi,
"six.moves": self._moves_mock
"winreg": self._winreg_mock
}
)
patcher.start()
self.addCleanup(patcher.stop)
self._winreg_mock = self._moves_mock.winreg
self._azureguestagent = importlib.import_module(MODPATH)
self._azureagentplugin = self._azureguestagent.AzureGuestAgentPlugin()
self.snatcher = testutils.LogSnatcher(MODPATH)

2
cloudbaseinit/tests/plugins/windows/test_bootconfig.py
View File

@ -33,12 +33,10 @@ class BootConfigPluginTest(unittest.TestCase):
def setUp(self):
self.mock_wmi = mock.MagicMock()
self._moves_mock = mock.MagicMock()
patcher = mock.patch.dict(
"sys.modules",
{
"wmi": self.mock_wmi,
"six.moves": self._moves_mock,
'ctypes': mock.MagicMock(),
'ctypes.windll': mock.MagicMock(),
'ctypes.wintypes': mock.MagicMock(),

4
cloudbaseinit/tests/plugins/windows/test_rdp.py
View File

@ -32,12 +32,12 @@ class RDPPluginTest(unittest.TestCase):
def setUp(self):
self.mock_wmi = mock.MagicMock()
self._moves_mock = mock.MagicMock()
self._winreg_mock = mock.MagicMock()
patcher = mock.patch.dict(
"sys.modules",
{
"wmi": self.mock_wmi,
"six.moves": self._moves_mock
"winreg": self._winreg_mock
}
)
patcher.start()

6
cloudbaseinit/tests/plugins/windows/test_winrmcertificateauth.py
View File

@ -32,19 +32,17 @@ class ConfigWinRMCertificateAuthPluginTests(unittest.TestCase):
self._ctypes_mock = mock.MagicMock()
self._win32com_mock = mock.MagicMock()
self._pywintypes_mock = mock.MagicMock()
self._moves_mock = mock.MagicMock()
self._winreg_mock = mock.MagicMock()
self._module_patcher = mock.patch.dict(
'sys.modules',
{'ctypes': self._ctypes_mock,
'win32com': self._win32com_mock,
'pywintypes': self._pywintypes_mock,
'six.moves': self._moves_mock})
'winreg': self._winreg_mock})
self._module_patcher.start()
self._winreg_mock = self._moves_mock.winreg
self.winrmcert = importlib.import_module(
'cloudbaseinit.plugins.windows.winrmcertificateauth')
self._certif_auth = self.winrmcert.ConfigWinRMCertificateAuthPlugin()

5
cloudbaseinit/tests/plugins/windows/test_winrmlistener.py
View File

@ -33,7 +33,7 @@ class ConfigWinRMListenerPluginTests(unittest.TestCase):
self._mock_wintypes = mock.MagicMock()
self._mock_pywintypes = mock.MagicMock()
self._mock_win32 = mock.MagicMock()
self._moves_mock = mock.MagicMock()
self._winreg_mock = mock.MagicMock()
self._module_patcher = mock.patch.dict(
'sys.modules',
@ -41,10 +41,9 @@ class ConfigWinRMListenerPluginTests(unittest.TestCase):
'ctypes.wintypes': self._mock_wintypes,
'pywintypes': self._mock_pywintypes,
'win32com': self._mock_win32,
'six.moves': self._moves_mock
'winreg': self._winreg_mock
})
self._module_patcher.start()
self._winreg_mock = self._moves_mock.winreg
winrmlistener = importlib.import_module('cloudbaseinit.plugins.'
'windows.winrmlistener')

3
cloudbaseinit/tests/test_version.py
View File

@ -16,7 +16,6 @@ import importlib
import unittest
import mock
import six
from cloudbaseinit.tests import testutils
@ -51,7 +50,7 @@ class TestVersion(unittest.TestCase):
result = self.version._read_url(mock_url)
headers = {'User-Agent': self.version._PRODUCT_NAME}
mock_get.assert_called_once_with(mock_url, verify=six.PY3,
mock_get.assert_called_once_with(mock_url, verify=True,
headers=headers)
request = mock_get.return_value
request.raise_for_status.assert_called_once_with()

4
cloudbaseinit/tests/utils/test_network.py
View File

@ -29,14 +29,14 @@ CONF = cloudbaseinit_conf.CONF
class NetworkUtilsTest(unittest.TestCase):
@mock.patch('six.moves.urllib.request.urlopen')
@mock.patch('urllib.request.urlopen')
def test_check_url(self, mock_url_open):
mock_url_open.return_value = None
self.assertTrue(network.check_url("fake_url"))
@mock.patch('sys.platform', new='win32')
@mock.patch('cloudbaseinit.osutils.factory.get_os_utils')
@mock.patch('six.moves.urllib.parse.urlparse')
@mock.patch('urllib.parse.urlparse')
def _test_check_metadata_ip_route(self, mock_urlparse, mock_get_os_utils,
side_effect):
mock_utils = mock.MagicMock()

5
cloudbaseinit/tests/utils/windows/storage/test_wsm_storage_manager.py
View File

@ -31,15 +31,14 @@ class TestWSMStorageManager(unittest.TestCase):
def setUp(self):
self._mock_ctypes = mock.MagicMock()
self.mock_wmi = mock.MagicMock()
self._moves_mock = mock.MagicMock()
self._winreg_mock = self._moves_mock.winreg
self._winreg_mock = mock.MagicMock()
self._kernel32_mock = mock.MagicMock()
patcher = mock.patch.dict(
"sys.modules",
{
"wmi": self.mock_wmi,
"six.moves": self._moves_mock,
"winreg": self._winreg_mock,
"ctypes": self._mock_ctypes,
"oslo_log": mock.MagicMock()
}

18
cloudbaseinit/tests/utils/windows/test_network.py
View File

@ -27,12 +27,12 @@ class WindowsNetworkUtilsTests(unittest.TestCase):
def setUp(self):
self._ctypes_mock = mock.MagicMock()
self._moves_mock = mock.MagicMock()
self._winreg_mock = mock.MagicMock()
self._module_patcher = mock.patch.dict(
'sys.modules',
{'ctypes': self._ctypes_mock,
'six.moves': self._moves_mock})
'winreg': self._winreg_mock})
self._module_patcher.start()
@ -89,10 +89,10 @@ class WindowsNetworkUtilsTests(unittest.TestCase):
def _test_get_registry_dhcp_server(self, dhcp_server, exception=None):
fake_adapter = mock.sentinel.fake_adapter_name
self._moves_mock.winreg.QueryValueEx.return_value = [dhcp_server]
self._winreg_mock.QueryValueEx.return_value = [dhcp_server]
if exception:
self._moves_mock.winreg.QueryValueEx.side_effect = [exception]
self._winreg_mock.QueryValueEx.side_effect = [exception]
if exception.errno != 2:
self.assertRaises(cbinit_exception.CloudbaseInitException,
@ -105,14 +105,14 @@ class WindowsNetworkUtilsTests(unittest.TestCase):
else:
self.assertEqual(dhcp_server, response)
self._moves_mock.winreg.OpenKey.assert_called_once_with(
self._moves_mock.winreg.HKEY_LOCAL_MACHINE,
self._winreg_mock.OpenKey.assert_called_once_with(
self._winreg_mock.HKEY_LOCAL_MACHINE,
"SYSTEM\\CurrentControlSet\\Services\\Tcpip\\Parameters\\"
"Interfaces\\%s" % fake_adapter, 0,
self._moves_mock.winreg.KEY_READ)
self._winreg_mock.KEY_READ)
self._moves_mock.winreg.QueryValueEx.assert_called_once_with(
self._moves_mock.winreg.OpenKey.return_value.__enter__(),
self._winreg_mock.QueryValueEx.assert_called_once_with(
self._winreg_mock.OpenKey.return_value.__enter__(),
"DhcpServer")
def test_get_registry_dhcp_server(self):

5
cloudbaseinit/tests/utils/windows/test_rdp.py
View File

@ -31,12 +31,11 @@ class RdpTest(unittest.TestCase):
def setUp(self):
self._wmi_mock = mock.MagicMock()
self._moves_mock = mock.MagicMock()
self._winreg_mock = mock.MagicMock()
self._module_patcher = mock.patch.dict(
'sys.modules', {
'wmi': self._wmi_mock,
'six.moves': self._moves_mock})
self._winreg_mock = self._moves_mock.winreg
'winreg': self._winreg_mock})
self.snatcher = testutils.LogSnatcher(MODPATH)
self._module_patcher.start()
self.rdp = importlib.import_module(MODPATH)

5
cloudbaseinit/tests/utils/windows/test_security.py
View File

@ -26,14 +26,13 @@ from cloudbaseinit.tests import testutils
class WindowsSecurityUtilsTests(unittest.TestCase):
def setUp(self):
self._moves_mock = mock.MagicMock()
self._winreg_mock = mock.MagicMock()
self._module_patcher = mock.patch.dict(
'sys.modules',
{'six.moves': self._moves_mock})
{'winreg': self._winreg_mock})
self._module_patcher.start()
self._winreg_mock = self._moves_mock.winreg
self.security = importlib.import_module(
"cloudbaseinit.utils.windows.security")

4
cloudbaseinit/tests/utils/windows/test_timezone.py
View File

@ -32,7 +32,6 @@ class FakeWindowsError(Exception):
class TestTimezone(unittest.TestCase):
def setUp(self):
self._mock_moves = mock.MagicMock()
self._mock_winreg = mock.Mock()
self._mock_ctypes = mock.Mock()
self._mock_win32security = mock.Mock()
@ -42,11 +41,10 @@ class TestTimezone(unittest.TestCase):
self._module_patcher = mock.patch.dict(
'sys.modules',
{'ctypes': self._mock_ctypes,
'six.moves': self._mock_moves,
'winreg': self._mock_winreg,
'win32process': self._mock_win32process,
'win32security': self._mock_win32security})
self._module_patcher.start()
self._mock_moves.winreg = self._mock_winreg
self._timezone_module = importlib.import_module(
'cloudbaseinit.utils.windows.timezone')
self._timezone_module.WindowsError = FakeWindowsError

7
cloudbaseinit/tests/utils/windows/test_x509.py
View File

@ -13,7 +13,6 @@
# under the License.
import importlib
import six
import unittest
try:
@ -231,7 +230,7 @@ class CryptoAPICertManagerTests(unittest.TestCase):
mock_CertOpenStore.assert_called_with(
self.x509.cryptoapi.CERT_STORE_PROV_SYSTEM, 0, 0,
self.x509.cryptoapi.CERT_SYSTEM_STORE_LOCAL_MACHINE,
six.text_type(self.x509.STORE_NAME_MY))
str(self.x509.STORE_NAME_MY))
mock_get_cert_thumprint.assert_called_once_with(
mock_CertCreateSelfSignCertificate())
mock_add_system_time_interval.assert_has_calls(
@ -396,7 +395,7 @@ class CryptoAPICertManagerTests(unittest.TestCase):
mock_CertOpenStore.assert_called_with(
self.x509.cryptoapi.CERT_STORE_PROV_SYSTEM, 0, 0,
self.x509.cryptoapi.CERT_SYSTEM_STORE_LOCAL_MACHINE,
six.text_type(self.x509.STORE_NAME_MY))
str(self.x509.STORE_NAME_MY))
mock_CertAddEncodedCertificateToStore.assert_called_with(
mock_CertOpenStore(),
@ -510,7 +509,7 @@ class CryptoAPICertManagerTests(unittest.TestCase):
flags = mock.sentinel.current_user
mock_cert_open_store.assert_called_once_with(
mock.sentinel.store_prov_system, 0, 0, flags,
six.text_type(self.x509.STORE_NAME_MY))
str(self.x509.STORE_NAME_MY))
if store_handle:
mock_add_cert_store.assert_called_once_with(
mock_cert_open_store.return_value, cert_context_p,

3
cloudbaseinit/utils/debiface.py
View File

@ -16,7 +16,6 @@
import re
from oslo_log import log as oslo_logging
import six
from cloudbaseinit.models import network as network_model
@ -109,7 +108,7 @@ def _add_nic(iface, nics):
def parse(data):
"""Parse the received content and obtain network details."""
if not data or not isinstance(data, six.string_types):
if not data or not isinstance(data, str):
LOG.error("Invalid Debian config to parse:\n%s", data)
return

11
cloudbaseinit/utils/encoding.py
View File

@ -12,8 +12,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import six
from oslo_log import log as oslo_logging
@ -21,7 +19,7 @@ LOG = oslo_logging.getLogger(__name__)
def get_as_string(value):
if value is None or isinstance(value, six.text_type):
if value is None or isinstance(value, str):
return value
else:
try:
@ -33,7 +31,7 @@ def get_as_string(value):
def write_file(target_path, data, mode='wb'):
if isinstance(data, six.text_type) and 'b' in mode:
if isinstance(data, str) and 'b' in mode:
data = data.encode()
with open(target_path, mode) as f:
@ -41,7 +39,4 @@ def write_file(target_path, data, mode='wb'):
def hex_to_bytes(value):
if six.PY2:
return value.decode("hex")
else:
return bytes.fromhex(value)
return bytes.fromhex(value)

3
cloudbaseinit/utils/log.py
View File

@ -14,7 +14,6 @@
import logging
import serial
import six
from oslo_log import formatters
from oslo_log import log
@ -29,7 +28,7 @@ def _safe_write(function):
"""Avoid issues related to unicode strings handling."""
def _wrapper(message):
# Unicode strings are not properly handled by the serial module
if isinstance(message, six.text_type):
if isinstance(message, str):
function(message.encode("utf-8"))
else:
function(message)

4
cloudbaseinit/utils/network.py
View File

@ -18,10 +18,10 @@ import netaddr
import socket
import struct
import sys
from urllib import parse
from urllib import request
from oslo_log import log as oslo_logging
from six.moves.urllib import parse
from six.moves.urllib import request
from cloudbaseinit.osutils import factory as osutils_factory

5
cloudbaseinit/utils/network_team.py
View File

@ -14,11 +14,8 @@
import abc
import six
@six.add_metaclass(abc.ABCMeta)
class BaseNetworkTeamManager(object):
class BaseNetworkTeamManager(object, metaclass=abc.ABCMeta):
@abc.abstractmethod
def create_team(self, team_name, mode, load_balancing_algorithm,
members, mac_address, primary_nic_name=None,

4
cloudbaseinit/utils/template_engine/base_template.py
View File

@ -14,11 +14,9 @@
import abc
import re
import six
@six.add_metaclass(abc.ABCMeta)
class BaseTemplateEngine(object):
class BaseTemplateEngine(object, metaclass=abc.ABCMeta):
def __init__(self):
self._template_matcher = re.compile(r"##\s*template:(.*)", re.I)

4
cloudbaseinit/utils/windows/disk.py
View File

@ -20,7 +20,6 @@ from ctypes import wintypes
import random
import re
import six
import winioctlcon
from cloudbaseinit import exception
@ -146,8 +145,7 @@ class Win32_DRIVE_LAYOUT_INFORMATION_EX(ctypes.Structure):
]
@six.add_metaclass(abc.ABCMeta)
class BaseDevice(object):
class BaseDevice(object, metaclass=abc.ABCMeta):
"""Base class for devices like disks and partitions.
It has common methods for getting physical disk geometry,

2
cloudbaseinit/utils/windows/network.py
View File

@ -13,9 +13,9 @@
# under the License.
import ctypes
import winreg
from ctypes import wintypes
from six.moves import winreg
from cloudbaseinit import exception
from cloudbaseinit.utils.windows import iphlpapi

3
cloudbaseinit/utils/windows/rdp.py
View File

@ -12,8 +12,9 @@
# License for the specific language governing permissions and limitations
# under the License.
import winreg
from oslo_log import log as oslo_logging
from six.moves import winreg
from cloudbaseinit import exception
from cloudbaseinit.utils.windows import wmi_loader

2
cloudbaseinit/utils/windows/security.py
View File

@ -12,7 +12,7 @@
# License for the specific language governing permissions and limitations
# under the License.
from six.moves import winreg
import winreg
class WindowsSecurityUtils(object):

5
cloudbaseinit/utils/windows/storage/base.py
View File

@ -14,16 +14,13 @@
import abc
import six
SAN_POLICY_UNKNOWN = 0
SAN_POLICY_ONLINE = 1
SAN_POLICY_OFFLINE_SHARED = 2
SAN_POLICY_OFFLINE = 3
@six.add_metaclass(abc.ABCMeta)
class BaseStorageManager(object):
class BaseStorageManager(object, metaclass=abc.ABCMeta):
@abc.abstractmethod
def extend_volumes(self, volume_indexes=None):

2
cloudbaseinit/utils/windows/storage/wsm_storage_manager.py
View File

@ -13,9 +13,9 @@
# under the License.
import ctypes
import winreg
from oslo_log import log as oslo_logging
from six.moves import winreg
from cloudbaseinit import exception
from cloudbaseinit.utils.windows import kernel32

2
cloudbaseinit/utils/windows/timezone.py
View File

@ -16,8 +16,8 @@ import ctypes
from ctypes import wintypes
import os
import struct
import winreg
from six.moves import winreg
import win32security
from cloudbaseinit import exception

14
cloudbaseinit/utils/windows/x509.py
View File

@ -17,8 +17,6 @@ import ctypes
from ctypes import wintypes
import uuid
import six
from cloudbaseinit.utils import encoding
from cloudbaseinit.utils.windows import cryptoapi
from cloudbaseinit.utils import x509constants
@ -226,8 +224,7 @@ class CryptoAPICertManager(object):
flags = cryptoapi.CERT_SYSTEM_STORE_CURRENT_USER
store_handle = cryptoapi.CertOpenStore(
cryptoapi.CERT_STORE_PROV_SYSTEM, 0, 0, flags,
six.text_type(store_name))
cryptoapi.CERT_STORE_PROV_SYSTEM, 0, 0, flags, str(store_name))
if not store_handle:
raise cryptoapi.CryptoAPIException()
@ -297,8 +294,7 @@ class CryptoAPICertManager(object):
flags |= cryptoapi.CERT_SYSTEM_STORE_CURRENT_USER
store_handle = cryptoapi.CertOpenStore(
cryptoapi.CERT_STORE_PROV_SYSTEM, 0, 0, flags,
six.text_type(store_name))
cryptoapi.CERT_STORE_PROV_SYSTEM, 0, 0, flags, str(store_name))
if not store_handle:
raise cryptoapi.CryptoAPIException()
@ -364,8 +360,7 @@ class CryptoAPICertManager(object):
flags = cryptoapi.CERT_SYSTEM_STORE_CURRENT_USER
store_handle = cryptoapi.CertOpenStore(
cryptoapi.CERT_STORE_PROV_SYSTEM, 0, 0, flags,
six.text_type(store_name))
cryptoapi.CERT_STORE_PROV_SYSTEM, 0, 0, flags, str(store_name))
if not store_handle:
raise cryptoapi.CryptoAPIException()
@ -500,8 +495,7 @@ class CryptoAPICertManager(object):
flags = cryptoapi.CERT_SYSTEM_STORE_CURRENT_USER
store_handle = cryptoapi.CertOpenStore(
cryptoapi.CERT_STORE_PROV_SYSTEM, 0, 0, flags,
six.text_type(store_name))
cryptoapi.CERT_STORE_PROV_SYSTEM, 0, 0, flags, str(store_name))
if not store_handle:
raise cryptoapi.CryptoAPIException()

5
cloudbaseinit/version.py
View File

@ -18,7 +18,6 @@ import threading
from oslo_log import log as oslo_logging
import pbr.version
import requests
import six
_UPDATE_CHECK_URL = 'https://www.cloudbase.it/checkupdates.php?p={0}&v={1}'
@ -27,9 +26,7 @@ LOG = oslo_logging.getLogger(__name__)
def _read_url(url):
# Disable certificate verification on Python 2 as
# requests' CA list is incomplete. Works fine on Python3.
req = requests.get(url, verify=six.PY3,
req = requests.get(url, verify=True,
headers={'User-Agent': _PRODUCT_NAME})
req.raise_for_status()
if req.text:

1
requirements.txt
View File

@ -4,7 +4,6 @@ netaddr>=0.7.6
pyserial
oslo.config
oslo.log
six>=1.7.0
Babel>=1.3
oauthlib
netifaces