MDS/openide
1
0
Fork 0
mirror of https://gitflic.ru/project/openide/openide.git synced 2026-09-27 10:03:11 +07:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master'

Browse Source
This commit is contained in:
Ilya.Kazakevich
2014-09-05 20:26:50 +04:00
parent b4178ab21d 5d13f8997e
commit cf56a0ce1b
179 changed files with 5142 additions and 1507 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/helpers/pydev/django_debug.py
-124
View File
@@ -1,124 +0,0 @@
import inspect
from django_frame import DjangoTemplateFrame
from pydevd_comm import CMD_SET_BREAK
from pydevd_constants import DJANGO_SUSPEND, GetThreadId, DictContains
from pydevd_file_utils import NormFileToServer
from pydevd_breakpoints import LineBreakpoint
import pydevd_vars
import traceback
class DjangoLineBreakpoint(LineBreakpoint):
def __init__(self, file, line, condition, func_name, expression):
self.file = file
LineBreakpoint.__init__(self, line, condition, func_name, expression)
def is_triggered(self, template_frame_file, template_frame_line):
return self.file == template_frame_file and self.line == template_frame_line
def __str__(self):
return "DjangoLineBreakpoint: %s-%d" %(self.file, self.line)
def inherits(cls, *names):
if cls.__name__ in names:
return True
inherits_node = False
for base in inspect.getmro(cls):
if base.__name__ in names:
inherits_node = True
break
return inherits_node
def is_django_render_call(frame):
try:
name = frame.f_code.co_name
if name != 'render':
return False
if not DictContains(frame.f_locals, 'self'):
return False
cls = frame.f_locals['self'].__class__
inherits_node = inherits(cls, 'Node')
if not inherits_node:
return False
clsname = cls.__name__
return clsname != 'TextNode' and clsname != 'NodeList'
except:
traceback.print_exc()
return False
def is_django_context_get_call(frame):
try:
if not DictContains(frame.f_locals, 'self'):
return False
cls = frame.f_locals['self'].__class__
return inherits(cls, 'BaseContext')
except:
traceback.print_exc()
return False
def is_django_resolve_call(frame):
try:
name = frame.f_code.co_name
if name != '_resolve_lookup':
return False
if not DictContains(frame.f_locals, 'self'):
return False
cls = frame.f_locals['self'].__class__
clsname = cls.__name__
return clsname == 'Variable'
except:
traceback.print_exc()
return False
def is_django_suspended(thread):
return thread.additionalInfo.suspend_type == DJANGO_SUSPEND
def suspend_django(py_db_frame, mainDebugger, thread, frame, cmd=CMD_SET_BREAK):
frame = DjangoTemplateFrame(frame)
if frame.f_lineno is None:
return None
#try:
# if thread.additionalInfo.filename == frame.f_code.co_filename and thread.additionalInfo.line == frame.f_lineno:
# return None # don't stay twice on the same line
#except AttributeError:
# pass
pydevd_vars.addAdditionalFrameById(GetThreadId(thread), {id(frame): frame})
py_db_frame.setSuspend(thread, cmd)
thread.additionalInfo.suspend_type = DJANGO_SUSPEND
thread.additionalInfo.filename = frame.f_code.co_filename
thread.additionalInfo.line = frame.f_lineno
return frame
def find_django_render_frame(frame):
while frame is not None and not is_django_render_call(frame):
frame = frame.f_back
return frame
python/helpers/pydev/django_frame.py
-132
View File
@@ -1,132 +0,0 @@
from pydevd_file_utils import GetFileNameAndBaseFromFile
import pydev_log
import traceback
from pydevd_constants import DictContains
def read_file(filename):
f = open(filename, "r")
try:
s = f.read()
finally:
f.close()
return s
def offset_to_line_number(text, offset):
curLine = 1
curOffset = 0
while curOffset < offset:
if curOffset == len(text):
return -1
c = text[curOffset]
if c == '\n':
curLine += 1
elif c == '\r':
curLine += 1
if curOffset < len(text) and text[curOffset + 1] == '\n':
curOffset += 1
curOffset += 1
return curLine
def get_source(frame):
try:
node = frame.f_locals['self']
if hasattr(node, 'source'):
return node.source
else:
pydev_log.error_once(
"WARNING: Template path is not available. Please set TEMPLATE_DEBUG=True "
"in your settings.py to make django template breakpoints working")
return None
except:
pydev_log.debug(traceback.format_exc())
return None
def get_template_file_name(frame):
try:
source = get_source(frame)
if source is None:
pydev_log.debug("Source is None\n")
return None
fname = source[0].name
if fname == '<unknown source>':
pydev_log.debug("Source name is %s\n" % fname)
return None
else:
filename, base = GetFileNameAndBaseFromFile(fname)
return filename
except:
pydev_log.debug(traceback.format_exc())
return None
def get_template_line(frame, template_frame_file):
source = get_source(frame)
try:
return offset_to_line_number(read_file(template_frame_file), source[1][0])
except:
return None
class DjangoTemplateFrame:
def __init__(
self,
frame,
template_frame_file=None,
template_frame_line=None):
if template_frame_file is None:
template_frame_file = get_template_file_name(frame)
self.back_context = frame.f_locals['context']
self.f_code = FCode('Django Template', template_frame_file)
if template_frame_line is None:
template_frame_line = get_template_line(frame, template_frame_file)
self.f_lineno = template_frame_line
self.f_back = frame
self.f_globals = {}
self.f_locals = self.collect_context()
self.f_trace = None
def collect_context(self):
res = {}
try:
for d in self.back_context.dicts:
res.update(d)
except AttributeError:
pass
return res
def changeVariable(self, name, value):
for d in self.back_context.dicts:
if DictContains(d, name):
d[name] = value
self.f_locals[name] = value
class FCode:
def __init__(self, name, filename):
self.co_name = name
self.co_filename = filename
def is_django_exception_break_context(frame):
try:
return frame.f_code.co_name in ['_resolve_lookup', 'find_template']
except:
return False
def just_raised(trace):
if trace is None:
return False
return trace.tb_next is None
python/helpers/pydev/pkgutil_old.py
+591
View File
@@ -0,0 +1,591 @@
"""Utilities to support packages."""
# NOTE: This module must remain compatible with Python 2.3, as it is shared
# by setuptools for distribution with Python 2.3 and up.
import os
import sys
import imp
import os.path
from types import ModuleType
__all__ = [
'get_importer', 'iter_importers', 'get_loader', 'find_loader',
'walk_packages', 'iter_modules', 'get_data',
'ImpImporter', 'ImpLoader', 'read_code', 'extend_path',
]
def read_code(stream):
# This helper is needed in order for the PEP 302 emulation to
# correctly handle compiled files
import marshal
magic = stream.read(4)
if magic != imp.get_magic():
return None
stream.read(4) # Skip timestamp
return marshal.load(stream)
def simplegeneric(func):
"""Make a trivial single-dispatch generic function"""
registry = {}
def wrapper(*args, **kw):
ob = args[0]
try:
cls = ob.__class__
except AttributeError:
cls = type(ob)
try:
mro = cls.__mro__
except AttributeError:
try:
class cls(cls, object):
pass
mro = cls.__mro__[1:]
except TypeError:
mro = object, # must be an ExtensionClass or some such :(
for t in mro:
if t in registry:
return registry[t](*args, **kw)
else:
return func(*args, **kw)
try:
wrapper.__name__ = func.__name__
except (TypeError, AttributeError):
pass # Python 2.3 doesn't allow functions to be renamed
def register(typ, func=None):
if func is None:
return lambda f: register(typ, f)
registry[typ] = func
return func
wrapper.__dict__ = func.__dict__
wrapper.__doc__ = func.__doc__
wrapper.register = register
return wrapper
def walk_packages(path=None, prefix='', onerror=None):
"""Yields (module_loader, name, ispkg) for all modules recursively
on path, or, if path is None, all accessible modules.
'path' should be either None or a list of paths to look for
modules in.
'prefix' is a string to output on the front of every module name
on output.
Note that this function must import all *packages* (NOT all
modules!) on the given path, in order to access the __path__
attribute to find submodules.
'onerror' is a function which gets called with one argument (the
name of the package which was being imported) if any exception
occurs while trying to import a package. If no onerror function is
supplied, ImportErrors are caught and ignored, while all other
exceptions are propagated, terminating the search.
Examples:
# list all modules python can access
walk_packages()
# list all submodules of ctypes
walk_packages(ctypes.__path__, ctypes.__name__+'.')
"""
def seen(p, m={}):
if p in m:
return True
m[p] = True
for importer, name, ispkg in iter_modules(path, prefix):
yield importer, name, ispkg
if ispkg:
try:
__import__(name)
except ImportError:
if onerror is not None:
onerror(name)
except Exception:
if onerror is not None:
onerror(name)
else:
raise
else:
path = getattr(sys.modules[name], '__path__', None) or []
# don't traverse path items we've seen before
path = [p for p in path if not seen(p)]
for item in walk_packages(path, name+'.', onerror):
yield item
def iter_modules(path=None, prefix=''):
"""Yields (module_loader, name, ispkg) for all submodules on path,
or, if path is None, all top-level modules on sys.path.
'path' should be either None or a list of paths to look for
modules in.
'prefix' is a string to output on the front of every module name
on output.
"""
if path is None:
importers = iter_importers()
else:
importers = map(get_importer, path)
yielded = {}
for i in importers:
for name, ispkg in iter_importer_modules(i, prefix):
if name not in yielded:
yielded[name] = 1
yield i, name, ispkg
#@simplegeneric
def iter_importer_modules(importer, prefix=''):
if not hasattr(importer, 'iter_modules'):
return []
return importer.iter_modules(prefix)
iter_importer_modules = simplegeneric(iter_importer_modules)
class ImpImporter:
"""PEP 302 Importer that wraps Python's "classic" import algorithm
ImpImporter(dirname) produces a PEP 302 importer that searches that
directory. ImpImporter(None) produces a PEP 302 importer that searches
the current sys.path, plus any modules that are frozen or built-in.
Note that ImpImporter does not currently support being used by placement
on sys.meta_path.
"""
def __init__(self, path=None):
self.path = path
def find_module(self, fullname, path=None):
# Note: we ignore 'path' argument since it is only used via meta_path
subname = fullname.split(".")[-1]
if subname != fullname and self.path is None:
return None
if self.path is None:
path = None
else:
path = [os.path.realpath(self.path)]
try:
file, filename, etc = imp.find_module(subname, path)
except ImportError:
return None
return ImpLoader(fullname, file, filename, etc)
def iter_modules(self, prefix=''):
if self.path is None or not os.path.isdir(self.path):
return
yielded = {}
import inspect
try:
filenames = os.listdir(self.path)
except OSError:
# ignore unreadable directories like import does
filenames = []
filenames.sort() # handle packages before same-named modules
for fn in filenames:
modname = inspect.getmodulename(fn)
if modname=='__init__' or modname in yielded:
continue
path = os.path.join(self.path, fn)
ispkg = False
if not modname and os.path.isdir(path) and '.' not in fn:
modname = fn
try:
dircontents = os.listdir(path)
except OSError:
# ignore unreadable directories like import does
dircontents = []
for fn in dircontents:
subname = inspect.getmodulename(fn)
if subname=='__init__':
ispkg = True
break
else:
continue # not a package
if modname and '.' not in modname:
yielded[modname] = 1
yield prefix + modname, ispkg
class ImpLoader:
"""PEP 302 Loader that wraps Python's "classic" import algorithm
"""
code = source = None
def __init__(self, fullname, file, filename, etc):
self.file = file
self.filename = filename
self.fullname = fullname
self.etc = etc
def load_module(self, fullname):
self._reopen()
try:
mod = imp.load_module(fullname, self.file, self.filename, self.etc)
finally:
if self.file:
self.file.close()
# Note: we don't set __loader__ because we want the module to look
# normal; i.e. this is just a wrapper for standard import machinery
return mod
def get_data(self, pathname):
return open(pathname, "rb").read()
def _reopen(self):
if self.file and self.file.closed:
mod_type = self.etc[2]
if mod_type==imp.PY_SOURCE:
self.file = open(self.filename, 'rU')
elif mod_type in (imp.PY_COMPILED, imp.C_EXTENSION):
self.file = open(self.filename, 'rb')
def _fix_name(self, fullname):
if fullname is None:
fullname = self.fullname
elif fullname != self.fullname:
raise ImportError("Loader for module %s cannot handle "
"module %s" % (self.fullname, fullname))
return fullname
def is_package(self, fullname):
fullname = self._fix_name(fullname)
return self.etc[2]==imp.PKG_DIRECTORY
def get_code(self, fullname=None):
fullname = self._fix_name(fullname)
if self.code is None:
mod_type = self.etc[2]
if mod_type==imp.PY_SOURCE:
source = self.get_source(fullname)
self.code = compile(source, self.filename, 'exec')
elif mod_type==imp.PY_COMPILED:
self._reopen()
try:
self.code = read_code(self.file)
finally:
self.file.close()
elif mod_type==imp.PKG_DIRECTORY:
self.code = self._get_delegate().get_code()
return self.code
def get_source(self, fullname=None):
fullname = self._fix_name(fullname)
if self.source is None:
mod_type = self.etc[2]
if mod_type==imp.PY_SOURCE:
self._reopen()
try:
self.source = self.file.read()
finally:
self.file.close()
elif mod_type==imp.PY_COMPILED:
if os.path.exists(self.filename[:-1]):
f = open(self.filename[:-1], 'rU')
self.source = f.read()
f.close()
elif mod_type==imp.PKG_DIRECTORY:
self.source = self._get_delegate().get_source()
return self.source
def _get_delegate(self):
return ImpImporter(self.filename).find_module('__init__')
def get_filename(self, fullname=None):
fullname = self._fix_name(fullname)
mod_type = self.etc[2]
if self.etc[2]==imp.PKG_DIRECTORY:
return self._get_delegate().get_filename()
elif self.etc[2] in (imp.PY_SOURCE, imp.PY_COMPILED, imp.C_EXTENSION):
return self.filename
return None
try:
import zipimport
from zipimport import zipimporter
def iter_zipimport_modules(importer, prefix=''):
dirlist = zipimport._zip_directory_cache[importer.archive].keys()
dirlist.sort()
_prefix = importer.prefix
plen = len(_prefix)
yielded = {}
import inspect
for fn in dirlist:
if not fn.startswith(_prefix):
continue
fn = fn[plen:].split(os.sep)
if len(fn)==2 and fn[1].startswith('__init__.py'):
if fn[0] not in yielded:
yielded[fn[0]] = 1
yield fn[0], True
if len(fn)!=1:
continue
modname = inspect.getmodulename(fn[0])
if modname=='__init__':
continue
if modname and '.' not in modname and modname not in yielded:
yielded[modname] = 1
yield prefix + modname, False
iter_importer_modules.register(zipimporter, iter_zipimport_modules)
except ImportError:
pass
def get_importer(path_item):
"""Retrieve a PEP 302 importer for the given path item
The returned importer is cached in sys.path_importer_cache
if it was newly created by a path hook.
If there is no importer, a wrapper around the basic import
machinery is returned. This wrapper is never inserted into
the importer cache (None is inserted instead).
The cache (or part of it) can be cleared manually if a
rescan of sys.path_hooks is necessary.
"""
try:
importer = sys.path_importer_cache[path_item]
except KeyError:
for path_hook in sys.path_hooks:
try:
importer = path_hook(path_item)
break
except ImportError:
pass
else:
importer = None
sys.path_importer_cache.setdefault(path_item, importer)
if importer is None:
try:
importer = ImpImporter(path_item)
except ImportError:
importer = None
return importer
def iter_importers(fullname=""):
"""Yield PEP 302 importers for the given module name
If fullname contains a '.', the importers will be for the package
containing fullname, otherwise they will be importers for sys.meta_path,
sys.path, and Python's "classic" import machinery, in that order. If
the named module is in a package, that package is imported as a side
effect of invoking this function.
Non PEP 302 mechanisms (e.g. the Windows registry) used by the
standard import machinery to find files in alternative locations
are partially supported, but are searched AFTER sys.path. Normally,
these locations are searched BEFORE sys.path, preventing sys.path
entries from shadowing them.
For this to cause a visible difference in behaviour, there must
be a module or package name that is accessible via both sys.path
and one of the non PEP 302 file system mechanisms. In this case,
the emulation will find the former version, while the builtin
import mechanism will find the latter.
Items of the following types can be affected by this discrepancy:
imp.C_EXTENSION, imp.PY_SOURCE, imp.PY_COMPILED, imp.PKG_DIRECTORY
"""
if fullname.startswith('.'):
raise ImportError("Relative module names not supported")
if '.' in fullname:
# Get the containing package's __path__
pkg = '.'.join(fullname.split('.')[:-1])
if pkg not in sys.modules:
__import__(pkg)
path = getattr(sys.modules[pkg], '__path__', None) or []
else:
for importer in sys.meta_path:
yield importer
path = sys.path
for item in path:
yield get_importer(item)
if '.' not in fullname:
yield ImpImporter()
def get_loader(module_or_name):
"""Get a PEP 302 "loader" object for module_or_name
If the module or package is accessible via the normal import
mechanism, a wrapper around the relevant part of that machinery
is returned. Returns None if the module cannot be found or imported.
If the named module is not already imported, its containing package
(if any) is imported, in order to establish the package __path__.
This function uses iter_importers(), and is thus subject to the same
limitations regarding platform-specific special import locations such
as the Windows registry.
"""
if module_or_name in sys.modules:
module_or_name = sys.modules[module_or_name]
if isinstance(module_or_name, ModuleType):
module = module_or_name
loader = getattr(module, '__loader__', None)
if loader is not None:
return loader
fullname = module.__name__
else:
fullname = module_or_name
return find_loader(fullname)
def find_loader(fullname):
"""Find a PEP 302 "loader" object for fullname
If fullname contains dots, path must be the containing package's __path__.
Returns None if the module cannot be found or imported. This function uses
iter_importers(), and is thus subject to the same limitations regarding
platform-specific special import locations such as the Windows registry.
"""
for importer in iter_importers(fullname):
loader = importer.find_module(fullname)
if loader is not None:
return loader
return None
def extend_path(path, name):
"""Extend a package's path.
Intended use is to place the following code in a package's __init__.py:
from pkgutil import extend_path
__path__ = extend_path(__path__, __name__)
This will add to the package's __path__ all subdirectories of
directories on sys.path named after the package. This is useful
if one wants to distribute different parts of a single logical
package as multiple directories.
It also looks for *.pkg files beginning where * matches the name
argument. This feature is similar to *.pth files (see site.py),
except that it doesn't special-case lines starting with 'import'.
A *.pkg file is trusted at face value: apart from checking for
duplicates, all entries found in a *.pkg file are added to the
path, regardless of whether they are exist the filesystem. (This
is a feature.)
If the input path is not a list (as is the case for frozen
packages) it is returned unchanged. The input path is not
modified; an extended copy is returned. Items are only appended
to the copy at the end.
It is assumed that sys.path is a sequence. Items of sys.path that
are not (unicode or 8-bit) strings referring to existing
directories are ignored. Unicode items of sys.path that cause
errors when used as filenames may cause this function to raise an
exception (in line with os.path.isdir() behavior).
"""
if not isinstance(path, list):
# This could happen e.g. when this is called from inside a
# frozen package. Return the path unchanged in that case.
return path
pname = os.path.join(*name.split('.')) # Reconstitute as relative path
# Just in case os.extsep != '.'
sname = os.extsep.join(name.split('.'))
sname_pkg = sname + os.extsep + "pkg"
init_py = "__init__" + os.extsep + "py"
path = path[:] # Start with a copy of the existing path
for dir in sys.path:
if not isinstance(dir, basestring) or not os.path.isdir(dir):
continue
subdir = os.path.join(dir, pname)
# XXX This may still add duplicate entries to path on
# case-insensitive filesystems
initfile = os.path.join(subdir, init_py)
if subdir not in path and os.path.isfile(initfile):
path.append(subdir)
# XXX Is this the right thing for subpackages like zope.app?
# It looks for a file named "zope.app.pkg"
pkgfile = os.path.join(dir, sname_pkg)
if os.path.isfile(pkgfile):
try:
f = open(pkgfile)
except IOError, msg:
sys.stderr.write("Can't open %s: %s\n" %
(pkgfile, msg))
else:
for line in f:
line = line.rstrip('\n')
if not line or line.startswith('#'):
continue
path.append(line) # Don't check for existence!
f.close()
return path
def get_data(package, resource):
"""Get a resource from a package.
This is a wrapper round the PEP 302 loader get_data API. The package
argument should be the name of a package, in standard module format
(foo.bar). The resource argument should be in the form of a relative
filename, using '/' as the path separator. The parent directory name '..'
is not allowed, and nor is a rooted name (starting with a '/').
The function returns a binary string, which is the contents of the
specified resource.
For packages located in the filesystem, which have already been imported,
this is the rough equivalent of
d = os.path.dirname(sys.modules[package].__file__)
data = open(os.path.join(d, resource), 'rb').read()
If the package cannot be located or loaded, or it uses a PEP 302 loader
which does not support get_data(), then None is returned.
"""
loader = get_loader(package)
if loader is None or not hasattr(loader, 'get_data'):
return None
mod = sys.modules.get(package) or loader.load_module(package)
if mod is None or not hasattr(mod, '__file__'):
return None
# Modify the resource name to be compatible with the loader.get_data
# signature - an os.path format "filename" starting with the dirname of
# the package's __file__
parts = resource.split('/')
parts.insert(0, os.path.dirname(mod.__file__))
resource_name = os.path.join(*parts)
return loader.get_data(resource_name)
python/helpers/pydev/pluginbase.py
+453
View File
@@ -0,0 +1,453 @@
# -*- coding: utf-8 -*-
"""
pluginbase
~~~~~~~~~~
Pluginbase is a module for Python that provides a system for building
plugin based applications.
:copyright: (c) Copyright 2014 by Armin Ronacher.
:license: BSD, see LICENSE for more details.
"""
import os
import sys
from pydevd_constants import IS_PY24, IS_PY27, IS_PY3K
if IS_PY24:
from uuid_old import uuid4
else:
from uuid import uuid4
if IS_PY3K:
import pkgutil
else:
import pkgutil_old as pkgutil
import errno
try:
from hashlib import md5
except ImportError:
from md5 import md5
import threading
from types import ModuleType
from weakref import ref as weakref
PY2 = sys.version_info[0] == 2
if PY2:
text_type = unicode
string_types = (unicode, str)
from cStringIO import StringIO as NativeBytesIO
else:
text_type = str
string_types = (str,)
from io import BytesIO as NativeBytesIO
_local = threading.local()
_internalspace = ModuleType(__name__ + '._internalspace')
_internalspace.__path__ = []
sys.modules[_internalspace.__name__] = _internalspace
def get_plugin_source(module=None, stacklevel=None):
"""Returns the :class:`PluginSource` for the current module or the given
module. The module can be provided by name (in which case an import
will be attempted) or as a module object.
If no plugin source can be discovered, the return value from this method
is `None`.
This function can be very useful if additional data has been attached
to the plugin source. For instance this could allow plugins to get
access to a back reference to the application that created them.
:param module: optionally the module to locate the plugin source of.
:param stacklevel: defines how many levels up the module should search
for before it discovers the plugin frame. The
default is 0. This can be useful for writing wrappers
around this function.
"""
if module is None:
frm = sys._getframe((stacklevel or 0) + 1)
name = frm.f_globals['__name__']
glob = frm.f_globals
elif isinstance(module, string_types):
frm = sys._getframe(1)
name = module
glob = __import__(module, frm.f_globals,
frm.f_locals, ['__dict__']).__dict__
else:
name = module.__name__
glob = module.__dict__
return _discover_space(name, glob)
def _discover_space(name, globals):
try:
return _local.space_stack[-1]
except (AttributeError, IndexError):
pass
if '__pluginbase_state__' in globals:
return globals['__pluginbase_state__'].source
mod_name = globals.get('__name__')
if mod_name is not None and \
mod_name.startswith(_internalspace.__name__ + '.'):
end = mod_name.find('.', len(_internalspace.__name__) + 1)
space = sys.modules.get(mod_name[:end])
if space is not None:
return space.__pluginbase_state__.source
def _shutdown_module(mod):
members = list(mod.__dict__.items())
for key, value in members:
if key[:1] != '_':
setattr(mod, key, None)
for key, value in members:
setattr(mod, key, None)
def _to_bytes(s):
if isinstance(s, text_type):
return s.encode('utf-8')
return s
class _IntentionallyEmptyModule(ModuleType):
def __getattr__(self, name):
try:
return ModuleType.__getattr__(self, name)
except AttributeError:
if name[:2] == '__':
raise
raise RuntimeError(
'Attempted to import from a plugin base module (%s) without '
'having a plugin source activated. To solve this error '
'you have to move the import into a "with" block of the '
'associated plugin source.' % self.__name__)
class _PluginSourceModule(ModuleType):
def __init__(self, source):
modname = '%s.%s' % (_internalspace.__name__, source.spaceid)
ModuleType.__init__(self, modname)
self.__pluginbase_state__ = PluginBaseState(source)
@property
def __path__(self):
try:
ps = self.__pluginbase_state__.source
except AttributeError:
return []
return ps.searchpath + ps.base.searchpath
def _setup_base_package(module_name):
try:
mod = __import__(module_name, None, None, ['__name__'])
except ImportError:
mod = None
if '.' in module_name:
parent_mod = __import__(module_name.rsplit('.', 1)[0],
None, None, ['__name__'])
else:
parent_mod = None
if mod is None:
mod = _IntentionallyEmptyModule(module_name)
if parent_mod is not None:
setattr(parent_mod, module_name.rsplit('.', 1)[-1], mod)
sys.modules[module_name] = mod
class PluginBase(object):
"""The plugin base acts as a control object around a dummy Python
package that acts as a container for plugins. Usually each
application creates exactly one base object for all plugins.
:param package: the name of the package that acts as the plugin base.
Usually this module does not exist. Unless you know
what you are doing you should not create this module
on the file system.
:param searchpath: optionally a shared search path for modules that
will be used by all plugin sources registered.
"""
def __init__(self, package, searchpath=None):
#: the name of the dummy package.
self.package = package
if searchpath is None:
searchpath = []
#: the default search path shared by all plugins as list.
self.searchpath = searchpath
_setup_base_package(package)
def make_plugin_source(self, *args, **kwargs):
"""Creats a plugin source for this plugin base and returns it.
All parameters are forwarded to :class:`PluginSource`.
"""
return PluginSource(self, *args, **kwargs)
class PluginSource(object):
"""The plugin source is what ultimately decides where plugins are
loaded from. Plugin bases can have multiple plugin sources which act
as isolation layer. While this is not a security system it generally
is not possible for plugins from different sources to accidentally
cross talk.
Once a plugin source has been created it can be used in a ``with``
statement to change the behavior of the ``import`` statement in the
block to define which source to load the plugins from::
plugin_source = plugin_base.make_plugin_source(
searchpath=['./path/to/plugins', './path/to/more/plugins'])
with plugin_source:
from myapplication.plugins import my_plugin
:param base: the base this plugin source belongs to.
:param identifier: optionally a stable identifier. If it's not defined
a random identifier is picked. It's useful to set this
to a stable value to have consistent tracebacks
between restarts and to support pickle.
:param searchpath: a list of paths where plugins are looked for.
:param persist: optionally this can be set to `True` and the plugins
will not be cleaned up when the plugin source gets
garbage collected.
"""
# Set these here to false by default so that a completely failing
# constructor does not fuck up the destructor.
persist = False
mod = None
def __init__(self, base, identifier=None, searchpath=None,
persist=False):
#: indicates if this plugin source persists or not.
self.persist = persist
if identifier is None:
identifier = str(uuid4())
#: the identifier for this source.
self.identifier = identifier
#: A reference to the plugin base that created this source.
self.base = base
#: a list of paths where plugins are searched in.
self.searchpath = searchpath
#: The internal module name of the plugin source as it appears
#: in the :mod:`pluginsource._internalspace`.
div = None
self.spaceid = '_sp' + md5(
_to_bytes(self.base.package) + _to_bytes('|') +
_to_bytes(self.identifier)
).hexdigest()
#: a reference to the module on the internal
#: :mod:`pluginsource._internalspace`.
self.mod = _PluginSourceModule(self)
if hasattr(_internalspace, self.spaceid):
raise RuntimeError('This plugin source already exists.')
sys.modules[self.mod.__name__] = self.mod
setattr(_internalspace, self.spaceid, self.mod)
def __del__(self):
if not self.persist:
self.cleanup()
def list_plugins(self):
"""Returns a sorted list of all plugins that are available in this
plugin source. This can be useful to automatically discover plugins
that are available and is usually used together with
:meth:`load_plugin`.
"""
rv = []
for _, modname, ispkg in pkgutil.iter_modules(self.mod.__path__):
rv.append(modname)
return sorted(rv)
def load_plugin(self, name):
"""This automatically loads a plugin by the given name from the
current source and returns the module. This is a convenient
alternative to the import statement and saves you from invoking
``__import__`` or a similar function yourself.
:param name: the name of the plugin to load.
"""
if '.' in name:
raise ImportError('Plugin names cannot contain dots.')
#with self:
# return __import__(self.base.package + '.' + name,
# globals(), {}, ['__name__'])
self.__assert_not_cleaned_up()
_local.__dict__.setdefault('space_stack', []).append(self)
try:
res = __import__(self.base.package + '.' + name,
globals(), {}, ['__name__'])
return res
finally:
try:
_local.space_stack.pop()
except (AttributeError, IndexError):
pass
def open_resource(self, plugin, filename):
"""This function locates a resource inside the plugin and returns
a byte stream to the contents of it. If the resource cannot be
loaded an :exc:`IOError` will be raised. Only plugins that are
real Python packages can contain resources. Plain old Python
modules do not allow this for obvious reasons.
.. versionadded:: 0.3
:param plugin: the name of the plugin to open the resource of.
:param filename: the name of the file within the plugin to open.
"""
mod = self.load_plugin(plugin)
fn = getattr(mod, '__file__', None)
if fn is not None:
if fn.endswith(('.pyc', '.pyo')):
fn = fn[:-1]
if os.path.isfile(fn):
return open(os.path.join(os.path.dirname(fn), filename), 'rb')
buf = pkgutil.get_data(self.mod.__name__ + '.' + plugin, filename)
if buf is None:
raise IOError(errno.ENOEXITS, 'Could not find resource')
return NativeBytesIO(buf)
def cleanup(self):
"""Cleans up all loaded plugins manually. This is necessary to
call only if :attr:`persist` is enabled. Otherwise this happens
automatically when the source gets garbage collected.
"""
self.__cleanup()
def __cleanup(self, _sys=sys, _shutdown_module=_shutdown_module):
# The default parameters are necessary because this can be fired
# from the destructor and so late when the interpreter shuts down
# that these functions and modules might be gone.
if self.mod is None:
return
modname = self.mod.__name__
self.mod.__pluginbase_state__ = None
self.mod = None
try:
delattr(_internalspace, self.spaceid)
except AttributeError:
pass
prefix = modname + '.'
_sys.modules.pop(modname)
for key, value in list(_sys.modules.items()):
if not key.startswith(prefix):
continue
mod = _sys.modules.pop(key, None)
if mod is None:
continue
_shutdown_module(mod)
def __assert_not_cleaned_up(self):
if self.mod is None:
raise RuntimeError('The plugin source was already cleaned up.')
def __enter__(self):
self.__assert_not_cleaned_up()
_local.__dict__.setdefault('space_stack', []).append(self)
return self
def __exit__(self, exc_type, exc_value, tb):
try:
_local.space_stack.pop()
except (AttributeError, IndexError):
pass
def _rewrite_module_path(self, modname):
self.__assert_not_cleaned_up()
if modname == self.base.package:
return self.mod.__name__
elif modname.startswith(self.base.package + '.'):
pieces = modname.split('.')
return self.mod.__name__ + '.' + '.'.join(
pieces[self.base.package.count('.') + 1:])
class PluginBaseState(object):
__slots__ = ('_source',)
def __init__(self, source):
if source.persist:
self._source = lambda: source
else:
self._source = weakref(source)
@property
def source(self):
rv = self._source()
if rv is None:
raise AttributeError('Plugin source went away')
return rv
class _ImportHook(ModuleType):
def __init__(self, name, system_import):
ModuleType.__init__(self, name)
self._system_import = system_import
self.enabled = True
def enable(self):
"""Enables the import hook which drives the plugin base system.
This is the default.
"""
self.enabled = True
def disable(self):
"""Disables the import hook and restores the default import system
behavior. This effectively breaks pluginbase but can be useful
for testing purposes.
"""
self.enabled = False
def plugin_import(self, name, globals=None, locals=None,
fromlist=None, level=-2):
import_name = name
if self.enabled:
ref_globals = globals
if ref_globals is None:
ref_globals = sys._getframe(1).f_globals
space = _discover_space(name, ref_globals)
if space is not None:
actual_name = space._rewrite_module_path(name)
if actual_name is not None:
import_name = actual_name
if level == -2:
# fake impossible value; default value depends on version
if IS_PY24:
# the level parameter was added in version 2.5
return self._system_import(import_name, globals, locals, fromlist)
elif IS_PY3K:
# default value for level parameter in python 3
level = 0
else:
# default value for level parameter in other versions
level = -1
return self._system_import(import_name, globals, locals,
fromlist, level)
try:
import __builtin__ as builtins
except ImportError:
import builtins
import_hook = _ImportHook(__name__ + '.import_hook', builtins.__import__)
builtins.__import__ = import_hook.plugin_import
sys.modules[import_hook.__name__] = import_hook
del builtins
python/helpers/pydev/pydevd.py
+104 -50
View File
@@ -9,8 +9,9 @@ pydev_monkey_qt.patch_qt()
import traceback
from django_debug import DjangoLineBreakpoint
from pydevd_frame import add_exception_to_frame
from pydevd_plugin_utils import load_plugins, NullProxy, PluginProxy
from pydevd_frame_utils import add_exception_to_frame
import pydev_imports
from pydevd_breakpoints import * #@UnusedWildImport
import fix_getpass
@@ -118,7 +119,7 @@ DONT_TRACE = {
'_pydev_threading':1,
'_pydev_Queue':1,
'django_debug.py':1,
'django_frame.py':1,
'jinja2_debug.py':1,
'pydev_log.py':1,
'pydev_monkey.py':1 ,
'pydevd.py':1 ,
@@ -303,17 +304,15 @@ class PyDB:
self._cmd_queue = {} # the hash of Queues. Key is thread id, value is thread
self.breakpoints = {}
self.django_breakpoints = {}
self.file_to_id_to_line_breakpoint = {}
self.file_to_id_to_django_breakpoint = {}
self.file_to_id_to_plugin_breakpoint = {}
# Note: breakpoints dict should not be mutated: a copy should be created
# and later it should be assigned back (to prevent concurrency issues).
self.break_on_uncaught_exceptions = {}
self.break_on_caught_exceptions = {}
self.django_exception_break = {}
self.readyToRun = False
self._main_lock = _pydev_thread.allocate_lock()
self._lock_running_thread_ids = _pydev_thread.allocate_lock()
@@ -346,6 +345,11 @@ class PyDB:
# This attribute holds the file-> lines which have an @IgnoreException.
self.filename_to_lines_where_exceptions_are_ignored = {}
#working with plugins
self.plugins = load_plugins('pydevd_plugins')
self.plugin = NullProxy()
def haveAliveThreads(self):
for t in threadingEnumerate():
@@ -845,15 +849,22 @@ class PyDB:
if len(expression) <= 0 or expression is None or expression == "None":
expression = None
supported_type = False
if type == 'python-line':
breakpoint = LineBreakpoint(line, condition, func_name, expression)
breakpoints = self.breakpoints
file_to_id_to_breakpoint = self.file_to_id_to_line_breakpoint
elif type == 'django-line':
breakpoint = DjangoLineBreakpoint(file, line, condition, func_name, expression)
breakpoints = self.django_breakpoints
file_to_id_to_breakpoint = self.file_to_id_to_django_breakpoint
supported_type = True
else:
result = self.add_plugin_breakpoint('add_line_breakpoint', type, file, line, condition, expression, func_name)
if result is not None:
supported_type = True
breakpoint, breakpoints = result
file_to_id_to_breakpoint = self.file_to_id_to_plugin_breakpoint
else:
supported_type = False
if not supported_type:
raise NameError(type)
if DebugInfoHolder.DEBUG_TRACE_BREAKPOINTS > 0:
@@ -886,27 +897,31 @@ class PyDB:
pydev_log.error('Error removing breakpoint. Expected breakpoint_id to be an int. Found: %s' % (breakpoint_id,))
else:
file_to_id_to_breakpoint = None
if breakpoint_type == 'python-line':
breakpoints = self.breakpoints
file_to_id_to_breakpoint = self.file_to_id_to_line_breakpoint
elif breakpoint_type == 'django-line':
breakpoints = self.django_breakpoints
file_to_id_to_breakpoint = self.file_to_id_to_django_breakpoint
else:
raise NameError(breakpoint_type)
result = self.plugin.get_breakpoints(self, breakpoint_type)
if result is not None:
file_to_id_to_breakpoint = self.file_to_id_to_plugin_breakpoint
breakpoints = result
try:
id_to_pybreakpoint = file_to_id_to_breakpoint.get(file, {})
if DebugInfoHolder.DEBUG_TRACE_BREAKPOINTS > 0:
existing = id_to_pybreakpoint[breakpoint_id]
sys.stderr.write('Removed breakpoint:%s - line:%s - func_name:%s (id: %s)\n' % (
file, existing.line, existing.func_name.encode('utf-8'), breakpoint_id))
if file_to_id_to_breakpoint is None:
pydev_log.error('Error removing breakpoint. Cant handle breakpoint of type %s' % breakpoint_type)
else:
try:
id_to_pybreakpoint = file_to_id_to_breakpoint.get(file, {})
if DebugInfoHolder.DEBUG_TRACE_BREAKPOINTS > 0:
existing = id_to_pybreakpoint[breakpoint_id]
sys.stderr.write('Removed breakpoint:%s - line:%s - func_name:%s (id: %s)\n' % (
file, existing.line, existing.func_name.encode('utf-8'), breakpoint_id))
del id_to_pybreakpoint[breakpoint_id]
self.consolidate_breakpoints(file, id_to_pybreakpoint, breakpoints)
except KeyError:
pydev_log.error("Error removing breakpoint: Breakpoint id not found: %s id: %s. Available ids: %s\n" % (
file, breakpoint_id, DictKeys(id_to_pybreakpoint)))
del id_to_pybreakpoint[breakpoint_id]
self.consolidate_breakpoints(file, id_to_pybreakpoint, breakpoints)
except KeyError:
pydev_log.error("Error removing breakpoint: Breakpoint id not found: %s id: %s. Available ids: %s\n" % (
file, breakpoint_id, DictKeys(id_to_pybreakpoint)))
elif cmd_id == CMD_EVALUATE_EXPRESSION or cmd_id == CMD_EXEC_EXPRESSION:
@@ -1021,29 +1036,58 @@ class PyDB:
pass
elif cmd_id == CMD_ADD_EXCEPTION_BREAK:
exception, notify_always, notify_on_terminate = text.split('\t', 2)
exception_breakpoint = self.add_break_on_exception(
exception,
notify_always=int(notify_always) > 0,
notify_on_terminate = int(notify_on_terminate) == 1,
notify_on_first_raise_only=int(notify_always) == 2
)
if exception_breakpoint is not None:
self.update_after_exceptions_added([exception_breakpoint])
if text.find(' ') != -1:
exception, notify_always, notify_on_terminate = text.split('\t', 2)
else:
exception, notify_always, notify_on_terminate = text, 0, 0
if exception.find('-') != -1:
type, exception = exception.split('-')
else:
type = 'python'
if type == 'python':
exception_breakpoint = self.add_break_on_exception(
exception,
notify_always=int(notify_always) > 0,
notify_on_terminate = int(notify_on_terminate) == 1,
notify_on_first_raise_only=int(notify_always) == 2
)
if exception_breakpoint is not None:
self.update_after_exceptions_added([exception_breakpoint])
else:
supported_type = self.add_plugin_breakpoint('add_exception_breakpoint', type, exception)
if not supported_type:
raise NameError(type)
elif cmd_id == CMD_REMOVE_EXCEPTION_BREAK:
exception = text
try:
cp = self.break_on_uncaught_exceptions.copy()
DictPop(cp, exception, None)
self.break_on_uncaught_exceptions = cp
if exception.find('-') != -1:
type, exception = exception.split('-')
else:
type = 'python'
cp = self.break_on_caught_exceptions.copy()
DictPop(cp, exception, None)
self.break_on_caught_exceptions = cp
except:
pydev_log.debug("Error while removing exception %s"%sys.exc_info()[0]);
update_exception_hook(self)
if type == 'python':
try:
cp = self.break_on_uncaught_exceptions.copy()
DictPop(cp, exception, None)
self.break_on_uncaught_exceptions = cp
cp = self.break_on_caught_exceptions.copy()
DictPop(cp, exception, None)
self.break_on_caught_exceptions = cp
except:
pydev_log.debug("Error while removing exception %s"%sys.exc_info()[0])
update_exception_hook(self)
else:
supported_type = self.plugin.remove_exception_breakpoint(self, type, exception)
if not supported_type:
raise NameError(type)
elif cmd_id == CMD_LOAD_SOURCE:
path = text
@@ -1057,16 +1101,13 @@ class PyDB:
elif cmd_id == CMD_ADD_DJANGO_EXCEPTION_BREAK:
exception = text
self.django_exception_break[exception] = True
self.setTracingForUntracedContexts()
self.add_plugin_breakpoint('add_exception_breakpoint', 'django', exception)
elif cmd_id == CMD_REMOVE_DJANGO_EXCEPTION_BREAK:
exception = text
try:
del self.django_exception_break[exception]
except :
pass
self.plugin.remove_exception_breakpoint(self, 'django', exception)
elif cmd_id == CMD_EVALUATE_CONSOLE_EXPRESSION:
# Command which takes care for the debug console communication
@@ -1547,6 +1588,19 @@ class PyDB:
cmd = self.cmdFactory.makeExitMessage()
self.writer.addCommand(cmd)
def add_plugin_breakpoint(self, func_name, *args, **kwargs):
# add breakpoint from plugin and remember plugin to use
for plugin in self.plugins:
if hasattr(plugin, func_name):
func = getattr(plugin, func_name)
result = func(self, *args, **kwargs)
if result:
# On adding breakpoint we override plugin that will be used
# That means that we don't work with more then 1 plugin at a time
self.plugin = plugin
return result
return None
def wait_for_commands(self, globals):
thread = threading.currentThread()
import pydevd_frame_utils
python/helpers/pydev/pydevd_constants.py
+1
View File
@@ -6,6 +6,7 @@ STATE_SUSPEND = 2
PYTHON_SUSPEND = 1
DJANGO_SUSPEND = 2
JINJA2_SUSPEND = 3
try:
__setFalse = False
python/helpers/pydev/pydevd_frame.py
+84 -200
View File
@@ -3,18 +3,14 @@ import os.path
import re
import traceback # @Reimport
from django_debug import find_django_render_frame
from django_debug import is_django_render_call, is_django_suspended, suspend_django, is_django_resolve_call, is_django_context_get_call
from django_frame import DjangoTemplateFrame
from django_frame import is_django_exception_break_context
from django_frame import just_raised, get_template_file_name, get_template_line
import pydev_log
from pydevd_breakpoints import get_exception_breakpoint, get_exception_name
from pydevd_comm import CMD_ADD_DJANGO_EXCEPTION_BREAK, \
CMD_STEP_CAUGHT_EXCEPTION, CMD_STEP_RETURN, CMD_STEP_OVER, CMD_SET_BREAK, \
from pydevd_comm import CMD_STEP_CAUGHT_EXCEPTION, CMD_STEP_RETURN, CMD_STEP_OVER, CMD_SET_BREAK, \
CMD_STEP_INTO, CMD_SMART_STEP_INTO, CMD_RUN_TO_LINE, CMD_SET_NEXT_STATEMENT
from pydevd_constants import * # @UnusedWildImport
from pydevd_file_utils import GetFilenameAndBase
from pydevd_frame_utils import add_exception_to_frame
try:
from pydevd_signature import sendSignatureCallTrace
except ImportError:
@@ -55,7 +51,7 @@ class PyDBFrame:
def doWaitSuspend(self, *args, **kwargs):
self._args[0].doWaitSuspend(*args, **kwargs)
def _is_django_render_call(self, frame):
def _is_django_render_call(self, frame): # todo: cache in plugins the same way
try:
return self._cached_is_django_render_call
except:
@@ -97,22 +93,11 @@ class PyDBFrame:
flag = False
else:
try:
if mainDebugger.django_exception_break and get_exception_name(exception) in [
'VariableDoesNotExist', 'TemplateDoesNotExist', 'TemplateSyntaxError'] \
and just_raised(trace) and is_django_exception_break_context(frame):
result = mainDebugger.plugin.exception_break(mainDebugger, self, frame, event, self._args, arg)
if result:
(flag, frame) = result
render_frame = find_django_render_frame(frame)
if render_frame:
suspend_frame = suspend_django(
self, mainDebugger, thread, render_frame, CMD_ADD_DJANGO_EXCEPTION_BREAK)
if suspend_frame:
add_exception_to_frame(suspend_frame, (exception, value, trace))
flag = True
thread.additionalInfo.message = 'VariableDoesNotExist'
suspend_frame.f_back = frame
frame = suspend_frame
except :
except:
flag = False
return flag, frame
@@ -253,7 +238,8 @@ class PyDBFrame:
sendSignatureCallTrace(main_debugger, frame, filename)
is_exception_event = event == 'exception'
has_exception_breakpoints = main_debugger.break_on_caught_exceptions or main_debugger.django_exception_break
has_exception_breakpoints = main_debugger.break_on_caught_exceptions \
or main_debugger.plugin.has_exception_breaks(main_debugger)
if is_exception_event:
if has_exception_breakpoints:
@@ -293,9 +279,8 @@ class PyDBFrame:
can_skip = (step_cmd is None and stop_frame is None)\
or (step_cmd in (CMD_STEP_RETURN, CMD_STEP_OVER) and stop_frame is not frame)
check_stop_on_django_render_call = main_debugger.django_breakpoints and self._is_django_render_call(frame)
if check_stop_on_django_render_call:
can_skip = False
if can_skip:
can_skip = main_debugger.plugin.can_skip(main_debugger, frame)
# Let's check to see if we are in a function that has a breakpoint. If we don't have a breakpoint,
# we will return nothing for the next trace
@@ -334,79 +319,56 @@ class PyDBFrame:
try:
line = frame.f_lineno
flag = False
if event == 'call' and info.pydev_state != STATE_SUSPEND and check_stop_on_django_render_call:
flag, frame = self.should_stop_on_django_breakpoint(frame, event, arg)
#return is not taken into account for breakpoint hit because we'd have a double-hit in this case
#(one for the line and the other for the return).
if not flag and event != 'return' and info.pydev_state != STATE_SUSPEND and breakpoints_for_file is not None\
and DictContains(breakpoints_for_file, line):
stop_info = {}
breakpoint = None
exist_result = False
stop_info['stop'] = False
if not flag and event != 'return' and info.pydev_state != STATE_SUSPEND and breakpoints_for_file is not None \
and DictContains(breakpoints_for_file, line):
breakpoint = breakpoints_for_file[line]
stop_info['stop'] = True
new_frame = frame
if info.pydev_step_cmd == CMD_STEP_OVER and info.pydev_step_stop is frame and event in ('line', 'return'):
stop_info['stop'] = False #we don't stop on breakpoint if we have to stop by step-over (it will be processed later)
else:
result = main_debugger.plugin.get_breakpoint(main_debugger, frame, event, self._args)
if result:
exist_result = True
(flag, breakpoint, new_frame) = result
if breakpoint:
#ok, hit breakpoint, now, we have to discover if it is a conditional breakpoint
# lets do the conditional stuff here
breakpoint = breakpoints_for_file[line]
stop = True
if step_cmd == CMD_STEP_OVER and stop_frame is frame and event in ('line', 'return'):
stop = False #we don't stop on breakpoint if we have to stop by step-over (it will be processed later)
else:
condition = breakpoint.condition
if condition is not None:
if stop_info['stop'] or exist_result:
if breakpoint.condition is not None:
try:
val = eval(condition, frame.f_globals, frame.f_locals)
val = eval(breakpoint.condition, new_frame.f_globals, new_frame.f_locals)
if not val:
return self.trace_dispatch
except:
if type(condition) != type(''):
if hasattr(condition, 'encode'):
condition = condition.encode('utf-8')
msg = 'Error while evaluating expression: %s\n' % (condition,)
sys.stderr.write(msg)
traceback.print_exc()
if not main_debugger.suspend_on_breakpoint_exception:
return self.trace_dispatch
else:
stop = True
try:
additional_info = None
try:
additional_info = thread.additionalInfo
except AttributeError:
pass #that's ok, no info currently set
if additional_info is not None:
# add exception_type and stacktrace into thread additional info
etype, value, tb = sys.exc_info()
try:
error = ''.join(traceback.format_exception_only(etype, value))
stack = traceback.extract_stack(f=tb.tb_frame.f_back)
# On self.setSuspend(thread, CMD_SET_BREAK) this info will be
# sent to the client.
additional_info.conditional_breakpoint_exception = \
('Condition:\n' + condition + '\n\nError:\n' + error, stack)
finally:
etype, value, tb = None, None, None
except:
traceback.print_exc()
pydev_log.info('Error while evaluating condition \'%s\': %s\n' % (breakpoint.condition, sys.exc_info()[1]))
return self.trace_dispatch
if breakpoint.expression is not None:
try:
try:
val = eval(breakpoint.expression, frame.f_globals, frame.f_locals)
val = eval(breakpoint.expression, new_frame.f_globals, new_frame.f_locals)
except:
val = sys.exc_info()[1]
finally:
if val is not None:
thread.additionalInfo.message = val
if stop:
self.setSuspend(thread, CMD_SET_BREAK)
if stop_info['stop']:
self.setSuspend(thread, CMD_SET_BREAK)
elif flag:
result = main_debugger.plugin.suspend(main_debugger, self, thread, frame)
if result:
frame = result
# if thread has a suspend flag, we suspend with a busy wait
if info.pydev_state == STATE_SUSPEND:
@@ -419,8 +381,6 @@ class PyDBFrame:
#step handling. We stop when we hit the right frame
try:
django_stop = False
should_skip = False
if pydevd_dont_trace.should_trace_hook is not None:
if not hasattr(self, 'should_skip'):
@@ -432,34 +392,18 @@ class PyDBFrame:
should_skip = self.should_skip
if should_skip:
stop = False
stop_info['stop'] = False
elif step_cmd == CMD_STEP_INTO:
stop = event in ('line', 'return')
if is_django_suspended(thread):
#django_stop = event == 'call' and is_django_render_call(frame)
stop = stop and is_django_resolve_call(frame.f_back) and not is_django_context_get_call(frame)
if stop:
info.pydev_django_resolve_frame = 1 #we remember that we've go into python code from django rendering frame
stop_info['stop'] = event in ('line', 'return')
main_debugger.plugin.cmd_step_into(main_debugger, frame, event, self._args, stop_info)
elif step_cmd == CMD_STEP_OVER:
if is_django_suspended(thread):
django_stop = event == 'call' and self._is_django_render_call(frame)
stop = False
else:
if event == 'return' and info.pydev_django_resolve_frame is not None and is_django_resolve_call(frame.f_back):
#we return to Django suspend mode and should not stop before django rendering frame
stop_frame = info.pydev_step_stop = info.pydev_django_resolve_frame
info.pydev_django_resolve_frame = None
thread.additionalInfo.suspend_type = DJANGO_SUSPEND
stop = stop_frame is frame and event in ('line', 'return')
stop_info['stop'] = info.pydev_step_stop is frame and event in ('line', 'return')
main_debugger.plugin.cmd_step_over(main_debugger, frame, event, self._args, stop_info)
elif step_cmd == CMD_SMART_STEP_INTO:
stop = False
stop_info['stop'] = False
if info.pydev_smart_step_stop is frame:
info.pydev_func_name = None
info.pydev_smart_step_stop = None
@@ -472,13 +416,13 @@ class PyDBFrame:
curr_func_name = ''
if curr_func_name == info.pydev_func_name:
stop = True
stop_info['stop'] = True
elif step_cmd == CMD_STEP_RETURN:
stop = event == 'return' and stop_frame is frame
stop_info['stop'] = event == 'return' and stop_frame is frame
elif step_cmd == CMD_RUN_TO_LINE or step_cmd == CMD_SET_NEXT_STATEMENT:
stop = False
stop_info['stop'] = False
if event == 'line' or event == 'exception':
#Yes, we can only act on line events (weird hum?)
@@ -493,50 +437,47 @@ class PyDBFrame:
if curr_func_name == info.pydev_func_name:
line = info.pydev_next_line
if frame.f_lineno == line:
stop = True
stop_info['stop'] = True
else:
if frame.f_trace is None:
frame.f_trace = self.trace_dispatch
frame.f_lineno = line
frame.f_trace = None
stop = True
stop_info['stop'] = True
else:
stop = False
stop_info['stop'] = False
if django_stop:
frame = suspend_django(self, main_debugger, thread, frame)
if frame:
self.doWaitSuspend(thread, frame, event, arg)
elif stop:
#event is always == line or return at this point
if event == 'line':
self.setSuspend(thread, step_cmd)
self.doWaitSuspend(thread, frame, event, arg)
else: #return event
back = frame.f_back
if back is not None:
#When we get to the pydevd run function, the debugging has actually finished for the main thread
#(note that it can still go on for other threads, but for this one, we just make it finish)
#So, just setting it to None should be OK
base = basename(back.f_code.co_filename)
if base == 'pydevd.py' and back.f_code.co_name == 'run':
back = None
elif base == 'pydevd_traceproperty.py':
# We dont want to trace the return event of pydevd_traceproperty (custom property for debugging)
#if we're in a return, we want it to appear to the user in the previous frame!
return None
if back is not None:
#if we're in a return, we want it to appear to the user in the previous frame!
if True in stop_info.values():
stopped_on_plugin = main_debugger.plugin.stop(main_debugger, frame, event, self._args, stop_info, arg, step_cmd)
if DictContains(stop_info, 'stop') and stop_info['stop'] and not stopped_on_plugin:
if event == 'line':
self.setSuspend(thread, step_cmd)
self.doWaitSuspend(thread, back, event, arg)
else:
#in jython we may not have a back frame
info.pydev_step_stop = None
info.pydev_step_cmd = None
info.pydev_state = STATE_RUN
self.doWaitSuspend(thread, frame, event, arg)
else: #return event
back = frame.f_back
if back is not None:
#When we get to the pydevd run function, the debugging has actually finished for the main thread
#(note that it can still go on for other threads, but for this one, we just make it finish)
#So, just setting it to None should be OK
base = basename(back.f_code.co_filename)
if base == 'pydevd.py' and back.f_code.co_name == 'run':
back = None
elif base == 'pydevd_traceproperty.py':
# We dont want to trace the return event of pydevd_traceproperty (custom property for debugging)
#if we're in a return, we want it to appear to the user in the previous frame!
return None
if back is not None:
#if we're in a return, we want it to appear to the user in the previous frame!
self.setSuspend(thread, step_cmd)
self.doWaitSuspend(thread, back, event, arg)
else:
#in jython we may not have a back frame
info.pydev_step_stop = None
info.pydev_step_cmd = None
info.pydev_state = STATE_RUN
except:
@@ -562,61 +503,4 @@ class PyDBFrame:
except ImportError:
if hasattr(sys, 'exc_clear'): #jython does not have it
sys.exc_clear() #don't keep the traceback
pass #ok, psyco not available
def should_stop_on_django_breakpoint(self, frame, event, arg):
mainDebugger = self._args[0]
thread = self._args[3]
flag = False
template_frame_file = get_template_file_name(frame)
#pydev_log.debug("Django is rendering a template: %s\n" % template_frame_file)
django_breakpoints_for_file = mainDebugger.django_breakpoints.get(template_frame_file)
if django_breakpoints_for_file:
#pydev_log.debug("Breakpoints for that file: %s\n" % django_breakpoints_for_file)
template_frame_line = get_template_line(frame, template_frame_file)
#pydev_log.debug("Tracing template line: %d\n" % template_frame_line)
if DictContains(django_breakpoints_for_file, template_frame_line):
django_breakpoint = django_breakpoints_for_file[template_frame_line]
if django_breakpoint.is_triggered(template_frame_file, template_frame_line):
#pydev_log.debug("Breakpoint is triggered.\n")
flag = True
new_frame = DjangoTemplateFrame(
frame,
template_frame_file=template_frame_file,
template_frame_line=template_frame_line,
)
if django_breakpoint.condition is not None:
try:
val = eval(django_breakpoint.condition, new_frame.f_globals, new_frame.f_locals)
if not val:
flag = False
pydev_log.debug("Condition '%s' is evaluated to %s. Not suspending.\n" % (django_breakpoint.condition, val))
except:
pydev_log.info(
'Error while evaluating condition \'%s\': %s\n' % (django_breakpoint.condition, sys.exc_info()[1]))
if django_breakpoint.expression is not None:
try:
try:
val = eval(django_breakpoint.expression, new_frame.f_globals, new_frame.f_locals)
except:
val = sys.exc_info()[1]
finally:
if val is not None:
thread.additionalInfo.message = val
if flag:
frame = suspend_django(self, mainDebugger, thread, frame)
return flag, frame
def add_exception_to_frame(frame, exception_info):
frame.f_locals['__exception__'] = exception_info
pass #ok, psyco not available
python/helpers/pydev/pydevd_frame_utils.py
+7 -2
View File
@@ -5,7 +5,7 @@ class Frame:
f_fileno,
f_code,
f_locals,
f_globals={},
f_globals=None,
f_trace=None):
self.f_back = f_back
self.f_lineno = f_fileno
@@ -14,8 +14,13 @@ class Frame:
self.f_globals = f_globals
self.f_trace = f_trace
if self.f_globals is None:
self.f_globals = {}
class FCode:
def __init__(self, name, filename):
self.co_name = name
self.co_filename = filename
self.co_filename = filename
def add_exception_to_frame(frame, exception_info):
frame.f_locals['__exception__'] = exception_info
python/helpers/pydev/pydevd_plugin_utils.py
+46
View File
@@ -0,0 +1,46 @@
import os
import pydev_log
from pluginbase import PluginBase
def load_plugins(package):
plugin_base = PluginBase(package=package)
plugin_source = plugin_base.make_plugin_source(searchpath=[os.path.dirname(os.path.realpath(__file__)) + '/' + package], persist=True)
plugins = []
for plugin in plugin_source.list_plugins():
loaded_plugin = None
try:
loaded_plugin = plugin_source.load_plugin(plugin)
except:
pydev_log.error("Failed to load plugin %s" % plugin)
if loaded_plugin:
plugins.append(loaded_plugin)
return plugins
class NullProxy(object):
def __init__(self):
def foo(*args, **kwargs):
return None
self.null_func = foo
def __getattr__(self, name):
return self.null_func
class PluginProxy(object):
def __init__(self, plugin):
self.plugin = plugin
self.cache = {}
def __getattr__(self, name):
if not hasattr(self.cache, name):
self.cache[name] = getattr(self.plugin, name)
return self.cache[name]
python/helpers/pydev/pydevd_plugins/__init__.py
View File
python/helpers/pydev/pydevd_plugins/django_debug.py
+363
View File
@@ -0,0 +1,363 @@
from pydevd_comm import CMD_SET_BREAK, CMD_ADD_EXCEPTION_BREAK
import inspect
from pydevd_constants import DJANGO_SUSPEND, STATE_SUSPEND, GetThreadId, DictContains
from pydevd_file_utils import NormFileToServer, GetFileNameAndBaseFromFile
from pydevd_breakpoints import LineBreakpoint, get_exception_name
import pydevd_vars
import traceback
import pydev_log
from pydevd_frame_utils import add_exception_to_frame, FCode
class DjangoLineBreakpoint(LineBreakpoint):
def __init__(self, file, line, condition, func_name, expression):
self.file = file
LineBreakpoint.__init__(self, line, condition, func_name, expression)
def is_triggered(self, template_frame_file, template_frame_line):
return self.file == template_frame_file and self.line == template_frame_line
def __str__(self):
return "DjangoLineBreakpoint: %s-%d" %(self.file, self.line)
def add_line_breakpoint(pydb, type, file, line, condition, expression, func_name):
if type == 'django-line':
breakpoint = DjangoLineBreakpoint(file, line, condition, func_name, expression)
if not hasattr(pydb, 'django_breakpoints'):
pydb.django_breakpoints = {}
return breakpoint, pydb.django_breakpoints
return None
def add_exception_breakpoint(pydb, type, exception):
if type == 'django':
if not hasattr(pydb, 'django_exception_break'):
pydb.django_exception_break = {}
pydb.django_exception_break[exception] = True
pydb.setTracingForUntracedContexts()
return True
return False
def remove_exception_breakpoint(pydb, type, exception):
if type == 'django':
try:
del pydb.django_exception_break[exception]
return True
except:
pass
return False
def get_breakpoints(pydb):
if type == 'django-line':
return pydb.django_breakpoints
return None
def inherits(cls, *names):
if cls.__name__ in names:
return True
inherits_node = False
for base in inspect.getmro(cls):
if base.__name__ in names:
inherits_node = True
break
return inherits_node
def is_django_render_call(frame):
try:
name = frame.f_code.co_name
if name != 'render':
return False
if not DictContains(frame.f_locals, 'self'):
return False
cls = frame.f_locals['self'].__class__
inherits_node = inherits(cls, 'Node')
if not inherits_node:
return False
clsname = cls.__name__
return clsname != 'TextNode' and clsname != 'NodeList'
except:
traceback.print_exc()
return False
def is_django_context_get_call(frame):
try:
if not DictContains(frame.f_locals, 'self'):
return False
cls = frame.f_locals['self'].__class__
return inherits(cls, 'BaseContext')
except:
traceback.print_exc()
return False
def is_django_resolve_call(frame):
try:
name = frame.f_code.co_name
if name != '_resolve_lookup':
return False
if not DictContains(frame.f_locals, 'self'):
return False
cls = frame.f_locals['self'].__class__
clsname = cls.__name__
return clsname == 'Variable'
except:
traceback.print_exc()
return False
def is_django_suspended(thread):
return thread.additionalInfo.suspend_type == DJANGO_SUSPEND
def suspend_django(py_db_frame, mainDebugger, thread, frame, cmd=CMD_SET_BREAK):
frame = DjangoTemplateFrame(frame)
if frame.f_lineno is None:
return None
#try:
# if thread.additionalInfo.filename == frame.f_code.co_filename and thread.additionalInfo.line == frame.f_lineno:
# return None # don't stay twice on the same line
#except AttributeError:
# pass
pydevd_vars.addAdditionalFrameById(GetThreadId(thread), {id(frame): frame})
py_db_frame.setSuspend(thread, cmd)
thread.additionalInfo.suspend_type = DJANGO_SUSPEND
thread.additionalInfo.filename = frame.f_code.co_filename
thread.additionalInfo.line = frame.f_lineno
return frame
def find_django_render_frame(frame):
while frame is not None and not is_django_render_call(frame):
frame = frame.f_back
return frame
#=======================================================================================================================
# Django Frame
#=======================================================================================================================
def read_file(filename):
f = open(filename, "r")
s = f.read()
f.close()
return s
def offset_to_line_number(text, offset):
curLine = 1
curOffset = 0
while curOffset < offset:
if curOffset == len(text):
return -1
c = text[curOffset]
if c == '\n':
curLine += 1
elif c == '\r':
curLine += 1
if curOffset < len(text) and text[curOffset + 1] == '\n':
curOffset += 1
curOffset += 1
return curLine
def get_source(frame):
try:
node = frame.f_locals['self']
if hasattr(node, 'source'):
return node.source
else:
pydev_log.error_once("WARNING: Template path is not available. Please set TEMPLATE_DEBUG=True in your settings.py to make "
" django template breakpoints working")
return None
except:
pydev_log.debug(traceback.format_exc())
return None
def get_template_file_name(frame):
try:
source = get_source(frame)
if source is None:
pydev_log.debug("Source is None\n")
return None
fname = source[0].name
if fname == '<unknown source>':
pydev_log.debug("Source name is %s\n" % fname)
return None
else:
filename, base = GetFileNameAndBaseFromFile(fname)
return filename
except:
pydev_log.debug(traceback.format_exc())
return None
def get_template_line(frame):
source = get_source(frame)
file_name = get_template_file_name(frame)
try:
return offset_to_line_number(read_file(file_name), source[1][0])
except:
return None
class DjangoTemplateFrame:
def __init__(self, frame):
file_name = get_template_file_name(frame)
self.back_context = frame.f_locals['context']
self.f_code = FCode('Django Template', file_name)
self.f_lineno = get_template_line(frame)
self.f_back = frame
self.f_globals = {}
self.f_locals = self.collect_context(self.back_context)
self.f_trace = None
def collect_context(self, context):
res = {}
try:
for d in context.dicts:
for k, v in d.items():
res[k] = v
except AttributeError:
pass
return res
def changeVariable(self, name, value):
for d in self.back_context.dicts:
for k, v in d.items():
if k == name:
d[k] = value
def is_django_exception_break_context(frame):
try:
name = frame.f_code.co_name
except:
name = None
return name in ['_resolve_lookup', 'find_template']
def just_raised(trace):
if trace is None:
return False
return trace.tb_next is None
#=======================================================================================================================
# Django Step Commands
#=======================================================================================================================
def can_skip(mainDebugger, frame):
if hasattr(mainDebugger, 'django_breakpoints') and mainDebugger.django_breakpoints and is_django_render_call(frame):
filename = get_template_file_name(frame)
django_breakpoints_for_file = mainDebugger.django_breakpoints.get(filename)
if django_breakpoints_for_file:
return False
return True
def has_exception_breaks(mainDebugger):
return hasattr(mainDebugger, 'django_exception_break') and mainDebugger.django_exception_break
def cmd_step_into(mainDebugger, frame, event, args, stop_info):
mainDebugger, filename, info, thread = args
if is_django_suspended(thread):
#stop_info['django_stop'] = event == 'call' and is_django_render_call(frame)
stop_info['stop'] = stop_info['stop'] and is_django_resolve_call(frame.f_back) and not is_django_context_get_call(frame)
if stop_info['stop']:
info.pydev_django_resolve_frame = 1 #we remember that we've go into python code from django rendering frame
def cmd_step_over(mainDebugger, frame, event, args, stop_info):
mainDebugger, filename, info, thread = args
if is_django_suspended(thread):
stop_info['django_stop'] = event == 'call' and is_django_render_call(frame)
stop_info['stop'] = False
return True
else:
if event == 'return' and info.pydev_django_resolve_frame is not None and is_django_resolve_call(frame.f_back):
#we return to Django suspend mode and should not stop before django rendering frame
info.pydev_step_stop = info.pydev_django_resolve_frame
info.pydev_django_resolve_frame = None
thread.additionalInfo.suspend_type = DJANGO_SUSPEND
stop_info['stop'] = info.pydev_step_stop is frame and event in ('line', 'return')
return False
def stop(mainDebugger, frame, event, args, stop_info, arg, step_cmd):
mainDebugger, filename, info, thread = args
if DictContains(stop_info, 'django_stop') and stop_info['django_stop']:
frame = suspend_django(mainDebugger, mainDebugger, thread, frame, step_cmd)
if frame:
mainDebugger.doWaitSuspend(thread, frame, event, arg)
return True
return False
def get_breakpoint(mainDebugger, frame, event, args):
mainDebugger, filename, info, thread = args
flag = False
django_breakpoint = None
new_frame = None
if event == 'call' and info.pydev_state != STATE_SUSPEND and hasattr(mainDebugger, 'django_breakpoints') and \
mainDebugger.django_breakpoints and is_django_render_call(frame):
filename = get_template_file_name(frame)
pydev_log.debug("Django is rendering a template: %s\n" % filename)
django_breakpoints_for_file = mainDebugger.django_breakpoints.get(filename)
if django_breakpoints_for_file:
pydev_log.debug("Breakpoints for that file: %s\n" % django_breakpoints_for_file)
template_line = get_template_line(frame)
pydev_log.debug("Tracing template line: %d\n" % template_line)
if DictContains(django_breakpoints_for_file, template_line):
django_breakpoint = django_breakpoints_for_file[template_line]
flag = True
new_frame = DjangoTemplateFrame(frame)
return flag, django_breakpoint, new_frame
def suspend(mainDebugger, pydb_frame, thread, frame):
return suspend_django(pydb_frame, mainDebugger, thread, frame)
def exception_break(mainDebugger, pydb_frame, frame, event, args, arg):
mainDebugger, filename, info, thread = args
exception, value, trace = arg
if hasattr(mainDebugger, 'django_exception_break') and mainDebugger.django_exception_break and \
get_exception_name(exception) in ['VariableDoesNotExist', 'TemplateDoesNotExist', 'TemplateSyntaxError'] and \
just_raised(trace) and is_django_exception_break_context(frame):
render_frame = find_django_render_frame(frame)
if render_frame:
suspend_frame = suspend_django(pydb_frame, mainDebugger, thread, render_frame, CMD_ADD_EXCEPTION_BREAK)
if suspend_frame:
add_exception_to_frame(suspend_frame, (exception, value, trace))
flag = True
thread.additionalInfo.message = 'VariableDoesNotExist'
suspend_frame.f_back = frame
frame = suspend_frame
return (flag, frame)
return None
python/helpers/pydev/pydevd_plugins/jinja2_debug.py
+338
View File
@@ -0,0 +1,338 @@
from pydevd_breakpoints import LineBreakpoint, get_exception_name
from pydevd_constants import JINJA2_SUSPEND, GetThreadId, STATE_SUSPEND, DictContains
from pydevd_comm import CMD_SET_BREAK, CMD_STEP_OVER, CMD_ADD_EXCEPTION_BREAK
import pydevd_vars
from pydevd_file_utils import GetFileNameAndBaseFromFile
from pydevd_frame_utils import add_exception_to_frame, FCode
class Jinja2LineBreakpoint(LineBreakpoint):
def __init__(self, file, line, condition, func_name, expression):
self.file = file
LineBreakpoint.__init__(self, line, condition, func_name, expression)
def is_triggered(self, template_frame_file, template_frame_line):
return self.file == template_frame_file and self.line == template_frame_line
def __str__(self):
return "Jinja2LineBreakpoint: %s-%d" %(self.file, self.line)
def add_line_breakpoint(pydb, type, file, line, condition, func_name, expression):
result = None
if type == 'jinja2-line':
breakpoint = Jinja2LineBreakpoint(file, line, condition, func_name, expression)
if not hasattr(pydb, 'jinja2_breakpoints'):
pydb.jinja2_breakpoints = {}
result = breakpoint, pydb.jinja2_breakpoints
return result
return result
def add_exception_breakpoint(pydb, type, exception):
if type == 'jinja2':
if not hasattr(pydb, 'jinja2_exception_break'):
pydb.jinja2_exception_break = {}
pydb.jinja2_exception_break[exception] = True
pydb.setTracingForUntracedContexts()
return True
return False
def remove_exception_breakpoint(pydb, type, exception):
if type == 'jinja2':
try:
del pydb.jinja2_exception_break[exception]
return True
except:
pass
return False
def get_breakpoints(pydb, type):
if type == 'jinja2-line':
return pydb.jinja2_breakpoints
return None
def is_jinja2_render_call(frame):
try:
name = frame.f_code.co_name
if DictContains(frame.f_globals, "__jinja_template__") and name in ("root", "loop", "macro") or name.startswith("block_"):
return True
return False
except:
traceback.print_exc()
return False
def suspend_jinja2(py_db_frame, mainDebugger, thread, frame, cmd=CMD_SET_BREAK):
frame = Jinja2TemplateFrame(frame)
if frame.f_lineno is None:
return None
pydevd_vars.addAdditionalFrameById(GetThreadId(thread), {id(frame): frame})
py_db_frame.setSuspend(thread, cmd)
thread.additionalInfo.suspend_type = JINJA2_SUSPEND
thread.additionalInfo.filename = frame.f_code.co_filename
thread.additionalInfo.line = frame.f_lineno
return frame
def is_jinja2_suspended(thread):
return thread.additionalInfo.suspend_type == JINJA2_SUSPEND
def is_jinja2_context_call(frame):
return DictContains(frame.f_locals, "_Context__obj")
def is_jinja2_internal_function(frame):
return DictContains(frame.f_locals, 'self') and frame.f_locals['self'].__class__.__name__ in \
('LoopContext', 'TemplateReference', 'Macro', 'BlockReference')
def find_jinja2_render_frame(frame):
while frame is not None and not is_jinja2_render_call(frame):
frame = frame.f_back
return frame
def change_variable(mainDebugger, frame, attr, expression):
if isinstance(frame, Jinja2TemplateFrame):
result = eval(expression, frame.f_globals, frame.f_locals)
frame.changeVariable(attr, result)
#=======================================================================================================================
# Jinja2 Frame
#=======================================================================================================================
class Jinja2TemplateFrame:
def __init__(self, frame):
file_name = get_jinja2_template_filename(frame)
self.back_context = None
if 'context' in frame.f_locals:
#sometimes we don't have 'context', e.g. in macros
self.back_context = frame.f_locals['context']
self.f_code = FCode('template', file_name)
self.f_lineno = get_jinja2_template_line(frame)
self.f_back = find_render_function_frame(frame)
self.f_globals = {}
self.f_locals = self.collect_context(frame)
self.f_trace = None
def collect_context(self, frame):
res = {}
if self.back_context is not None:
for k, v in self.back_context.items():
res[k] = v
for k, v in frame.f_locals.items():
if not k.startswith('l_'):
if not k in res:
#local variables should shadow globals from context
res[k] = v
elif v and not is_missing(v):
res[k[2:]] = v
return res
def changeVariable(self, name, value):
for k, v in self.back_context.items():
if k == name:
self.back_context.vars[k] = value
def is_missing(item):
if item.__class__.__name__ is 'MissingType':
return True
return False
def find_render_function_frame(frame):
#in order to hide internal rendering functions
old_frame = frame
try:
while not (DictContains(frame.f_locals, 'self') and frame.f_locals['self'].__class__.__name__ == 'Template' and \
frame.f_code.co_name == 'render'):
frame = frame.f_back
if frame is None:
return old_frame
return frame
except:
return old_frame
def get_jinja2_template_line(frame):
debug_info = None
if DictContains(frame.f_globals,'__jinja_template__'):
_debug_info = frame.f_globals['__jinja_template__']._debug_info
if _debug_info is not '':
#sometimes template contains only plain text
debug_info = frame.f_globals['__jinja_template__'].debug_info
if debug_info is None:
return None
lineno = frame.f_lineno
for pair in debug_info:
if pair[1] == lineno:
return pair[0]
return None
def get_jinja2_template_filename(frame):
if DictContains(frame.f_globals, '__jinja_template__'):
fname = frame.f_globals['__jinja_template__'].filename
filename, base = GetFileNameAndBaseFromFile(fname)
return filename
return None
#=======================================================================================================================
# Jinja2 Step Commands
#=======================================================================================================================
def has_exception_breaks(mainDebugger):
return hasattr(mainDebugger, 'jinja2_exception_break') and mainDebugger.jinja2_exception_break
def can_skip(mainDebugger, frame):
if hasattr(mainDebugger, 'jinja2_breakpoints') and mainDebugger.jinja2_breakpoints and is_jinja2_render_call(frame):
filename = get_jinja2_template_filename(frame)
jinja2_breakpoints_for_file = mainDebugger.jinja2_breakpoints.get(filename)
if jinja2_breakpoints_for_file:
return False
return True
def cmd_step_into(mainDebugger, frame, event, args, stop_info):
mainDebugger, filename, info, thread = args
if not hasattr(info, 'pydev_call_from_jinja2'):
info.pydev_call_from_jinja2 = None
if not hasattr(info, 'pydev_call_inside_jinja2'):
info.pydev_call_inside_jinja2 = None
if is_jinja2_suspended(thread):
stop_info['jinja2_stop'] = event in ('call', 'line') and is_jinja2_render_call(frame)
stop_info['stop'] = False
if info.pydev_call_from_jinja2 is not None:
if is_jinja2_internal_function(frame):
#if internal Jinja2 function was called, we sould continue debugging inside template
info.pydev_call_from_jinja2 = None
else:
#we go into python code from Jinja2 rendering frame
stop_info['stop'] = True
if event == 'call' and is_jinja2_context_call(frame.f_back):
#we called function from context, the next step will be in function
info.pydev_call_from_jinja2 = 1
if event == 'return' and is_jinja2_context_call(frame.f_back):
#we return from python code to Jinja2 rendering frame
info.pydev_step_stop = info.pydev_call_from_jinja2
info.pydev_call_from_jinja2 = None
thread.additionalInfo.suspend_type = JINJA2_SUSPEND
stop_info['stop'] = False
#print "info.pydev_call_from_jinja2", info.pydev_call_from_jinja2, "stop_info", stop_info, \
# "thread.additionalInfo.suspend_type", thread.additionalInfo.suspend_type
#print "event", event, "farme.locals", frame.f_locals
def cmd_step_over(mainDebugger, frame, event, args, stop_info):
mainDebugger, filename, info, thread = args
if not hasattr(info, 'pydev_call_from_jinja2'):
info.pydev_call_from_jinja2 = None
if not hasattr(info, 'pydev_call_inside_jinja2'):
info.pydev_call_inside_jinja2 = None
if is_jinja2_suspended(thread):
stop_info['stop'] = False
if info.pydev_call_inside_jinja2 is None:
if is_jinja2_render_call(frame):
if event == 'call':
info.pydev_call_inside_jinja2 = frame.f_back
if event in ('line', 'return'):
info.pydev_call_inside_jinja2 = frame
else:
if event == 'line':
if is_jinja2_render_call(frame) and info.pydev_call_inside_jinja2 is frame:
stop_info['jinja2_stop'] = True
if event == 'return':
if frame is info.pydev_call_inside_jinja2 and not DictContains(frame.f_back.f_locals,'event'):
info.pydev_call_inside_jinja2 = find_jinja2_render_frame(frame.f_back)
return True
else:
if event == 'return' and is_jinja2_context_call(frame.f_back):
#we return from python code to Jinja2 rendering frame
info.pydev_call_from_jinja2 = None
info.pydev_call_inside_jinja2 = find_jinja2_render_frame(frame)
thread.additionalInfo.suspend_type = JINJA2_SUSPEND
stop_info['stop'] = False
return True
#print "info.pydev_call_from_jinja2", info.pydev_call_from_jinja2, "stop", stop, "jinja_stop", jinja2_stop, \
# "thread.additionalInfo.suspend_type", thread.additionalInfo.suspend_type
#print "event", event, "info.pydev_call_inside_jinja2", info.pydev_call_inside_jinja2
#print "frame", frame, "frame.f_back", frame.f_back, "step_stop", info.pydev_step_stop
#print "is_context_call", is_jinja2_context_call(frame)
#print "render", is_jinja2_render_call(frame)
#print "-------------"
return False
def stop(mainDebugger, frame, event, args, stop_info, arg, step_cmd):
mainDebugger, filename, info, thread = args
if DictContains(stop_info, 'jinja2_stop') and stop_info['jinja2_stop']:
frame = suspend_jinja2(mainDebugger, mainDebugger, thread, frame, step_cmd)
if frame:
mainDebugger.doWaitSuspend(thread, frame, event, arg)
return True
return False
def get_breakpoint(mainDebugger, frame, event, args):
mainDebugger, filename, info, thread = args
new_frame = None
jinja2_breakpoint = None
flag = False
if event in ('line', 'call') and info.pydev_state != STATE_SUSPEND and hasattr(mainDebugger, 'jinja2_breakpoints') and \
mainDebugger.jinja2_breakpoints and is_jinja2_render_call(frame):
filename = get_jinja2_template_filename(frame)
jinja2_breakpoints_for_file = mainDebugger.jinja2_breakpoints.get(filename)
new_frame = Jinja2TemplateFrame(frame)
if jinja2_breakpoints_for_file:
lineno = frame.f_lineno
template_lineno = get_jinja2_template_line(frame)
if template_lineno is not None and DictContains(jinja2_breakpoints_for_file, template_lineno):
jinja2_breakpoint = jinja2_breakpoints_for_file[template_lineno]
flag = True
new_frame = Jinja2TemplateFrame(frame)
return flag, jinja2_breakpoint, new_frame
def suspend(mainDebugger, pydb_frame, thread, frame):
return suspend_jinja2(pydb_frame, mainDebugger, thread, frame)
def exception_break(mainDebugger, pydb_frame, frame, event, args, arg):
mainDebugger, filename, info, thread = args
exception, value, trace = arg
if hasattr(mainDebugger, 'jinja2_exception_break') and mainDebugger.jinja2_exception_break:
if get_exception_name(exception) in ('UndefinedError', 'TemplateNotFound', 'TemplatesNotFound'):
#errors in rendering
render_frame = find_jinja2_render_frame(frame)
if render_frame:
suspend_frame = suspend_jinja2(pydb_frame, mainDebugger, thread, render_frame, CMD_ADD_EXCEPTION_BREAK)
if suspend_frame:
add_exception_to_frame(suspend_frame, (exception, value, trace))
flag = True
suspend_frame.f_back = frame
frame = suspend_frame
return (flag, frame)
elif get_exception_name(exception) in ('TemplateSyntaxError', 'TemplateAssertionError'):
#errors in compile time
name = frame.f_code.co_name
if name in ('template', 'top-level template code') or name.startswith('block '):
#Jinja2 translates exception info and creates fake frame on his own
pydb_frame.setSuspend(thread, CMD_ADD_EXCEPTION_BREAK)
add_exception_to_frame(frame, (exception, value, trace))
thread.additionalInfo.suspend_type = JINJA2_SUSPEND
flag = True
return (flag, frame)
return None
python/helpers/pydev/pydevd_vars.py
+4 -5
View File
@@ -2,7 +2,6 @@
resolution/conversion to XML.
"""
import pickle
from django_frame import DjangoTemplateFrame
from pydevd_constants import * #@UnusedWildImport
from types import * #@UnusedWildImport
@@ -360,10 +359,10 @@ def changeAttrExpression(thread_id, frame_id, attr, expression):
try:
expression = expression.replace('@LINE@', '\n')
if isinstance(frame, DjangoTemplateFrame):
result = eval(expression, frame.f_globals, frame.f_locals)
frame.changeVariable(attr, result)
return
# if isinstance(frame, DjangoTemplateFrame): # TODO: implemente for plugins
# result = eval(expression, frame.f_globals, frame.f_locals)
# frame.changeVariable(attr, result)
# return
if attr[:7] == "Globals":
attr = attr[8:]
python/helpers/pydev/test_debug.py
+3 -1
View File
@@ -5,16 +5,18 @@ import os
test_data_path = os.path.abspath(os.path.join(os.path.dirname(os.path.realpath(__file__)), '..', '..', 'testData', 'debug'))
class PyDevTestCase(unittest.TestCase):
def testZipFileExits(self):
from pydevd_file_utils import exists
self.assertTrue(exists(test_data_path +'/zipped_lib.zip/zipped_module.py'))
self.assertTrue(exists(test_data_path + '/zipped_lib.zip/zipped_module.py'))
self.assertFalse(exists(test_data_path + '/zipped_lib.zip/zipped_module2.py'))
self.assertFalse(exists(test_data_path + '/zipped_lib2.zip/zipped_module.py'))
def testEggFileExits(self):
from pydevd_file_utils import exists
self.assertTrue(exists(test_data_path + '/pycharm-debug.egg/pydev/pydevd.py'))
self.assertFalse(exists(test_data_path + '/pycharm-debug.egg/pydev/pydevd2.py'))
python/helpers/pydev/tests/check_pydevconsole.py
+110
View File
@@ -0,0 +1,110 @@
import sys
import os
# Put pydevconsole in the path.
sys.argv[0] = os.path.dirname(sys.argv[0])
sys.path.insert(1, os.path.join(os.path.dirname(sys.argv[0])))
print('Running tests with:', sys.executable)
print('PYTHONPATH:')
print('\n'.join(sorted(sys.path)))
import threading
import unittest
import pydevconsole
from pydev_imports import xmlrpclib, SimpleXMLRPCServer
try:
raw_input
raw_input_name = 'raw_input'
except NameError:
raw_input_name = 'input'
#=======================================================================================================================
# Test
#=======================================================================================================================
class Test(unittest.TestCase):
def startClientThread(self, client_port):
class ClientThread(threading.Thread):
def __init__(self, client_port):
threading.Thread.__init__(self)
self.client_port = client_port
def run(self):
class HandleRequestInput:
def RequestInput(self):
return 'RequestInput: OK'
handle_request_input = HandleRequestInput()
import pydev_localhost
print('Starting client with:', pydev_localhost.get_localhost(), self.client_port)
client_server = SimpleXMLRPCServer((pydev_localhost.get_localhost(), self.client_port), logRequests=False)
client_server.register_function(handle_request_input.RequestInput)
client_server.serve_forever()
client_thread = ClientThread(client_port)
client_thread.setDaemon(True)
client_thread.start()
return client_thread
def getFreeAddresses(self):
import socket
s = socket.socket()
s.bind(('', 0))
port0 = s.getsockname()[1]
s1 = socket.socket()
s1.bind(('', 0))
port1 = s1.getsockname()[1]
s.close()
s1.close()
return port0, port1
def testServer(self):
client_port, server_port = self.getFreeAddresses()
class ServerThread(threading.Thread):
def __init__(self, client_port, server_port):
threading.Thread.__init__(self)
self.client_port = client_port
self.server_port = server_port
def run(self):
import pydev_localhost
print('Starting server with:', pydev_localhost.get_localhost(), self.server_port, self.client_port)
pydevconsole.StartServer(pydev_localhost.get_localhost(), self.server_port, self.client_port)
server_thread = ServerThread(client_port, server_port)
server_thread.setDaemon(True)
server_thread.start()
client_thread = self.startClientThread(client_port) #@UnusedVariable
import time
time.sleep(.3) #let's give it some time to start the threads
import pydev_localhost
server = xmlrpclib.Server('http://%s:%s' % (pydev_localhost.get_localhost(), server_port))
server.addExec("import sys; print('Running with: %s %s' % (sys.executable or sys.platform, sys.version))")
server.addExec('class Foo:')
server.addExec(' pass')
server.addExec('')
server.addExec('foo = Foo()')
server.addExec('a = %s()' % raw_input_name)
server.addExec('print (a)')
#=======================================================================================================================
# main
#=======================================================================================================================
if __name__ == '__main__':
unittest.main()
python/helpers/pydev/tests/test_pydev_ipython_010.py
+80
View File
@@ -0,0 +1,80 @@
# TODO: This test no longer works (check if it should be fixed or removed altogether).
#import unittest
#import sys
#import os
##make it as if we were executing from the directory above this one
#sys.argv[0] = os.path.dirname(sys.argv[0])
##twice the dirname to get the previous level from this file.
#sys.path.insert(1, os.path.join(os.path.dirname(sys.argv[0])))
#
#from pydev_localhost import get_localhost
#
#
#IS_JYTHON = sys.platform.find('java') != -1
#
##=======================================================================================================================
## TestCase
##=======================================================================================================================
#class TestCase(unittest.TestCase):
#
# def setUp(self):
# unittest.TestCase.setUp(self)
#
# def tearDown(self):
# unittest.TestCase.tearDown(self)
#
# def testIPython(self):
# try:
# from pydev_ipython_console import PyDevFrontEnd
# except:
# if IS_JYTHON:
# return
# front_end = PyDevFrontEnd(get_localhost(), 0)
#
# front_end.input_buffer = 'if True:'
# self.assert_(not front_end._on_enter())
#
# front_end.input_buffer = 'if True:\n' + \
# front_end.continuation_prompt() + ' a = 10\n'
# self.assert_(not front_end._on_enter())
#
#
# front_end.input_buffer = 'if True:\n' + \
# front_end.continuation_prompt() + ' a = 10\n\n'
# self.assert_(front_end._on_enter())
#
#
## front_end.input_buffer = ' print a'
## self.assert_(not front_end._on_enter())
## front_end.input_buffer = ''
## self.assert_(front_end._on_enter())
#
#
## front_end.input_buffer = 'a.'
## front_end.complete_current_input()
## front_end.input_buffer = 'if True:'
## front_end._on_enter()
# front_end.input_buffer = 'a = 30'
# front_end._on_enter()
# front_end.input_buffer = 'print a'
# front_end._on_enter()
# front_end.input_buffer = 'a?'
# front_end._on_enter()
# print front_end.complete('%')
# print front_end.complete('%e')
# print front_end.complete('cd c:/t')
# print front_end.complete('cd c:/temp/')
## front_end.input_buffer = 'print raw_input("press enter\\n")'
## front_end._on_enter()
##
#
##=======================================================================================================================
## main
##=======================================================================================================================
#if __name__ == '__main__':
# if sys.platform.find('java') == -1:
# #IPython not available for Jython
# unittest.main()
# else:
# print('not supported on Jython')
python/helpers/pydev/uuid_old.py
+541
View File
@@ -0,0 +1,541 @@
r"""UUID objects (universally unique identifiers) according to RFC 4122.
This module provides immutable UUID objects (class UUID) and the functions
uuid1(), uuid3(), uuid4(), uuid5() for generating version 1, 3, 4, and 5
UUIDs as specified in RFC 4122.
If all you want is a unique ID, you should probably call uuid1() or uuid4().
Note that uuid1() may compromise privacy since it creates a UUID containing
the computer's network address. uuid4() creates a random UUID.
Typical usage:
>>> import uuid
# make a UUID based on the host ID and current time
>>> uuid.uuid1()
UUID('a8098c1a-f86e-11da-bd1a-00112444be1e')
# make a UUID using an MD5 hash of a namespace UUID and a name
>>> uuid.uuid3(uuid.NAMESPACE_DNS, 'python.org')
UUID('6fa459ea-ee8a-3ca4-894e-db77e160355e')
# make a random UUID
>>> uuid.uuid4()
UUID('16fd2706-8baf-433b-82eb-8c7fada847da')
# make a UUID using a SHA-1 hash of a namespace UUID and a name
>>> uuid.uuid5(uuid.NAMESPACE_DNS, 'python.org')
UUID('886313e1-3b8a-5372-9b90-0c9aee199e5d')
# make a UUID from a string of hex digits (braces and hyphens ignored)
>>> x = uuid.UUID('{00010203-0405-0607-0809-0a0b0c0d0e0f}')
# convert a UUID to a string of hex digits in standard form
>>> str(x)
'00010203-0405-0607-0809-0a0b0c0d0e0f'
# get the raw 16 bytes of the UUID
>>> x.bytes
'\x00\x01\x02\x03\x04\x05\x06\x07\x08\t\n\x0b\x0c\r\x0e\x0f'
# make a UUID from a 16-byte string
>>> uuid.UUID(bytes=x.bytes)
UUID('00010203-0405-0607-0809-0a0b0c0d0e0f')
"""
__author__ = 'Ka-Ping Yee <ping@zesty.ca>'
RESERVED_NCS, RFC_4122, RESERVED_MICROSOFT, RESERVED_FUTURE = [
'reserved for NCS compatibility', 'specified in RFC 4122',
'reserved for Microsoft compatibility', 'reserved for future definition']
class UUID(object):
"""Instances of the UUID class represent UUIDs as specified in RFC 4122.
UUID objects are immutable, hashable, and usable as dictionary keys.
Converting a UUID to a string with str() yields something in the form
'12345678-1234-1234-1234-123456789abc'. The UUID constructor accepts
five possible forms: a similar string of hexadecimal digits, or a tuple
of six integer fields (with 32-bit, 16-bit, 16-bit, 8-bit, 8-bit, and
48-bit values respectively) as an argument named 'fields', or a string
of 16 bytes (with all the integer fields in big-endian order) as an
argument named 'bytes', or a string of 16 bytes (with the first three
fields in little-endian order) as an argument named 'bytes_le', or a
single 128-bit integer as an argument named 'int'.
UUIDs have these read-only attributes:
bytes the UUID as a 16-byte string (containing the six
integer fields in big-endian byte order)
bytes_le the UUID as a 16-byte string (with time_low, time_mid,
and time_hi_version in little-endian byte order)
fields a tuple of the six integer fields of the UUID,
which are also available as six individual attributes
and two derived attributes:
time_low the first 32 bits of the UUID
time_mid the next 16 bits of the UUID
time_hi_version the next 16 bits of the UUID
clock_seq_hi_variant the next 8 bits of the UUID
clock_seq_low the next 8 bits of the UUID
node the last 48 bits of the UUID
time the 60-bit timestamp
clock_seq the 14-bit sequence number
hex the UUID as a 32-character hexadecimal string
int the UUID as a 128-bit integer
urn the UUID as a URN as specified in RFC 4122
variant the UUID variant (one of the constants RESERVED_NCS,
RFC_4122, RESERVED_MICROSOFT, or RESERVED_FUTURE)
version the UUID version number (1 through 5, meaningful only
when the variant is RFC_4122)
"""
def __init__(self, hex=None, bytes=None, bytes_le=None, fields=None,
int=None, version=None):
r"""Create a UUID from either a string of 32 hexadecimal digits,
a string of 16 bytes as the 'bytes' argument, a string of 16 bytes
in little-endian order as the 'bytes_le' argument, a tuple of six
integers (32-bit time_low, 16-bit time_mid, 16-bit time_hi_version,
8-bit clock_seq_hi_variant, 8-bit clock_seq_low, 48-bit node) as
the 'fields' argument, or a single 128-bit integer as the 'int'
argument. When a string of hex digits is given, curly braces,
hyphens, and a URN prefix are all optional. For example, these
expressions all yield the same UUID:
UUID('{12345678-1234-5678-1234-567812345678}')
UUID('12345678123456781234567812345678')
UUID('urn:uuid:12345678-1234-5678-1234-567812345678')
UUID(bytes='\x12\x34\x56\x78'*4)
UUID(bytes_le='\x78\x56\x34\x12\x34\x12\x78\x56' +
'\x12\x34\x56\x78\x12\x34\x56\x78')
UUID(fields=(0x12345678, 0x1234, 0x5678, 0x12, 0x34, 0x567812345678))
UUID(int=0x12345678123456781234567812345678)
Exactly one of 'hex', 'bytes', 'bytes_le', 'fields', or 'int' must
be given. The 'version' argument is optional; if given, the resulting
UUID will have its variant and version set according to RFC 4122,
overriding the given 'hex', 'bytes', 'bytes_le', 'fields', or 'int'.
"""
if [hex, bytes, bytes_le, fields, int].count(None) != 4:
raise TypeError('need one of hex, bytes, bytes_le, fields, or int')
if hex is not None:
hex = hex.replace('urn:', '').replace('uuid:', '')
hex = hex.strip('{}').replace('-', '')
if len(hex) != 32:
raise ValueError('badly formed hexadecimal UUID string')
int = long(hex, 16)
if bytes_le is not None:
if len(bytes_le) != 16:
raise ValueError('bytes_le is not a 16-char string')
bytes = (bytes_le[3] + bytes_le[2] + bytes_le[1] + bytes_le[0] +
bytes_le[5] + bytes_le[4] + bytes_le[7] + bytes_le[6] +
bytes_le[8:])
if bytes is not None:
if len(bytes) != 16:
raise ValueError('bytes is not a 16-char string')
int = long(('%02x'*16) % tuple(map(ord, bytes)), 16)
if fields is not None:
if len(fields) != 6:
raise ValueError('fields is not a 6-tuple')
(time_low, time_mid, time_hi_version,
clock_seq_hi_variant, clock_seq_low, node) = fields
if not 0 <= time_low < 1<<32L:
raise ValueError('field 1 out of range (need a 32-bit value)')
if not 0 <= time_mid < 1<<16L:
raise ValueError('field 2 out of range (need a 16-bit value)')
if not 0 <= time_hi_version < 1<<16L:
raise ValueError('field 3 out of range (need a 16-bit value)')
if not 0 <= clock_seq_hi_variant < 1<<8L:
raise ValueError('field 4 out of range (need an 8-bit value)')
if not 0 <= clock_seq_low < 1<<8L:
raise ValueError('field 5 out of range (need an 8-bit value)')
if not 0 <= node < 1<<48L:
raise ValueError('field 6 out of range (need a 48-bit value)')
clock_seq = (clock_seq_hi_variant << 8L) | clock_seq_low
int = ((time_low << 96L) | (time_mid << 80L) |
(time_hi_version << 64L) | (clock_seq << 48L) | node)
if int is not None:
if not 0 <= int < 1<<128L:
raise ValueError('int is out of range (need a 128-bit value)')
if version is not None:
if not 1 <= version <= 5:
raise ValueError('illegal version number')
# Set the variant to RFC 4122.
int &= ~(0xc000 << 48L)
int |= 0x8000 << 48L
# Set the version number.
int &= ~(0xf000 << 64L)
int |= version << 76L
self.__dict__['int'] = int
def __cmp__(self, other):
if isinstance(other, UUID):
return cmp(self.int, other.int)
return NotImplemented
def __hash__(self):
return hash(self.int)
def __int__(self):
return self.int
def __repr__(self):
return 'UUID(%r)' % str(self)
def __setattr__(self, name, value):
raise TypeError('UUID objects are immutable')
def __str__(self):
hex = '%032x' % self.int
return '%s-%s-%s-%s-%s' % (
hex[:8], hex[8:12], hex[12:16], hex[16:20], hex[20:])
def get_bytes(self):
bytes = ''
for shift in range(0, 128, 8):
bytes = chr((self.int >> shift) & 0xff) + bytes
return bytes
bytes = property(get_bytes)
def get_bytes_le(self):
bytes = self.bytes
return (bytes[3] + bytes[2] + bytes[1] + bytes[0] +
bytes[5] + bytes[4] + bytes[7] + bytes[6] + bytes[8:])
bytes_le = property(get_bytes_le)
def get_fields(self):
return (self.time_low, self.time_mid, self.time_hi_version,
self.clock_seq_hi_variant, self.clock_seq_low, self.node)
fields = property(get_fields)
def get_time_low(self):
return self.int >> 96L
time_low = property(get_time_low)
def get_time_mid(self):
return (self.int >> 80L) & 0xffff
time_mid = property(get_time_mid)
def get_time_hi_version(self):
return (self.int >> 64L) & 0xffff
time_hi_version = property(get_time_hi_version)
def get_clock_seq_hi_variant(self):
return (self.int >> 56L) & 0xff
clock_seq_hi_variant = property(get_clock_seq_hi_variant)
def get_clock_seq_low(self):
return (self.int >> 48L) & 0xff
clock_seq_low = property(get_clock_seq_low)
def get_time(self):
return (((self.time_hi_version & 0x0fffL) << 48L) |
(self.time_mid << 32L) | self.time_low)
time = property(get_time)
def get_clock_seq(self):
return (((self.clock_seq_hi_variant & 0x3fL) << 8L) |
self.clock_seq_low)
clock_seq = property(get_clock_seq)
def get_node(self):
return self.int & 0xffffffffffff
node = property(get_node)
def get_hex(self):
return '%032x' % self.int
hex = property(get_hex)
def get_urn(self):
return 'urn:uuid:' + str(self)
urn = property(get_urn)
def get_variant(self):
if not self.int & (0x8000 << 48L):
return RESERVED_NCS
elif not self.int & (0x4000 << 48L):
return RFC_4122
elif not self.int & (0x2000 << 48L):
return RESERVED_MICROSOFT
else:
return RESERVED_FUTURE
variant = property(get_variant)
def get_version(self):
# The version bits are only meaningful for RFC 4122 UUIDs.
if self.variant == RFC_4122:
return int((self.int >> 76L) & 0xf)
version = property(get_version)
def _find_mac(command, args, hw_identifiers, get_index):
import os
for dir in ['', '/sbin/', '/usr/sbin']:
executable = os.path.join(dir, command)
if not os.path.exists(executable):
continue
try:
# LC_ALL to get English output, 2>/dev/null to
# prevent output on stderr
cmd = 'LC_ALL=C %s %s 2>/dev/null' % (executable, args)
pipe = os.popen(cmd)
except IOError:
continue
for line in pipe:
words = line.lower().split()
for i in range(len(words)):
if words[i] in hw_identifiers:
return int(words[get_index(i)].replace(':', ''), 16)
return None
def _ifconfig_getnode():
"""Get the hardware address on Unix by running ifconfig."""
# This works on Linux ('' or '-a'), Tru64 ('-av'), but not all Unixes.
for args in ('', '-a', '-av'):
mac = _find_mac('ifconfig', args, ['hwaddr', 'ether'], lambda i: i+1)
if mac:
return mac
import socket
ip_addr = socket.gethostbyname(socket.gethostname())
# Try getting the MAC addr from arp based on our IP address (Solaris).
mac = _find_mac('arp', '-an', [ip_addr], lambda i: -1)
if mac:
return mac
# This might work on HP-UX.
mac = _find_mac('lanscan', '-ai', ['lan0'], lambda i: 0)
if mac:
return mac
return None
def _ipconfig_getnode():
"""Get the hardware address on Windows by running ipconfig.exe."""
import os, re
dirs = ['', r'c:\windows\system32', r'c:\winnt\system32']
try:
import ctypes
buffer = ctypes.create_string_buffer(300)
ctypes.windll.kernel32.GetSystemDirectoryA(buffer, 300)
dirs.insert(0, buffer.value.decode('mbcs'))
except:
pass
for dir in dirs:
try:
pipe = os.popen(os.path.join(dir, 'ipconfig') + ' /all')
except IOError:
continue
for line in pipe:
value = line.split(':')[-1].strip().lower()
if re.match('([0-9a-f][0-9a-f]-){5}[0-9a-f][0-9a-f]', value):
return int(value.replace('-', ''), 16)
def _netbios_getnode():
"""Get the hardware address on Windows using NetBIOS calls.
See http://support.microsoft.com/kb/118623 for details."""
import win32wnet, netbios
ncb = netbios.NCB()
ncb.Command = netbios.NCBENUM
ncb.Buffer = adapters = netbios.LANA_ENUM()
adapters._pack()
if win32wnet.Netbios(ncb) != 0:
return
adapters._unpack()
for i in range(adapters.length):
ncb.Reset()
ncb.Command = netbios.NCBRESET
ncb.Lana_num = ord(adapters.lana[i])
if win32wnet.Netbios(ncb) != 0:
continue
ncb.Reset()
ncb.Command = netbios.NCBASTAT
ncb.Lana_num = ord(adapters.lana[i])
ncb.Callname = '*'.ljust(16)
ncb.Buffer = status = netbios.ADAPTER_STATUS()
if win32wnet.Netbios(ncb) != 0:
continue
status._unpack()
bytes = map(ord, status.adapter_address)
return ((bytes[0]<<40L) + (bytes[1]<<32L) + (bytes[2]<<24L) +
(bytes[3]<<16L) + (bytes[4]<<8L) + bytes[5])
# Thanks to Thomas Heller for ctypes and for his help with its use here.
# If ctypes is available, use it to find system routines for UUID generation.
_uuid_generate_random = _uuid_generate_time = _UuidCreate = None
try:
import ctypes, ctypes.util
_buffer = ctypes.create_string_buffer(16)
# The uuid_generate_* routines are provided by libuuid on at least
# Linux and FreeBSD, and provided by libc on Mac OS X.
for libname in ['uuid', 'c']:
try:
lib = ctypes.CDLL(ctypes.util.find_library(libname))
except:
continue
if hasattr(lib, 'uuid_generate_random'):
_uuid_generate_random = lib.uuid_generate_random
if hasattr(lib, 'uuid_generate_time'):
_uuid_generate_time = lib.uuid_generate_time
# On Windows prior to 2000, UuidCreate gives a UUID containing the
# hardware address. On Windows 2000 and later, UuidCreate makes a
# random UUID and UuidCreateSequential gives a UUID containing the
# hardware address. These routines are provided by the RPC runtime.
# NOTE: at least on Tim's WinXP Pro SP2 desktop box, while the last
# 6 bytes returned by UuidCreateSequential are fixed, they don't appear
# to bear any relationship to the MAC address of any network device
# on the box.
try:
lib = ctypes.windll.rpcrt4
except:
lib = None
_UuidCreate = getattr(lib, 'UuidCreateSequential',
getattr(lib, 'UuidCreate', None))
except:
pass
def _unixdll_getnode():
"""Get the hardware address on Unix using ctypes."""
_uuid_generate_time(_buffer)
return UUID(bytes=_buffer.raw).node
def _windll_getnode():
"""Get the hardware address on Windows using ctypes."""
if _UuidCreate(_buffer) == 0:
return UUID(bytes=_buffer.raw).node
def _random_getnode():
"""Get a random node ID, with eighth bit set as suggested by RFC 4122."""
import random
return random.randrange(0, 1<<48L) | 0x010000000000L
_node = None
def getnode():
"""Get the hardware address as a 48-bit positive integer.
The first time this runs, it may launch a separate program, which could
be quite slow. If all attempts to obtain the hardware address fail, we
choose a random 48-bit number with its eighth bit set to 1 as recommended
in RFC 4122.
"""
global _node
if _node is not None:
return _node
import sys
if sys.platform == 'win32':
getters = [_windll_getnode, _netbios_getnode, _ipconfig_getnode]
else:
getters = [_unixdll_getnode, _ifconfig_getnode]
for getter in getters + [_random_getnode]:
try:
_node = getter()
except:
continue
if _node is not None:
return _node
_last_timestamp = None
def uuid1(node=None, clock_seq=None):
"""Generate a UUID from a host ID, sequence number, and the current time.
If 'node' is not given, getnode() is used to obtain the hardware
address. If 'clock_seq' is given, it is used as the sequence number;
otherwise a random 14-bit sequence number is chosen."""
# When the system provides a version-1 UUID generator, use it (but don't
# use UuidCreate here because its UUIDs don't conform to RFC 4122).
if _uuid_generate_time and node is clock_seq is None:
_uuid_generate_time(_buffer)
return UUID(bytes=_buffer.raw)
global _last_timestamp
import time
nanoseconds = int(time.time() * 1e9)
# 0x01b21dd213814000 is the number of 100-ns intervals between the
# UUID epoch 1582-10-15 00:00:00 and the Unix epoch 1970-01-01 00:00:00.
timestamp = int(nanoseconds/100) + 0x01b21dd213814000L
if timestamp <= _last_timestamp:
timestamp = _last_timestamp + 1
_last_timestamp = timestamp
if clock_seq is None:
import random
clock_seq = random.randrange(1<<14L) # instead of stable storage
time_low = timestamp & 0xffffffffL
time_mid = (timestamp >> 32L) & 0xffffL
time_hi_version = (timestamp >> 48L) & 0x0fffL
clock_seq_low = clock_seq & 0xffL
clock_seq_hi_variant = (clock_seq >> 8L) & 0x3fL
if node is None:
node = getnode()
return UUID(fields=(time_low, time_mid, time_hi_version,
clock_seq_hi_variant, clock_seq_low, node), version=1)
def uuid3(namespace, name):
"""Generate a UUID from the MD5 hash of a namespace UUID and a name."""
import md5
hash = md5.md5(namespace.bytes + name).digest()
return UUID(bytes=hash[:16], version=3)
def uuid4():
"""Generate a random UUID."""
# When the system provides a version-4 UUID generator, use it.
if _uuid_generate_random:
_uuid_generate_random(_buffer)
return UUID(bytes=_buffer.raw)
# Otherwise, get randomness from urandom or the 'random' module.
try:
import os
return UUID(bytes=os.urandom(16), version=4)
except:
import random
bytes = [chr(random.randrange(256)) for i in range(16)]
return UUID(bytes=bytes, version=4)
def uuid5(namespace, name):
"""Generate a UUID from the SHA-1 hash of a namespace UUID and a name."""
import sha
hash = sha.sha(namespace.bytes + name).digest()
return UUID(bytes=hash[:16], version=5)
# The following standard UUIDs are for use with uuid3() or uuid5().
NAMESPACE_DNS = UUID('6ba7b810-9dad-11d1-80b4-00c04fd430c8')
NAMESPACE_URL = UUID('6ba7b811-9dad-11d1-80b4-00c04fd430c8')
NAMESPACE_OID = UUID('6ba7b812-9dad-11d1-80b4-00c04fd430c8')
NAMESPACE_X500 = UUID('6ba7b814-9dad-11d1-80b4-00c04fd430c8')