tests.tracer_impl() - Code Metrics - Inspection of "Remove duplicate docstring." - ionelmc/python-hunter - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( af5503...cad96e )

by Ionel Cristian

created 2015-12-21 21:53 UTC

tests.tracer_impl() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	3
dl	0
loc	6
rs	9.4286

from __future__ import print_function

import inspect
import os
import platform
import subprocess
import sys

try:
    from StringIO import StringIO
except ImportError:
    from io import StringIO
try:
    from itertools import izip_longest
except ImportError:
    from itertools import zip_longest as izip_longest

import pytest

import hunter
from hunter import And
from hunter import Not
from hunter import Or
from hunter import Q
from hunter import Query
from hunter import stop
from hunter import trace
from hunter import When

from hunter import CodePrinter
from hunter import Debugger
from hunter import VarsPrinter

pytest_plugins = 'pytester',


@pytest.yield_fixture(autouse=True, scope="function")
def auto_stop():
    try:
        yield
    finally:
        stop()


def _get_func_spec(func):
    spec = inspect.getargspec(func)
    return inspect.formatargspec(spec.args, spec.varargs)


def test_pth_activation():
    module_name = os.path.__name__
    expected_module = "{0}.py".format(module_name)
    hunter_env = "module={!r},function=\"join\"".format(module_name)
    func_spec = _get_func_spec(os.path.join)
    expected_call = "call      def join{0}:".format(func_spec)

    output = subprocess.check_output(
        [sys.executable, os.path.join(os.path.dirname(__file__), 'sample.py')],
        env=dict(os.environ, PYTHONHUNTER=hunter_env),
        stderr=subprocess.STDOUT,
    )
    assert expected_module.encode() in output
    assert expected_call.encode() in output


def test_pth_sample4():
    env = dict(os.environ, PYTHONHUNTER="CodePrinter")
    env.pop('COVERAGE_PROCESS_START', None)
    env.pop('COV_CORE_SOURCE', None)
    output = subprocess.check_output(
        [sys.executable, os.path.join(os.path.dirname(__file__), 'sample4.py')],
        env=env,
        stderr=subprocess.STDOUT,
    )
    assert output


def test_pth_sample2(LineMatcher):
    env = dict(os.environ, PYTHONHUNTER="module='__main__'")
    env.pop('COVERAGE_PROCESS_START', None)
    env.pop('COV_CORE_SOURCE', None)
    output = subprocess.check_output(
        [sys.executable, os.path.join(os.path.dirname(__file__), 'sample2.py')],
        env=env,
        stderr=subprocess.STDOUT,
    )
    lm = LineMatcher(output.decode('utf-8').splitlines())
    lm.fnmatch_lines([
        '*tests*sample2.py:* call      if __name__ == "__main__":  #*',
        '*tests*sample2.py:* line      if __name__ == "__main__":  #*',
        '*tests*sample2.py:* line          import functools',
        '*tests*sample2.py:* line          def deco(opt):',
        '*tests*sample2.py:* line          @deco(1)',
        '*tests*sample2.py:* call          def deco(opt):',
        '*tests*sample2.py:* line              def decorator(func):',
        '*tests*sample2.py:* line              return decorator',
        '*tests*sample2.py:* return            return decorator',
        '*                 * ...       return value: <function deco*',
        '*tests*sample2.py:* line          @deco(2)',
        '*tests*sample2.py:* call          def deco(opt):',
        '*tests*sample2.py:* line              def decorator(func):',
        '*tests*sample2.py:* line              return decorator',
        '*tests*sample2.py:* return            return decorator',
        '*                 * ...       return value: <function deco*',
        '*tests*sample2.py:* line          @deco(3)',
        '*tests*sample2.py:* call          def deco(opt):',
        '*tests*sample2.py:* line              def decorator(func):',
        '*tests*sample2.py:* line              return decorator',
        '*tests*sample2.py:* return            return decorator',
        '*                 * ...       return value: <function deco*',
        '*tests*sample2.py:* call              def decorator(func):',
        '*tests*sample2.py:* line                  @functools.wraps(func)',
        '*tests*sample2.py:* line                  return wrapper',
        '*tests*sample2.py:* return                return wrapper',
        '*                 * ...       return value: <function foo *',
        '*tests*sample2.py:* call              def decorator(func):',
        '*tests*sample2.py:* line                  @functools.wraps(func)',
        '*tests*sample2.py:* line                  return wrapper',
        '*tests*sample2.py:* return                return wrapper',
        '*                 * ...       return value: <function foo *',
        '*tests*sample2.py:* call              def decorator(func):',
        '*tests*sample2.py:* line                  @functools.wraps(func)',
        '*tests*sample2.py:* line                  return wrapper',
        '*tests*sample2.py:* return                return wrapper',
        '*                 * ...       return value: <function foo *',
        '*tests*sample2.py:* line          foo(',
        "*tests*sample2.py:* line              'a*',",
        "*tests*sample2.py:* line              'b'",
        '*tests*sample2.py:* call                  @functools.wraps(func)',
        '*                 *    |                  def wrapper(*args):',
        '*tests*sample2.py:* line                      return func(*args)',
        '*tests*sample2.py:* call                  @functools.wraps(func)',
        '*                 *    |                  def wrapper(*args):',
        '*tests*sample2.py:* line                      return func(*args)',
        '*tests*sample2.py:* call                  @functools.wraps(func)',
        '*                 *    |                  def wrapper(*args):',
        '*tests*sample2.py:* line                      return func(*args)',
        '*tests*sample2.py:* call          @deco(1)',
        '*                 *    |          @deco(2)',
        '*                 *    |          @deco(3)',
        '*                 *    |          def foo(*args):',
        '*tests*sample2.py:* line              return args',
        '*tests*sample2.py:* return            return args',
        "*                 * ...       return value: ('a*', 'b')",
        "*tests*sample2.py:* return                    return func(*args)",
        "*                 * ...       return value: ('a*', 'b')",
        "*tests*sample2.py:* return                    return func(*args)",
        "*                 * ...       return value: ('a*', 'b')",
        "*tests*sample2.py:* return                    return func(*args)",
        "*                 * ...       return value: ('a*', 'b')",
        "*tests*sample2.py:* line          try:",
        "*tests*sample2.py:* line              None(",
        "*tests*sample2.py:* line                  'a',",
        "*tests*sample2.py:* line                  'b'",
        "*tests*sample2.py:* exception             'b'",
        "*                 * ...       exception value: *",
        "*tests*sample2.py:* line          except:",
        "*tests*sample2.py:* line              pass",
        "*tests*sample2.py:* return            pass",
        "*                   ...       return value: None",
    ])


