VimPathFinder - Code Metrics - Inspection of "[WIP] catch requests from invalid thread" - neovim/python-client - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#255)

by Björn

created 2017-11-08 17:41 UTC

VimPathFinder A

↳ Parent: path_hook()

Complexity

Total Complexity

Size/Duplication

Total Lines	14
Duplicated Lines	0 %

Test Coverage

Coverage

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
c	1
b	0
f	0
dl	0
loc	14
ccs	0
cts	9
cp	0
rs	10
wmc	3

"""Legacy python/python3-vim emulation."""
import imp
import io
import logging
import os
import sys

from .decorators import plugin, rpc_export
from ..api import Nvim, walk
from ..msgpack_rpc import ErrorResponse
from ..util import format_exc_skip

__all__ = ('ScriptHost',)


logger = logging.getLogger(__name__)
debug, info, warn = (logger.debug, logger.info, logger.warn,)

IS_PYTHON3 = sys.version_info >= (3, 0)

if IS_PYTHON3:
    basestring = str

    if sys.version_info >= (3, 4):
        from importlib.machinery import PathFinder


@plugin
class ScriptHost(object):

    """Provides an environment for running python plugins created for Vim."""

    def __init__(self, nvim):
        """Initialize the legacy python-vim environment."""
        self.setup(nvim)
        # context where all code will run
        self.module = imp.new_module('__main__')
        nvim.script_context = self.module
        # it seems some plugins assume 'sys' is already imported, so do it now
        exec('import sys', self.module.__dict__)

        self.legacy_vim = LegacyVim.from_nvim(nvim)
        sys.modules['vim'] = self.legacy_vim

    def setup(self, nvim):
        """Setup import hooks and global streams.

        This will add import hooks for importing modules from runtime
        directories and patch the sys module so 'print' calls will be
        forwarded to Nvim.
        """
        self.nvim = nvim
        info('install import hook/path')
        self.hook = path_hook(nvim)
        sys.path_hooks.append(self.hook)
        nvim.VIM_SPECIAL_PATH = '_vim_path_'
        sys.path.append(nvim.VIM_SPECIAL_PATH)
        info('redirect sys.stdout and sys.stderr')
        self.saved_stdout = sys.stdout
        self.saved_stderr = sys.stderr
        sys.stdout = RedirectStream(lambda data: nvim.out_write(data))

        sys.stderr = RedirectStream(lambda data: nvim.err_write(data))


    def teardown(self):
        """Restore state modified from the `setup` call."""
        nvim = self.nvim
        info('uninstall import hook/path')
        sys.path.remove(nvim.VIM_SPECIAL_PATH)
        sys.path_hooks.remove(self.hook)
        info('restore sys.stdout and sys.stderr')
        sys.stdout = self.saved_stdout
        sys.stderr = self.saved_stderr

    @rpc_export('python_execute', sync=True)
    def python_execute(self, script, range_start, range_stop):
        """Handle the `python` ex command."""
        self._set_current_range(range_start, range_stop)
        try:
            exec(script, self.module.__dict__)

        except Exception:
            raise ErrorResponse(format_exc_skip(1))

    @rpc_export('python_execute_file', sync=True)
    def python_execute_file(self, file_path, range_start, range_stop):
        """Handle the `pyfile` ex command."""
        self._set_current_range(range_start, range_stop)
        with open(file_path) as f:
            script = compile(f.read(), file_path, 'exec')
            try:
                exec(script, self.module.__dict__)

            except Exception:
                raise ErrorResponse(format_exc_skip(1))

    @rpc_export('python_do_range', sync=True)
    def python_do_range(self, start, stop, code):
        """Handle the `pydo` ex command."""
        self._set_current_range(start, stop)
        nvim = self.nvim
        start -= 1
        fname = '_vim_pydo'

        # define the function
        function_def = 'def %s(line, linenr):\n %s' % (fname, code,)
        exec(function_def, self.module.__dict__)

        # get the function
        function = self.module.__dict__[fname]
        while start < stop:
            # Process batches of 5000 to avoid the overhead of making multiple
            # API calls for every line. Assuming an average line length of 100
            # bytes, approximately 488 kilobytes will be transferred per batch,
            # which can be done very quickly in a single API call.
            sstart = start
            sstop = min(start + 5000, stop)
            lines = nvim.current.buffer.api.get_lines(sstart, sstop, True)

            exception = None
            newlines = []
            linenr = sstart + 1
            for i, line in enumerate(lines):

                result = function(line, linenr)
                if result is None:
                    # Update earlier lines, and skip to the next
                    if newlines:
                        end = sstart + len(newlines) - 1
                        nvim.current.buffer.api.set_lines(sstart, end,
                                                          True, newlines)
                    sstart += len(newlines) + 1
                    newlines = []
                    pass

                elif isinstance(result, basestring):
                    newlines.append(result)
                else:
                    exception = TypeError('pydo should return a string ' +
                                          'or None, found %s instead'
                                          % result.__class__.__name__)
                    break
                linenr += 1

            start = sstop
            if newlines:
                end = sstart + len(newlines)
                nvim.current.buffer.api.set_lines(sstart, end, True, newlines)
            if exception:
                raise exception

        # delete the function
        del self.module.__dict__[fname]

    @rpc_export('python_eval', sync=True)
    def python_eval(self, expr):
        """Handle the `pyeval` vim function."""
        return eval(expr, self.module.__dict__)


    def _set_current_range(self, start, stop):
        current = self.legacy_vim.current
        current.range = current.buffer.range(start, stop)


