root/test_pyconquer.py

"""Test pyconquer.py"""

import os
localDir = os.path.dirname(__file__)
import pprint
try:
    from cStringIO import StringIO
except ImportError:
    from StringIO import StringIO
import sys
import threading
import unittest

import pyconquer


class Counter:
    def __init__(self):
        self.value = 0

def plus1(arg):
    newval = arg + 1
    return newval


class LoggerTests(unittest.TestCase):
    
    def testNormalFlow(self):
        c = Counter()
        
        tr = pyconquer.Logger("!(threading.py)")
        self.assertEqual(tr.events, set(['call', 'return', 'exception']))
        tr.watch(c, "value", name='c.value')
        tr.out = StringIO()
        try:
            tr.start()
            def inc():
                c.value += 1
            for i in xrange(3):
                threading.Thread(target=inc).start()
        finally:
            tr.stop()
        
        self.assertEqual(c.value, 3)
        self.assertEqual(tr.out.getvalue(), """

________ pyconquer tracing started ________

= c.value: 0

[ MainThread ]
* testNormalFlow (test_pyconquer:%(a)s) New Thread: Thread-1

        [ Thread-1 ]
        > inc (test_pyconquer:%(b)s)
        < inc (test_pyconquer:%(c)s)
        = c.value: 1
        X Thread terminated: Thread-1

[ MainThread ]
* testNormalFlow (test_pyconquer:%(a)s) New Thread: Thread-2

        [ Thread-2 ]
        > inc (test_pyconquer:%(b)s)
        < inc (test_pyconquer:%(c)s)
        = c.value: 2
        X Thread terminated: Thread-2

[ MainThread ]
* testNormalFlow (test_pyconquer:%(a)s) New Thread: Thread-3

        [ Thread-3 ]
        > inc (test_pyconquer:%(b)s)
        < inc (test_pyconquer:%(c)s)
        = c.value: 3
        X Thread terminated: Thread-3

________ pyconquer tracing stopped ________
""" % {'a': tr._caller_lineno + 4,
       'b': tr._caller_lineno + 1,
       'c': tr._caller_lineno + 2})
        
        # Test 1) a start() after stop(),
        #      2) lines in this frame, and
        #      3) a call inside a line
        tr.events.add('line')
        tr.watches.clear()
        tr.out = StringIO()
        try:
            tr.start()
            c.value = 0
            c.value = plus1(plus1(1))
        finally:
            tr.stop()
        
        self.assertEqual(c.value, 3)
        self.assertEqual(tr.out.getvalue(), """

________ pyconquer tracing started ________


[ MainThread ]
  testNormalFlow (test_pyconquer:%(x)s)
  testNormalFlow (test_pyconquer:%(y)s)
> plus1 (test_pyconquer:%(a)s)
  plus1 (test_pyconquer:%(b)s)
  plus1 (test_pyconquer:%(c)s)
< plus1 (test_pyconquer:%(c)s): 2
> plus1 (test_pyconquer:%(a)s)
  plus1 (test_pyconquer:%(b)s)
  plus1 (test_pyconquer:%(c)s)
< plus1 (test_pyconquer:%(c)s): 3
  testNormalFlow (test_pyconquer:%(z)s)

________ pyconquer tracing stopped ________
""" % {'a': 21, 'b': 22, 'c': 23,
       'x': tr._caller_lineno + 1,
       'y': tr._caller_lineno + 2,
       'z': tr._caller_lineno + 4})
    
    def testCEvents(self):
        c = Counter()
        tr = pyconquer.Logger(events=('call', 'return', 'exception',
                                      'c_call', 'c_return', 'c_exception'))
        tr.out = StringIO()
        try:
            tr.start()
            # Trace c_call and c_return
            c.value = sys.getrecursionlimit()
            # Raise a c_exception
            try:
                c.value = chr('a')
            except TypeError:
                pass
        finally:
            tr.stop()
        
        self.assertEqual(tr.out.getvalue(), """

________ pyconquer tracing started ________


[ MainThread ]
>>testCEvents (test_pyconquer:%(a)s): <built-in function getrecursionlimit>
<<testCEvents (test_pyconquer:%(a)s): <built-in function getrecursionlimit>
>>testCEvents (test_pyconquer:%(b)s): <built-in function chr>
EEtestCEvents (test_pyconquer:%(b)s): <built-in function chr>
E testCEvents (test_pyconquer:%(b)s): TypeError

________ pyconquer tracing stopped ________
""" % {'a': tr._caller_lineno + 2,
       'b': tr._caller_lineno + 5})


class VerifierTests(unittest.TestCase):
    
    def test_0_Tree(self):
        pt = pyconquer.Verifier()
        
        def three():
            return 3
        pt.compile(three)
        
        # Assert the first four Nodes.
        level1 = pt.tree.children
        self.assertEqual(len(level1), 2)
        self.assertEqual(len(level1[0].children), 2)
        self.assertEqual(len(level1[1].children), 2)
        
        def a_is_1():
            a = 1
            b = a
        pt.compile(a_is_1)
        
        # Assert the first four Nodes.
        level1 = pt.tree.children
        self.assertEqual(len(level1), 2)
        self.assertEqual(len(level1[0].children), 2)
        self.assertEqual(len(level1[1].children), 2)
    
    def test_1_SequenceTests(self):
        c = Counter()
        a = [c.value]
        def test():
            a[0] += 2 # One for each thread!
            return c.value == a[0]
        
        pt = pyconquer.Verifier()
##        pt = pyconquer.Verifier(pyconquer.all_events)
        
        # Test a "safe" increment function. This isn't truly safe,
        # because Python threads may switch even between the
        # bytecodes LOAD_ATTR (value), and LOAD_CONST (1).
        # [But LOAD_CONST should "goto fast_next_opcode" on its
        # way to INPLACE_ADD and skip releasing the GIL].
        # But it passes our test. ;)
        def safe_inc():
            c.value += 1
        try:
            pt.compile(safe_inc)
        except pyconquer.VerificationError, x:
            print
            for node in pt.trail:
                print node
            print "TREE:", pt.tree
            self.fail("safe_inc failed")


if __name__ == '__main__':
    unittest.main()