tests.test_predicate_bad_query() - Code Metrics - Inspection of "Tweak a bit the new perf test." - ionelmc/python-hunter - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 67ac21...36843e )

by Ionel Cristian

created 2015-12-23 01:54 UTC

tests.test_predicate_bad_query() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
dl	0
loc	11
rs	9.4286
cc	1

from __future__ import print_function

import inspect
import os
import platform
import subprocess
import sys
import tokenize

try:
    from cStringIO 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_eq={'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_eq={'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_eq={'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_eq={'module': 'a'}>, " 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'))"

    assert "predicates.And: predicates=(<hunter." in repr(Q(module='a') & Q(module='b'))
    assert "predicates.Query: query_eq={'module': 'a'}>," 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')) == "And(Query(module='a'), Query(module='b'))"


def test_predicate_q_nest_1():
    assert repr(Q(Q(module='a'))).endswith("predicates.Query: query_eq={'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_startswith': "abc"}, {'module': "abcd"}, True),
    ({'module__startswith': "abc"}, {'module': "abcd"}, True),
    ({'module_contains': "bc"}, {'module': "abcd"}, True),

    ({'module_endswith': "abc"}, {'module': "abcd"}, False),
    ({'module__endswith': "bcd"}, {'module': "abcd"}, True),

    ({'module_in': "abcd"}, {'module': "bc"}, True),
    ({'module': "abcd"}, {'module': "bc"}, False),
    ({'module': ["abcd"]}, {'module': "bc"}, False),
    ({'module_in': ["abcd"]}, {'module': "bc"}, False),
    ({'module_in': ["a", "bc", "d"]}, {'module': "bc"}, True),

    ({'module': "abcd"}, {'module': "abc"}, False),

    ({'module_startswith': ("abc", "xyz")}, {'module': "abc"}, True),
    ({'module_startswith': {"abc", "xyz"}}, {'module': "abc"}, True),
    ({'module_startswith': ["abc", "xyz"]}, {'module': "abc"}, True),
    ({'module_startswith': ("abc", "xyz")}, {'module': "abcd"}, True),

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

    ({'module': "abc"}, {'module': 1}, False),

    ({'module_regex': r"(re|sre.*)\b"}, {'module': "regex"}, False),
    ({'module_regex': r"(re|sre.*)\b"}, {'module': "re.gex"}, True),
    ({'module_regex': r"(re|sre.*)\b"}, {'module': "sregex"}, True),
    ({'module_regex': r"(re|sre.*)\b"}, {'module': "re"}, True),
])
def test_predicate_matching(expr, inp, expected):
    assert Query(**expr)(inp) == expected


@pytest.mark.parametrize('exc_type,expr', [
    (TypeError, {'module_1': 1}),
    (TypeError, {'module1': 1}),
    (ValueError, {'module_startswith': 1}),
    (ValueError, {'module_startswith': {1: 2}}),
    (ValueError, {'module_endswith': 1}),
    (ValueError, {'module_endswith': {1: 2}}),
    (TypeError, {'module_foo': 1}),
])
def test_predicate_bad_query(expr, exc_type):
    pytest.raises(exc_type, Query, **expr)


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 _tokenize():
    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)


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

    @benchmark
    def run():
        with t.trace(Q(module="does-not-exist") | Q(module="does not exist".split())):
            _tokenize()


def test_perf_actions(tracer_impl, benchmark):
    t = tracer_impl()

    @benchmark
    def run():
        output = StringIO()
        with t.trace(Q(
            ~Q(module_in=['re', 'sre', 'sre_parse']) & ~Q(module_startswith='namedtuple') & Q(kind="call"),
            actions=[
                CodePrinter(
                    stream=output
                ),
                VarsPrinter(
                    'line',
                    globals=True,
                    stream=output
                )
            ]
        )):
            _tokenize()


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

ionelmc / python-hunter

Push — master ( 67ac21...36843e )

tests.test_predicate_bad_query() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like