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_thread_filtering(LineMatcher, query):
lines = StringIO()
idents = set()
names = set()
started = threading.Event()
def record(event):
idents.add(event.threadid)
names.add(event.threadname)
return True
with hunter.trace(~Q(**query), record,
actions=[CodePrinter(stream=lines), VarsPrinter('a', stream=lines), CallPrinter(stream=lines)],
threading_support=True):
def foo(a=1):
started.set()
print(a)
def main():
foo()
t = threading.Thread(target=foo)
t.start()
started.wait(10)
main()
lm = LineMatcher(lines.getvalue().splitlines())
print(lines.getvalue())
def test_threading_support(LineMatcher):
lines = StringIO()
idents = set()
names = set()
started = threading.Event()
def record(event):
idents.add(event.threadid)
names.add(event.threadname)
return True
with hunter.trace(record,
actions=[CodePrinter(stream=lines), VarsPrinter('a', stream=lines), CallPrinter(stream=lines)],
threading_support=True):
def foo(a=1):
started.set()
print(a)
def main():
foo()
t = threading.Thread(target=foo)
t.start()
started.wait(10)
main()
lm = LineMatcher(lines.getvalue().splitlines())
assert idents - {t.ident} == {None}
assert 'MainThread' in names
def test_threading_support(LineMatcher):
lines = StringIO()
idents = set()
names = set()
started = threading.Event()
def record(event):
idents.add(event.threadid)
names.add(event.threadname)
return True
with hunter.trace(record,
actions=[CodePrinter(stream=lines), VarsPrinter('a', stream=lines), CallPrinter(stream=lines)],
threading_support=True):
def foo(a=1):
started.set()
print(a)
def main():
foo()
t = threading.Thread(target=foo)
t.start()
started.wait(10)
main()
lm = LineMatcher(lines.getvalue().splitlines())
assert idents - {t.ident} == {None}
def test_thread_filtering(LineMatcher, query):
lines = StringIO()
idents = set()
names = set()
started = threading.Event()
def record(event):
idents.add(event.threadid)
names.add(event.threadname)
return True
with hunter.trace(~Q(**query), record,
actions=[CodePrinter(stream=lines), VarsPrinter('a', stream=lines), CallPrinter(stream=lines)],
threading_support=True):
def foo(a=1):
started.set()
print(a)
def main():
foo()
t = threading.Thread(target=foo)
t.start()
started.wait(10)
main()
lm = LineMatcher(lines.getvalue().splitlines())
print(lines.getvalue())
assert None not in idents
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 test_predicate_no_inf_recursion(mockevent):
assert Or(And(1)) == 1
assert Or(Or(1)) == 1
assert And(Or(1)) == 1
assert And(And(1)) == 1
predicate = Q(Q(lambda ev: 1, module='wat'))
print('predicate:', predicate)
predicate(mockevent)
called = [[], []]
predicate = (
Q(module=__name__, action=lambda ev: called[0].append(ev)) |
Q(module='foo', action=lambda ev: called[1].append(ev))
)
assert predicate(mockevent) == True
assert called == [[mockevent], []]
assert predicate(mockevent) == True
assert called == [[mockevent, mockevent], []]
called = [[], []]
predicate = (
Q(module=__name__, action=lambda ev: called[0].append(ev)) &
Q(function='mockevent', action=lambda ev: called[1].append(ev))
)
assert predicate(mockevent) == True
assert called == [[mockevent], [mockevent]]
def test_from_predicate_no_predicate(LineMatcher):
buff = StringIO()
from sample7 import one
with trace(From(Q(function='five')), action=CallPrinter(stream=buff)):
one()
output = buff.getvalue()
lm = LineMatcher(output.splitlines())
lm.fnmatch_lines([
"* call => five()",
"* line for i in range(1): # five",
"* line return i",
"* return <= five: 0",
])
assert 'four' not in output
assert 'three' not in output
assert 'two' not in output
assert 'one' not in output
def run():
output = StringIO()
with t.trace(Q(
~Q(module_contains='pytest'),
~Q(module_contains='hunter'),
~Q(filename=''),
~Q(filename=''),
stdlib=False,
action=CodePrinter(stream=output)
)):
_bulky_func_that_use_stdlib()
return output
def test_predicate_str_repr():
assert repr(Q(module='a', function='b')).endswith("predicates.Query: query_eq=(('function', 'b'), ('module', 'a'))>")
assert str(Q(module='a', function='b')) == "Query(function='b', module='a')"
assert repr(Q(module='a')).endswith("predicates.Query: query_eq=(('module', 'a'),)>")
assert str(Q(module='a')) == "Query(module='a')"
assert "predicates.When: condition=, actions=('foo',)>" in repr(Q(module='a', action=C('foo')))
assert str(Q(module='a', action=C('foo'))) == "When(Query(module='a'), 'foo')"
assert "predicates.Not: predicate=>" in repr(~Q(module='a'))
assert str(~Q(module='a')) == "Not(Query(module='a'))"
assert "predicates.Or: predicates=(, " in repr(Q(module='a') | Q(module='b'))
assert repr(Q(module='a') | Q(module='b')).endswith("predicates.Query: query_eq=(('module', 'b'),)>)>")
assert str(Q(module='a') | Q(module='b')) == "Or(Query(module='a'), Query(module='b'))"