Use output of rpm --eval command, in case import of rpm module fails (#139)

Michal Cyprian · mcyprian · commit 20433cd4e440 · 2017-10-10T10:45:25.000+02:00
* Use output of rpm --eval command, in case import of rpm module fails Redesign utils module to prevent circular dependency Resolves #117
diff --git a/pyp2rpm/metadata_extractors.py b/pyp2rpm/metadata_extractors.py
@@ -40,6 +40,50 @@ def cut_to_length(text, length, delim):
         return text
 
 
+def get_interpreter_path(version=None):
+    """Return the executable of a specified or current version."""
+    if version and version != str(sys.version_info[0]):
+        return settings.PYTHON_INTERPRETER + version
+    else:
+        return sys.executable
+
+
+def license_from_trove(trove):
+    """Finds out license from list of trove classifiers.
+    Args:
+        trove: list of trove classifiers
+    Returns:
+        Fedora name of the package license or empty string, if no licensing
+        information is found in trove classifiers.
+    """
+    license = []
+    for classifier in trove:
+        if 'License' in classifier:
+            stripped = classifier.strip()
+            # if taken from EGG-INFO, begins with Classifier:
+            stripped = stripped[stripped.find('License'):]
+            if stripped in settings.TROVE_LICENSES:
+                license.append(settings.TROVE_LICENSES[stripped])
+    return ' and '.join(license)
+
+
+def versions_from_trove(trove):
+    """Finds out python version from list of trove classifiers.
+    Args:
+        trove: list of trove classifiers
+    Returns:
+        python version string
+    """
+    versions = set()
+    for classifier in trove:
+        if 'Programming Language :: Python ::' in classifier:
+            ver = classifier.split('::')[-1]
+            major = ver.split('.')[0].strip()
+            if major:
+                versions.add(major)
+    return sorted([v for v in versions if v.replace('.', '', 1).isdigit()])
+
+
 def pypi_metadata_extension(extraction_fce):
     """Extracts data from PyPI and merges them with data from extraction method."""
 
@@ -65,7 +109,7 @@ def inner(self, client=None):
             data_dict[data_field] = release_data.get(data_field, '')
 
         # we usually get better license representation from trove classifiers
-        data_dict["license"] = utils.license_from_trove(release_data.get('classifiers', ''))
+        data_dict["license"] = license_from_trove(release_data.get('classifiers', ''))
         data.set_from(data_dict, update=True)
         return data
     return inner
@@ -261,7 +305,7 @@ def _get_metadata(self, temp_dir):
             *settings.EXTRACT_DIST_COMMAND_ARGS + ['--stdout'])
 
         current_version = self.base_python_version or str(sys.version_info[0])
