Merge pull request #116 from TomasKorbar/changes
Add tests and fix some errors
This commit is contained in:
Jaroslav Škarvada
GitHub
commit
8910d0daba
23 changed files with 926 additions and 251 deletions
24
.travis.yml
Normal file
24
.travis.yml
Normal file
|
|
@ -0,0 +1,24 @@
|
|||
language: generic
|
||||
sudo: required
|
||||
services:
|
||||
- docker
|
||||
matrix:
|
||||
include:
|
||||
- name: "fedora 28 python 2 test suite"
|
||||
env:
|
||||
- python=2
|
||||
- os=fedora:28
|
||||
- package_manager=dnf
|
||||
- name: "fedora 28 python 3 test suite"
|
||||
env:
|
||||
- python=3
|
||||
- os=fedora:28
|
||||
- package_manager=dnf
|
||||
- name: "centos 7 test suite"
|
||||
env:
|
||||
- python=2
|
||||
- os=centos:7
|
||||
- package_manager=yum
|
||||
script:
|
||||
- docker build --build-arg PYTHON=${python} --build-arg OS=${os} --build-arg PACKAGE_MANAGER=${package_manager} -t tuned_image -f tests/Dockerfile .
|
||||
- docker run tuned_image /bin/sh -c "make test PYTHON=/usr/bin/python${python}"
|
||||
27
tests/Dockerfile
Normal file
27
tests/Dockerfile
Normal file
|
|
@ -0,0 +1,27 @@
|
|||
ARG OS
|
||||
|
||||
FROM $OS
|
||||
|
||||
WORKDIR /test_dir
|
||||
|
||||
ADD ./ /test_dir
|
||||
|
||||
ARG PACKAGE_MANAGER=dnf
|
||||
|
||||
ENV PKM=$PACKAGE_MANAGER
|
||||
|
||||
ARG OS
|
||||
RUN if [[ $OS == "centos:7" ]]; then yum install -y epel-release; fi;
|
||||
|
||||
RUN ${PKM} install -y virt-what ethtool gawk hdparm util-linux dbus polkit make
|
||||
ARG PYTHON
|
||||
RUN ${PKM} install -y python$PYTHON-flexmock
|
||||
|
||||
ARG OS
|
||||
ARG PYTHON
|
||||
RUN if [[ $OS == "centos:7" ]]; then export py=python; \
|
||||
else export py=python$PYTHON; fi; \
|
||||
${PKM} install -y ${py}-dbus \
|
||||
${py}-decorator ${py}-pyudev ${py}-configobj ${py}-schedutils \
|
||||
${py}-linux-procfs ${py}-perf ${py}-unittest2 ${py}-gobject-base;
|
||||
|
||||
0
tests/exports/__init__.py
Normal file
0
tests/exports/__init__.py
Normal file
88
tests/exports/test_controller.py
Normal file
88
tests/exports/test_controller.py
Normal file
|
|
@ -0,0 +1,88 @@
|
|||
import unittest2
|
||||
import flexmock
|
||||
|
||||
from tuned.exports.controller import ExportsController
|
||||
import tuned.exports as exports
|
||||
|
||||
class ControllerTestCase(unittest2.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls._controller = ExportsController()
|
||||
|
||||
def test_is_exportable_method(self):
|
||||
self.assertFalse(self._controller._is_exportable_method( \
|
||||
MockClass().NonExportableObject))
|
||||
|
||||
self.assertTrue(self._controller._is_exportable_method( \
|
||||
MockClass().ExportableMethod))
|
||||
|
||||
def test_is_exportable_signal(self):
|
||||
self.assertFalse(self._controller._is_exportable_signal( \
|
||||
MockClass().NonExportableObject))
|
||||
|
||||
self.assertTrue(self._controller._is_exportable_signal( \
|
||||
MockClass().ExportableSignal))
|
||||
|
||||
def test_initialize_exports(self):
|
||||
local_controller = ExportsController()
|
||||
exporter = MockExporter()
|
||||
instance = MockClass()
|
||||
local_controller.register_exporter(exporter)
|
||||
local_controller.register_object(instance)
|
||||
local_controller._initialize_exports()
|
||||
self.assertEqual(exporter.exported_methods[0].method,\
|
||||
instance.ExportableMethod)
|
||||
self.assertEqual(exporter.exported_methods[0].args[0],\
|
||||
"method_param1")
|
||||
self.assertEqual(exporter.exported_methods[0].kwargs['kword'],\
|
||||
"method_param2")
|
||||
self.assertEqual(exporter.exported_signals[0].method,\
|
||||
instance.ExportableSignal)
|
||||
self.assertEqual(exporter.exported_signals[0].args[0],\
|
||||
"signal_param1")
|
||||
self.assertEqual(exporter.exported_signals[0].kwargs['kword'],\
|
||||
"signal_param2")
|
||||
|
||||
def test_start_stop(self):
|
||||
local_controller = ExportsController()
|
||||
exporter = MockExporter()
|
||||
local_controller.register_exporter(exporter)
|
||||
local_controller.start()
|
||||
self.assertTrue(exporter.is_running)
|
||||
local_controller.stop()
|
||||
self.assertFalse(exporter.is_running)
|
||||
|
||||
|
||||
class MockExporter(object):
|
||||
def __init__(self):
|
||||
self.exported_methods = []
|
||||
self.exported_signals = []
|
||||
self.is_running = False
|
||||
|
||||
def export(self,method,*args,**kwargs):
|
||||
object_to_export = flexmock.flexmock(\
|
||||
method = method, args = args, kwargs = kwargs)
|
||||
self.exported_methods.append(object_to_export)
|
||||
|
||||
def signal(self,method,*args,**kwargs):
|
||||
object_to_export = flexmock.flexmock(\
|
||||
method = method, args = args, kwargs = kwargs)
|
||||
self.exported_signals.append(object_to_export)
|
||||
|
||||
def start(self):
|
||||
self.is_running = True
|
||||
|
||||
def stop(self):
|
||||
self.is_running = False
|
||||
|
||||
class MockClass(object):
|
||||
@exports.export('method_param1', kword = 'method_param2')
|
||||
def ExportableMethod(self):
|
||||
return True
|
||||
|
||||
def NonExportableObject(self):
|
||||
pass
|
||||
|
||||
@exports.signal('signal_param1', kword = 'signal_param2')
|
||||
def ExportableSignal(self):
|
||||
return True
|
||||
|
|
@ -1,7 +1,7 @@
|
|||
import unittest
|
||||
import unittest2
|
||||
from tuned.hardware.device_matcher import DeviceMatcher
|
||||
|
||||
class DeviceMatcherTestCase(unittest.TestCase):
|
||||
class DeviceMatcherTestCase(unittest2.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.matcher = DeviceMatcher()
|
||||
|
|
|
|||
36
tests/hardware/test_device_matcher_udev.py
Normal file
36
tests/hardware/test_device_matcher_udev.py
Normal file
|
|
@ -0,0 +1,36 @@
|
|||
import unittest2
|
||||
import pyudev
|
||||
|
||||
from tuned.hardware.device_matcher_udev import DeviceMatcherUdev
|
||||
|
||||
class DeviceMatcherUdevTestCase(unittest2.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.udev_context = pyudev.Context()
|
||||
cls.matcher = DeviceMatcherUdev()
|
||||
|
||||
def test_simple_search(self):
|
||||
try:
|
||||
device = pyudev.Devices.from_sys_path(self.udev_context,
|
||||
"/sys/devices/virtual/tty/tty0")
|
||||
except AttributeError:
|
||||
device = pyudev.Device.from_sys_path(self.udev_context,
|
||||
"/sys/devices/virtual/tty/tty0")
|
||||
self.assertTrue(self.matcher.match("tty0", device))
|
||||
try:
|
||||
device = pyudev.Devices.from_sys_path(self.udev_context,
|
||||
"/sys/devices/virtual/tty/tty1")
|
||||
except AttributeError:
|
||||
device = pyudev.Device.from_sys_path(self.udev_context,
|
||||
"/sys/devices/virtual/tty/tty1")
|
||||
self.assertFalse(self.matcher.match("tty0", device))
|
||||
|
||||
def test_regex_search(self):
|
||||
try:
|
||||
device = pyudev.Devices.from_sys_path(self.udev_context,
|
||||
"/sys/devices/virtual/tty/tty0")
|
||||
except AttributeError:
|
||||
device = pyudev.Device.from_sys_path(self.udev_context,
|
||||
"/sys/devices/virtual/tty/tty0")
|
||||
self.assertTrue(self.matcher.match("tty.", device))
|
||||
self.assertFalse(self.matcher.match("tty[1-9]", device))
|
||||
60
tests/hardware/test_inventory.py
Normal file
60
tests/hardware/test_inventory.py
Normal file
|
|
@ -0,0 +1,60 @@
|
|||
import unittest2
|
||||
from flexmock import flexmock
|
||||
import pyudev
|
||||
|
||||
from tuned.hardware.inventory import Inventory
|
||||
|
||||
subsystem_name = "test subsystem"
|
||||
|
||||
class InventoryTestCase(unittest2.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls._context = pyudev.Context()
|
||||
cls._inventory = Inventory(set_receive_buffer_size=False)
|
||||
cls._dummy = DummyPlugin()
|
||||
cls._dummier = DummyPlugin()
|
||||
|
||||
def test_get_device(self):
|
||||
try:
|
||||
device1 = pyudev.Devices.from_name(self._context, "tty", "tty0")
|
||||
except AttributeError:
|
||||
device1 = pyudev.Device.from_name(self._context, "tty", "tty0")
|
||||
device2 = self._inventory.get_device("tty", "tty0")
|
||||
self.assertEqual(device1,device2)
|
||||
|
||||
def test_get_devices(self):
|
||||
device_list1 = self._context.list_devices(subsystem = "tty")
|
||||
device_list2 = self._inventory.get_devices("tty")
|
||||
self.assertItemsEqual(device_list1,device_list2)
|
||||
|
||||
def test_subscribe(self):
|
||||
self._inventory.subscribe(self._dummy,subsystem_name,
|
||||
self._dummy.TestCallback)
|
||||
self._inventory.subscribe(self._dummier,subsystem_name,
|
||||
self._dummier.TestCallback)
|
||||
device = flexmock(subsystem = subsystem_name)
|
||||
self._inventory._handle_udev_event("test event", device)
|
||||
self.assertTrue(self._dummy.CallbackWasCalled)
|
||||
self.assertTrue(self._dummier.CallbackWasCalled)
|
||||
|
||||
def test_unsubscribe(self):
|
||||
self._dummy.CallbackWasCalled = False
|
||||
self._dummier.CallbackWasCalled = False
|
||||
self._inventory.unsubscribe(self._dummy)
|
||||
device = flexmock(subsystem = subsystem_name)
|
||||
self._inventory._handle_udev_event("test event", device)
|
||||
self.assertFalse(self._dummy.CallbackWasCalled)
|
||||
self.assertTrue(self._dummier.CallbackWasCalled)
|
||||
self._dummier.CallbackWasCalled = False
|
||||
self._inventory.unsubscribe(self._dummier)
|
||||
self._inventory._handle_udev_event("test event", device)
|
||||
self.assertFalse(self._dummy.CallbackWasCalled)
|
||||
self.assertFalse(self._dummier.CallbackWasCalled)
|
||||
self.assertIsNone(self._inventory._monitor_observer)
|
||||
|
||||
class DummyPlugin():
|
||||
def __init__(self):
|
||||
self.CallbackWasCalled = False
|
||||
|
||||
def TestCallback(self, event, device):
|
||||
self.CallbackWasCalled = True
|
||||
|
|
@ -1,4 +1,4 @@
|
|||
import unittest
|
||||
import unittest2
|
||||
import tests.globals
|
||||
import tuned.monitors.base
|
||||
|
||||
|
|
@ -13,7 +13,7 @@ class MockMonitor(tuned.monitors.base.Monitor):
|
|||
cls._load.setdefault(device, 0)
|
||||
cls._load[device] += 1
|
||||
|
||||
class MonitorBaseClassTestCase(unittest.TestCase):
|
||||
class MonitorBaseClassTestCase(unittest2.TestCase):
|
||||
def test_fail_base_class_init(self):
|
||||
with self.assertRaises(NotImplementedError):
|
||||
tuned.monitors.base.Monitor()
|
||||
|
|
|
|||
|
|
@ -1,122 +1,243 @@
|
|||
import unittest
|
||||
import tests.globals
|
||||
from flexmock import flexmock
|
||||
from tuned.plugins.base import Plugin as PluginBase
|
||||
import tuned.plugins.decorators
|
||||
from collections import Mapping
|
||||
import tempfile
|
||||
import unittest2
|
||||
import flexmock
|
||||
|
||||
from tuned.monitors.repository import Repository
|
||||
import tuned.plugins.decorators as decorators
|
||||
from tuned.plugins.base import Plugin
|
||||
import tuned.hardware as hardware
|
||||
import tuned.monitors as monitors
|
||||
import tuned.profiles as profiles
|
||||
import tuned.plugins as plugins
|
||||
import tuned.consts as consts
|
||||
from tuned import storage
|
||||
import tuned.plugins.base
|
||||
|
||||
tuned.plugins.base.log = flexmock.flexmock(info = lambda *args: None,\
|
||||
error = lambda *args: None,debug = lambda *args: None,\
|
||||
warn = lambda *args: None)
|
||||
|
||||
temp_storage_file = tempfile.TemporaryFile(mode = 'r')
|
||||
consts.DEFAULT_STORAGE_FILE = temp_storage_file.name
|
||||
monitors_repository = monitors.Repository()
|
||||
hardware_inventory = hardware.Inventory(set_receive_buffer_size=False)
|
||||
device_matcher = hardware.DeviceMatcher()
|
||||
device_matcher_udev = hardware.DeviceMatcherUdev()
|
||||
plugin_instance_factory = plugins.instance.Factory()
|
||||
storage_provider = storage.PickleProvider()
|
||||
storage_factory = storage.Factory(storage_provider)
|
||||
|
||||
class PluginBaseTestCase(unittest2.TestCase):
|
||||
def setUp(self):
|
||||
self._plugin = DummyPlugin(monitors_repository,storage_factory,\
|
||||
hardware_inventory,device_matcher,device_matcher_udev,\
|
||||
plugin_instance_factory,None,None)
|
||||
|
||||
self._commands_plugin = CommandsPlugin(monitors_repository,\
|
||||
storage_factory,hardware_inventory,device_matcher,\
|
||||
device_matcher_udev,plugin_instance_factory,None,\
|
||||
profiles.variables.Variables())
|
||||
|
||||
def test_get_effective_options(self):
|
||||
self.assertEqual(self._plugin._get_effective_options(\
|
||||
{'default_option1':'default_value2'}),\
|
||||
{'default_option1': 'default_value2',\
|
||||
'default_option2': 'default_value2'})
|
||||
|
||||
def test_option_bool(self):
|
||||
self.assertTrue(self._plugin._option_bool(True))
|
||||
self.assertTrue(self._plugin._option_bool('true'))
|
||||
self.assertFalse(self._plugin._option_bool('false'))
|
||||
|
||||
def test_create_instance(self):
|
||||
instance = self._plugin.create_instance(\
|
||||
'first_instance','test','test','test','test',\
|
||||
{'default_option1':'default_value2'})
|
||||
self.assertIsNotNone(instance)
|
||||
|
||||
def test_destroy_instance(self):
|
||||
instance = self._plugin.create_instance(\
|
||||
'first_instance','test','test','test','test',\
|
||||
{'default_option1':'default_value2'})
|
||||
|
||||
self._plugin.destroy_instance(instance)
|
||||
self.assertIn(instance,self._plugin.cleaned_instances)
|
||||
|
||||
def test_get_matching_devices(self):
|
||||
""" without udev regex """
|
||||
instance = self._plugin.create_instance(\
|
||||
'first_instance','right_device*',None,'test','test',\
|
||||
{'default_option1':'default_value2'})
|
||||
|
||||
self.assertEqual(self._plugin._get_matching_devices(\
|
||||
instance,['bad_device','right_device1','right_device2']),\
|
||||
set(['right_device1','right_device2']))
|
||||
|
||||
""" with udev regex """
|
||||
instance = self._plugin.create_instance(\
|
||||
'second_instance','right_device*','device[1-2]','test','test',\
|
||||
{'default_option1':'default_value2'})
|
||||
|
||||
device1 = DummyDevice('device1',{'name':'device1'})
|
||||
device2 = DummyDevice('device2',{'name':'device2'})
|
||||
device3 = DummyDevice('device3',{'name':'device3'})
|
||||
|
||||
self.assertEqual(self._plugin._get_matching_devices(\
|
||||
instance,[device1,device2,device3]),set(['device1','device2']))
|
||||
|
||||
def test_autoregister_commands(self):
|
||||
self._commands_plugin._autoregister_commands()
|
||||
self.assertEqual(self._commands_plugin._commands['size']['set'],\
|
||||
self._commands_plugin._set_size)
|
||||
self.assertEqual(self._commands_plugin._commands['size']['get'],\
|
||||
self._commands_plugin._get_size)
|
||||
self.assertEqual(\
|
||||
self._commands_plugin._commands['custom_name']['custom'],
|
||||
self._commands_plugin.the_most_custom_command)
|
||||
|
||||
def test_check_commands(self):
|
||||
self._commands_plugin._check_commands()
|
||||
|
||||
with self.assertRaises(TypeError):
|
||||
bad_plugin = BadCommandsPlugin(monitors_repository,storage_factory,\
|
||||
hardware_inventory,device_matcher,device_matcher_udev,\
|
||||
plugin_instance_factory,None,None)
|
||||
|
||||
def test_execute_all_non_device_commands(self):
|
||||
instance = self._commands_plugin.create_instance('test_instance','',\
|
||||
'','','',{'size':'XXL'})
|
||||
|
||||
self._commands_plugin._execute_all_non_device_commands(instance)
|
||||
|
||||
self.assertEqual(self._commands_plugin._size,'XXL')
|
||||
|
||||
def test_execute_all_device_commands(self):
|
||||
instance = self._commands_plugin.create_instance('test_instance','',\
|
||||
'','','',{'device_setting':'010'})
|
||||
|
||||
device1 = DummyDevice('device1',{})
|
||||
device2 = DummyDevice('device2',{})
|
||||
|
||||
self._commands_plugin._execute_all_device_commands(instance,\
|
||||
[device1,device2])
|
||||
|
||||
self.assertEqual(device1.setting,'010')
|
||||
self.assertEqual(device2.setting,'010')
|
||||
|
||||
def test_process_assignment_modifiers(self):
|
||||
self.assertEqual(self._plugin._process_assignment_modifiers('100',None)\
|
||||
,'100')
|
||||
self.assertEqual(self._plugin._process_assignment_modifiers(\
|
||||
'>100','200'),None)
|
||||
self.assertEqual(self._plugin._process_assignment_modifiers(\
|
||||
'<100','200'),'100')
|
||||
|
||||
def test_get_current_value(self):
|
||||
instance = self._commands_plugin.create_instance('test_instance','',\
|
||||
'','','',{})
|
||||
|
||||
command = [com for com in self._commands_plugin._commands.values()\
|
||||
if com['name'] == 'size'][0]
|
||||
|
||||
self.assertEqual(self._commands_plugin._get_current_value(command),'S')
|
||||
|
||||
def test_norm_value(self):
|
||||
self.assertEqual(self._plugin._norm_value('"000000021"'),'21')
|
||||
|
||||
def test_verify_value(self):
|
||||
self.assertEqual(self._plugin._verify_value(\
|
||||
'test_value','1',None,True),True)
|
||||
|
||||
self.assertEqual(self._plugin._verify_value(\
|
||||
'test_value','1',None,False),False)
|
||||
|
||||
self.assertEqual(self._plugin._verify_value(\
|
||||
'test_value','00001','001',False),True)
|
||||
|
||||
self.assertEqual(self._plugin._verify_value(\
|
||||
'test_value','0x1a','0x1a',False),True)
|
||||
|
||||
self.assertEqual(self._plugin._verify_value(\
|
||||
'test_value','0x1a','0x1b',False),False)
|
||||
|
||||
class MockPlugin(PluginBase):
|
||||
@classmethod
|
||||
def _get_default_options(cls):
|
||||
return { 'color': 'blue', 'size': 'XXL' }
|
||||
def tearDownClass(cls):
|
||||
temp_storage_file.close()
|
||||
|
||||
class InvalidCommandPlugin(MockPlugin):
|
||||
@tuned.plugins.decorators.command_set('color')
|
||||
def _set_color(self, new_color):
|
||||
pass
|
||||
class DummyPlugin(Plugin):
|
||||
def __init__(self,*args,**kwargs):
|
||||
super(DummyPlugin,self).__init__(*args,**kwargs)
|
||||
self.cleaned_instances = []
|
||||
|
||||
class CommandPlugin(MockPlugin):
|
||||
@classmethod
|
||||
def tunable_devices(cls):
|
||||
return ['a', 'b']
|
||||
def _get_config_options(self):
|
||||
return {'default_option1':'default_value1',\
|
||||
'default_option2':'default_value2'}
|
||||
|
||||
def _post_init(self):
|
||||
self._size = 'M'
|
||||
self._color = { 'a': 'green', 'b': 'pink' }
|
||||
def _instance_cleanup(self, instance):
|
||||
self.cleaned_instances.append(instance)
|
||||
|
||||
@tuned.plugins.decorators.command_set('size')
|
||||
def _set_size(self, new_size):
|
||||
def _get_device_objects(self, devices):
|
||||
objects = []
|
||||
for device in devices:
|
||||
objects.append({'name':device})
|
||||
return devices
|
||||
|
||||
class DummyDevice(Mapping):
|
||||
def __init__(self,sysname,dictionary,*args,**kwargs):
|
||||
super(DummyDevice,self).__init__(*args,**kwargs)
|
||||
self.dictionary = dictionary
|
||||
self.properties = dictionary
|
||||
self.sys_name = sysname
|
||||
self.setting = '101'
|
||||
|
||||
def __getitem__(self,prop):
|
||||
return self.dictionary.__getitem__(prop)
|
||||
|
||||
def __len__(self):
|
||||
return self.dictionary.__len__()
|
||||
|
||||
def __iter__(self):
|
||||
return self.dictionary.__iter__()
|
||||
|
||||
class CommandsPlugin(Plugin):
|
||||
def __init__(self,*args,**kwargs):
|
||||
super(CommandsPlugin,self).__init__(*args,**kwargs)
|
||||
self._size = 'S'
|
||||
|
||||
@classmethod
|
||||
def _get_config_options(self):
|
||||
"""Default configuration options for the plugin."""
|
||||
return {'size':'S','device_setting':'101'}
|
||||
|
||||
@decorators.command_set('size')
|
||||
def _set_size(self, new_size, sim):
|
||||
self._size = new_size
|
||||
return new_size
|
||||
|
||||
@tuned.plugins.decorators.command_get('size')
|
||||
@decorators.command_get('size')
|
||||
def _get_size(self):
|
||||
return self._size
|
||||
|
||||
@tuned.plugins.decorators.command_set('color', per_device=True)
|
||||
def _set_color(self, device, new_color):
|
||||
self._color[device] = new_color
|
||||
@decorators.command_set('device_setting',per_device = True)
|
||||
def _set_device_setting(self,value,device,sim):
|
||||
device.setting = value
|
||||
return device.setting
|
||||
|
||||
@tuned.plugins.decorators.command_get('color')
|
||||
def _get_color(self, device):
|
||||
return self._color[device]
|
||||
@decorators.command_get('device_setting')
|
||||
def _get_device_setting(self,device,ignore_missing = False):
|
||||
return device.setting
|
||||
|
||||
class PluginBaseClassTestCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.storage_factory = flexmock(create = lambda name: None)
|
||||
self.monitor_repository = None
|
||||
self.plugin = MockPlugin(self.monitor_repository, self.storage_factory)
|
||||
@decorators.command_custom('custom_name')
|
||||
def the_most_custom_command(self):
|
||||
return True
|
||||
|
||||
def test_init(self):
|
||||
self.storage_factory.should_receive('create').and_return(None).times(2)
|
||||
plugin = MockPlugin(self.monitor_repository, self.storage_factory, None, None)
|
||||
plugin = MockPlugin(self.monitor_repository, self.storage_factory)
|
||||
class BadCommandsPlugin(Plugin):
|
||||
def __init__(self,*args,**kwargs):
|
||||
super(BadCommandsPlugin,self).__init__(*args,**kwargs)
|
||||
self._size = 'S'
|
||||
|
||||
def test_cleanup(self):
|
||||
self.plugin.cleanup()
|
||||
|
||||
def test_update_tuning_not_implemented(self):
|
||||
with self.assertRaises(NotImplementedError):
|
||||
self.plugin.update_tuning()
|
||||
|
||||
def test_class_properties(self):
|
||||
self.assertIs(PluginBase.tunable_devices(), None)
|
||||
self.assertTrue(PluginBase.is_supported())
|
||||
|
||||
def test_instance_properties(self):
|
||||
self.assertTrue(self.plugin.dynamic_tuning)
|
||||
|
||||
def test_merge_unknown_options(self):
|
||||
plugin1 = PluginBase(self.monitor_repository, self.storage_factory, None, None)
|
||||
plugin2 = PluginBase(self.monitor_repository, self.storage_factory, None, {})
|
||||
plugin3 = PluginBase(self.monitor_repository, self.storage_factory, None, {'unknown': 'test'})
|
||||
self.assertDictEqual(plugin1._options, {})
|
||||
self.assertDictEqual(plugin2._options, {})
|
||||
self.assertDictEqual(plugin3._options, {})
|
||||
|
||||
def test_merge_known_options(self):
|
||||
plugin1 = MockPlugin(self.monitor_repository, self.storage_factory, None, None)
|
||||
plugin2 = MockPlugin(self.monitor_repository, self.storage_factory, None, {'color': 'red'})
|
||||
plugin3 = MockPlugin(self.monitor_repository, self.storage_factory, None, {'size': 'S', 'fabric': 'cotton'})
|
||||
self.assertDictEqual(plugin1._options, {'size': 'XXL', 'color': 'blue'})
|
||||
self.assertDictEqual(plugin2._options, {'size': 'XXL', 'color': 'red'})
|
||||
self.assertDictEqual(plugin3._options, {'size': 'S', 'color': 'blue'})
|
||||
|
||||
def test_classs_with_invalid_commands(self):
|
||||
with self.assertRaises(TypeError):
|
||||
plugin = InvalidCommandPlugin(self.monitor_repository, self.storage_factory)
|
||||
|
||||
def test_storage_with_device_independent_commands(self):
|
||||
storage = flexmock()
|
||||
storage_factory = flexmock()
|
||||
storage_factory.should_receive('create').and_return(storage)
|
||||
|
||||
storage.should_receive('set').with_args('size', 'M').once.ordered
|
||||
storage.should_receive('get').with_args('size').and_return('M').once.ordered
|
||||
storage.should_receive('unset').with_args('size').once.ordered
|
||||
|
||||
plugin = CommandPlugin(self.monitor_repository, storage_factory, ['b'], {'size': 'XXS', 'color': None})
|
||||
plugin.execute_commands()
|
||||
plugin.cleanup_commands()
|
||||
|
||||
def test_storage_with_per_device_commands(self):
|
||||
storage = flexmock()
|
||||
storage_factory = flexmock()
|
||||
storage_factory.should_receive('create').and_return(storage)
|
||||
|
||||
storage.should_receive('set').with_args('color@b', 'pink').once.ordered
|
||||
storage.should_receive('get').with_args('color@b').and_return('pink').once.ordered
|
||||
storage.should_receive('unset').with_args('color@b').once.ordered
|
||||
|
||||
plugin = CommandPlugin(self.monitor_repository, storage_factory, ['b'], {'size': None, 'color': 'white'})
|
||||
plugin.execute_commands()
|
||||
plugin.cleanup_commands()
|
||||
|
||||
def test_exception_with_per_device_commands_when_no_devices_specified(self):
|
||||
storage = flexmock(set=lambda key: None, get=lambda key, value: None, unset=lambda key: None)
|
||||
storage_factory = flexmock()
|
||||
storage_factory.should_receive('create').and_return(storage)
|
||||
|
||||
plugin = CommandPlugin(self.monitor_repository, storage_factory)
|
||||
with self.assertRaises(TypeError):
|
||||
plugin.execute_commands()
|
||||
with self.assertRaises(TypeError):
|
||||
plugin.cleanup_commands()
|
||||
@decorators.command_set('size')
|
||||
def _set_size(self, new_size):
|
||||
self._size = new_size
|
||||
|
|
|
|||
|
|
@ -1,111 +1,114 @@
|
|||
import unittest
|
||||
import unittest2
|
||||
import flexmock
|
||||
import tempfile
|
||||
import shutil
|
||||
import os.path
|
||||
import tuned.profiles.exceptions
|
||||
from tuned.profiles.loader import Loader
|
||||
from flexmock import flexmock
|
||||
import os
|
||||
|
||||
class MockProfile(object):
|
||||
def __init__(self, name, config):
|
||||
self.name = name
|
||||
self.options = {}
|
||||
self.units = {}
|
||||
self.test_config = config
|
||||
|
||||
class MockProfileFactory(object):
|
||||
def create(self, name, config):
|
||||
return MockProfile(name, config)
|
||||
|
||||
class MockProfileMerger(object):
|
||||
def merge(self, profiles):
|
||||
new = MockProfile("merged", {})
|
||||
new.test_merged = profiles
|
||||
return new
|
||||
|
||||
class LoaderTestCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.factory = MockProfileFactory()
|
||||
self.merger = MockProfileMerger()
|
||||
self.loader = Loader(self._tmp_load_dirs, self.factory, self.merger)
|
||||
import tuned.profiles as profiles
|
||||
from tuned.profiles.exceptions import InvalidProfileException
|
||||
|
||||
class LoaderTestCase(unittest2.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
tmpdir1 = tempfile.mkdtemp()
|
||||
tmpdir2 = tempfile.mkdtemp()
|
||||
cls._tmp_load_dirs = [tmpdir1, tmpdir2]
|
||||
profiles.loader.log = flexmock.flexmock(info = lambda *args: None,\
|
||||
error = lambda *args: None,debug = lambda *args: None,\
|
||||
warn = lambda *args: None)
|
||||
cls._test_dir = tempfile.mkdtemp()
|
||||
cls._profiles_dir = cls._test_dir + '/test_profiles'
|
||||
cls._dummy_profile_dir = cls._profiles_dir + '/dummy'
|
||||
cls._dummy_profile_dir2 = cls._profiles_dir + '/dummy2'
|
||||
cls._dummy_profile_dir3 = cls._profiles_dir + '/dummy3'
|
||||
cls._dummy_profile_dir4 = cls._profiles_dir + '/dummy4'
|
||||
try:
|
||||
os.mkdir(cls._profiles_dir)
|
||||
os.mkdir(cls._dummy_profile_dir)
|
||||
os.mkdir(cls._dummy_profile_dir2)
|
||||
os.mkdir(cls._dummy_profile_dir3)
|
||||
os.mkdir(cls._dummy_profile_dir4)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
cls._create_profile(tmpdir1, "default", "[main]\n\n[network]\ntype=net\ndevices=em*\n\n[disk]\nenabled=false\n")
|
||||
cls._create_profile(tmpdir1, "invalid", "INVALID")
|
||||
cls._create_profile(tmpdir1, "expand", "[expand]\ntype=script\nscript=runme.sh\n")
|
||||
cls._create_profile(tmpdir2, "empty", "")
|
||||
with open(cls._dummy_profile_dir + '/tuned.conf','w') as f:
|
||||
f.write('[main]\nsummary=dummy profile\n')
|
||||
f.write('[test_unit]\ntest_option=hello\n')
|
||||
f.write('random_option=random\n')
|
||||
|
||||
cls._create_profile(tmpdir1, "custom", "[custom]\ntype=one\n")
|
||||
cls._create_profile(tmpdir2, "custom", "[custom]\ntype=two\n")
|
||||
with open(cls._dummy_profile_dir2 + '/tuned.conf','w') as f:
|
||||
f.write(\
|
||||
'[main]\nsummary=second dummy profile\n')
|
||||
f.write('[test_unit]\ntest_option=hello world\n')
|
||||
f.write('secondary_option=whatever\n')
|
||||
|
||||
with open(cls._dummy_profile_dir3 + '/tuned.conf','w') as f:
|
||||
f.write('[main]\nsummary=another profile\ninclude=dummy\n')
|
||||
f.write('[test_unit]\ntest_option=bye bye\n')
|
||||
f.write('new_option=add this\n')
|
||||
|
||||
with open(cls._dummy_profile_dir4 + '/tuned.conf','w') as f:
|
||||
f.write(\
|
||||
'[main]\nsummary=dummy profile for configuration read test\n')
|
||||
f.write('file_path=${i:PROFILE_DIR}/whatever\n')
|
||||
f.write('script=random_name.sh\n')
|
||||
f.write('[test_unit]\ntest_option=hello world\n')
|
||||
|
||||
def setUp(self):
|
||||
locator = profiles.Locator([self._profiles_dir])
|
||||
factory = profiles.Factory()
|
||||
merger = profiles.Merger()
|
||||
self._loader = profiles.Loader(locator,factory,merger,None,\
|
||||
profiles.variables.Variables())
|
||||
|
||||
def test_safe_name(self):
|
||||
self.assertFalse(self._loader.safe_name('*'))
|
||||
self.assertFalse(self._loader.safe_name('$'))
|
||||
self.assertTrue(self._loader.safe_name('Allowed_ch4rs.-'))
|
||||
|
||||
def test_load_without_include(self):
|
||||
merged_profile = self._loader.load(['dummy','dummy2'])
|
||||
|
||||
self.assertEqual(merged_profile.name, 'dummy dummy2')
|
||||
self.assertEqual(merged_profile.options['summary'],\
|
||||
'second dummy profile')
|
||||
self.assertEqual(merged_profile.units['test_unit'].\
|
||||
options['test_option'],'hello world')
|
||||
self.assertEqual(merged_profile.units['test_unit'].\
|
||||
options['secondary_option'],'whatever')
|
||||
|
||||
with self.assertRaises(InvalidProfileException):
|
||||
self._loader.load([])
|
||||
|
||||
with self.assertRaises(InvalidProfileException):
|
||||
self._loader.load(['invalid'])
|
||||
|
||||
def test_load_with_include(self):
|
||||
merged_profile = self._loader.load(['dummy3'])
|
||||
|
||||
self.assertEqual(merged_profile.name,'dummy3')
|
||||
self.assertEqual(merged_profile.options['summary'],'another profile')
|
||||
self.assertEqual(merged_profile.units['test_unit'].\
|
||||
options['test_option'],'bye bye')
|
||||
self.assertEqual(merged_profile.units['test_unit'].\
|
||||
options['new_option'],'add this')
|
||||
self.assertEqual(merged_profile.units['test_unit'].\
|
||||
options['random_option'],'random')
|
||||
|
||||
def test_expand_profile_dir(self):
|
||||
self.assertEqual(self._loader._expand_profile_dir(\
|
||||
'/hello/world','${i:PROFILE_DIR}/file'),'/hello/world/file')
|
||||
|
||||
def test_load_config_data(self):
|
||||
config = self._loader._load_config_data(\
|
||||
self._dummy_profile_dir4 + '/tuned.conf')
|
||||
|
||||
self.assertEqual(config['main']['script'][0],\
|
||||
self._dummy_profile_dir4 + '/random_name.sh')
|
||||
|
||||
self.assertEqual(config['main']['file_path'],\
|
||||
self._dummy_profile_dir4 + '/whatever')
|
||||
|
||||
self.assertEqual(config['test_unit']['test_option'],\
|
||||
'hello world')
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
for tmp_dir in cls._tmp_load_dirs:
|
||||
shutil.rmtree(tmp_dir, True)
|
||||
|
||||
@classmethod
|
||||
def _create_profile(cls, load_dir, profile_name, tuned_conf_content):
|
||||
profile_dir = os.path.join(load_dir, profile_name)
|
||||
conf_name = os.path.join(profile_dir, "tuned.conf")
|
||||
os.mkdir(profile_dir)
|
||||
with open(conf_name, "w") as conf_file:
|
||||
conf_file.write(tuned_conf_content)
|
||||
|
||||
def test_init(self):
|
||||
Loader([], None, None)
|
||||
Loader(["/tmp"], None, None)
|
||||
Loader(["/foo", "/bar"], None, None)
|
||||
|
||||
def test_init_wrong_type(self):
|
||||
with self.assertRaises(TypeError):
|
||||
Loader(False, self.factory, self.merger)
|
||||
|
||||
def test_load(self):
|
||||
profile = self.loader.load("default")
|
||||
self.assertIn("main", profile.test_config)
|
||||
self.assertIn("disk", profile.test_config)
|
||||
self.assertEqual(profile.test_config["network"]["devices"], "em*")
|
||||
|
||||
def test_load_empty(self):
|
||||
profile = self.loader.load("empty")
|
||||
self.assertDictEqual(profile.test_config, {})
|
||||
|
||||
def test_load_invalid(self):
|
||||
with self.assertRaises(tuned.profiles.exceptions.InvalidProfileException):
|
||||
invalid_config = self.loader.load("invalid")
|
||||
|
||||
def test_load_nonexistent(self):
|
||||
with self.assertRaises(tuned.profiles.exceptions.InvalidProfileException):
|
||||
config = self.loader.load("nonexistent")
|
||||
|
||||
def test_load_order(self):
|
||||
profile = self.loader.load("custom")
|
||||
self.assertEqual(profile.test_config["custom"]["type"], "two")
|
||||
|
||||
def test_default_load(self):
|
||||
profile = self.loader.load("empty")
|
||||
self.assertIs(type(profile), MockProfile)
|
||||
|
||||
def test_script_expand_names(self):
|
||||
profile = self.loader.load("expand")
|
||||
expected_name = os.path.join(self._tmp_load_dirs[0], "expand", "runme.sh")
|
||||
self.assertEqual(profile.test_config["expand"]["script"], expected_name)
|
||||
|
||||
def test_load_multiple_profiles(self):
|
||||
profile = self.loader.load(["default", "expand"])
|
||||
self.assertEqual(len(profile.test_merged), 2)
|
||||
|
||||
def test_include_directive(self):
|
||||
profile1 = MockProfile("first", {})
|
||||
profile1.options = {"include": "default"}
|
||||
profile2 = MockProfile("second", {})
|
||||
|
||||
flexmock(self.factory).should_receive("create").and_return(profile1).and_return(profile2).twice()
|
||||
profile = self.loader.load("empty")
|
||||
|
||||
self.assertEqual(len(profile.test_merged), 2)
|
||||
shutil.rmtree(cls._test_dir)
|
||||
|
|
|
|||
|
|
@ -1,10 +1,10 @@
|
|||
import unittest
|
||||
import unittest2
|
||||
import os
|
||||
import shutil
|
||||
import tempfile
|
||||
from tuned.profiles.locator import Locator
|
||||
|
||||
class LocatorTestCase(unittest.TestCase):
|
||||
class LocatorTestCase(unittest2.TestCase):
|
||||
def setUp(self):
|
||||
self.locator = Locator(self._tmp_load_dirs)
|
||||
|
||||
|
|
|
|||
|
|
@ -1,49 +1,62 @@
|
|||
import unittest
|
||||
import unittest2
|
||||
from tuned.profiles.merger import Merger
|
||||
from tuned.profiles.profile import Profile
|
||||
from collections import OrderedDict
|
||||
|
||||
class MergerTestCase(unittest.TestCase):
|
||||
class MergerTestCase(unittest2.TestCase):
|
||||
def test_merge_without_replace(self):
|
||||
merger = Merger()
|
||||
config1 = OrderedDict([
|
||||
("main", OrderedDict()),
|
||||
("net", { "devices": "em0", "custom": "option"}),
|
||||
("main", {"test_option" : "test_value1"}),
|
||||
("net", { "devices": "em0", "custom": "custom_value"}),
|
||||
])
|
||||
profile1 = Profile('test_profile1',config1)
|
||||
config2 = OrderedDict([
|
||||
("main", OrderedDict()),
|
||||
("net", { "devices": "em1" }),
|
||||
('main', {'test_option' : 'test_value2'}),
|
||||
('net', { 'devices': 'em1' }),
|
||||
])
|
||||
config = merger.merge([config1, config2])
|
||||
profile2 = Profile("test_profile2",config2)
|
||||
|
||||
self.assertIn("main", config)
|
||||
self.assertIn("net", config)
|
||||
self.assertEqual(config["net"]["custom"], "option")
|
||||
self.assertEqual(config["net"]["devices"], "em1")
|
||||
merged_profile = merger.merge([profile1, profile2])
|
||||
|
||||
self.assertEqual(merged_profile.options["test_option"],"test_value2")
|
||||
self.assertIn("net", merged_profile.units)
|
||||
self.assertEqual(merged_profile.units["net"].options["custom"],\
|
||||
"custom_value")
|
||||
self.assertEqual(merged_profile.units["net"].devices, "em1")
|
||||
|
||||
def test_merge_with_replace(self):
|
||||
merger = Merger()
|
||||
config1 = OrderedDict([
|
||||
("main", OrderedDict()),
|
||||
("main", {"test_option" : "test_value1"}),
|
||||
("net", { "devices": "em0", "custom": "option"}),
|
||||
])
|
||||
profile1 = Profile('test_profile1',config1)
|
||||
config2 = OrderedDict([
|
||||
("main", OrderedDict()),
|
||||
("main", {"test_option" : "test_value2"}),
|
||||
("net", { "devices": "em1", "replace": True }),
|
||||
])
|
||||
config = merger.merge([config1, config2])
|
||||
profile2 = Profile('test_profile2',config2)
|
||||
merged_profile = merger.merge([profile1, profile2])
|
||||
|
||||
self.assertIn("main", config)
|
||||
self.assertIn("net", config)
|
||||
self.assertNotIn("custom", config["net"])
|
||||
self.assertEqual(config["net"]["devices"], "em1")
|
||||
self.assertEqual(merged_profile.options["test_option"],"test_value2")
|
||||
self.assertIn("net", merged_profile.units)
|
||||
self.assertNotIn("custom", merged_profile.units["net"].options)
|
||||
self.assertEqual(merged_profile.units["net"].devices, "em1")
|
||||
|
||||
def test_merge_multiple_order(self):
|
||||
merger = Merger()
|
||||
config1 = OrderedDict([ ("main", OrderedDict()), ("net", { "devices": "em0" }) ])
|
||||
config2 = OrderedDict([ ("main", OrderedDict()), ("net", { "devices": "em1" }) ])
|
||||
config3 = OrderedDict([ ("main", OrderedDict()), ("net", { "devices": "em2" }) ])
|
||||
config = merger.merge([config1, config2, config3])
|
||||
config1 = OrderedDict([ ("main", {"test_option" : "test_value1"}),\
|
||||
("net", { "devices": "em0" }) ])
|
||||
profile1 = Profile('test_profile1',config1)
|
||||
config2 = OrderedDict([ ("main", {"test_option" : "test_value2"}),\
|
||||
("net", { "devices": "em1" }) ])
|
||||
profile2 = Profile('test_profile2',config2)
|
||||
config3 = OrderedDict([ ("main", {"test_option" : "test_value3"}),\
|
||||
("net", { "devices": "em2" }) ])
|
||||
profile3 = Profile('test_profile3',config3)
|
||||
merged_profile = merger.merge([profile1, profile2, profile3])
|
||||
|
||||
self.assertIn("main", config)
|
||||
self.assertIn("net", config)
|
||||
self.assertEqual(config["net"]["devices"], "em2")
|
||||
self.assertEqual(merged_profile.options["test_option"],"test_value3")
|
||||
self.assertIn("net", merged_profile.units)
|
||||
self.assertEqual(merged_profile.units["net"].devices, "em2")
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
import unittest
|
||||
import unittest2
|
||||
import tuned.profiles
|
||||
import collections
|
||||
|
||||
|
|
@ -6,7 +6,7 @@ class MockProfile(tuned.profiles.profile.Profile):
|
|||
def _create_unit(self, name, config):
|
||||
return (name, config)
|
||||
|
||||
class ProfileTestCase(unittest.TestCase):
|
||||
class ProfileTestCase(unittest2.TestCase):
|
||||
|
||||
def test_init(self):
|
||||
MockProfile("test", {})
|
||||
|
|
@ -20,7 +20,7 @@ class ProfileTestCase(unittest.TestCase):
|
|||
|
||||
self.assertIs(type(profile.units), collections.OrderedDict)
|
||||
self.assertEqual(len(profile.units), 2)
|
||||
self.assertListEqual(sorted([name_config[0] for name_config in profile.units]), sorted(["network", "storage"]))
|
||||
self.assertListEqual(sorted([name_config for name_config in profile.units]), sorted(["network", "storage"]))
|
||||
|
||||
def test_create_units_empty(self):
|
||||
profile = MockProfile("test", {"main":{}})
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
import unittest
|
||||
import unittest2
|
||||
from tuned.profiles import Unit
|
||||
|
||||
class UnitTestCase(unittest.TestCase):
|
||||
class UnitTestCase(unittest2.TestCase):
|
||||
|
||||
def test_default_options(self):
|
||||
unit = Unit("sample", {})
|
||||
|
|
|
|||
|
|
@ -1,7 +1,12 @@
|
|||
import unittest
|
||||
import os.path
|
||||
import tempfile
|
||||
|
||||
import tuned.storage
|
||||
import tuned.consts as consts
|
||||
|
||||
temp_storage_file = tempfile.TemporaryFile(mode='r')
|
||||
consts.DEFAULT_STORAGE_FILE = temp_storage_file.name
|
||||
|
||||
class StoragePickleProviderTestCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
|
|
@ -17,7 +22,7 @@ class StoragePickleProviderTestCase(unittest.TestCase):
|
|||
self.assertEqual(self._temp_filename, provider._path)
|
||||
|
||||
provider = tuned.storage.PickleProvider()
|
||||
self.assertEqual("/run/tuned/save.pickle", provider._path)
|
||||
self.assertEqual(temp_storage_file.name, provider._path)
|
||||
|
||||
def test_memory_persistence(self):
|
||||
provider = tuned.storage.PickleProvider(self._temp_filename)
|
||||
|
|
@ -66,3 +71,7 @@ class StoragePickleProviderTestCase(unittest.TestCase):
|
|||
provider.load()
|
||||
self.assertIsNone(provider.get("ns1", "opt1"))
|
||||
self.assertIsNone(provider.get("ns2", "opt2"))
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
temp_storage_file.close()
|
||||
|
|
|
|||
0
tests/utils/__init__.py
Normal file
0
tests/utils/__init__.py
Normal file
244
tests/utils/test_commands.py
Normal file
244
tests/utils/test_commands.py
Normal file
|
|
@ -0,0 +1,244 @@
|
|||
import unittest2
|
||||
import tempfile
|
||||
import flexmock
|
||||
import shutil
|
||||
import re
|
||||
import os
|
||||
|
||||
from tuned.utils.commands import commands
|
||||
import tuned.consts as consts
|
||||
from tuned.exceptions import TunedException
|
||||
import tuned.utils.commands
|
||||
|
||||
tuned.utils.commands.log = flexmock.flexmock(info = lambda *args: None,\
|
||||
error = lambda *args: None,debug = lambda *args: None,\
|
||||
warn = lambda *args: None)
|
||||
|
||||
class CommandsTestCase(unittest2.TestCase):
|
||||
def setUp(self):
|
||||
self._commands = commands()
|
||||
self._test_dir = tempfile.mkdtemp()
|
||||
self._test_file = tempfile.NamedTemporaryFile(mode='r',dir = self._test_dir)
|
||||
|
||||
def test_get_bool(self):
|
||||
positive_values = ['y','yes','t','true']
|
||||
negative_values = ['n','no','f','false']
|
||||
|
||||
for val in positive_values:
|
||||
self.assertEqual(self._commands.get_bool(val),"1")
|
||||
|
||||
for val in negative_values:
|
||||
self.assertEqual(self._commands.get_bool(val),"0")
|
||||
|
||||
self.assertEqual(self._commands.get_bool('bad_value'),'bad_value')
|
||||
|
||||
def test_remove_ws(self):
|
||||
self.assertEqual(self._commands.remove_ws(' a bc '),'a bc')
|
||||
|
||||
def test_unquote(self):
|
||||
self.assertEqual(self._commands.unquote('"whatever"'),'whatever')
|
||||
|
||||
def test_escape(self):
|
||||
self.assertEqual(self._commands.escape('\\'),'\\\\')
|
||||
|
||||
def test_unescape(self):
|
||||
self.assertEqual(self._commands.unescape('\\'),'')
|
||||
|
||||
def test_align_str(self):
|
||||
self.assertEqual(self._commands.align_str('abc',5,'def'),'abc def')
|
||||
|
||||
def test_dict2list(self):
|
||||
dictionary = {'key1':1,'key2':2,'key3':3}
|
||||
self.assertEqual(self._commands.dict2list(dictionary)\
|
||||
,['key1',1,'key2',2,'key3',3])
|
||||
|
||||
def test_re_lookup_compile(self):
|
||||
pattern = re.compile(r'([1-9])')
|
||||
dictionary = {'[1-9]':''}
|
||||
self.assertEqual(self._commands.re_lookup_compile(dictionary).pattern\
|
||||
,pattern.pattern)
|
||||
self.assertIsNone(self._commands.re_lookup_compile(None))
|
||||
|
||||
def test_multiple_re_replace(self):
|
||||
text = 'abcd1234'
|
||||
dictionary = {'abc':'gfh'}
|
||||
pattern = self._commands.re_lookup_compile(dictionary)
|
||||
self.assertEqual(self._commands.multiple_re_replace(dictionary,text)\
|
||||
,'gfhd1234')
|
||||
self.assertEqual(self._commands.multiple_re_replace(\
|
||||
dictionary,text,pattern),'gfhd1234')
|
||||
|
||||
def test_re_lookup(self):
|
||||
dictionary = {'abc':'abc','mno':'mno'}
|
||||
text1 = 'abc def'
|
||||
text2 = 'jkl mno'
|
||||
text12 = 'abc mno'
|
||||
text3 = 'whatever'
|
||||
self.assertEqual(self._commands.re_lookup(dictionary,text1),'abc')
|
||||
self.assertEqual(self._commands.re_lookup(dictionary,text2),'mno')
|
||||
self.assertEqual(self._commands.re_lookup(dictionary,text12),'abc')
|
||||
self.assertIsNone(self._commands.re_lookup(dictionary,text3),None)
|
||||
|
||||
def test_write_to_file(self):
|
||||
self.assertTrue(self._commands.write_to_file(self._test_file.name,\
|
||||
'hello world'))
|
||||
with open(self._test_file.name,'r') as f:
|
||||
self.assertEqual(f.read(),'hello world')
|
||||
|
||||
self.assertTrue(self._commands.write_to_file(self._test_file.name,\
|
||||
'world hello'))
|
||||
with open(self._test_file.name,'r') as f:
|
||||
self.assertEqual(f.read(),'world hello')
|
||||
|
||||
local_test_file = self._test_dir + '/dir' +'/self._test_file'
|
||||
self.assertTrue(self._commands.write_to_file(local_test_file,\
|
||||
'hello world',True))
|
||||
with open(local_test_file,'r') as f:
|
||||
self.assertEqual(f.read(),'hello world')
|
||||
|
||||
shutil.rmtree(os.path.dirname(local_test_file))
|
||||
|
||||
self.assertFalse(self._commands.write_to_file(local_test_file,\
|
||||
'hello world'))
|
||||
|
||||
def test_read_file(self):
|
||||
with open(self._test_file.name,'w') as f:
|
||||
f.write('hello world')
|
||||
self.assertEqual(self._commands.read_file(self._test_file.name),\
|
||||
'hello world')
|
||||
self.assertEqual(self._commands.read_file('/bad_name','error'),\
|
||||
'error')
|
||||
|
||||
def test_rmtree(self):
|
||||
test_tree = self._test_dir + '/one/two'
|
||||
os.makedirs(test_tree)
|
||||
test_tree = self._test_dir + '/one'
|
||||
self.assertTrue(self._commands.rmtree(test_tree))
|
||||
self.assertFalse(os.path.isdir(test_tree))
|
||||
self.assertTrue(self._commands.rmtree(test_tree))
|
||||
|
||||
def test_unlink(self):
|
||||
local_test_file = self._test_dir + 'file_to_delete'
|
||||
open(local_test_file,'w').close()
|
||||
self.assertTrue(os.path.exists(local_test_file))
|
||||
self.assertTrue(self._commands.unlink(local_test_file))
|
||||
self.assertFalse(os.path.exists(local_test_file))
|
||||
self.assertTrue(self._commands.unlink(local_test_file))
|
||||
|
||||
def test_rename(self):
|
||||
rename_test_file = self._test_dir + '/bad_name'
|
||||
open(rename_test_file,'w').close()
|
||||
self.assertTrue(self._commands.rename(rename_test_file,\
|
||||
self._test_dir + '/right_name'))
|
||||
self.assertTrue(os.path.exists(self._test_dir + '/right_name'))
|
||||
os.remove(self._test_dir + '/right_name')
|
||||
|
||||
def test_copy(self):
|
||||
copy_test_file = self._test_dir + '/origo'
|
||||
with open(copy_test_file,'w') as f:
|
||||
f.write('hello world')
|
||||
self.assertTrue(self._commands.copy(copy_test_file,\
|
||||
self._test_dir + '/copy'))
|
||||
self.assertTrue(os.path.exists(self._test_dir + '/copy'))
|
||||
self.assertTrue(os.path.exists(self._test_dir + '/origo'))
|
||||
with open(self._test_dir + '/copy','r') as f:
|
||||
self.assertEqual(f.read(),'hello world')
|
||||
os.remove(self._test_dir + '/origo')
|
||||
os.remove(self._test_dir + '/copy')
|
||||
|
||||
def test_replace_in_file(self):
|
||||
with open(self._test_file.name,'w') as f:
|
||||
f.write('hello world')
|
||||
|
||||
self.assertTrue(self._commands.replace_in_file(self._test_file.name,\
|
||||
'hello','bye'))
|
||||
with open(self._test_file.name,'r') as f:
|
||||
self.assertEqual(f.read(),'bye world')
|
||||
|
||||
def test_multiple_replace_in_file(self):
|
||||
dictionary = {'abc':'123','ghi':'456'}
|
||||
|
||||
with open(self._test_file.name,'w') as f:
|
||||
f.write('abc def ghi')
|
||||
|
||||
self.assertTrue(self._commands.multiple_replace_in_file(\
|
||||
self._test_file.name,dictionary))
|
||||
with open(self._test_file.name,'r') as f:
|
||||
self.assertEqual(f.read(),'123 def 456')
|
||||
|
||||
def test_add_modify_option_in_file(self):
|
||||
with open(self._test_file.name,'w') as f:
|
||||
f.write('option1="123"\noption2="456"\n')
|
||||
|
||||
dictionary = {'option3':789,'option1':321}
|
||||
self.assertTrue(self._commands.add_modify_option_in_file(\
|
||||
self._test_file.name,dictionary))
|
||||
with open(self._test_file.name,'r') as f:
|
||||
self.assertEqual(f.read(),\
|
||||
'option1="321"\noption2="456"\noption3="789"\n')
|
||||
|
||||
def test_get_active_option(self):
|
||||
self.assertEqual(self._commands.get_active_option('opt1 [opt2] opt3'),\
|
||||
'opt2')
|
||||
self.assertEqual(self._commands.get_active_option('opt1 opt2 opt3'),\
|
||||
'opt1')
|
||||
self.assertEqual(self._commands.get_active_option(\
|
||||
'opt1 opt2 opt3',False),'opt1 opt2 opt3')
|
||||
|
||||
def test_hex2cpulist(self):
|
||||
self.assertEqual(self._commands.hex2cpulist('0xf'),[0,1,2,3])
|
||||
self.assertEqual(self._commands.hex2cpulist('0x1,0000,0001'),[0,32])
|
||||
|
||||
def test_cpulist_unpack(self):
|
||||
cpus = '4-8,^6,0xf00,,'
|
||||
self.assertEqual(self._commands.cpulist_unpack(cpus),[4,5,7,8,9,10,11])
|
||||
|
||||
def test_cpulist_pack(self):
|
||||
self.assertEqual(self._commands.cpulist_pack([0,1,3,4,5,6,8,9,32]),\
|
||||
['0-1','3-6','8-9','32'])
|
||||
|
||||
def test_cpulist2hex(self):
|
||||
self.assertEqual(self._commands.cpulist2hex('1-3,5,32'),\
|
||||
'00000001,0000002e')
|
||||
|
||||
def test_cpulist2bitmask(self):
|
||||
self.assertEqual(self._commands.cpulist2bitmask([1,2,3]),0b1110)
|
||||
self.assertEqual(self._commands.cpulist2bitmask([2,4,6]),0b1010100)
|
||||
|
||||
def test_get_size(self):
|
||||
self.assertEqual(self._commands.get_size('100KB'),102400)
|
||||
self.assertEqual(self._commands.get_size('100Kb'),102400)
|
||||
self.assertEqual(self._commands.get_size('100kb'),102400)
|
||||
self.assertEqual(self._commands.get_size('1MB'),1024 * 1024)
|
||||
self.assertEqual(self._commands.get_size('1GB'),1024 * 1024 * 1024)
|
||||
|
||||
def test_get_active_profile(self):
|
||||
consts.ACTIVE_PROFILE_FILE = self._test_dir + '/active_profile'
|
||||
consts.PROFILE_MODE_FILE = self._test_dir + '/profile_mode'
|
||||
with open(consts.ACTIVE_PROFILE_FILE,'w') as f:
|
||||
f.write('test_profile')
|
||||
with open(consts.PROFILE_MODE_FILE,'w') as f:
|
||||
f.write('auto')
|
||||
(profile,mode) = self._commands.get_active_profile()
|
||||
self.assertEqual(profile,'test_profile')
|
||||
self.assertEqual(mode,False)
|
||||
os.remove(consts.ACTIVE_PROFILE_FILE)
|
||||
os.remove(consts.PROFILE_MODE_FILE)
|
||||
(profile,mode) = self._commands.get_active_profile()
|
||||
self.assertEqual(profile,None)
|
||||
self.assertEqual(mode,None)
|
||||
|
||||
def test_save_active_profile(self):
|
||||
consts.ACTIVE_PROFILE_FILE = self._test_dir + '/active_profile'
|
||||
consts.PROFILE_MODE_FILE = self._test_dir + '/profile_mode'
|
||||
self._commands.save_active_profile('test_profile',False)
|
||||
with open(consts.ACTIVE_PROFILE_FILE) as f:
|
||||
self.assertEqual(f.read(),'test_profile\n')
|
||||
with open(consts.PROFILE_MODE_FILE) as f:
|
||||
self.assertEqual(f.read(),'auto\n')
|
||||
os.remove(consts.ACTIVE_PROFILE_FILE)
|
||||
os.remove(consts.PROFILE_MODE_FILE)
|
||||
|
||||
def tearDown(self):
|
||||
self._test_file.close()
|
||||
shutil.rmtree(self._test_dir)
|
||||
42
tests/utils/test_global_config.py
Normal file
42
tests/utils/test_global_config.py
Normal file
|
|
@ -0,0 +1,42 @@
|
|||
import unittest2
|
||||
import flexmock
|
||||
import tempfile
|
||||
import shutil
|
||||
import os
|
||||
|
||||
import tuned.consts as consts
|
||||
import tuned.utils.global_config as global_config
|
||||
|
||||
class GlobalConfigTestCase(unittest2.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
global_config.log = flexmock.flexmock(info = lambda *args: None,\
|
||||
error = lambda *args: None,debug = lambda *args: None,\
|
||||
warn = lambda *args: None)
|
||||
|
||||
cls.test_dir = tempfile.mkdtemp()
|
||||
with open(cls.test_dir + '/test_config','w') as f:
|
||||
f.write('test_option = hello\ntest_bool = 1\ntest_size = 12MB\n'\
|
||||
+ 'false_bool=0\n')
|
||||
|
||||
cls._global_config = global_config.GlobalConfig(\
|
||||
cls.test_dir + '/test_config')
|
||||
|
||||
def test_get(self):
|
||||
self.assertEqual(self._global_config.get('test_option'), 'hello')
|
||||
|
||||
def test_get_bool(self):
|
||||
self.assertTrue(self._global_config.get_bool('test_bool'))
|
||||
self.assertFalse(self._global_config.get_bool('false_bool'))
|
||||
|
||||
def test_get_size(self):
|
||||
self.assertEqual(self._global_config.get_size('test_size'),\
|
||||
12*1024*1024)
|
||||
|
||||
self._global_config.set('test_size','bad_value')
|
||||
|
||||
self.assertIsNone(self._global_config.get_size('test_size'))
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
shutil.rmtree(cls.test_dir)
|
||||
|
|
@ -73,7 +73,7 @@ class Admin(object):
|
|||
action = getattr(self, "_action_" + action_name)
|
||||
except AttributeError as e:
|
||||
if not self._dbus:
|
||||
self._error(e + ", action '%s' is not implemented" % action_name)
|
||||
self._error(str(e) + ", action '%s' is not implemented" % action_name)
|
||||
return False
|
||||
if self._dbus:
|
||||
try:
|
||||
|
|
|
|||
|
|
@ -12,7 +12,13 @@ class DeviceMatcherUdev(device_matcher.DeviceMatcher):
|
|||
"""
|
||||
|
||||
properties = ''
|
||||
for key, val in list(device.items()):
|
||||
|
||||
try:
|
||||
items = device.properties.items()
|
||||
except AttributeError:
|
||||
items = device.items()
|
||||
|
||||
for key, val in list(items):
|
||||
properties += key + '=' + val + '\n'
|
||||
|
||||
return re.search(regex, properties, re.MULTILINE) is not None
|
||||
|
|
|
|||
|
|
@ -12,7 +12,7 @@ class Inventory(object):
|
|||
about related hardware events.
|
||||
"""
|
||||
|
||||
def __init__(self, udev_context=None, udev_monitor_cls=None, monitor_observer_factory=None, buffer_size=None):
|
||||
def __init__(self, udev_context=None, udev_monitor_cls=None, monitor_observer_factory=None, buffer_size=None, set_receive_buffer_size=True):
|
||||
if udev_context is not None:
|
||||
self._udev_context = udev_context
|
||||
else:
|
||||
|
|
@ -23,7 +23,9 @@ class Inventory(object):
|
|||
self._udev_monitor = udev_monitor_cls.from_netlink(self._udev_context)
|
||||
if buffer_size is None:
|
||||
buffer_size = consts.CFG_DEF_UDEV_BUFFER_SIZE
|
||||
self._udev_monitor.set_receive_buffer_size(buffer_size)
|
||||
|
||||
if (set_receive_buffer_size):
|
||||
self._udev_monitor.set_receive_buffer_size(buffer_size)
|
||||
|
||||
if monitor_observer_factory is None:
|
||||
monitor_observer_factory = _MonitorObserverFactory()
|
||||
|
|
|
|||
|
|
@ -31,14 +31,14 @@ class PickleProvider(interfaces.Provider):
|
|||
def save(self):
|
||||
try:
|
||||
log.debug("Saving %s" % str(self._data))
|
||||
with open(self._path, "w") as f:
|
||||
with open(self._path, "wb") as f:
|
||||
pickle.dump(self._data, f)
|
||||
except (OSError, IOError) as e:
|
||||
log.error("Error saving storage file '%s': %s" % (self._path, e))
|
||||
|
||||
def load(self):
|
||||
try:
|
||||
with open(self._path, "r") as f:
|
||||
with open(self._path, "rb") as f:
|
||||
self._data = pickle.load(f)
|
||||
except (OSError, IOError) as e:
|
||||
log.debug("Error loading storage file '%s': %s" % (self._path, e))
|
||||
|
|
|
|||
|
|
@ -16,9 +16,9 @@ class GlobalConfig():
|
|||
"update_interval = integer(default=%s)" % consts.CFG_DEF_UPDATE_INTERVAL,
|
||||
"recommend_command = boolean(default=%s)" % consts.CFG_DEF_RECOMMEND_COMMAND]
|
||||
|
||||
def __init__(self):
|
||||
def __init__(self,config_file = consts.GLOBAL_CONFIG_FILE):
|
||||
self._cfg = {}
|
||||
self.load_config()
|
||||
self.load_config(file_name=config_file)
|
||||
self._cmd = commands()
|
||||
|
||||
def load_config(self, file_name = consts.GLOBAL_CONFIG_FILE):
|
||||
|
|
|
|||
Loading…
Reference in a new issue