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_and_or_kwargs():
assert And(1, function=2) == And(1, Query(function=2))
assert Or(1, function=2) == Or(1, Query(function=2))
target_obj, target_func_name = self.target.rsplit(".", 1)
is_unwrapped = base.func_code.co_name == target_func_name
if instruments_by_event["end"]:
# We have instruments that require evaluation in the local
# context of the function. Call sys.settrace() to gain access.
predicate = hunter.When(
hunter.Query(
# Only trace returns (this will include exceptions!)...
kind="return",
# ...and only in the given function...
function=target_func_name,
# ...(no deeper).
depth=0,
)
if is_unwrapped
else hunter.Query(
# Only trace returns (this will include exceptions!)...
kind="return",
# ...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,
),
TraceHandler(self, instruments_by_event["end"]),
)
tracer = hunter.Tracer(
# There's no need to call threading.settrace() because
# a) we're targeting a function we're about to call
continue
if I.check_call(self, *args, **kwargs):
instruments_by_event[I.event].append(I)
if I.event in ("call", "return"):
if "hotspots" in I.value or "hotspots" in (
I.custom.get("tags") or ""
):
hotspots.enabled = True
target_obj, target_func_name = self.target.rsplit(".", 1)
is_unwrapped = base.func_code.co_name == target_func_name
if instruments_by_event["end"]:
# We have instruments that require evaluation in the local
# context of the function. Call sys.settrace() to gain access.
predicate = hunter.When(
hunter.Query(
# Only trace returns (this will include exceptions!)...
kind="return",
# ...and only in the given function...
function=target_func_name,
# ...(no deeper).
depth=0,
)
if is_unwrapped
else hunter.Query(
# Only trace returns (this will include exceptions!)...
kind="return",
# ...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
threading_support=False
).trace(predicate)
elif hotspots.enabled:
# We have instruments that require timing internal lines.
# Call sys.settrace() to gain access.
predicate = hunter.When(
hunter.Query(
# Only trace lines...
kind="line",
# ...and only in the given function...
function=target_func_name,
# ...(no deeper).
depth=1,
)
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