def test_predicate_str_repr():
    assert repr(Q(module='a')).endswith("predicates.Query: query={'module': 'a'}>")
    assert str(Q(module='a')) == "Query(module='a')"

    assert "predicates.When: condition=<hunter." in repr(Q(module='a', action='foo'))
    assert "predicates.Query: query={'module': 'a'}>, actions=['foo']>" in repr(Q(module='a', action='foo'))
    assert str(Q(module='a', action='foo')) == "When(Query(module='a'), 'foo')"

    assert "predicates.Not: predicate=<hunter." in repr(~Q(module='a'))
    assert "predicates.Query: query={'module': 'a'}>>" in repr(~Q(module='a'))
    assert str(~Q(module='a')) == "Not(Query(module='a'))"

    assert "predicates.Or: predicates=(<hunter." in repr(Q(module='a') | Q(module='b'))
    assert "predicates.Query: query={'module': 'a'}>, " in repr(Q(module='a') | Q(module='b'))
    assert repr(Q(module='a') | Q(module='b')).endswith("predicates.Query: query={'module': 'b'}>)>")
    assert str(Q(module='a') | Q(module='b')) == "Or(Query(module='a'), Query(module='b'))"

    assert "predicates.And: predicates=(<hunter." in repr(Q(module='a') & Q(module='b'))
    assert "predicates.Query: query={'module': 'a'}>," in repr(Q(module='a') & Q(module='b'))
    assert repr(Q(module='a') & Q(module='b')).endswith("predicates.Query: query={'module': 'b'}>)>")
    assert str(Q(module='a') & Q(module='b')) == "And(Query(module='a'), Query(module='b'))"


def test_predicate_q_nest_1():
    assert repr(Q(Q(module='a'))).endswith("predicates.Query: query={'module': 'a'}>")


def test_predicate_q_expansion():
    assert Q(1, 2, module=3) == Or(1, 2, Q(module=3))
    assert Q(1, 2, module=3, action=4) == When(Or(1, 2, Q(module=3)), 4)
    assert Q(1, 2, module=3, actions=[4, 5]) == When(Or(1, 2, Q(module=3)), 4, 5)


def test_predicate_and():
    assert And(1, 2) == And(1, 2)
    assert Q(module=1) & Q(module=2) == And(Q(module=1), Q(module=2))
    assert Q(module=1) & Q(module=2) & Q(module=3) == And(Q(module=1), Q(module=2), Q(module=3))

    assert (Q(module=1) & Q(module=2))({'module': 3}) == False
    assert (Q(module=1) & Q(function=2))({'module': 1, 'function': 2}) == True

    assert And(1, 2) | 3 == Or(And(1, 2), 3)


def test_predicate_or():
    assert Q(module=1) | Q(module=2) == Or(Q(module=1), Q(module=2))
    assert Q(module=1) | Q(module=2) | Q(module=3) == Or(Q(module=1), Q(module=2), Q(module=3))

    assert (Q(module=1) | Q(module=2))({'module': 3}) == False
    assert (Q(module=1) | Q(module=2))({'module': 2}) == True

    assert Or(1, 2) & 3 == And(Or(1, 2), 3)


def test_tracing_bare(LineMatcher):
    lines = StringIO()
    with trace(CodePrinter(stream=lines)):
        def a():
            return 1

        b = a()
        b = 2
        try:
            raise Exception("BOOM!")
        except Exception:
            pass
    print(lines.getvalue())
    lm = LineMatcher(lines.getvalue().splitlines())
    lm.fnmatch_lines([
        "* ...       return value: <hunter.*tracer.Tracer *",
        "*test_hunter.py* call              def a():",
        "*test_hunter.py* line                  return 1",
        "*test_hunter.py* return                return 1",
        "* ...       return value: 1",
    ])


