← Back to branch summary

~free.ekanayaka/storm/any-expr

« back to all changes in this revision

Viewing changes to tests/tracer.py

  • Committer: James Henstridge
  • Date: 2009-02-04 01:53:02 UTC
  • mto: This revision was merged to the branch mainline in revision 295.
  • Revision ID: james@jamesh.id.au-20090204015302-i1spb0m92tjsuyro
Change the tracer tests to not mock sys.stderr, as the test 
infrastructure itself uses sys.stderr.  Instead use a custom stream for 
the majority of tests and test that sys.stderr is used if a custom 
stream is not provided.

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/tracer.py
88
88
 
89
89
    def setUp(self):
90
90
        super(DebugTracerTest, self).setUp()
91
 
        self.tracer = DebugTracer()
 
91
        self.stream = self.mocker.mock(file)
 
92
        self.tracer = DebugTracer(self.stream)
92
93
 
93
94
        datetime_mock = self.mocker.replace("datetime.datetime")
94
95
        datetime_mock.now()
98
99
        self.variable = MockVariable("PARAM")
99
100
 
100
101
    def tearDown(self):
 
102
        del _tracers[:]
101
103
        super(DebugTracerTest, self).tearDown()
102
 
        del _tracers[:]
 
104
 
 
105
    def test_wb_debug_tracer_uses_stderr_by_default(self):
 
106
        self.mocker.replay()
 
107
 
 
108
        tracer = DebugTracer()
 
109
        self.assertEqual(tracer._stream, sys.stderr)
 
110
 
 
111
    def test_wb_debug_tracer_uses_first_arg_as_stream(self):
 
112
        self.mocker.replay()
 
113
 
 
114
        marker = object()
 
115
        tracer = DebugTracer(marker)
 
116
        self.assertEqual(tracer._stream, marker)
103
117
 
104
118
    def test_connection_raw_execute(self):
105
 
        stderr = self.mocker.replace("sys.stderr")
106
 
        stderr.write("[04:05:06.000007] EXECUTE: 'STATEMENT', ('PARAM',)\n")
107
 
        stderr.flush()
 
119
        self.stream.write(
 
120
            "[04:05:06.000007] EXECUTE: 'STATEMENT', ('PARAM',)\n")
 
121
        self.stream.flush()
108
122
        self.mocker.replay()
109
123
 
110
124
        connection = "CONNECTION"
116
130
                                           statement, params)
117
131
 
118
132
    def test_connection_raw_execute_with_non_variable(self):
119
 
        stderr = self.mocker.replace("sys.stderr")
120
 
        stderr.write("[04:05:06.000007] EXECUTE: 'STATEMENT', ('PARAM', 1)\n")
121
 
        stderr.flush()
 
133
        self.stream.write(
 
134
            "[04:05:06.000007] EXECUTE: 'STATEMENT', ('PARAM', 1)\n")
 
135
        self.stream.flush()
122
136
        self.mocker.replay()
123
137
 
124
138
        connection = "CONNECTION"
130
144
                                           statement, params)
131
145
 
132
146
    def test_connection_raw_execute_error(self):
133
 
        stderr = self.mocker.replace("sys.stderr")
134
 
        stderr.write("[04:05:06.000007] ERROR: ERROR\n")
135
 
        stderr.flush()
 
147
        self.stream.write("[04:05:06.000007] ERROR: ERROR\n")
 
148
        self.stream.flush()
136
149
        self.mocker.replay()
137
150
 
138
151
        connection = "CONNECTION"
145
158
                                                 statement, params, error)
146
159
 
147
160
    def test_connection_raw_execute_success(self):
148
 
        stderr = self.mocker.replace("sys.stderr")
149
 
        stderr.write("[04:05:06.000007] DONE\n")
150
 
        stderr.flush()
 
161
        self.stream.write("[04:05:06.000007] DONE\n")
 
162
        self.stream.flush()
151
163
        self.mocker.replay()
152
164
 
153
165
        connection = "CONNECTION"
158
170
        self.tracer.connection_raw_execute_success(connection, raw_cursor,
159
171
                                                   statement, params)
160
172
 
161
 
    def test_custom_stream(self):
162
 
        self.tracer = DebugTracer(sys.stdout)
163
 
 
164
 
        stdout = self.mocker.replace("sys.stdout")
165
 
        stdout.write("[04:05:06.000007] EXECUTE: 'STATEMENT', ('PARAM',)\n")
166
 
        stdout.flush()
167
 
        self.mocker.replay()
168
 
 
169
 
        connection = "CONNECTION"
170
 
        raw_cursor = "RAW_CURSOR"
171
 
        statement = "STATEMENT"
172
 
        params = [self.variable]
173
 
 
174
 
        self.tracer.connection_raw_execute(connection, raw_cursor,
175
 
                                           statement, params)
176
 
 
177
173
 
178
174
class TimeoutTracerTestBase(TestHelper):
179
175