feat: adds function for installing packages via pip

.gitignore

@@ -3,4 +3,5 @@
 /dist/
 /.pytest_cache/
 tests/test-results.xml
-*.egg-info/
+*.egg-info/
+__pycache__

neon_utils/packaging_utils.py

@@ -29,7 +29,8 @@
 import sys
 import re
 import importlib.util
-from typing import Tuple, Optional
+from typing import Tuple, Optional, List
+from tempfile import mkstemp
 
 import pkg_resources
 import sysconfig
@@ -265,3 +266,33 @@ def get_skill_license(): # TODO: Implement OSM version of this
  skill_data = dict_merge(default_skill, skill_data)
  skill_data["requirements"]["python"].sort()
  return dict(dict_merge(skill_data, readme_data))
+
+def install_packages_from_pip(core_module: str, packages: List[str]) -> int:
+ """
+ Install a Python package using pip
+ :param core_module: string neon core module to install dependency for
+ :param packages: List(string) list of packages to install
+ :returns: int pip exit code
+ """
+ import pip
+ _, tmp_constraints_file = mkstemp()
+ _, tmp_requirements_file = mkstemp()
+
+ install_str = " ".join(packages)
+
+ with open(tmp_constraints_file, 'w', encoding="utf8") as f:
+ constraints = '\n'.join(get_package_dependencies(core_module))
+ f.write(constraints)
+ LOG.info(f"Constraints={constraints}")
+
+ with open(tmp_requirements_file, "w", encoding="utf8") as f:
+ for pkg in packages:
+ f.write(f"{pkg}\n")
+
+ LOG.info(f"Requested installation of plugins: {install_str}")
+ pip_args = ['install', '-r', tmp_requirements_file, '-c', tmp_constraints_file]
+ result = pip.main(pip_args) if hasattr(pip, 'main') else pip._internal.main(pip_args)
+
+ if result != 0:
+ return result
+ return 0

tests/packaging_util_tests.py

@@ -29,6 +29,7 @@
 import os
 import sys
 import unittest
+from unittest.mock import patch
 
 sys.path.append(os.path.dirname(os.path.dirname(os.path.realpath(__file__))))
 from neon_utils.packaging_utils import *
@@ -145,6 +146,28 @@ def test_build_skill_spec(self):
  self.assertEqual(valid_spec, skill_spec)
 
  # TODO: Actually validate exception cases? DM
+
+ def test_install_packages_from_pip(self):
+ import pip
+ from neon_utils.packaging_utils import install_packages_from_pip
+
+ with patch.object(pip, 'main', return_value=0) as mock_method:
+ test_result = install_packages_from_pip("neon-utils", ["pip-install-test"])
+
+ args, kwargs = mock_method.call_args
+ self.assertEqual(0, test_result)
+ mock_method.assert_called_once_with(['install', '-r', args[0][2], '-c', args[0][4]])
+
+ with open(args[0][2], "r", encoding="utf8") as f:
+ line = f.readline()
+ self.assertEqual("pip-install-test\n", line)
+
+ mock_method.reset_mock()
+ test_result = install_packages_from_pip("neon-utils", ["pip-install-test", "pip-install-another"])
+
+ self.assertEqual(0, test_result)
+ mock_method.assert_called_once()
+
 
 
 if __name__ == '__main__':