Merge pull request #15 from adhikasp/no-split-from-import

Prevent `from * import *` to be splitted

.gitignore

@@ -10,3 +10,6 @@ build/
 dist/
 htmlcov/
 autoflake.egg-info/
+/.coverage
+/pypi_tmp/
+/tmp.*/

AUTHORS.rst

@@ -2,6 +2,7 @@ Author
 ------
 - Steven Myint (https://github.com/myint)
 
-Patches
--------
+Contributors
+------------
 - tell-k (https://github.com/tell-k)
+- Adhika Setya Pramudita (https://github.com/adhikasp)

MANIFEST.in

@@ -5,3 +5,4 @@ include test_autoflake.py
 exclude .travis.yml
 exclude Makefile
 exclude test_acid.py
+exclude test_acid_pypi.py

Makefile

@@ -3,6 +3,7 @@ check:
 	pylint \
 		--reports=no \
 		--disable=invalid-name,no-member,too-few-public-methods,no-else-return \
+		--disable=redefined-builtin \
 		--rcfile=/dev/null \
 		autoflake.py setup.py
 	pydocstyle autoflake.py setup.py

autoflake.py

@@ -28,6 +28,7 @@
 
 import ast
 import difflib
+from collections import defaultdict
 import io
 import os
 import re
@@ -106,6 +107,17 @@ def unused_import_line_numbers(messages):
             yield message.lineno
 
 
+def unused_import_module_name(messages):
+    """Yield line number and module name of unused imports."""
+    pattern = r'\'(.+?)\''
+    for message in messages:
+        if isinstance(message, pyflakes.messages.UnusedImport):
+            module_name = re.search(pattern, str(message))
+            module_name = module_name.group()[1:-1]
+            if module_name:
+                yield (message.lineno, module_name)
+
+
 def unused_variable_line_numbers(messages):
     """Yield line numbers of unused variables."""
     for message in messages:
@@ -203,13 +215,48 @@ def multiline_statement(line, previous_line=''):
         return True
 
 
+def filter_from_import(line, unused_module):
+    """Parse and filter ``from something import a, b, c``.
+
+    Return line without unused import modules, or `pass` if all of the
+    module in import is unused.
+
+    """
+    (indentation, imports) = re.split(pattern=r'\bimport\b',
+                                      string=line, maxsplit=1)
+    base_module = re.search(pattern=r'from\s+([^ ]+)',
+                            string=indentation).group(1)
+
+    # Create an imported module list with base module name
+    # ex ``from a import b, c as d`` -> ``['a.b', 'a.c as d']``
+    imports = re.split(pattern=r',', string=imports.strip())
+    imports = [base_module + '.' + x.strip() for x in imports]
+
+    # We compare full module name (``a.module`` not `module`) to
+    # guarantee the exact same module as detected from pyflakes.
+    filtered_imports = [x.replace(base_module + '.', '')
+                        for x in imports if x not in unused_module]
+
+    # All of the import in this statement is unused
+    if not filtered_imports:
+        return get_indentation(line) + 'pass' + get_line_ending(line)
+
+    indentation += 'import '
+
+    return (
+        indentation +
+        ', '.join(sorted(filtered_imports)) +
+        get_line_ending(line))
+
+
 def break_up_import(line):
     """Return line with imports on separate lines."""
     assert '\\' not in line
     assert '(' not in line
     assert ')' not in line
     assert ';' not in line
     assert '#' not in line
+    assert not re.match(line, r'^\s*from\s')
 
     newline = get_line_ending(line)
     if not newline:
@@ -238,6 +285,9 @@ def filter_code(source, additional_imports=None,
 
     marked_import_line_numbers = frozenset(
         unused_import_line_numbers(messages))
+    marked_unused_module = defaultdict(lambda: [])
+    for line_number, module_name in unused_import_module_name(messages):
+        marked_unused_module[line_number].append(module_name)
 
     if remove_unused_variables:
         marked_variable_line_numbers = frozenset(
@@ -253,6 +303,7 @@ def filter_code(source, additional_imports=None,
         elif line_number in marked_import_line_numbers:
             yield filter_unused_import(
                 line,
+                unused_module=marked_unused_module[line_number],
                 remove_all_unused_imports=remove_all_unused_imports,
                 imports=imports,
                 previous_line=previous_line)
@@ -264,25 +315,32 @@ def filter_code(source, additional_imports=None,
         previous_line = line
 
 
-def filter_unused_import(line, remove_all_unused_imports, imports,
-                         previous_line=''):
+def filter_unused_import(line, unused_module, remove_all_unused_imports,
+                         imports, previous_line=''):
     """Return line if used, otherwise return None."""
     if multiline_import(line, previous_line):
         return line
-    elif ',' in line:
+
+    is_from_import = re.match(r'^\s*from\s', line)
+
+    if ',' in line and not is_from_import:
         return break_up_import(line)
+
+    package = extract_package_name(line)
+    if not remove_all_unused_imports and package not in imports:
+        return line
+
+    if ',' in line:
+        assert is_from_import
+        return filter_from_import(line, unused_module)
     else:
-        package = extract_package_name(line)
-        if not remove_all_unused_imports and package not in imports:
-            return line
-        else:
-            # We need to replace import with "pass" in case the import is the
-            # only line inside a block. For example,
-            # "if True:\n    import os". In such cases, if the import is
-            # removed, the block will be left hanging with no body.
-            return (get_indentation(line) +
-                    'pass' +
-                    get_line_ending(line))
+        # We need to replace import with "pass" in case the import is the
+        # only line inside a block. For example,
+        # "if True:\n    import os". In such cases, if the import is
+        # removed, the block will be left hanging with no body.
+        return (get_indentation(line) +
+                'pass' +
+                get_line_ending(line))
 
 
 def filter_unused_variable(line, previous_line=''):

test_acid.py

@@ -93,7 +93,7 @@ def run(filename, command, verbose=False, options=None):
                 try:
                     check_syntax(temp_filename, raise_error=True)
                 except (SyntaxError, TypeError,
-                        UnicodeDecodeError) as exception:
+                        UnicodeDecodeError, ValueError) as exception:
                     sys.stderr.write('autoflake broke ' + filename + '\n' +
                                      str(exception) + '\n')
                     return False
@@ -121,7 +121,7 @@ def check_syntax(filename, raise_error=False):
         try:
             compile(input_file.read(), '<string>', 'exec', dont_inherit=True)
             return True
-        except (SyntaxError, TypeError, UnicodeDecodeError):
+        except (SyntaxError, TypeError, UnicodeDecodeError, ValueError):
             if raise_error:
                 raise
             else:

test_acid_pypi.py

@@ -0,0 +1,142 @@
+#!/usr/bin/env python
+
+"""Run acid test against latest packages on PyPI."""
+
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import os
+import subprocess
+import sys
+import tarfile
+import zipfile
+
+import test_acid
+
+
+TMP_DIR = os.path.join(os.path.abspath(os.path.dirname(__file__)),
+                       'pypi_tmp')
+
+
+def latest_packages(last_hours):
+    """Return names of latest released packages on PyPI."""
+    process = subprocess.Popen(
+        ['yolk', '--latest-releases={hours}'.format(hours=last_hours)],
+        stdout=subprocess.PIPE)
+
+    for line in process.communicate()[0].decode('utf-8').split('\n'):
+        if line:
+            yield line.split()[0]
+
+
+def download_package(name, output_directory):
+    """Download package to output_directory.
+
+    Raise CalledProcessError on failure.
+
+    """
+    subprocess.check_call(['yolk', '--fetch-package={name}'.format(name=name)],
+                          cwd=output_directory)
+
+
+def extract_package(path, output_directory):
+    """Extract package at path."""
+    if path.lower().endswith('.tar.gz'):
+        try:
+            tar = tarfile.open(path)
+            tar.extractall(path=output_directory)
+            tar.close()
+            return True
+        except (tarfile.ReadError, IOError):
+            return False
+    elif path.lower().endswith('.zip'):
+        try:
+            archive = zipfile.ZipFile(path)
+            archive.extractall(path=output_directory)
+            archive.close()
+        except (zipfile.BadZipfile, IOError):
+            return False
+        return True
+
+    return False
+
+
+def main():
+    """Run main."""
+    try:
+        os.mkdir(TMP_DIR)
+    except OSError:
+        pass
+
+    args = test_acid.process_args()
+    if args.files:
+        # Copy
+        names = list(args.files)
+    else:
+        names = None
+
+    checked_packages = []
+    skipped_packages = []
+    last_hours = 1
+    while True:
+        if args.files:
+            if not names:
+                break
+        else:
+            while not names:
+                # Continually populate if user did not specify a package
+                # explicitly.
+                names = [p for p in latest_packages(last_hours)
+                         if p not in checked_packages and
+                         p not in skipped_packages]
+
+                if not names:
+                    last_hours *= 2
+
+        package_name = names.pop(0)
+        print(package_name, file=sys.stderr)
+
+        package_tmp_dir = os.path.join(TMP_DIR, package_name)
+        try:
+            os.mkdir(package_tmp_dir)
+        except OSError:
+            print('Skipping already checked package', file=sys.stderr)
+            skipped_packages.append(package_name)
+            continue
+
+        try:
+            download_package(
+                package_name,
+                output_directory=package_tmp_dir)
+        except subprocess.CalledProcessError:
+            print('yolk fetch failed', file=sys.stderr)
+            continue
+
+        for download_name in os.listdir(package_tmp_dir):
+            try:
+                if not extract_package(
+                        os.path.join(package_tmp_dir, download_name),
+                        output_directory=package_tmp_dir):
+                    print('Could not extract package', file=sys.stderr)
+                    continue
+            except UnicodeDecodeError:
+                print('Could not extract package', file=sys.stderr)
+                continue
+
+            args.files = [package_tmp_dir]
+            if test_acid.check(args):
+                checked_packages.append(package_name)
+            else:
+                return 1
+
+    if checked_packages:
+        print('\nTested packages:\n    ' + '\n    '.join(checked_packages),
+              file=sys.stderr)
+
+
+if __name__ == '__main__':
+    try:
+        sys.exit(main())
+    except KeyboardInterrupt:
+        sys.exit(1)