class RedirectStream(io.IOBase):
    def __init__(self, redirect_handler):
        self.redirect_handler = redirect_handler

    def write(self, data):
        self.redirect_handler(data)

    def writelines(self, seq):
        self.redirect_handler('\n'.join(seq))


if IS_PYTHON3:
    num_types = (int, float)
else:
    num_types = (int, long, float)



def num_to_str(obj):
    if isinstance(obj, num_types):
        return str(obj)
    else:
        return obj


class LegacyVim(Nvim):
    def eval(self, expr):
        obj = self.request("vim_eval", expr)
        return walk(num_to_str, obj)


# This was copied/adapted from nvim-python help
def path_hook(nvim):
    def _get_paths():
        if nvim._thread_invalid():
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent
            return []
        return discover_runtime_directories(nvim)

    def _find_module(fullname, oldtail, path):
        idx = oldtail.find('.')
        if idx > 0:
            name = oldtail[:idx]
            tail = oldtail[idx + 1:]
            fmr = imp.find_module(name, path)
            module = imp.find_module(fullname[:-len(oldtail)] + name, *fmr)
            return _find_module(fullname, tail, module.__path__)

        else:
            return imp.find_module(fullname, path)

    class VimModuleLoader(object):
        def __init__(self, module):
            self.module = module

        def load_module(self, fullname, path=None):

            # Check sys.modules, required for reload (see PEP302).
            if fullname in sys.modules:
                return sys.modules[fullname]
            return imp.load_module(fullname, *self.module)

    class VimPathFinder(object):
        @staticmethod
        def find_module(fullname, path=None):
            """Method for Python 2.7 and 3.3."""
            try:
                return VimModuleLoader(
                    _find_module(fullname, fullname, path or _get_paths()))
            except ImportError:
                return None

        @staticmethod
        def find_spec(fullname, path=None, target=None):
            """Method for Python 3.4+."""
            return PathFinder.find_spec(fullname, path or _get_paths(), target)

    def hook(path):
        if path == nvim.VIM_SPECIAL_PATH:
            return VimPathFinder
        else:
            raise ImportError

    return hook


def discover_runtime_directories(nvim):
    rv = []
    for path in nvim.list_runtime_paths():
        if not os.path.exists(path):
            continue
        path1 = os.path.join(path, 'pythonx')
        if IS_PYTHON3:
            path2 = os.path.join(path, 'python3')
        else:
            path2 = os.path.join(path, 'python2')
        if os.path.exists(path1):
            rv.append(path1)
        if os.path.exists(path2):
            rv.append(path2)
    return rv


