Skip to content

Commit

Permalink
Merge pull request #1115 from fledge-iot/FOGL-7910
Browse files Browse the repository at this point in the history
  • Loading branch information
@ashish-jabble
ashish-jabble authored Jul 18, 2023
2 parents cbe59e2 + 69c9a6d commit 8eadda1
Show file tree
Hide file tree
Showing 2 changed files with 64 additions and 75 deletions.
65 changes: 30 additions & 35 deletions python/fledge/services/core/api/utils.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand All @@ -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
74 changes: 34 additions & 40 deletions tests/unit/python/fledge/services/core/api/test_api_utils.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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", [
Expand All @@ -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')

0 comments on commit 8eadda1

Please sign in to comment.