text: from __future__ import with_statement
  
  import logging
  import webob
  from exam import fixture
  
  from raven.base import Client
  from raven.middleware import Sentry
  from raven.utils.compat import Iterator
  from raven.utils.testutils import TestCase
  
  
  class TempStoreClient(Client):
      def __init__(self, **kwargs):
          self.events = []
          super(TempStoreClient, self).__init__(**kwargs)
  
      def is_enabled(self):
          return True
  
      def send(self, **kwargs):
          self.events.append(kwargs)
  
  
  class ErroringIterable(Iterator):
      def __init__(self):
          self.closed = False
  
      def __iter__(self):
          return self
  
      def __next__(self):
          raise ValueError('hello world')
  
      def close(self):
          self.closed = True
  
  
  class ExitingIterable(ErroringIterable):
      def __init__(self, exc_func):
          self._exc_func = exc_func
  
      def __iter__(self):
          return self
  
      def __next__(self):
          raise self._exc_func()
  
  
  class SimpleIteratable(Iterator):
      def __init__(self):
          self.closed = False
          self._iter = iter(['a'])
  
      def __iter__(self):
          return self
  
      def __next__(self):
          return next(self._iter)
  
      def close(self):
          self.closed = True
  
  
  class ExampleApp(object):
      def __init__(self, iterable):
          self.iterable = iterable
  
      def __call__(self, environ, start_response):
          return self.iterable
  
  
  class MiddlewareTestCase(TestCase):
      @fixture
      def client(self):
          return TempStoreClient()
  
      @fixture
      def request(self):
          return webob.Request.blank('/an-error?foo=bar')
  
      def test_captures_error_in_iteration(self):
          iterable = ErroringIterable()
          app = ExampleApp(iterable)
          middleware = Sentry(app, client=self.client)
  
          response = middleware(self.request.environ, lambda *args: None)
  
          with self.assertRaises(ValueError):
              response = list(response)
  
          self.assertEquals(len(self.client.events), 1)
          event = self.client.events.pop(0)
  
          assert 'exception' in event
          exc = event['exception']['values'][-1]
          self.assertEquals(exc['type'], 'ValueError')
          self.assertEquals(exc['value'], 'hello world')
          self.assertEquals(event['level'], logging.ERROR)
          self.assertEquals(event['message'], 'ValueError: hello world')
  
          assert 'request' in event
          http = event['request']
          self.assertEquals(http['url'], 'http://localhost/an-error')
          self.assertEquals(http['query_string'], 'foo=bar')
          self.assertEquals(http['method'], 'GET')
          # self.assertEquals(http['data'], {'foo': 'bar'})
          headers = http['headers']
          self.assertTrue('Host' in headers, headers.keys())
          self.assertEquals(headers['Host'], 'localhost:80')
          env = http['env']
          self.assertTrue('SERVER_NAME' in env, env.keys())
          self.assertEquals(env['SERVER_NAME'], 'localhost')
          self.assertTrue('SERVER_PORT' in env, env.keys())
          self.assertEquals(env['SERVER_PORT'], '80')
  
      def test_systemexit_0_is_ignored(self):
          iterable = ExitingIterable(lambda: SystemExit(0))
          app = ExampleApp(iterable)
          middleware = Sentry(app, client=self.client)
  
          response = middleware(self.request.environ, lambda *args: None)
  
          with self.assertRaises(SystemExit):
              response = list(response)
  
          self.assertEquals(len(self.client.events), 0)
  
      def test_systemexit_is_captured(self):
          iterable = ExitingIterable(lambda: SystemExit(1))
          app = ExampleApp(iterable)
          middleware = Sentry(app, client=self.client)
  
          response = middleware(self.request.environ, lambda *args: None)
  
          with self.assertRaises(SystemExit):
              response = list(response)
  
          self.assertEquals(len(self.client.events), 1)
          event = self.client.events.pop(0)
  
          assert 'exception' in event
          exc = event['exception']['values'][-1]
          self.assertEquals(exc['type'], 'SystemExit')
          self.assertEquals(exc['value'], '1')
          self.assertEquals(event['level'], logging.ERROR)
          self.assertEquals(event['message'], 'SystemExit: 1')
  
      def test_keyboard_interrupt_is_captured(self):
          iterable = ExitingIterable(lambda: KeyboardInterrupt())
          app = ExampleApp(iterable)
          middleware = Sentry(app, client=self.client)
  
          response = middleware(self.request.environ, lambda *args: None)
  
          with self.assertRaises(KeyboardInterrupt):
              response = list(response)
  
          self.assertEquals(len(self.client.events), 1)
          event = self.client.events.pop(0)
  
          assert 'exception' in event
          exc = event['exception']['values'][-1]
          self.assertEquals(exc['type'], 'KeyboardInterrupt')
          self.assertEquals(exc['value'], '')
          self.assertEquals(event['level'], logging.ERROR)
  
      def test_close(self):
          iterable = SimpleIteratable()
          app = ExampleApp(iterable)
          middleware = Sentry(app, client=self.client)
  
          response = middleware(self.request.environ, lambda *args: None)
          list(response)  # exhaust iterator
          response.close()
          self.assertTrue(iterable.closed, True)
  