FOGL-7910 improvements in finding C plugins with python utility

python/fledge/services/core/api/utils.py

@@ -4,25 +4,25 @@
 # See: http://fledge-iot.readthedocs.io/
 # FLEDGE_END
 
-
 import subprocess
 import os
 import json
-
 from fledge.common.common import _FLEDGE_ROOT, _FLEDGE_PLUGIN_PATH
 from fledge.common.logger import FLCoreLogger
 
 _logger = FLCoreLogger().get_logger(__name__)
 _lib_path = _FLEDGE_ROOT + "/" + "plugins"
 
+C_PLUGIN_UTIL_PATH = _FLEDGE_ROOT + "/extras/C/get_plugin_info" if os.path.isdir(_FLEDGE_ROOT + "/extras/C") \
+ else _FLEDGE_ROOT + "/cmake_build/C/plugins/utils/get_plugin_info"
+
 
 def get_plugin_info(name, dir):
  try:
- arg1 = _find_c_util('get_plugin_info')
  arg2 = _find_c_lib(name, dir)
  if arg2 is None:
  raise ValueError('The plugin {} does not exist'.format(name))
- cmd_with_args = [arg1, arg2, "plugin_info"]
+ cmd_with_args = [C_PLUGIN_UTIL_PATH, arg2, "plugin_info"]
  p = subprocess.Popen(cmd_with_args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  out, err = p.communicate()
  res = out.decode("utf-8")
@@ -43,58 +43,53 @@ def get_plugin_info(name, dir):
  return jdoc
 
 
-def _find_c_lib(name, dir):
- _path = [_lib_path + "/" + dir]
+def _find_c_lib(name, installed_dir):
+ _path = [_lib_path + "/" + installed_dir]
  _path = _find_plugins_from_env(_path)
+ lib_path = None
+
  for fp in _path:
  for path, subdirs, files in os.walk(fp):
  for fname in files:
  # C-binary file
  if fname.endswith("lib{}.so".format(name)):
- return os.path.join(path, fname)
- return None
-
-
-def _find_c_util(name):
- for path, subdirs, files in os.walk(_FLEDGE_ROOT):
- for fname in files:
- # C-utility file
- if fname == name:
- return os.path.join(path, fname)
- return None
+ lib_path = os.path.join(path, fname)
+ break
+ else:
+ continue
+ break
+ return lib_path
 
 
 def find_c_plugin_libs(direction):
  libraries = []
  _path = [_lib_path]
  _path = _find_plugins_from_env(_path)
  for fp in _path:
- for root, dirs, files in os.walk(fp + "/" + direction):
- for name in dirs:
- p = os.path.join(root, name)
- for path, subdirs, f in os.walk(p):
- for fname in f:
- # C-binary file
- if fname.endswith('.so'):
- # Replace lib and .so from fname
- libraries.append((fname.replace("lib", "").replace(".so", ""), 'binary'))
- # For Hybrid plugins
- if direction == 'south' and fname.endswith('.json'):
- libraries.append((fname.replace(".json", ""), 'json'))
+ if os.path.isdir(fp + "/" + direction):
+ for name in os.listdir(fp + "/" + direction):
+ p = fp + "/" + direction + "/" + name
+ for fname in os.listdir(p):
+ if fname.endswith('.so'):
+ # Replace lib and .so from fname
+ libraries.append((fname.replace("lib", "").replace(".so", ""), 'binary'))
+ # For Hybrid plugins
+ if direction == 'south' and fname.endswith('.json'):
+ libraries.append((fname.replace(".json", ""), 'json'))
  return libraries
 
 
 def _find_plugins_from_env(_plugin_path: list) -> list:
  if _FLEDGE_PLUGIN_PATH:
  my_list = _FLEDGE_PLUGIN_PATH.split(";")
- for l in my_list:
- dir_found = os.path.isdir(l)
+ for ml in my_list:
+ dir_found = os.path.isdir(ml)
  if dir_found:
- subdirs = [dirs for x, dirs, files in os.walk(l)]
+ subdirs = [dirs for x, dirs, files in os.walk(ml)]
  if subdirs[0]:
- _plugin_path.append(l)
+ _plugin_path.append(ml)
  else:
- _logger.warning("{} subdir type not found.".format(l))
+ _logger.warning("{} subdir type not found.".format(ml))
  else:
- _logger.warning("{} dir path not found.".format(l))
+ _logger.warning("{} dir path not found.".format(ml))
  return _plugin_path

tests/unit/python/fledge/services/core/api/test_api_utils.py

@@ -15,10 +15,10 @@
 @pytest.allure.story("api", "utils")
 class TestUtils:
 
- @pytest.mark.parametrize("direction", ['south', 'north'])
+ @pytest.mark.parametrize("direction", ['south', 'north', 'filter', 'notificationDelivery', 'notificationRule'])
  def test_find_c_plugin_libs_if_empty(self, direction):
- with patch('os.walk') as mockwalk:
- mockwalk.return_value = [([], [], [])]
+ with patch('os.listdir') as mockwalk:
+ mockwalk.return_value = []
  assert [] == utils.find_c_plugin_libs(direction)
 
  @pytest.mark.parametrize("direction, plugin_name, plugin_type, libs", [
@@ -35,48 +35,42 @@ def test_find_c_plugin_libs(self, direction, plugin_name, plugin_type, libs):
 
  def test_get_plugin_info_value_error(self):
  plugin_name = 'Random'
- with patch.object(utils, '_find_c_util', return_value='plugins/utils/get_plugin_info') as patch_util:
- with patch.object(utils, '_find_c_lib', return_value=None) as patch_lib:
- with patch.object(utils._logger, 'error') as patch_logger:
- assert {} == utils.get_plugin_info(plugin_name, dir='south')
- assert 1 == patch_logger.call_count
- args = patch_logger.call_args
- assert '{} C plugin get info failed.'.format(plugin_name) == args[0][1]
- patch_lib.assert_called_once_with(plugin_name, 'south')
- patch_util.assert_called_once_with('get_plugin_info')
+ with patch.object(utils, '_find_c_lib', return_value=None) as patch_lib:
+ with patch.object(utils._logger, 'error') as patch_logger:
+ assert {} == utils.get_plugin_info(plugin_name, dir='south')
+ assert 1 == patch_logger.call_count
+ args = patch_logger.call_args
+ assert '{} C plugin get info failed.'.format(plugin_name) == args[0][1]
+ patch_lib.assert_called_once_with(plugin_name, 'south')
 
  @pytest.mark.parametrize("exc_name", [Exception, OSError, subprocess.CalledProcessError])
  def test_get_plugin_info_exception(self, exc_name):
  plugin_name = 'OMF'
  plugin_lib_path = 'fledge/plugins/north/{}/lib{}'.format(plugin_name, plugin_name)
- with patch.object(utils, '_find_c_util', return_value='plugins/utils/get_plugin_info') as patch_util:
- with patch.object(utils, '_find_c_lib', return_value=plugin_lib_path) as patch_lib:
- with patch.object(utils.subprocess, "Popen", side_effect=exc_name):
- with patch.object(utils._logger, 'error') as patch_logger:
- assert {} == utils.get_plugin_info(plugin_name, dir='south')
- assert 1 == patch_logger.call_count
- args = patch_logger.call_args
- assert '{} C plugin get info failed.'.format(plugin_name) == args[0][1]
- patch_lib.assert_called_once_with(plugin_name, 'south')
- patch_util.assert_called_once_with('get_plugin_info')
+ with patch.object(utils, '_find_c_lib', return_value=plugin_lib_path) as patch_lib:
+ with patch.object(utils.subprocess, "Popen", side_effect=exc_name):
+ with patch.object(utils._logger, 'error') as patch_logger:
+ assert {} == utils.get_plugin_info(plugin_name, dir='south')
+ assert 1 == patch_logger.call_count
+ args = patch_logger.call_args
+ assert '{} C plugin get info failed.'.format(plugin_name) == args[0][1]
+ patch_lib.assert_called_once_with(plugin_name, 'south')
 
  @patch('subprocess.Popen')
  def test_get_plugin_info(self, mock_subproc_popen):
- with patch.object(utils, '_find_c_util', return_value='plugins/utils/get_plugin_info') as patch_util:
- with patch.object(utils, '_find_c_lib', return_value='fledge/plugins/south/Random/libRandom') as patch_lib:
- process_mock = MagicMock()
- attrs = {'communicate.return_value': (b'{"name": "Random", "version": "1.0.0", "type": "south", '
- b'"interface": "1.0.0", "config": {"plugin" : '
- b'{ "description" : "Random C south plugin", "type" : "string", '
- b'"default" : "Random" }, "asset" : { "description" : '
- b'"Asset name", "type" : "string", '
- b'"default" : "Random" } } }\n', 'error')}
- process_mock.configure_mock(**attrs)
- mock_subproc_popen.return_value = process_mock
- j = utils.get_plugin_info('Random', dir='south')
- assert {'name': 'Random', 'type': 'south', 'version': '1.0.0', 'interface': '1.0.0',
- 'config': {'plugin': {'description': 'Random C south plugin', 'type': 'string',
- 'default': 'Random'},
- 'asset': {'description': 'Asset name', 'type': 'string', 'default': 'Random'}}} == j
- patch_lib.assert_called_once_with('Random', 'south')
- patch_util.assert_called_once_with('get_plugin_info')
+ with patch.object(utils, '_find_c_lib', return_value='fledge/plugins/south/Random/libRandom') as patch_lib:
+ process_mock = MagicMock()
+ attrs = {'communicate.return_value': (b'{"name": "Random", "version": "1.0.0", "type": "south", '
+ b'"interface": "1.0.0", "config": {"plugin" : '
+ b'{ "description" : "Random C south plugin", "type" : "string", '
+ b'"default" : "Random" }, "asset" : { "description" : '
+ b'"Asset name", "type" : "string", '
+ b'"default" : "Random" } } }\n', 'error')}
+ process_mock.configure_mock(**attrs)
+ mock_subproc_popen.return_value = process_mock
+ j = utils.get_plugin_info('Random', dir='south')
+ assert {'name': 'Random', 'type': 'south', 'version': '1.0.0', 'interface': '1.0.0',
+ 'config': {'plugin': {'description': 'Random C south plugin', 'type': 'string',
+ 'default': 'Random'},
+ 'asset': {'description': 'Asset name', 'type': 'string', 'default': 'Random'}}} == j
+ patch_lib.assert_called_once_with('Random', 'south')