1		"""Legacy python/python3-vim emulation."""
2	5	import imp
3	5	import io
4	5	import logging
5	5	import os
6	5	import sys
7
8	5	from .decorators import plugin, rpc_export
9	5	from ..api import Nvim, walk
10	5	from ..msgpack_rpc import ErrorResponse
11	5	from ..util import format_exc_skip
12
13	5	__all__ = ('ScriptHost',)
14
15
16	5	logger = logging.getLogger(__name__)
17	5	debug, info, warn = (logger.debug, logger.info, logger.warn,)
18
19	5	IS_PYTHON3 = sys.version_info >= (3, 0)
20
21	5	if IS_PYTHON3:
22	3	basestring = str
23
24	3	if sys.version_info >= (3, 4):
25	2	from importlib.machinery import PathFinder
26
27
28	5	@plugin
29	5	class ScriptHost(object):
30
31		"""Provides an environment for running python plugins created for Vim."""
32
33	5	def __init__(self, nvim):
34		"""Initialize the legacy python-vim environment."""
35		self.setup(nvim)
36		# context where all code will run
37		self.module = imp.new_module('__main__')
38		nvim.script_context = self.module
39		# it seems some plugins assume 'sys' is already imported, so do it now
40		exec('import sys', self.module.__dict__)
		0 ignored issues – show Coding Style Security introduced 2016-03-06 13:35 UTC by Report Bug Copy Issue Report The use of `exec` is discouraged. Execution of dynamic code might introduce security vulnerabilities. It is generally recommended to use this feature with care and only when necessary. Loading history...
41		self.legacy_vim = LegacyVim.from_nvim(nvim)
42		sys.modules['vim'] = self.legacy_vim
43
44	5	def setup(self, nvim):
45		"""Setup import hooks and global streams.
46
47		This will add import hooks for importing modules from runtime
48		directories and patch the sys module so 'print' calls will be
49		forwarded to Nvim.
50		"""
51		self.nvim = nvim
52		info('install import hook/path')
53		self.hook = path_hook(nvim)
54		sys.path_hooks.append(self.hook)
55		nvim.VIM_SPECIAL_PATH = '_vim_path_'
56		sys.path.append(nvim.VIM_SPECIAL_PATH)
57		info('redirect sys.stdout and sys.stderr')
58		self.saved_stdout = sys.stdout
59		self.saved_stderr = sys.stderr
60		sys.stdout = RedirectStream(lambda data: nvim.out_write(data))
		0 ignored issues – show Unused Code introduced 2016-03-06 13:35 UTC by Report Bug Copy Issue Report This lambda might be unnecessary. Loading history...
61		sys.stderr = RedirectStream(lambda data: nvim.err_write(data))
		0 ignored issues – show Unused Code introduced 2016-03-06 13:35 UTC by Report Bug Copy Issue Report This lambda might be unnecessary. Loading history...
62
63	5	def teardown(self):
64		"""Restore state modified from the `setup` call."""
65		nvim = self.nvim
66		info('uninstall import hook/path')
67		sys.path.remove(nvim.VIM_SPECIAL_PATH)
68		sys.path_hooks.remove(self.hook)
69		info('restore sys.stdout and sys.stderr')
70		sys.stdout = self.saved_stdout
71		sys.stderr = self.saved_stderr
72
73	5	@rpc_export('python_execute', sync=True)
74		def python_execute(self, script, range_start, range_stop):
75		"""Handle the `python` ex command."""
76		self._set_current_range(range_start, range_stop)
77		try:
78		exec(script, self.module.__dict__)
		0 ignored issues – show Coding Style Security introduced 2017-04-16 08:05 UTC by Report Bug Copy Issue Report The use of `exec` is discouraged. Execution of dynamic code might introduce security vulnerabilities. It is generally recommended to use this feature with care and only when necessary. Loading history...
