Use 'importlib' module in lieu of deprecated 'imp' module (#6832)
Migrate from using the `imp` module to using the `importlib` module. As of Python 3, the `imp` module has been deprecated in favor of the `importlib` module.
This commit is contained in:
parent
e9b27f9450
commit
72c420320f
@ -72,7 +72,7 @@ class ProcDockerStats(daemon_base.DaemonBase):
|
|||||||
UNITS_MiB = 'MiB'
|
UNITS_MiB = 'MiB'
|
||||||
UNITS_GiB = 'GiB'
|
UNITS_GiB = 'GiB'
|
||||||
|
|
||||||
res = re.match('(\d+\.?\d*)([a-zA-Z]+)', value)
|
res = re.match(r'(\d+\.?\d*)([a-zA-Z]+)', value)
|
||||||
value = float(res.groups()[0])
|
value = float(res.groups()[0])
|
||||||
units = res.groups()[1]
|
units = res.groups()[1]
|
||||||
if units.lower() == UNITS_KB.lower():
|
if units.lower() == UNITS_KB.lower():
|
||||||
|
@ -1,4 +1,4 @@
|
|||||||
import imp
|
import importlib
|
||||||
import sys
|
import sys
|
||||||
import os
|
import os
|
||||||
import pytest
|
import pytest
|
||||||
@ -28,6 +28,15 @@ modules_path = os.path.dirname(test_path)
|
|||||||
scripts_path = os.path.join(modules_path, "scripts")
|
scripts_path = os.path.join(modules_path, "scripts")
|
||||||
sys.path.insert(0, modules_path)
|
sys.path.insert(0, modules_path)
|
||||||
|
|
||||||
|
# Load the file under test
|
||||||
|
determine_reboot_cause_path = os.path.join(scripts_path, 'determine-reboot-cause')
|
||||||
|
loader = importlib.machinery.SourceFileLoader('determine_reboot_cause', determine_reboot_cause_path)
|
||||||
|
spec = importlib.util.spec_from_loader(loader.name, loader)
|
||||||
|
determine_reboot_cause = importlib.util.module_from_spec(spec)
|
||||||
|
loader.exec_module(determine_reboot_cause)
|
||||||
|
sys.modules['determine_reboot_cause'] = determine_reboot_cause
|
||||||
|
|
||||||
|
|
||||||
PROC_CMDLINE_CONTENTS = """\
|
PROC_CMDLINE_CONTENTS = """\
|
||||||
BOOT_IMAGE=/image-20191130.52/boot/vmlinuz-4.9.0-11-2-amd64 root=/dev/sda4 rw console=tty0 console=ttyS1,9600n8 quiet net.ifnames=0 biosdevname=0 loop=image-20191130.52/fs.squashfs loopfstype=squashfs apparmor=1 security=apparmor varlog_size=4096 usbcore.autosuspend=-1 module_blacklist=gpio_ich SONIC_BOOT_TYPE=warm"""
|
BOOT_IMAGE=/image-20191130.52/boot/vmlinuz-4.9.0-11-2-amd64 root=/dev/sda4 rw console=tty0 console=ttyS1,9600n8 quiet net.ifnames=0 biosdevname=0 loop=image-20191130.52/fs.squashfs loopfstype=squashfs apparmor=1 security=apparmor varlog_size=4096 usbcore.autosuspend=-1 module_blacklist=gpio_ich SONIC_BOOT_TYPE=warm"""
|
||||||
|
|
||||||
@ -55,20 +64,14 @@ EXPECTED_HARDWARE_REBOOT_CAUSE = {"warm-reboot", ""}
|
|||||||
EXPECTED_WATCHDOG_REBOOT_CAUSE_DICT = {'comment': '', 'gen_time': '2020_10_22_03_15_08', 'cause': 'Watchdog', 'user': 'N/A', 'time': 'N/A'}
|
EXPECTED_WATCHDOG_REBOOT_CAUSE_DICT = {'comment': '', 'gen_time': '2020_10_22_03_15_08', 'cause': 'Watchdog', 'user': 'N/A', 'time': 'N/A'}
|
||||||
EXPECTED_USER_REBOOT_CAUSE_DICT = {'comment': '', 'gen_time': '2020_10_22_03_14_07', 'cause': 'reboot', 'user': 'admin', 'time': 'Thu Oct 22 03:11:08 UTC 2020'}
|
EXPECTED_USER_REBOOT_CAUSE_DICT = {'comment': '', 'gen_time': '2020_10_22_03_14_07', 'cause': 'reboot', 'user': 'admin', 'time': 'Thu Oct 22 03:11:08 UTC 2020'}
|
||||||
|
|
||||||
imp.load_source('determine_reboot_cause', scripts_path + '/determine-reboot-cause')
|
|
||||||
from determine_reboot_cause import *
|
|
||||||
|
|
||||||
class TestDetermineRebootCause(object):
|
class TestDetermineRebootCause(object):
|
||||||
@classmethod
|
|
||||||
def setup_class(cls):
|
|
||||||
print("SETUP")
|
|
||||||
|
|
||||||
def test_parse_warmfast_reboot_from_proc_cmdline(self):
|
def test_parse_warmfast_reboot_from_proc_cmdline(self):
|
||||||
with mock.patch("os.path.isfile") as mock_isfile:
|
with mock.patch("os.path.isfile") as mock_isfile:
|
||||||
mock_isfile.return_value = True
|
mock_isfile.return_value = True
|
||||||
open_mocked = mock.mock_open(read_data=PROC_CMDLINE_CONTENTS)
|
open_mocked = mock.mock_open(read_data=PROC_CMDLINE_CONTENTS)
|
||||||
with mock.patch("{}.open".format(BUILTINS), open_mocked):
|
with mock.patch("{}.open".format(BUILTINS), open_mocked):
|
||||||
result = parse_warmfast_reboot_from_proc_cmdline()
|
result = determine_reboot_cause.parse_warmfast_reboot_from_proc_cmdline()
|
||||||
assert result == EXPECTED_PARSE_WARMFAST_REBOOT_FROM_PROC_CMDLINE
|
assert result == EXPECTED_PARSE_WARMFAST_REBOOT_FROM_PROC_CMDLINE
|
||||||
open_mocked.assert_called_once_with("/proc/cmdline")
|
open_mocked.assert_called_once_with("/proc/cmdline")
|
||||||
|
|
||||||
@ -77,41 +80,36 @@ class TestDetermineRebootCause(object):
|
|||||||
mock_isfile.return_value = True
|
mock_isfile.return_value = True
|
||||||
open_mocked = mock.mock_open(read_data=REBOOT_CAUSE_CONTENTS)
|
open_mocked = mock.mock_open(read_data=REBOOT_CAUSE_CONTENTS)
|
||||||
with mock.patch("{}.open".format(BUILTINS), open_mocked):
|
with mock.patch("{}.open".format(BUILTINS), open_mocked):
|
||||||
result = find_software_reboot_cause_from_reboot_cause_file()
|
result = determine_reboot_cause.find_software_reboot_cause_from_reboot_cause_file()
|
||||||
assert result == EXPECTED_FIND_SOFTWARE_REBOOT_CAUSE_USER
|
assert result == EXPECTED_FIND_SOFTWARE_REBOOT_CAUSE_USER
|
||||||
open_mocked.assert_called_once_with("/host/reboot-cause/reboot-cause.txt")
|
open_mocked.assert_called_once_with("/host/reboot-cause/reboot-cause.txt")
|
||||||
|
|
||||||
def test_find_software_reboot_cause_first_boot(self):
|
def test_find_software_reboot_cause_first_boot(self):
|
||||||
with mock.patch("sonic_py_common.device_info.get_sonic_version_info", return_value=GET_SONIC_VERSION_INFO):
|
with mock.patch("sonic_py_common.device_info.get_sonic_version_info", return_value=GET_SONIC_VERSION_INFO):
|
||||||
result = find_first_boot_version()
|
result = determine_reboot_cause.find_first_boot_version()
|
||||||
assert result == EXPECTED_FIND_FIRSTBOOT_VERSION
|
assert result == EXPECTED_FIND_FIRSTBOOT_VERSION
|
||||||
|
|
||||||
def test_find_software_reboot_cause(self):
|
def test_find_software_reboot_cause(self):
|
||||||
with mock.patch("determine_reboot_cause.find_software_reboot_cause_from_reboot_cause_file", return_value="Unknown"):
|
with mock.patch("determine_reboot_cause.find_software_reboot_cause_from_reboot_cause_file", return_value="Unknown"):
|
||||||
with mock.patch("os.path.isfile") as mock_isfile:
|
with mock.patch("os.path.isfile") as mock_isfile:
|
||||||
mock_isfile.return_value = False
|
mock_isfile.return_value = False
|
||||||
result = find_software_reboot_cause()
|
result = determine_reboot_cause.find_software_reboot_cause()
|
||||||
assert result == "Unknown"
|
assert result == "Unknown"
|
||||||
|
|
||||||
def test_find_proc_cmdline_reboot_cause(self):
|
def test_find_proc_cmdline_reboot_cause(self):
|
||||||
with mock.patch("determine_reboot_cause.parse_warmfast_reboot_from_proc_cmdline", return_value="fast-reboot"):
|
with mock.patch("determine_reboot_cause.parse_warmfast_reboot_from_proc_cmdline", return_value="fast-reboot"):
|
||||||
result = find_proc_cmdline_reboot_cause()
|
result = determine_reboot_cause.find_proc_cmdline_reboot_cause()
|
||||||
assert result == "fast-reboot"
|
assert result == "fast-reboot"
|
||||||
|
|
||||||
def test_find_hardware_reboot_cause(self):
|
def test_find_hardware_reboot_cause(self):
|
||||||
with mock.patch("determine_reboot_cause.get_reboot_cause_from_platform", return_value=("Powerloss", None)):
|
with mock.patch("determine_reboot_cause.get_reboot_cause_from_platform", return_value=("Powerloss", None)):
|
||||||
result = find_hardware_reboot_cause()
|
result = determine_reboot_cause.find_hardware_reboot_cause()
|
||||||
assert result == "Powerloss (None)"
|
assert result == "Powerloss (None)"
|
||||||
|
|
||||||
def test_get_reboot_cause_dict_watchdog(self):
|
def test_get_reboot_cause_dict_watchdog(self):
|
||||||
reboot_cause_dict = get_reboot_cause_dict(REBOOT_CAUSE_WATCHDOG, "", GEN_TIME_WATCHDOG)
|
reboot_cause_dict = determine_reboot_cause.get_reboot_cause_dict(REBOOT_CAUSE_WATCHDOG, "", GEN_TIME_WATCHDOG)
|
||||||
assert reboot_cause_dict == EXPECTED_WATCHDOG_REBOOT_CAUSE_DICT
|
assert reboot_cause_dict == EXPECTED_WATCHDOG_REBOOT_CAUSE_DICT
|
||||||
|
|
||||||
def test_get_reboot_cause_dict_user(self):
|
def test_get_reboot_cause_dict_user(self):
|
||||||
reboot_cause_dict = get_reboot_cause_dict(REBOOT_CAUSE_USER, "", GEN_TIME_USER)
|
reboot_cause_dict = determine_reboot_cause.get_reboot_cause_dict(REBOOT_CAUSE_USER, "", GEN_TIME_USER)
|
||||||
assert reboot_cause_dict == EXPECTED_USER_REBOOT_CAUSE_DICT
|
assert reboot_cause_dict == EXPECTED_USER_REBOOT_CAUSE_DICT
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def teardown_class(cls):
|
|
||||||
print("TEARDOWN")
|
|
||||||
|
|
||||||
|
@ -1,4 +1,4 @@
|
|||||||
import imp
|
import importlib
|
||||||
import sys
|
import sys
|
||||||
import os
|
import os
|
||||||
import pytest
|
import pytest
|
||||||
@ -14,8 +14,13 @@ modules_path = os.path.dirname(test_path)
|
|||||||
scripts_path = os.path.join(modules_path, "scripts")
|
scripts_path = os.path.join(modules_path, "scripts")
|
||||||
sys.path.insert(0, modules_path)
|
sys.path.insert(0, modules_path)
|
||||||
|
|
||||||
imp.load_source('procdockerstatsd', scripts_path + '/procdockerstatsd')
|
# Load the file under test
|
||||||
from procdockerstatsd import *
|
procdockerstatsd_path = os.path.join(scripts_path, 'procdockerstatsd')
|
||||||
|
loader = importlib.machinery.SourceFileLoader('procdockerstatsd', procdockerstatsd_path)
|
||||||
|
spec = importlib.util.spec_from_loader(loader.name, loader)
|
||||||
|
procdockerstatsd = importlib.util.module_from_spec(spec)
|
||||||
|
loader.exec_module(procdockerstatsd)
|
||||||
|
sys.modules['procdockerstatsd'] = procdockerstatsd
|
||||||
|
|
||||||
class TestProcDockerStatsDaemon(object):
|
class TestProcDockerStatsDaemon(object):
|
||||||
def test_convert_to_bytes(self):
|
def test_convert_to_bytes(self):
|
||||||
@ -35,7 +40,7 @@ class TestProcDockerStatsDaemon(object):
|
|||||||
('7.751GiB', 8322572878)
|
('7.751GiB', 8322572878)
|
||||||
]
|
]
|
||||||
|
|
||||||
pdstatsd = ProcDockerStats(SYSLOG_IDENTIFIER)
|
pdstatsd = procdockerstatsd.ProcDockerStats(procdockerstatsd.SYSLOG_IDENTIFIER)
|
||||||
|
|
||||||
for test_input, expected_output in test_data:
|
for test_input, expected_output in test_data:
|
||||||
res = pdstatsd.convert_to_bytes(test_input)
|
res = pdstatsd.convert_to_bytes(test_input)
|
||||||
|
@ -1,4 +1,4 @@
|
|||||||
import imp
|
import importlib
|
||||||
import signal
|
import signal
|
||||||
import sys
|
import sys
|
||||||
|
|
||||||
@ -25,10 +25,21 @@ def db_connect(db_name, namespace=EMPTY_NAMESPACE):
|
|||||||
from swsscommon import swsscommon
|
from swsscommon import swsscommon
|
||||||
return swsscommon.DBConnector(db_name, REDIS_TIMEOUT_MSECS, True, namespace)
|
return swsscommon.DBConnector(db_name, REDIS_TIMEOUT_MSECS, True, namespace)
|
||||||
|
|
||||||
|
|
||||||
|
def _load_module_from_file(module_name, file_path):
|
||||||
|
loader = importlib.machinery.SourceFileLoader(module_name, file_path)
|
||||||
|
spec = importlib.util.spec_from_loader(loader.name, loader)
|
||||||
|
module = importlib.util.module_from_spec(spec)
|
||||||
|
loader.exec_module(module)
|
||||||
|
sys.modules[module_name] = module
|
||||||
|
return module
|
||||||
|
|
||||||
|
|
||||||
#
|
#
|
||||||
# DaemonBase ===================================================================
|
# DaemonBase ===================================================================
|
||||||
#
|
#
|
||||||
|
|
||||||
|
|
||||||
class DaemonBase(Logger):
|
class DaemonBase(Logger):
|
||||||
def __init__(self, log_identifier):
|
def __init__(self, log_identifier):
|
||||||
super(DaemonBase, self).__init__(
|
super(DaemonBase, self).__init__(
|
||||||
@ -68,7 +79,7 @@ class DaemonBase(Logger):
|
|||||||
|
|
||||||
try:
|
try:
|
||||||
module_file = "/".join([platform_path, "plugins", module_name + ".py"])
|
module_file = "/".join([platform_path, "plugins", module_name + ".py"])
|
||||||
module = imp.load_source(module_name, module_file)
|
module = _load_module_from_file(module_name, module_file)
|
||||||
except IOError as e:
|
except IOError as e:
|
||||||
raise IOError("Failed to load platform module '%s': %s" % (module_name, str(e)))
|
raise IOError("Failed to load platform module '%s': %s" % (module_name, str(e)))
|
||||||
|
|
||||||
|
Reference in New Issue
Block a user