avocado.utils: Add linux_modules module

This library adds utility functions for linux kernel modules.
The functionality covers listing, loading and removing
loadable modules.
Signed-off-by: NLucas Meneghel Rodrigues <lmr@redhat.com>

avocado/utils/linux_modules.py

+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+#
+# See LICENSE for more details.
+#
+# This code was inspired in the autotest project,
+#
+# client/base_utils.py
+# Original author: Ross Brattain <ross.b.brattain@intel.com>
+
+"""
+APIs to list and load/unload linux kernel modules.
+"""
+
+import re
+import logging
+from . import process
+
+LOG = logging.getLogger('avocado.test')
+
+
+def load_module(module_name):
+    # Checks if a module has already been loaded
+    if module_is_loaded(module_name):
+        return False
+
+    process.system('/sbin/modprobe ' + module_name)
+    return True
+
+
+def parse_lsmod_for_module(l_raw, module_name, escape=True):
+    """
+    Use a regexp to parse raw lsmod output and get module information
+    :param l_raw: raw output of lsmod
+    :type l_raw:  str
+    :param module_name: Name of module to search for
+    :type module_name: str
+    :param escape: Escape regexp tokens in module_name, default True
+    :type escape: bool
+    :return: Dictionary of module info, name, size, submodules if present
+    :rtype: dict
+    """
+    # re.escape the module name for safety
+    if escape:
+        module_search = re.escape(module_name)
+    else:
+        module_search = module_name
+    # ^module_name spaces size spaces used optional spaces optional submodules
+    # use multiline regex to scan the entire output as one string without
+    # having to splitlines use named matches so we can extract the dictionary
+    # with groupdict
+    pattern = (r"^(?P<name>%s)\s+(?P<size>\d+)\s+(?P<used>\d+)"
+               "\s*(?P<submodules>\S+)?$")
+    lsmod = re.search(pattern % module_search, l_raw, re.M)
+    if lsmod:
+        # default to empty list if no submodules
+        module_info = lsmod.groupdict([])
+        # convert size to integer because it is an integer
+        module_info['size'] = int(module_info['size'])
+        module_info['used'] = int(module_info['used'])
+        if module_info['submodules']:
+            module_info['submodules'] = module_info['submodules'].split(',')
+        return module_info
+    else:
+        # return empty dict to be consistent
+        return {}
+
+
+def loaded_module_info(module_name):
+    """
+    Get loaded module details: Size and Submodules.
+
+    :param module_name: Name of module to search for
+    :type module_name: str
+    :return: Dictionary of module info, name, size, submodules if present
+    :rtype: dict
+    """
+    l_raw = process.system_output('/sbin/lsmod')
+    return parse_lsmod_for_module(l_raw, module_name)
+
+
+def get_submodules(module_name):
+    """
+    Get all submodules of the module.
+
+    :param module_name: Name of module to search for
+    :type module_name: str
+    :return: List of the submodules
+    :rtype: list
+    """
+    module_info = loaded_module_info(module_name)
+    module_list = []
+    try:
+        submodules = module_info["submodules"]
+    except KeyError:
+        LOG.info("Module %s is not loaded" % module_name)
+    else:
+        module_list = submodules
+        for module in submodules:
+            module_list += get_submodules(module)
+    return module_list
+
+
+def unload_module(module_name):
+    """
+    Removes a module. Handles dependencies. If even then it's not possible
+    to remove one of the modules, it will throw an error.CmdError exception.
+
+    :param module_name: Name of the module we want to remove.
+    :type module_name: str
+    """
+    module_info = loaded_module_info(module_name)
+    try:
+        submodules = module_info['submodules']
+    except KeyError:
+        LOG.info("Module %s is already unloaded" % module_name)
+    else:
+        for module in submodules:
+            unload_module(module)
+        module_info = loaded_module_info(module_name)
+        try:
+            module_used = module_info['used']
+        except KeyError:
+            LOG.info("Module %s is already unloaded" % module_name)
+            return
+        if module_used != 0:
+            raise RuntimeError("Module %s is still in use. "
+                               "Can not unload it." % module_name)
+        process.system("/sbin/modprobe -r %s" % module_name)
+        LOG.info("Module %s unloaded" % module_name)
+
+
+def module_is_loaded(module_name):
+    """
+    Is module loaded
+
+    :param module_name: Name of module to search for
+    :type module_name: str
+    :return: True is module is loaded
+    :rtype: bool
+    """
+    module_name = module_name.replace('-', '_')
+    return bool(loaded_module_info(module_name))
+
+
+def get_loaded_modules():
+    lsmod_output = process.system_output('/sbin/lsmod').splitlines()[1:]
+    return [line.split(None, 1)[0] for line in lsmod_output]

selftests/all/unit/avocado/utils_linux_modules_unittest.py

+import unittest
+import os
+import sys
+
+# simple magic for using scripts within a source tree
+basedir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+basedir = os.path.dirname(basedir)
+if os.path.isdir(os.path.join(basedir, 'avocado')):
+    sys.path.append(basedir)
+
+from avocado.utils import linux_modules
+
+
+class TestLsmod(unittest.TestCase):
+
+    LSMOD_OUT = """\
+Module                  Size  Used by
+ccm                    17773  2
+ip6t_rpfilter          12546  1
+ip6t_REJECT            12939  2
+xt_conntrack           12760  9
+ebtable_nat            12807  0
+ebtable_broute         12731  0
+bridge                110862  1 ebtable_broute
+stp                    12868  1 bridge
+llc                    13941  2 stp,bridge
+ebtable_filter         12827  0
+ebtables               30758  3 ebtable_broute,ebtable_nat,ebtable_filter
+ip6table_nat           13015  1
+nf_conntrack_ipv6      18738  6
+nf_defrag_ipv6         34712  1 nf_conntrack_ipv6
+nf_nat_ipv6            13213  1 ip6table_nat
+ip6table_mangle        12700  1
+ip6table_security      12710  1
+ip6table_raw           12683  1
+ip6table_filter        12815  1
+"""
+
+    def test_parse_lsmod(self):
+        lsmod_info = linux_modules.parse_lsmod_for_module(
+            self.LSMOD_OUT, "ebtables")
+        submodules = ['ebtable_broute', 'ebtable_nat', 'ebtable_filter']
+        assert lsmod_info['submodules'] == submodules
+        assert lsmod_info == {
+            'name': "ebtables",
+            'size': 30758,
+            'used': 3,
+            'submodules': submodules
+        }
+
+    @staticmethod
+    def test_parse_lsmod_is_empty():
+        lsmod_info = linux_modules.parse_lsmod_for_module("", "ebtables")
+        assert lsmod_info == {}
+
+    def test_parse_lsmod_no_submodules(self):
+        lsmod_info = linux_modules.parse_lsmod_for_module(self.LSMOD_OUT, "ccm")
+        submodules = []
+        assert lsmod_info['submodules'] == submodules
+        assert lsmod_info == {
+            'name': "ccm",
+            'size': 17773,
+            'used': 2,
+            'submodules': submodules
+        }
+
+    def test_parse_lsmod_single_submodules(self):
+        lsmod_info = linux_modules.parse_lsmod_for_module(
+            self.LSMOD_OUT, "bridge")
+        submodules = ['ebtable_broute']
+        assert lsmod_info['submodules'] == submodules
+        assert lsmod_info == {
+            'name': "bridge",
+            'size': 110862,
+            'used': 1,
+            'submodules': submodules
+        }
+
+
+if __name__ == '__main__':
+    unittest.main()