79		except Exception:
80		raise ErrorResponse(format_exc_skip(1))
81
82	5	@rpc_export('python_execute_file', sync=True)
83		def python_execute_file(self, file_path, range_start, range_stop):
84		"""Handle the `pyfile` ex command."""
85		self._set_current_range(range_start, range_stop)
86		with open(file_path) as f:
87		script = compile(f.read(), file_path, 'exec')
88		try:
89		exec(script, self.module.__dict__)
		0 ignored issues – show Coding Style Security introduced 2016-06-08 19:50 UTC by Report Bug Copy Issue Report The use of `exec` is discouraged. Execution of dynamic code might introduce security vulnerabilities. It is generally recommended to use this feature with care and only when necessary. Loading history...
90		except Exception:
91		raise ErrorResponse(format_exc_skip(1))
92
93	5	@rpc_export('python_do_range', sync=True)
94		def python_do_range(self, start, stop, code):
95		"""Handle the `pydo` ex command."""
96		self._set_current_range(start, stop)
97		nvim = self.nvim
98		start -= 1
99		fname = '_vim_pydo'
100
101		# define the function
102		function_def = 'def %s(line, linenr):\n %s' % (fname, code,)
103		exec(function_def, self.module.__dict__)
		0 ignored issues – show Coding Style Security introduced 2016-03-06 13:35 UTC by Report Bug Copy Issue Report The use of `exec` is discouraged. Execution of dynamic code might introduce security vulnerabilities. It is generally recommended to use this feature with care and only when necessary. Loading history...
104		# get the function
105		function = self.module.__dict__[fname]
106		while start < stop:
107		# Process batches of 5000 to avoid the overhead of making multiple
108		# API calls for every line. Assuming an average line length of 100
109		# bytes, approximately 488 kilobytes will be transferred per batch,
110		# which can be done very quickly in a single API call.
111		sstart = start
112		sstop = min(start + 5000, stop)
113		lines = nvim.current.buffer.api.get_lines(sstart, sstop, True)
114
115		exception = None
116		newlines = []
117		linenr = sstart + 1
118		for i, line in enumerate(lines):
		0 ignored issues – show Unused Code introduced 2016-03-06 13:35 UTC by Report Bug Copy Issue Report The variable `i` seems to be unused. Loading history...
119		result = function(line, linenr)
120		if result is None:
121		# Update earlier lines, and skip to the next
122		if newlines:
123		end = sstart + len(newlines) - 1
124		nvim.current.buffer.api.set_lines(sstart, end,
125		True, newlines)
126		sstart += len(newlines) + 1
127		newlines = []
128		pass
		0 ignored issues – show Unused Code introduced 2016-03-06 13:35 UTC by Report Bug Copy Issue Report Unnecessary pass statement Loading history...
129		elif isinstance(result, basestring):
130		newlines.append(result)
131		else:
132		exception = TypeError('pydo should return a string ' +
133		'or None, found %s instead'
134		% result.__class__.__name__)
135		break
136		linenr += 1
137
138		start = sstop
139		if newlines:
140		end = sstart + len(newlines)
141		nvim.current.buffer.api.set_lines(sstart, end, True, newlines)
142		if exception:
143		raise exception
		0 ignored issues – show Bug introduced 2016-03-06 13:35 UTC by Report Bug Copy Issue Report Raising NoneType while only classes or instances are allowed Loading history...
144		# delete the function
145		del self.module.__dict__[fname]
146
147	5	@rpc_export('python_eval', sync=True)
148		def python_eval(self, expr):
149		"""Handle the `pyeval` vim function."""
150		return eval(expr, self.module.__dict__)
		0 ignored issues – show Security Best Practice introduced 2016-03-06 13:35 UTC by Report Bug Copy Issue Report `eval` should generally only be used if absolutely necessary. The usage of `eval` might allow for executing arbitrary code if a user manages to inject dynamic input. Please use this language feature with care and only when you are sure of the input. Loading history...
151
152	5	def _set_current_range(self, start, stop):
153		current = self.legacy_vim.current
154		current.range = current.buffer.range(start, stop)
155
156
157	5	class RedirectStream(io.IOBase):
158	5	def __init__(self, redirect_handler):
159		self.redirect_handler = redirect_handler
160
161	5	def write(self, data):
162		self.redirect_handler(data)
163
164	5	def writelines(self, seq):
165		self.redirect_handler('\n'.join(seq))
166
167
168	5	if IS_PYTHON3:
169	3	num_types = (int, float)
170		else:
171	2	num_types = (int, long, float)
		0 ignored issues – show Comprehensibility Best Practice introduced 2016-04-08 13:03 UTC by Report Bug Copy Issue Report Undefined variable 'long' Loading history...
