pytest plugin metadata loader

Author: jscotkahat

fmf/base.py

@@ -16,16 +16,10 @@
 import fmf.utils as utils
 from io import open
 from fmf.utils import log, dict_to_yaml
+from fmf.constants import SUFFIX, IGNORED_DIRECTORIES, MAIN
+from fmf.plugin_loader import get_suffixes, get_plugin_for_file
 from pprint import pformat as pretty
 
-# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-#  Constants
-# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-
-SUFFIX = ".fmf"
-MAIN = "main" + SUFFIX
-IGNORED_DIRECTORIES = ['/dev', '/proc', '/sys']
-
 # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 #  YAML
 # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
@@ -95,7 +89,6 @@ def __init__(self, data, name=None, parent=None):
         # Track whether the data dictionary has been updated
         # (needed to prevent removing nodes with an empty dict).
         self._updated = False
-
         # Special handling for top parent
         if self.parent is None:
             self.name = "/"
@@ -453,7 +446,7 @@ def grow(self, path):
             return
         # Investigate main.fmf as the first file (for correct inheritance)
         filenames = sorted(
-            [filename for filename in filenames if filename.endswith(SUFFIX)])
+            [filename for filename in filenames if any(filter(filename.endswith, get_suffixes()))])
         try:
             filenames.insert(0, filenames.pop(filenames.index(MAIN)))
         except ValueError:
@@ -465,8 +458,12 @@ def grow(self, path):
             fullpath = os.path.abspath(os.path.join(dirpath, filename))
             log.info("Checking file {0}".format(fullpath))
             try:
-                with open(fullpath, encoding='utf-8') as datafile:
-                    data = yaml.load(datafile, Loader=YamlLoader)
+                if fullpath.endswith(SUFFIX):
+                    with open(fullpath, encoding='utf-8') as datafile:
+                        data = yaml.load(datafile, Loader=YamlLoader)
+                else:
+                    plugin = get_plugin_for_file(fullpath)
+                    data = plugin().get_data(fullpath)
             except yaml.error.YAMLError as error:
                     raise(utils.FileError("Failed to parse '{0}'.\n{1}".format(
                             fullpath, error)))

fmf/constants.py

@@ -0,0 +1,9 @@
+# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+#  Constants
+# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+SUFFIX = ".fmf"
+MAIN = "main" + SUFFIX
+IGNORED_DIRECTORIES = ['/dev', '/proc', '/sys']
+# comma separated list for plugin env var
+PLUGIN_ENV = "PLUGINS"

fmf/plugin_loader.py

@@ -0,0 +1,50 @@
+from functools import lru_cache
+import inspect
+from fmf.constants import PLUGIN_ENV, SUFFIX
+from fmf.utils import log
+import importlib
+import os
+
+
+
+class Plugin:
+    """
+    Main abstact class for FMF plugins
+    """
+    # you have to define extension list as class attribute e.g. [".py"]
+    extensions = list()
+
+    def get_data(self, filename):
+        raise NotImplemented("Define own impementation")
+
+
+@lru_cache(maxsize=1)
+def enabled_plugins():
+    plugins = os.getenv(PLUGIN_ENV).split(",") if os.getenv(PLUGIN_ENV) else []
+    plugin_list = list()
+    for item in plugins:
+        loader = importlib.machinery.SourceFileLoader(os.path.basename(item), item)
+        module = importlib.util.module_from_spec(
+            importlib.util.spec_from_loader(loader.name, loader)
+        )
+        loader.exec_module(module)
+        for name, item in inspect.getmembers(module):
+            if inspect.isclass(item) and issubclass(item, Plugin):
+                plugin_list.append(item)
+                log.info("Loaded plugin {}".format(item))
+    return plugin_list
+
+
+def get_suffixes():
+    output = [SUFFIX]
+    for item in enabled_plugins():
+        output += item.extensions
+    return output
+
+
+def get_plugin_for_file(filename):
+    extension = "." + filename.rsplit(".", 1)[1]
+    for item in enabled_plugins():
+        if extension in item.extensions:
+            log.debug("File {} parsed by by plugin {}".format(filename, item))
+            return item

fmf/plugins/pytest.py

@@ -0,0 +1,39 @@
+from fmf.plugin_loader import Plugin
+from fmf.utils import log
+from fmf_metadata.pytest_collector import collect
+from fmf_metadata.constants import PYTEST_DEFAULT_CONF
+from fmf_metadata.base import _Test, _TestCls, define_undefined
+from fmf_metadata.base import FMF
+import re
+import os
+
+_ = FMF
+
+def update_data(store_dict, func, config):
+    keys = []
+    filename = os.path.basename(func.fspath)
+    if func.cls:
+        cls = _TestCls(func.cls, filename)
+        keys.append(cls.name)
+    else:
+        cls = _TestCls(None, filename)
+    test = _Test(func)
+    # normalise test name to pytest identifier
+    test.name = re.search(
+        f".*({os.path.basename(func.function.__name__)}.*)", func.name
+    ).group(1)
+    # TODO: removed str_normalise(...) will see what happen
+    keys.append(test.name)
+    define_undefined(store_dict, keys, config, filename, cls, test)
+    return store_dict
+
+
+class Pytest(Plugin):
+    extensions = [".py"]
+
+    def get_data(self, file_name):
+        out = dict()
+        for item in collect([file_name]):
+            update_data(store_dict=out, func=item, config=PYTEST_DEFAULT_CONF)
+            log.info("Processing Item: {}".format(item))
+        return out

tests/tests_plugin/.fmf/version

@@ -0,0 +1 @@
+1

tests/tests_plugin/main.fmf

@@ -0,0 +1,5 @@
+author: Jan Scotka <jscotka@redhat.com>
+
+/pure_fmf:
+  test: ./runtest.sh
+  summary: Pure FMF test case

tests/tests_plugin/test_basic.py

@@ -0,0 +1,28 @@
+import pytest
+import unittest
+from fmf_metadata import FMF
+
+
+@FMF.tag("Tier1")
+@FMF.summary("This is basic testcase")
+def test_pass():
+    assert True
+
+
+def test_fail():
+    assert False
+
+
+@pytest.mark.skip
+def test_skip():
+    assert True
+
+
+@pytest.mark.parametrize("test_input", ["a", "b", "c"])
+def test_parametrize(test_input):
+    assert bool(test_input)
+
+
+class TestCls(unittest.TestCase):
+    def test(self):
+        self.assertTrue(True)