def test_tracing_printing_failures(LineMatcher):
    lines = StringIO()
    with trace(CodePrinter(stream=lines), VarsPrinter("x", stream=lines)):
        class Bad(Exception):
            def __repr__(self):
                raise RuntimeError("I'm a bad class!")

        def a():
            x = Bad()
            return x

        def b():
            x = Bad()
            raise x

        a()
        try:
            b()
        except Exception as exc:
            pass
    lm = LineMatcher(lines.getvalue().splitlines())
    lm.fnmatch_lines([
        """* ...       return value: <hunter.*tracer.Tracer *""",
        """*tests*test_hunter.py:* call              class Bad(Exception):""",
        """*tests*test_hunter.py:* line              class Bad(Exception):""",
        """*tests*test_hunter.py:* line                  def __repr__(self):""",
        """*tests*test_hunter.py:* return                def __repr__(self):""",
        """* ...       return value: *""",
        """*tests*test_hunter.py:* call              def a():""",
        """*tests*test_hunter.py:* line                  x = Bad()""",
        """*tests*test_hunter.py:* line                  return x""",
        """* vars      x => !!! FAILED REPR: RuntimeError("I'm a bad class!",)""",
        """*tests*test_hunter.py:* return                return x""",
        """* ...       return value: !!! FAILED REPR: RuntimeError("I'm a bad class!",)""",
        """* vars      x => !!! FAILED REPR: RuntimeError("I'm a bad class!",)""",
        """*tests*test_hunter.py:* call              def b():""",
        """*tests*test_hunter.py:* line                  x = Bad()""",
        """*tests*test_hunter.py:* line                  raise x""",
        """* vars      x => !!! FAILED REPR: RuntimeError("I'm a bad class!",)""",
        """*tests*test_hunter.py:* exception             raise x""",
        """* ...       exception value: !!! FAILED REPR: RuntimeError("I'm a bad class!",)""",
        """* vars      x => !!! FAILED REPR: RuntimeError("I'm a bad class!",)""",
        """*tests*test_hunter.py:* return                raise x""",
        """* ...       return value: None""",
        """* vars      x => !!! FAILED REPR: RuntimeError("I'm a bad class!",)""",
    ])


def test_tracing_vars(LineMatcher):
    lines = StringIO()
    with trace(actions=[VarsPrinter('b', stream=lines), CodePrinter(stream=lines)]):
        def a():
            b = 1
            b = 2
            return 1

        b = a()
        b = 2
        try:
            raise Exception("BOOM!")
        except Exception:
            pass
    print(lines.getvalue())
    lm = LineMatcher(lines.getvalue().splitlines())
    lm.fnmatch_lines([
        "* ...       return value: <hunter.*tracer.Tracer *",
        "*test_hunter.py* call              def a():",
        "*test_hunter.py* line                  b = 1",
        "* vars      b => 1",
        "*test_hunter.py* line                  b = 2",
        "* vars      b => 2",
        "*test_hunter.py* line                  return 1",
        "* vars      b => 2",
        "*test_hunter.py* return                return 1",
        "* ...       return value: 1",
    ])


def test_trace_merge():
    trace(function="a")
    trace(function="b")
    assert trace(function="c")._handler == When(Q(function="c"), CodePrinter)


def test_trace_api_expansion():
    # simple use
    with trace(function="foobar") as t:
        assert t._handler == When(Q(function="foobar"), CodePrinter)

    # "or" by expression
    with trace(module="foo", function="foobar") as t:
        assert t._handler == When(Q(module="foo", function="foobar"), CodePrinter)

    # pdb.set_trace
    with trace(function="foobar", action=Debugger) as t:
        assert t._handler == When(Q(function="foobar"), Debugger)

    # pdb.set_trace on any hits
    with trace(module="foo", function="foobar", action=Debugger) as t:
        assert t._handler == When(Q(module="foo", function="foobar"), Debugger)

    # pdb.set_trace when function is foobar, otherwise just print when module is foo
    with trace(Q(function="foobar", action=Debugger), module="foo") as t:
        assert t._handler == When(Or(
            When(Q(function="foobar"), Debugger),
            Q(module="foo")
        ), CodePrinter)

    # dumping variables from stack
    with trace(Q(function="foobar", action=VarsPrinter("foobar")), module="foo") as t:
        assert t._handler == When(Or(
            When(Q(function="foobar"), VarsPrinter("foobar")),
            Q(module="foo"),
        ), CodePrinter)

    with trace(Q(function="foobar", action=VarsPrinter("foobar", "mumbojumbo")), module="foo") as t:
        assert t._handler == When(Or(
            When(Q(function="foobar"), VarsPrinter("foobar", "mumbojumbo")),
            Q(module="foo"),
        ), CodePrinter)

    # multiple actions
    with trace(Q(function="foobar", actions=[VarsPrinter("foobar"), Debugger]), module="foo") as t:
        assert t._handler == When(Or(
            When(Q(function="foobar"), VarsPrinter("foobar"), Debugger),
            Q(module="foo"),
        ), CodePrinter)

    # customization
    assert trace(lambda event: event.locals.get("node") == "Foobar",
                 module="foo", function="foobar")
    assert trace(Q(lambda event: event.locals.get("node") == "Foobar",
                   function="foobar", actions=[VarsPrinter("foobar"), Debugger]), module="foo", )
    assert trace(Q(function="foobar", actions=[VarsPrinter("foobar"),
                                               lambda event: print("some custom output")]), module="foo", )


def test_trace_with_class_actions():
    with trace(CodePrinter):
        def a():
            pass

        a()


def test_predicate_no_inf_recursion():
    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('foo')


def test_predicate_compression():
    assert Or(Or(1, 2), And(3)) == Or(1, 2, 3)
    assert Or(Or(1, 2), 3) == Or(1, 2, 3)
    assert Or(1, Or(2, 3), 4) == Or(1, 2, 3, 4)
    assert And(1, 2, Or(3, 4)).predicates == (1, 2, Or(3, 4))

    assert repr(Or(Or(1, 2), And(3))) == repr(Or(1, 2, 3))
    assert repr(Or(Or(1, 2), 3)) == repr(Or(1, 2, 3))
    assert repr(Or(1, Or(2, 3), 4)) == repr(Or(1, 2, 3, 4))