172
173
174	5	def num_to_str(obj):
175		if isinstance(obj, num_types):
176		return str(obj)
177		else:
178		return obj
179
180
181	5	class LegacyVim(Nvim):
182	5	def eval(self, expr):
183		obj = self.request("vim_eval", expr)
184		return walk(num_to_str, obj)
185
186
187		# This was copied/adapted from nvim-python help
188	5	def path_hook(nvim):
189		def _get_paths():
190		if nvim._thread_invalid():
		0 ignored issues – show Coding Style Best Practice introduced 2017-04-16 08:05 UTC by Report Bug Copy Issue Report It seems like `_thread_invalid` was declared protected and should not be accessed from this context. Prefixing a member variable `_` is usually regarded as the equivalent of declaring it with protected visibility that exists in other languages. Consequentially, such a member should only be accessed from the same class or a child class: class MyParent: def __init__(self): self._x = 1; self.y = 2; class MyChild(MyParent): def some_method(self): return self._x # Ok, since accessed from a child class class AnotherClass: def some_method(self, instance_of_my_child): return instance_of_my_child._x # Would be flagged as AnotherClass is not # a child class of MyParent Loading history...
191		return []
192		return discover_runtime_directories(nvim)
193
194		def _find_module(fullname, oldtail, path):
195		idx = oldtail.find('.')
196		if idx > 0:
197		name = oldtail[:idx]
198		tail = oldtail[idx + 1:]
199		fmr = imp.find_module(name, path)
200		module = imp.find_module(fullname[:-len(oldtail)] + name, *fmr)
201		return _find_module(fullname, tail, module.__path__)
		0 ignored issues – show Bug introduced 2016-03-06 13:35 UTC by Report Bug Copy Issue Report The Instance of `tuple` does not seem to have a member named `__path__`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
202		else:
203		return imp.find_module(fullname, path)
204
205		class VimModuleLoader(object):
206		def __init__(self, module):
207		self.module = module
208
209		def load_module(self, fullname, path=None):
		0 ignored issues – show Unused Code introduced 2016-03-06 13:35 UTC by Report Bug Copy Issue Report The argument `path` seems to be unused. Loading history...
210		# Check sys.modules, required for reload (see PEP302).
211		if fullname in sys.modules:
212		return sys.modules[fullname]
213		return imp.load_module(fullname, *self.module)
214
215		class VimPathFinder(object):
216		@staticmethod
217		def find_module(fullname, path=None):
218		"""Method for Python 2.7 and 3.3."""
219		try:
220		return VimModuleLoader(
221		_find_module(fullname, fullname, path or _get_paths()))
222		except ImportError:
223		return None
224
225		@staticmethod
226		def find_spec(fullname, path=None, target=None):
227		"""Method for Python 3.4+."""
228		return PathFinder.find_spec(fullname, path or _get_paths(), target)
229
230		def hook(path):
231		if path == nvim.VIM_SPECIAL_PATH:
232		return VimPathFinder
233		else:
234		raise ImportError
235
236		return hook
237	5
238
239		def discover_runtime_directories(nvim):
240		rv = []
241		for path in nvim.list_runtime_paths():
242		if not os.path.exists(path):
243		continue
244		path1 = os.path.join(path, 'pythonx')
245		if IS_PYTHON3:
246		path2 = os.path.join(path, 'python3')
247		else:
248		path2 = os.path.join(path, 'python2')
249		if os.path.exists(path1):
250		rv.append(path1)
251		if os.path.exists(path2):
252		rv.append(path2)
253		return rv
254

neovim / python-client

Pull Request — master (#255)

VimPathFinder A

Complexity

Size/Duplication

Test Coverage

Importance

Duplication Side-by-Side

Filter issues like