Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_depth_limit(LineMatcher, tracer_impl, depth):
buff = StringIO()
from sample7 import one
tracer = hunter.Tracer()
predicate = When(Q(depth_lt=depth), CallPrinter(stream=buff))
try:
tracer.trace(predicate)
one()
finally:
tracer.stop()
output = buff.getvalue()
lm = LineMatcher(output.splitlines())
lm.fnmatch_lines([
"* call => one()",
"* line for i in range(1): # one",
"* line two()",
"* call => two()",
"* return <= two: None",
"* line for i in range(1): # one",
"* return <= one: None",
def tracing_wrapper(*args, **kwargs):
# create the Tracer manually to avoid spending time in likely useless things like:
# - loading PYTHONHUNTERCONFIG
# - setting up the clear_env_var or thread_support options
# - atexit cleanup registration
with hunter.Tracer().trace(hunter.When(hunter.Query(**filters), *actions)):
return func(*args, **kwargs)
def tracer_impl(request):
if request.param == 'pure':
Tracer = pytest.importorskip('hunter.tracer').Tracer
elif request.param == 'cython':
Tracer = pytest.importorskip('hunter._tracer').Tracer
if Tracer is not hunter.Tracer:
pytest.skip("Not %s in this environment" % Tracer)
return Tracer
def test_proper_backend():
if os.environ.get('PUREPYTHONHUNTER') or platform.python_implementation() == 'PyPy':
assert 'hunter.tracer.Tracer' in repr(hunter.Tracer)
else:
assert 'hunter._tracer.Tracer' in repr(hunter.Tracer)
if is_unwrapped
else hunter.Query(
# Only trace lines...
kind="line",
# ...and only in the given function...
function=target_func_name,
# ...but we don't know how many times it's been wrapped.
# Use the module instead as an approximate match.
# This may catch other functions with the same name
# in the same module, but not much we can do about
# that without a custom Cython Query.
module_in=target_obj,
),
hotspots,
)
tracer = hunter.Tracer(
# There's no need to call threading.settrace() because
# a) we're targeting a function we're about to call
# in the same thread,
# b) we're going to undo it immediately after, and
# c) it would collide with other threads if they did
# the same concurrently.
threading_support=False
).trace(predicate)
else:
tracer = None
try:
if instruments_by_event["call"] or instruments_by_event["return"]:
start = time.time()
_locals = {
"start": start,