-        paths_to_attempt = (utils.get_interpreter_path(version=ver) for ver in (
+        paths_to_attempt = (get_interpreter_path(version=ver) for ver in (
             current_version,  # the version provided with `-b` option or default
             '2' if current_version == '3' else '3'  # alternative Python version
         ))
@@ -381,7 +425,7 @@ def versions_from_archive(self):
         Returns:
             (str, list) base_python_version, python_versions
         """
-        py_vers = utils.versions_from_trove(self.metadata['classifiers'])
+        py_vers = versions_from_trove(self.metadata['classifiers'])
         if self.unsupported_version in py_vers:
             py_vers.remove(self.unsupported_version)
 
@@ -528,7 +572,7 @@ def license(self):
 
     @property
     def versions_from_archive(self):
-        py_vers = utils.versions_from_trove(self.classifiers)
+        py_vers = versions_from_trove(self.classifiers)
         return (py_vers[0] if py_vers else settings.DEFAULT_PYTHON_VERSION,
                 py_vers[1:] if py_vers else [settings.DEFAULT_ADDITIONAL_VERSION])
 
diff --git a/pyp2rpm/settings.py b/pyp2rpm/settings.py
@@ -1,16 +1,12 @@
-try:
-    import rpm
-    DEFAULT_PKG_SAVE_PATH = rpm.expandMacro("%{_topdir}")
-except ImportError:
-    import os
-    DEFAULT_PKG_SAVE_PATH = os.path.expanduser('~/rpmbuild')
+from pyp2rpm import utils
 
 DEFAULT_PYTHON_VERSION = '2'
 DEFAULT_ADDITIONAL_VERSION = '3'
 DEFAULT_PKG_SOURCE = 'pypi'
 DEFAULT_METADATA_SOURCE = 'pypi'
 DEFAULT_TEMPLATE = 'fedora'
 DEFAULT_DISTRO = 'fedora'
+DEFAULT_PKG_SAVE_PATH = utils.get_default_save_path()
 KNOWN_DISTROS = ['fedora', 'mageia', 'pld']
 ARCHIVE_SUFFIXES = ['.tar', '.tgz', '.tar.gz', '.tar.bz2',
                     '.gz', '.bz2', '.xz', '.zip', '.egg', '.whl']
@@ -23,7 +19,6 @@
 PYPI_USABLE_DATA = ['description', 'summary', 'license', 'home_page', 'requires']
 PYTHON_INTERPRETER = '/usr/bin/python'
 EXTRACT_DIST_COMMAND_ARGS = ['--quiet', '--command-packages', 'pyp2rpm.command', 'extract_dist']
-CONSOLE_LOGGING = True
 
 TROVE_LICENSES = {'License :: OSI Approved :: Academic Free License (AFL)': 'AFL',
                   'License :: OSI Approved :: Apache Software License': 'ASL %(TODO: version)s',
diff --git a/pyp2rpm/utils.py b/pyp2rpm/utils.py
@@ -9,8 +9,10 @@
 import copy
 import itertools
 
-from pyp2rpm import settings
-
+try:
+    import rpm
+except ImportError:
+    rpm = None
 
 logger = logging.getLogger(__name__)
 
@@ -39,41 +41,6 @@ def __exit__(self, type, value, traceback):  # TODO handle exception
         os.chdir(self.primary_path)
 
 
-class RedirectStdStreams(object):
-    """Temporarily redirect stdout/stderr"""
-
-    def __init__(self, stdout=None, stderr=None):
-        if settings.CONSOLE_LOGGING:
-            self.enabled = False
-        else:
-            self.enabled = True
-            if isinstance(stdout, str):
-                stdout = self.stdout_descriptor = open(stdout, "w")
-            if isinstance(stderr, str):
-                stderr = self.stdout_descriptor = open(stderr, "w")
-
-            self._stdout = stdout or sys.stdout
-            self._stderr = stderr or sys.stderr
-
-    def __enter__(self):
-        if self.enabled:
-            self.old_stdout, self.old_stderr = sys.stdout, sys.stderr
-            self.old_stdout.flush()
-            self.old_stderr.flush()
-            sys.stdout, sys.stderr = self._stdout, self._stderr
-
-    def __exit__(self, exc_type, exc_value, traceback):
-        if self.enabled:
-            self._stdout.flush()
-            self._stderr.flush()
-            sys.stdout = self.old_stdout
-            sys.stderr = self.old_stderr
-            if hasattr(self, 'stdout_descriptor'):
-                self.stdout_descriptor.close()
-            if hasattr(self, 'stderr_descriptor'):
-                self.stderr_descriptor.close()
-
-
 def memoize_by_args(func):
     """Memoizes return value of a func based on args."""
     memory = {}
@@ -89,42 +56,6 @@ def memoized(*args):
     return memoized
 
 
-def license_from_trove(trove):
-    """Finds out license from list of trove classifiers.
-    Args:
-        trove: list of trove classifiers
-    Returns:
-        Fedora name of the package license or empty string, if no licensing
-        information is found in trove classifiers.
-    """
-    license = []
-    for classifier in trove:
-        if 'License' in classifier != -1:
-            stripped = classifier.strip()
-            # if taken from EGG-INFO, begins with Classifier:
-            stripped = stripped[stripped.find('License'):]
-            if stripped in settings.TROVE_LICENSES:
-                license.append(settings.TROVE_LICENSES[stripped])
-    return ' and '.join(license)
-
-
-def versions_from_trove(trove):
-    """Finds out python version from list of trove classifiers.
-    Args:
-        trove: list of trove classifiers
-    Returns:
-        python version string
-    """
-    versions = set()
-    for classifier in trove:
-        if 'Programming Language :: Python ::' in classifier:
-            ver = classifier.split('::')[-1]
-            major = ver.split('.')[0].strip()
-            if major:
-                versions.add(major)
-    return sorted([v for v in versions if v.replace('.', '', 1).isdigit()])
-
-
 def build_srpm(specfile, save_dir):
     """Builds a srpm from given specfile using rpmbuild.
     Generated srpm is stored in directory specified by save_dir.
@@ -134,7 +65,7 @@ def build_srpm(specfile, save_dir):
         save_dir: path to source and build tree
     """
     logger.info('Starting rpmbuild to build: {0} SRPM.'.format(specfile))
-    if save_dir != settings.DEFAULT_PKG_SAVE_PATH:
+    if save_dir != get_default_save_path():
         try:
             msg = subprocess.Popen(['rpmbuild',
                                     '--define', '_sourcedir {0}'.format(save_dir),
@@ -185,14 +116,6 @@ def unique_deps(deps):
     return list(k for k, _ in itertools.groupby(deps))
 
 
-def get_interpreter_path(version=None):
-    """Return the executable of a specified or current version."""
-    if version and version != str(sys.version_info[0]):
-        return settings.PYTHON_INTERPRETER + version
-    else:
-        return sys.executable
-
-
 if PY3:
     def console_to_str(s):
         try:
@@ -211,3 +134,30 @@ def c_time_locale():
     locale.setlocale(locale.LC_TIME, 'C')
     yield
     locale.setlocale(locale.LC_TIME, old_time_locale)
+
+
+def rpm_eval(macro):
+    """Get value of given macro using rpm tool"""
+    try:
+        value = subprocess.Popen(
+            ['rpm', '--eval', macro],
+            stdout=subprocess.PIPE).communicate()[0].strip()
+    except OSError:
+        logger.error('Failed to get value of {0} rpm macro'.format(
+            macro), exc_info=True)
+        value = b''
+    return console_to_str(value)
+
+
+def get_default_save_path():
+    """Return default save path for the packages"""
+    macro = '%{_topdir}'
+    if rpm:
+        save_path = rpm.expandMacro(macro)
+    else:
+        save_path = rpm_eval(macro)
+        if not save_path:
+            logger.warn("rpm tools are missing, using default save path "
+                        "~/rpmbuild/.")
+            save_path = os.path.expanduser('~/rpmbuild')
+    return save_path
diff --git a/tests/test_metadata_extractors.py b/tests/test_metadata_extractors.py
@@ -379,3 +379,15 @@ def setup_method(self, method):
     def test_extract(self, i, what, expected):
         data = self.e[i].extract_data()
         assert getattr(data, what) == expected
+
+    @pytest.mark.parametrize(("input", "expected"), [
+       ([], ""),
+       (['License :: OSI Approved :: Python Software Foundation License'],
+        'Python'),
+       (['Classifier: License :: OSI Approved :: Python Software Foundation License'],
+        'Python'),
+       (['License :: OSI Approved :: GNU General Public License v2 or later (GPLv2+)',
+         'License :: OSI Approved :: MIT License'], 'GPLv2+ and MIT'),
+    ])
+    def test_license_from_trove(self, input, expected):
+        assert me.license_from_trove(input) == expected
diff --git a/tests/test_utils.py b/tests/test_utils.py
@@ -1,7 +1,12 @@
 import pytest
+import os
+from flexmock import flexmock
+try:
+    import rpm
+except ImportError:
+    rpm = None
 
 from pyp2rpm import utils
-from pyp2rpm import settings
 
 
 class TestUtils(object):
@@ -20,16 +25,6 @@ def memoized(self, num):
 
         return num
 
-    @pytest.mark.parametrize(("input", "expected"), [
-        ([], ""),
-        (['License :: OSI Approved :: Python Software Foundation License'], 'Python'),
-        (['Classifier: License :: OSI Approved :: Python Software Foundation License'], 'Python'),
-        (['License :: OSI Approved :: GNU General Public License v2 or later (GPLv2+)',
-          'License :: OSI Approved :: MIT License'], 'GPLv2+ and MIT'),
-    ])
-    def test_license_from_trove(self, input, expected):
-        assert utils.license_from_trove(input) == expected
-
     @pytest.mark.parametrize(('input', 'expected'), [
         (['script', 'script2', 'script-0.1'], ['script', 'script2']),
         ([], []),
@@ -63,3 +58,27 @@ def test_runtime_to_build(self, input, expected):
     ])
     def test_unique_deps(self, input, expected):
         assert utils.unique_deps(input) == expected
+
+    def test_rpm_eval(self):
+        if os.path.exists('/usr/bin/rpm'):
+            assert utils.rpm_eval('macro') == 'macro'
+        else:
+            assert utils.rpm_eval('macro') == ''
+
+    def test_get_default_save_path_eval_success(self):
+        if rpm:
+            flexmock(rpm).should_receive(
+                'expandMacro').once().and_return('foo')
+        else:
+            flexmock(utils).should_receive('rpm_eval').once().and_return('foo')
+        assert utils.get_default_save_path() == 'foo'
+
+    def test_get_default_save_path_eval_fail(self):
+        if rpm:
+            flexmock(rpm).should_receive(
+                'expandMacro').once().and_return('foo')
+        else:
+            flexmock(utils).should_receive('rpm_eval').once().and_return('')
+            flexmock(os).should_receive('path.expanduser').once(
+            ).and_return('foo')
+        assert utils.get_default_save_path() == 'foo'