def test_predicate_not():
    assert Not(1).predicate == 1
    assert ~Or(1, 2) == Not(Or(1, 2))
    assert ~And(1, 2) == Not(And(1, 2))

    assert ~Not(1) == 1

    assert ~Query(module=1) | ~Query(module=2) == Not(And(Query(module=1), Query(module=2)))
    assert ~Query(module=1) & ~Query(module=2) == Not(Or(Query(module=1), Query(module=2)))

    assert ~Query(module=1) | Query(module=2) == Or(Not(Query(module=1)), Query(module=2))
    assert ~Query(module=1) & Query(module=2) == And(Not(Query(module=1)), Query(module=2))

    assert ~(Query(module=1) & Query(module=2)) == Not(And(Query(module=1), Query(module=2)))
    assert ~(Query(module=1) | Query(module=2)) == Not(Or(Query(module=1), Query(module=2)))

    assert repr(~Or(1, 2)) == repr(Not(Or(1, 2)))
    assert repr(~And(1, 2)) == repr(Not(And(1, 2)))

    assert repr(~Query(module=1) | ~Query(module=2)) == repr(Not(And(Query(module=1), Query(module=2))))
    assert repr(~Query(module=1) & ~Query(module=2)) == repr(Not(Or(Query(module=1), Query(module=2))))

    assert repr(~(Query(module=1) & Query(module=2))) == repr(Not(And(Query(module=1), Query(module=2))))
    assert repr(~(Query(module=1) | Query(module=2))) == repr(Not(Or(Query(module=1), Query(module=2))))

    assert Not(Q(module=1))({'module': 1}) == False


def test_predicate_query_allowed():
    pytest.raises(TypeError, Query, 1)
    pytest.raises(TypeError, Query, a=1)


def test_predicate_when_allowed():
    pytest.raises(TypeError, When, 1)


@pytest.mark.parametrize('expr,inp,expected', [
    ({'module': "abc"}, {'module': "abc"}, True),
    ({'module': "abcd"}, {'module': "abc"}, False),
    ({'module': "abcd"}, {'module': "abce"}, False),
    ({'module': "abc"}, {'module': "abcd"}, False),
    ({'module': ("abc", "xyz")}, {'module': "abc"}, True),
    ({'module': ("abc", "xyz")}, {'module': "abcd"}, True),
    ({'module': ("abc", "xyz")}, {'module': "xyzw"}, True),
    ({'module': ("abc", "xyz")}, {'module': "fooabc"}, False),
    ({'module': ("abc", "xyz")}, {'module': 1}, False),

    ({'module': ["abc", "xyz"]}, {'module': "abc"}, True),
    ({'module': ["abc", "xyz"]}, {'module': "abcd"}, True),
    ({'module': ["abc", "xyz"]}, {'module': "xyzw"}, True),
    ({'module': ["abc", "xyz"]}, {'module': "fooabc"}, False),
    ({'module': ["abc", "xyz"]}, {'module': 1}, False),

    ({'module': {"abc", "xyz"}}, {'module': "abc"}, True),
    ({'module': {"abc", "xyz"}}, {'module': "abcd"}, True),
    ({'module': {"abc", "xyz"}}, {'module': "xyzw"}, True),
    ({'module': {"abc", "xyz"}}, {'module': "fooabc"}, False),
    ({'module': {"abc", "xyz"}}, {'module': 1}, False),

    ({'module': "abc"}, {'module': 1}, False),
])
def test_predicate_matching(expr, inp, expected):
    assert Query(**expr)(inp) == expected


def test_predicate_when():
    called = []
    assert When(Q(module=1), lambda ev: called.append(ev))({'module': 2}) == False
    assert called == []

    assert When(Q(module=1), lambda ev: called.append(ev))({'module': 1}) == True
    assert called == [{'module': 1}]

    called = []
    assert Q(module=1, action=lambda ev: called.append(ev))({'module': 1}) == True
    assert called == [{'module': 1}]

    called = [[], []]
    predicate = (
        Q(module=1, action=lambda ev: called[0].append(ev)) |
        Q(module=2, action=lambda ev: called[1].append(ev))
    )
    assert predicate({'module': 1}) == True
    assert called == [[{'module': 1}], []]

    assert predicate({'module': 2}) == True
    assert called == [[{'module': 1}], [{'module': 2}]]

    called = [[], []]
    predicate = (
        Q(module=1, action=lambda ev: called[0].append(ev)) &
        Q(function=2, action=lambda ev: called[1].append(ev))
    )
    assert predicate({'module': 2}) == False
    assert called == [[], []]

    assert predicate({'module': 1, 'function': 2}) == True
    assert called == [[{'module': 1, 'function': 2}], [{'module': 1, 'function': 2}]]


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)


@pytest.fixture(scope="session", params=['pure', 'cython'])
def tracer_impl(request):
    if request.param == 'pure':
        return pytest.importorskip('hunter.tracer').Tracer
    elif request.param == 'cython':
        return pytest.importorskip('hunter._tracer').Tracer


def test_perf_filter(tracer_impl, benchmark):
    t = tracer_impl()
    import tokenize

    @benchmark
    def run():
        with t.trace(Q(module="does-not-exist") | Q(module="does not exist".split())):
            with open(tokenize.__file__, 'rb') as fh:
                toks = []
                try:
                    for tok in tokenize.tokenize(fh.readline):
                        toks.append(tok)
                except tokenize.TokenError as exc:
                    toks.append(exc)


1			from __future__ import print_function
2
3			import inspect
4			import os
5			import platform
6			import subprocess
7			import sys
8
9			try:
10			from StringIO import StringIO
11			except ImportError:
12			from io import StringIO
13			try:
14			from itertools import izip_longest
15			except ImportError:
16			from itertools import zip_longest as izip_longest
17
18			import pytest
19
20			import hunter
21			from hunter import And
22			from hunter import Not
23			from hunter import Or
24			from hunter import Q
25			from hunter import Query
26			from hunter import stop
27			from hunter import trace
28			from hunter import When
29
30			from hunter import CodePrinter
31			from hunter import Debugger
32			from hunter import VarsPrinter
33
34			pytest_plugins = 'pytester',
35
36
37			@pytest.yield_fixture(autouse=True, scope="function")
38			def auto_stop():
39			try:
40			yield
41			finally:
42			stop()
43
44
45			def _get_func_spec(func):
46			spec = inspect.getargspec(func)
47			return inspect.formatargspec(spec.args, spec.varargs)
48
49
50			def test_pth_activation():
51			module_name = os.path.__name__
52			expected_module = "{0}.py".format(module_name)
53			hunter_env = "module={!r},function=\"join\"".format(module_name)
54			func_spec = _get_func_spec(os.path.join)
55			expected_call = "call def join{0}:".format(func_spec)
56
57			output = subprocess.check_output(
58			[sys.executable, os.path.join(os.path.dirname(__file__), 'sample.py')],
59			env=dict(os.environ, PYTHONHUNTER=hunter_env),
60			stderr=subprocess.STDOUT,
61			)
62			assert expected_module.encode() in output
63			assert expected_call.encode() in output
64
65
66			def test_pth_sample4():
67			env = dict(os.environ, PYTHONHUNTER="CodePrinter")
68			env.pop('COVERAGE_PROCESS_START', None)
69			env.pop('COV_CORE_SOURCE', None)
70			output = subprocess.check_output(
71			[sys.executable, os.path.join(os.path.dirname(__file__), 'sample4.py')],
72			env=env,
73			stderr=subprocess.STDOUT,
74			)
75			assert output
76
77
78			def test_pth_sample2(LineMatcher):
79			env = dict(os.environ, PYTHONHUNTER="module='__main__'")
80			env.pop('COVERAGE_PROCESS_START', None)
81			env.pop('COV_CORE_SOURCE', None)
82			output = subprocess.check_output(
83			[sys.executable, os.path.join(os.path.dirname(__file__), 'sample2.py')],
84			env=env,
85			stderr=subprocess.STDOUT,
86			)
87			lm = LineMatcher(output.decode('utf-8').splitlines())
88			lm.fnmatch_lines([
89			'testssample2.py:* call if __name__ == "__main__": #*',
90			'testssample2.py:* line if __name__ == "__main__": #*',
91			'testssample2.py:* line import functools',
92			'testssample2.py:* line def deco(opt):',
93			'testssample2.py:* line @deco(1)',
94			'testssample2.py:* call def deco(opt):',
95			'testssample2.py:* line def decorator(func):',
96			'testssample2.py:* line return decorator',
97			'testssample2.py:* return return decorator',
98			'* * ... return value: <function deco*',
99			'testssample2.py:* line @deco(2)',
100			'testssample2.py:* call def deco(opt):',
101			'testssample2.py:* line def decorator(func):',
102			'testssample2.py:* line return decorator',
103			'testssample2.py:* return return decorator',
104			'* * ... return value: <function deco*',
105			'testssample2.py:* line @deco(3)',
106			'testssample2.py:* call def deco(opt):',
107			'testssample2.py:* line def decorator(func):',
108			'testssample2.py:* line return decorator',
109			'testssample2.py:* return return decorator',
110			'* * ... return value: <function deco*',
111			'testssample2.py:* call def decorator(func):',
112			'testssample2.py:* line @functools.wraps(func)',
113			'testssample2.py:* line return wrapper',
114			'testssample2.py:* return return wrapper',
115			'* * ... return value: <function foo *',
116			'testssample2.py:* call def decorator(func):',
117			'testssample2.py:* line @functools.wraps(func)',
118			'testssample2.py:* line return wrapper',
119			'testssample2.py:* return return wrapper',
120			'* * ... return value: <function foo *',
121			'testssample2.py:* call def decorator(func):',
122			'testssample2.py:* line @functools.wraps(func)',
123			'testssample2.py:* line return wrapper',
124			'testssample2.py:* return return wrapper',
125			'* * ... return value: <function foo *',
126			'testssample2.py:* line foo(',
127			"testssample2.py:* line 'a*',",
128			"testssample2.py:* line 'b'",
129			'testssample2.py:* call @functools.wraps(func)',
130			'* * \| def wrapper(*args):',
131			'testssample2.py:* line return func(*args)',
132			'testssample2.py:* call @functools.wraps(func)',
133			'* * \| def wrapper(*args):',
134			'testssample2.py:* line return func(*args)',
135			'testssample2.py:* call @functools.wraps(func)',
136			'* * \| def wrapper(*args):',
137			'testssample2.py:* line return func(*args)',
138			'testssample2.py:* call @deco(1)',
139			'* * \| @deco(2)',
140			'* * \| @deco(3)',
141			'* * \| def foo(*args):',
142			'testssample2.py:* line return args',
143			'testssample2.py:* return return args',
144			"* * ... return value: ('a*', 'b')",
145			"testssample2.py:* return return func(*args)",
146			"* * ... return value: ('a*', 'b')",
147			"testssample2.py:* return return func(*args)",
148			"* * ... return value: ('a*', 'b')",
149			"testssample2.py:* return return func(*args)",
150			"* * ... return value: ('a*', 'b')",
151			"testssample2.py:* line try:",
152			"testssample2.py:* line None(",
153			"testssample2.py:* line 'a',",
154			"testssample2.py:* line 'b'",
155			"testssample2.py:* exception 'b'",
156			"* * ... exception value: *",
157			"testssample2.py:* line except:",
158			"testssample2.py:* line pass",
159			"testssample2.py:* return pass",
160			"* ... return value: None",
161			])
162
163
164			def test_predicate_str_repr():
165			assert repr(Q(module='a')).endswith("predicates.Query: query={'module': 'a'}>")
166			assert str(Q(module='a')) == "Query(module='a')"
167
168			assert "predicates.When: condition=<hunter." in repr(Q(module='a', action='foo'))
169			assert "predicates.Query: query={'module': 'a'}>, actions=['foo']>" in repr(Q(module='a', action='foo'))
170			assert str(Q(module='a', action='foo')) == "When(Query(module='a'), 'foo')"
171
172			assert "predicates.Not: predicate=<hunter." in repr(~Q(module='a'))
173			assert "predicates.Query: query={'module': 'a'}>>" in repr(~Q(module='a'))
174			assert str(~Q(module='a')) == "Not(Query(module='a'))"
175
176			assert "predicates.Or: predicates=(<hunter." in repr(Q(module='a') \| Q(module='b'))
177			assert "predicates.Query: query={'module': 'a'}>, " in repr(Q(module='a') \| Q(module='b'))
178			assert repr(Q(module='a') \| Q(module='b')).endswith("predicates.Query: query={'module': 'b'}>)>")
179			assert str(Q(module='a') \| Q(module='b')) == "Or(Query(module='a'), Query(module='b'))"
180
181			assert "predicates.And: predicates=(<hunter." in repr(Q(module='a') & Q(module='b'))
182			assert "predicates.Query: query={'module': 'a'}>," in repr(Q(module='a') & Q(module='b'))
183			assert repr(Q(module='a') & Q(module='b')).endswith("predicates.Query: query={'module': 'b'}>)>")
184			assert str(Q(module='a') & Q(module='b')) == "And(Query(module='a'), Query(module='b'))"
185
186
187			def test_predicate_q_nest_1():
188			assert repr(Q(Q(module='a'))).endswith("predicates.Query: query={'module': 'a'}>")
189
190
191			def test_predicate_q_expansion():
192			assert Q(1, 2, module=3) == Or(1, 2, Q(module=3))
193			assert Q(1, 2, module=3, action=4) == When(Or(1, 2, Q(module=3)), 4)
194			assert Q(1, 2, module=3, actions=[4, 5]) == When(Or(1, 2, Q(module=3)), 4, 5)
195
196
197			def test_predicate_and():
198			assert And(1, 2) == And(1, 2)
199			assert Q(module=1) & Q(module=2) == And(Q(module=1), Q(module=2))
200			assert Q(module=1) & Q(module=2) & Q(module=3) == And(Q(module=1), Q(module=2), Q(module=3))
201
202			assert (Q(module=1) & Q(module=2))({'module': 3}) == False
203			assert (Q(module=1) & Q(function=2))({'module': 1, 'function': 2}) == True
204
205			assert And(1, 2) \| 3 == Or(And(1, 2), 3)
206
207
208			def test_predicate_or():
209			assert Q(module=1) \| Q(module=2) == Or(Q(module=1), Q(module=2))
210			assert Q(module=1) \| Q(module=2) \| Q(module=3) == Or(Q(module=1), Q(module=2), Q(module=3))
211
212			assert (Q(module=1) \| Q(module=2))({'module': 3}) == False
213			assert (Q(module=1) \| Q(module=2))({'module': 2}) == True
214
215			assert Or(1, 2) & 3 == And(Or(1, 2), 3)
216
217
218			def test_tracing_bare(LineMatcher):
219			lines = StringIO()
220			with trace(CodePrinter(stream=lines)):
221			def a():
222			return 1
223
224			b = a()
225			b = 2
226			try:
227			raise Exception("BOOM!")
228			except Exception:
229			pass
230			print(lines.getvalue())
231			lm = LineMatcher(lines.getvalue().splitlines())
232			lm.fnmatch_lines([
233			"* ... return value: <hunter.tracer.Tracer ",
234			"test_hunter.py call def a():",
235			"test_hunter.py line return 1",
236			"test_hunter.py return return 1",
237			"* ... return value: 1",
238			])
239
240
241			def test_tracing_printing_failures(LineMatcher):
242			lines = StringIO()
243			with trace(CodePrinter(stream=lines), VarsPrinter("x", stream=lines)):
244			class Bad(Exception):
245			def __repr__(self):
246			raise RuntimeError("I'm a bad class!")
247
248			def a():
249			x = Bad()
250			return x
251
252			def b():
253			x = Bad()
254			raise x
255
256			a()
257			try:
258			b()
259			except Exception as exc:
260			pass
261			lm = LineMatcher(lines.getvalue().splitlines())
262			lm.fnmatch_lines([
263			"""* ... return value: <hunter.tracer.Tracer """,
264			"""teststest_hunter.py:* call class Bad(Exception):""",
265			"""teststest_hunter.py:* line class Bad(Exception):""",
266			"""teststest_hunter.py:* line def __repr__(self):""",
267			"""teststest_hunter.py:* return def __repr__(self):""",
268			"""* ... return value: *""",
269			"""teststest_hunter.py:* call def a():""",
270			"""teststest_hunter.py:* line x = Bad()""",
271			"""teststest_hunter.py:* line return x""",
272			"""* vars x => !!! FAILED REPR: RuntimeError("I'm a bad class!",)""",
273			"""teststest_hunter.py:* return return x""",
274			"""* ... return value: !!! FAILED REPR: RuntimeError("I'm a bad class!",)""",
275			"""* vars x => !!! FAILED REPR: RuntimeError("I'm a bad class!",)""",
276			"""teststest_hunter.py:* call def b():""",
277			"""teststest_hunter.py:* line x = Bad()""",
278			"""teststest_hunter.py:* line raise x""",
279			"""* vars x => !!! FAILED REPR: RuntimeError("I'm a bad class!",)""",
280			"""teststest_hunter.py:* exception raise x""",
281			"""* ... exception value: !!! FAILED REPR: RuntimeError("I'm a bad class!",)""",
282			"""* vars x => !!! FAILED REPR: RuntimeError("I'm a bad class!",)""",
283			"""teststest_hunter.py:* return raise x""",
284			"""* ... return value: None""",
285			"""* vars x => !!! FAILED REPR: RuntimeError("I'm a bad class!",)""",
286			])
287
288
289			def test_tracing_vars(LineMatcher):
290			lines = StringIO()
291			with trace(actions=[VarsPrinter('b', stream=lines), CodePrinter(stream=lines)]):
292			def a():
293			b = 1
294			b = 2
295			return 1
296
297			b = a()
298			b = 2
299			try:
300			raise Exception("BOOM!")
301			except Exception:
302			pass
303			print(lines.getvalue())
304			lm = LineMatcher(lines.getvalue().splitlines())
305			lm.fnmatch_lines([
306			"* ... return value: <hunter.tracer.Tracer ",
307			"test_hunter.py call def a():",
308			"test_hunter.py line b = 1",
309			"* vars b => 1",
310			"test_hunter.py line b = 2",
311			"* vars b => 2",
312			"test_hunter.py line return 1",
313			"* vars b => 2",
314			"test_hunter.py return return 1",
315			"* ... return value: 1",
316			])
317
318
319			def test_trace_merge():
320			trace(function="a")
321			trace(function="b")
322			assert trace(function="c")._handler == When(Q(function="c"), CodePrinter)
323
324
325			def test_trace_api_expansion():
326			# simple use
327			with trace(function="foobar") as t:
328			assert t._handler == When(Q(function="foobar"), CodePrinter)
329
330			# "or" by expression
331			with trace(module="foo", function="foobar") as t:
332			assert t._handler == When(Q(module="foo", function="foobar"), CodePrinter)
333
334			# pdb.set_trace
335			with trace(function="foobar", action=Debugger) as t:
336			assert t._handler == When(Q(function="foobar"), Debugger)
337
338			# pdb.set_trace on any hits
339			with trace(module="foo", function="foobar", action=Debugger) as t:
340			assert t._handler == When(Q(module="foo", function="foobar"), Debugger)
341
342			# pdb.set_trace when function is foobar, otherwise just print when module is foo
343			with trace(Q(function="foobar", action=Debugger), module="foo") as t:
344			assert t._handler == When(Or(
345			When(Q(function="foobar"), Debugger),
346			Q(module="foo")
347			), CodePrinter)
348
349			# dumping variables from stack
350			with trace(Q(function="foobar", action=VarsPrinter("foobar")), module="foo") as t:
351			assert t._handler == When(Or(
352			When(Q(function="foobar"), VarsPrinter("foobar")),
353			Q(module="foo"),
354			), CodePrinter)
355
356			with trace(Q(function="foobar", action=VarsPrinter("foobar", "mumbojumbo")), module="foo") as t:
357			assert t._handler == When(Or(
358			When(Q(function="foobar"), VarsPrinter("foobar", "mumbojumbo")),
359			Q(module="foo"),
360			), CodePrinter)
361
362			# multiple actions
363			with trace(Q(function="foobar", actions=[VarsPrinter("foobar"), Debugger]), module="foo") as t:
364			assert t._handler == When(Or(
365			When(Q(function="foobar"), VarsPrinter("foobar"), Debugger),
366			Q(module="foo"),
367			), CodePrinter)
368
369			# customization
370			assert trace(lambda event: event.locals.get("node") == "Foobar",
371			module="foo", function="foobar")
372			assert trace(Q(lambda event: event.locals.get("node") == "Foobar",
373			function="foobar", actions=[VarsPrinter("foobar"), Debugger]), module="foo", )
374			assert trace(Q(function="foobar", actions=[VarsPrinter("foobar"),
375			lambda event: print("some custom output")]), module="foo", )
376
377
378			def test_trace_with_class_actions():
379			with trace(CodePrinter):
380			def a():
381			pass
382
383			a()
384
385
386			def test_predicate_no_inf_recursion():
387			assert Or(And(1)) == 1
388			assert Or(Or(1)) == 1
389			assert And(Or(1)) == 1
390			assert And(And(1)) == 1
391			predicate = Q(Q(lambda ev: 1, module='wat'))
392			print('predicate:', predicate)
393			predicate('foo')
394
395
396			def test_predicate_compression():
397			assert Or(Or(1, 2), And(3)) == Or(1, 2, 3)
398			assert Or(Or(1, 2), 3) == Or(1, 2, 3)
399			assert Or(1, Or(2, 3), 4) == Or(1, 2, 3, 4)
400			assert And(1, 2, Or(3, 4)).predicates == (1, 2, Or(3, 4))
401
402			assert repr(Or(Or(1, 2), And(3))) == repr(Or(1, 2, 3))
403			assert repr(Or(Or(1, 2), 3)) == repr(Or(1, 2, 3))
404			assert repr(Or(1, Or(2, 3), 4)) == repr(Or(1, 2, 3, 4))
405
406
407			def test_predicate_not():
408			assert Not(1).predicate == 1
409			assert ~Or(1, 2) == Not(Or(1, 2))
410			assert ~And(1, 2) == Not(And(1, 2))
411
412			assert ~Not(1) == 1
413
414			assert ~Query(module=1) \| ~Query(module=2) == Not(And(Query(module=1), Query(module=2)))
415			assert ~Query(module=1) & ~Query(module=2) == Not(Or(Query(module=1), Query(module=2)))
416
417			assert ~Query(module=1) \| Query(module=2) == Or(Not(Query(module=1)), Query(module=2))
418			assert ~Query(module=1) & Query(module=2) == And(Not(Query(module=1)), Query(module=2))
419
420			assert ~(Query(module=1) & Query(module=2)) == Not(And(Query(module=1), Query(module=2)))
421			assert ~(Query(module=1) \| Query(module=2)) == Not(Or(Query(module=1), Query(module=2)))
422
423			assert repr(~Or(1, 2)) == repr(Not(Or(1, 2)))
424			assert repr(~And(1, 2)) == repr(Not(And(1, 2)))
425
426			assert repr(~Query(module=1) \| ~Query(module=2)) == repr(Not(And(Query(module=1), Query(module=2))))
427			assert repr(~Query(module=1) & ~Query(module=2)) == repr(Not(Or(Query(module=1), Query(module=2))))
428
429			assert repr(~(Query(module=1) & Query(module=2))) == repr(Not(And(Query(module=1), Query(module=2))))
430			assert repr(~(Query(module=1) \| Query(module=2))) == repr(Not(Or(Query(module=1), Query(module=2))))
431
432			assert Not(Q(module=1))({'module': 1}) == False
433
434
435			def test_predicate_query_allowed():
436			pytest.raises(TypeError, Query, 1)
437			pytest.raises(TypeError, Query, a=1)
438
439
440			def test_predicate_when_allowed():
441			pytest.raises(TypeError, When, 1)
442
443
444			@pytest.mark.parametrize('expr,inp,expected', [
445			({'module': "abc"}, {'module': "abc"}, True),
446			({'module': "abcd"}, {'module': "abc"}, False),
447			({'module': "abcd"}, {'module': "abce"}, False),
448			({'module': "abc"}, {'module': "abcd"}, False),
449			({'module': ("abc", "xyz")}, {'module': "abc"}, True),
450			({'module': ("abc", "xyz")}, {'module': "abcd"}, True),
451			({'module': ("abc", "xyz")}, {'module': "xyzw"}, True),
452			({'module': ("abc", "xyz")}, {'module': "fooabc"}, False),
453			({'module': ("abc", "xyz")}, {'module': 1}, False),
454
455			({'module': ["abc", "xyz"]}, {'module': "abc"}, True),
456			({'module': ["abc", "xyz"]}, {'module': "abcd"}, True),
457			({'module': ["abc", "xyz"]}, {'module': "xyzw"}, True),
458			({'module': ["abc", "xyz"]}, {'module': "fooabc"}, False),
459			({'module': ["abc", "xyz"]}, {'module': 1}, False),
460
461			({'module': {"abc", "xyz"}}, {'module': "abc"}, True),
462			({'module': {"abc", "xyz"}}, {'module': "abcd"}, True),
463			({'module': {"abc", "xyz"}}, {'module': "xyzw"}, True),
464			({'module': {"abc", "xyz"}}, {'module': "fooabc"}, False),
465			({'module': {"abc", "xyz"}}, {'module': 1}, False),
466
467			({'module': "abc"}, {'module': 1}, False),
468			])
469			def test_predicate_matching(expr, inp, expected):
470			assert Query(**expr)(inp) == expected
471
472
473			def test_predicate_when():
474			called = []
475			assert When(Q(module=1), lambda ev: called.append(ev))({'module': 2}) == False
476			assert called == []
477
478			assert When(Q(module=1), lambda ev: called.append(ev))({'module': 1}) == True
479			assert called == [{'module': 1}]
480
481			called = []
482			assert Q(module=1, action=lambda ev: called.append(ev))({'module': 1}) == True
483			assert called == [{'module': 1}]
484
485			called = [[], []]
486			predicate = (
487			Q(module=1, action=lambda ev: called[0].append(ev)) \|
488			Q(module=2, action=lambda ev: called[1].append(ev))
489			)
490			assert predicate({'module': 1}) == True
491			assert called == [[{'module': 1}], []]
492
493			assert predicate({'module': 2}) == True
494			assert called == [[{'module': 1}], [{'module': 2}]]
495
496			called = [[], []]
497			predicate = (
498			Q(module=1, action=lambda ev: called[0].append(ev)) &
499			Q(function=2, action=lambda ev: called[1].append(ev))
500			)
501			assert predicate({'module': 2}) == False
502			assert called == [[], []]
503
504			assert predicate({'module': 1, 'function': 2}) == True
505			assert called == [[{'module': 1, 'function': 2}], [{'module': 1, 'function': 2}]]
506
507
508			def test_proper_backend():
509			if os.environ.get('PUREPYTHONHUNTER') or platform.python_implementation() == 'PyPy':
510			assert 'hunter.tracer.Tracer' in repr(hunter.Tracer)
511			else:
512			assert 'hunter._tracer.Tracer' in repr(hunter.Tracer)
513
514
515			@pytest.fixture(scope="session", params=['pure', 'cython'])
516			def tracer_impl(request):
517			if request.param == 'pure':
518			return pytest.importorskip('hunter.tracer').Tracer
519			elif request.param == 'cython':
520			return pytest.importorskip('hunter._tracer').Tracer
521
522
523			def test_perf_filter(tracer_impl, benchmark):
524			t = tracer_impl()
525			import tokenize
526
527			@benchmark
528			def run():
529			with t.trace(Q(module="does-not-exist") \| Q(module="does not exist".split())):
530			with open(tokenize.__file__, 'rb') as fh:
531			toks = []
532			try:
533			for tok in tokenize.tokenize(fh.readline):
534			toks.append(tok)
535			except tokenize.TokenError as exc:
536			toks.append(exc)
537

ionelmc / python-hunter

Push — master ( af5503...cad96e )

tests.tracer_impl() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like