text: import unittest
  import io
  
  class TestThreadedTaskDispatcher(unittest.TestCase):
  
      def _makeOne(self):
          from waitress.task import ThreadedTaskDispatcher
          return ThreadedTaskDispatcher()
  
      def test_handler_thread_task_is_None(self):
          inst = self._makeOne()
          inst.threads[0] = True
          inst.queue.put(None)
          inst.handler_thread(0)
          self.assertEqual(inst.stop_count, -1)
          self.assertEqual(inst.threads, {})
  
      def test_handler_thread_task_raises(self):
          from waitress.task import JustTesting
          inst = self._makeOne()
          inst.threads[0] = True
          inst.logger = DummyLogger()
          task = DummyTask(JustTesting)
          inst.logger = DummyLogger()
          inst.queue.put(task)
          inst.handler_thread(0)
          self.assertEqual(inst.stop_count, -1)
          self.assertEqual(inst.threads, {})
          self.assertEqual(len(inst.logger.logged), 1)
  
      def test_set_thread_count_increase(self):
          inst = self._makeOne()
          L = []
          inst.start_new_thread = lambda *x: L.append(x)
          inst.set_thread_count(1)
          self.assertEqual(L, [(inst.handler_thread, (0,))])
  
      def test_set_thread_count_increase_with_existing(self):
          inst = self._makeOne()
          L = []
          inst.threads = {0: 1}
          inst.start_new_thread = lambda *x: L.append(x)
          inst.set_thread_count(2)
          self.assertEqual(L, [(inst.handler_thread, (1,))])
  
      def test_set_thread_count_decrease(self):
          inst = self._makeOne()
          inst.threads = {'a': 1, 'b': 2}
          inst.set_thread_count(1)
          self.assertEqual(inst.queue.qsize(), 1)
          self.assertEqual(inst.queue.get(), None)
  
      def test_set_thread_count_same(self):
          inst = self._makeOne()
          L = []
          inst.start_new_thread = lambda *x: L.append(x)
          inst.threads = {0: 1}
          inst.set_thread_count(1)
          self.assertEqual(L, [])
  
      def test_add_task(self):
          task = DummyTask()
          inst = self._makeOne()
          inst.add_task(task)
          self.assertEqual(inst.queue.qsize(), 1)
          self.assertTrue(task.deferred)
  
      def test_add_task_defer_raises(self):
          task = DummyTask(ValueError)
          inst = self._makeOne()
          self.assertRaises(ValueError, inst.add_task, task)
          self.assertEqual(inst.queue.qsize(), 0)
          self.assertTrue(task.deferred)
          self.assertTrue(task.cancelled)
  
      def test_shutdown_one_thread(self):
          inst = self._makeOne()
          inst.threads[0] = 1
          inst.logger = DummyLogger()
          task = DummyTask()
          inst.queue.put(task)
          self.assertEqual(inst.shutdown(timeout=.01), True)
          self.assertEqual(inst.logger.logged, ['1 thread(s) still running'])
          self.assertEqual(task.cancelled, True)
  
      def test_shutdown_no_threads(self):
          inst = self._makeOne()
          self.assertEqual(inst.shutdown(timeout=.01), True)
  
      def test_shutdown_no_cancel_pending(self):
          inst = self._makeOne()
          self.assertEqual(inst.shutdown(cancel_pending=False, timeout=.01),
                           False)
  
  class TestTask(unittest.TestCase):
  
      def _makeOne(self, channel=None, request=None):
          if channel is None:
              channel = DummyChannel()
          if request is None:
              request = DummyParser()
          from waitress.task import Task
          return Task(channel, request)
  
      def test_ctor_version_not_in_known(self):
          request = DummyParser()
          request.version = '8.4'
          inst = self._makeOne(request=request)
          self.assertEqual(inst.version, '1.0')
  
      def test_cancel(self):
          inst = self._makeOne()
          inst.cancel()
          self.assertTrue(inst.close_on_finish)
  
      def test_defer(self):
          inst = self._makeOne()
          self.assertEqual(inst.defer(), None)
  
      def test_build_response_header_bad_http_version(self):
          inst = self._makeOne()
          inst.request = DummyParser()
          inst.version = '8.4'
          self.assertRaises(AssertionError, inst.build_response_header)
  
      def test_build_response_header_v10_keepalive_no_content_length(self):
          inst = self._makeOne()
          inst.request = DummyParser()
          inst.request.headers['CONNECTION'] = 'keep-alive'
          inst.version = '1.0'
          result = inst.build_response_header()
          lines = filter_lines(result)
          self.assertEqual(len(lines), 4)
          self.assertEqual(lines[0], b'HTTP/1.0 200 OK')
          self.assertEqual(lines[1], b'Connection: close')
          self.assertTrue(lines[2].startswith(b'Date:'))
          self.assertEqual(lines[3], b'Server: waitress')
          self.assertEqual(inst.close_on_finish, True)
          self.assertTrue(('Connection', 'close') in inst.response_headers)
  
      def test_build_response_header_v10_keepalive_with_content_length(self):
          inst = self._makeOne()
          inst.request = DummyParser()
          inst.request.headers['CONNECTION'] = 'keep-alive'
          inst.response_headers = [('Content-Length', '10')]
          inst.version = '1.0'
          inst.content_length = 0
          result = inst.build_response_header()
          lines = filter_lines(result)
          self.assertEqual(len(lines), 5)
          self.assertEqual(lines[0], b'HTTP/1.0 200 OK')
          self.assertEqual(lines[1], b'Connection: Keep-Alive')
          self.assertEqual(lines[2], b'Content-Length: 10')
          self.assertTrue(lines[3].startswith(b'Date:'))
          self.assertEqual(lines[4], b'Server: waitress')
          self.assertEqual(inst.close_on_finish, False)
  
      def test_build_response_header_v11_connection_closed_by_client(self):
          inst = self._makeOne()
          inst.request = DummyParser()
          inst.version = '1.1'
          inst.request.headers['CONNECTION'] = 'close'
          result = inst.build_response_header()
          lines = filter_lines(result)
          self.assertEqual(len(lines), 5)
          self.assertEqual(lines[0], b'HTTP/1.1 200 OK')
          self.assertEqual(lines[1], b'Connection: close')
          self.assertTrue(lines[2].startswith(b'Date:'))
          self.assertEqual(lines[3], b'Server: waitress')
          self.assertEqual(lines[4], b'Transfer-Encoding: chunked')
          self.assertTrue(('Connection', 'close') in inst.response_headers)
          self.assertEqual(inst.close_on_finish, True)
  
      def test_build_response_header_v11_connection_keepalive_by_client(self):
          inst = self._makeOne()
          inst.request = DummyParser()
          inst.request.headers['CONNECTION'] = 'keep-alive'
          inst.version = '1.1'
          result = inst.build_response_header()
          lines = filter_lines(result)
          self.assertEqual(len(lines), 5)
          self.assertEqual(lines[0], b'HTTP/1.1 200 OK')
          self.assertEqual(lines[1], b'Connection: close')
          self.assertTrue(lines[2].startswith(b'Date:'))
          self.assertEqual(lines[3], b'Server: waitress')
          self.assertEqual(lines[4], b'Transfer-Encoding: chunked')
          self.assertTrue(('Connection', 'close') in inst.response_headers)
          self.assertEqual(inst.close_on_finish, True)
  
      def test_build_response_header_v11_200_no_content_length(self):
          inst = self._makeOne()
          inst.request = DummyParser()
          inst.version = '1.1'
          result = inst.build_response_header()
          lines = filter_lines(result)
          self.assertEqual(len(lines), 5)
          self.assertEqual(lines[0], b'HTTP/1.1 200 OK')
          self.assertEqual(lines[1], b'Connection: close')
          self.assertTrue(lines[2].startswith(b'Date:'))
          self.assertEqual(lines[3], b'Server: waitress')
          self.assertEqual(lines[4], b'Transfer-Encoding: chunked')
          self.assertEqual(inst.close_on_finish, True)
          self.assertTrue(('Connection', 'close') in inst.response_headers)
  
      def test_build_response_header_via_added(self):
          inst = self._makeOne()
          inst.request = DummyParser()
          inst.version = '1.0'
          inst.response_headers = [('Server', 'abc')]
          result = inst.build_response_header()
          lines = filter_lines(result)
          self.assertEqual(len(lines), 5)
          self.assertEqual(lines[0], b'HTTP/1.0 200 OK')
          self.assertEqual(lines[1], b'Connection: close')
          self.assertTrue(lines[2].startswith(b'Date:'))
          self.assertEqual(lines[3], b'Server: abc')
          self.assertEqual(lines[4], b'Via: waitress')
  
      def test_build_response_header_date_exists(self):
          inst = self._makeOne()
          inst.request = DummyParser()
          inst.version = '1.0'
          inst.response_headers = [('Date', 'date')]
          result = inst.build_response_header()
          lines = filter_lines(result)
          self.assertEqual(len(lines), 4)
          self.assertEqual(lines[0], b'HTTP/1.0 200 OK')
          self.assertEqual(lines[1], b'Connection: close')
          self.assertTrue(lines[2].startswith(b'Date:'))
          self.assertEqual(lines[3], b'Server: waitress')
  
      def test_build_response_header_preexisting_content_length(self):
          inst = self._makeOne()
          inst.request = DummyParser()
          inst.version = '1.1'
          inst.content_length = 100
          result = inst.build_response_header()
          lines = filter_lines(result)
          self.assertEqual(len(lines), 4)
          self.assertEqual(lines[0], b'HTTP/1.1 200 OK')
          self.assertEqual(lines[1], b'Content-Length: 100')
          self.assertTrue(lines[2].startswith(b'Date:'))
          self.assertEqual(lines[3], b'Server: waitress')
  
      def test_remove_content_length_header(self):
          inst = self._makeOne()
          inst.response_headers = [('Content-Length', '70')]
          inst.remove_content_length_header()
          self.assertEqual(inst.response_headers, [])
  
      def test_start(self):
          inst = self._makeOne()
          inst.start()
          self.assertTrue(inst.start_time)
  
      def test_finish_didnt_write_header(self):
          inst = self._makeOne()
          inst.wrote_header = False
          inst.complete = True
          inst.finish()
          self.assertTrue(inst.channel.written)
  
      def test_finish_wrote_header(self):
          inst = self._makeOne()
          inst.wrote_header = True
          inst.finish()
          self.assertFalse(inst.channel.written)
  
      def test_finish_chunked_response(self):
          inst = self._makeOne()
          inst.wrote_header = True
          inst.chunked_response = True
          inst.finish()
          self.assertEqual(inst.channel.written, b'0\r\n\r\n')
  
      def test_write_wrote_header(self):
          inst = self._makeOne()
          inst.wrote_header = True
          inst.complete = True
          inst.content_length = 3
          inst.write(b'abc')
          self.assertEqual(inst.channel.written, b'abc')
  
      def test_write_header_not_written(self):
          inst = self._makeOne()
          inst.wrote_header = False
          inst.complete = True
          inst.write(b'abc')
          self.assertTrue(inst.channel.written)
          self.assertEqual(inst.wrote_header, True)
  
      def test_write_start_response_uncalled(self):
          inst = self._makeOne()
          self.assertRaises(RuntimeError, inst.write, b'')
  
      def test_write_chunked_response(self):
          inst = self._makeOne()
          inst.wrote_header = True
          inst.chunked_response = True
          inst.complete = True
          inst.write(b'abc')
          self.assertEqual(inst.channel.written, b'3\r\nabc\r\n')
  
      def test_write_preexisting_content_length(self):
          inst = self._makeOne()
          inst.wrote_header = True
          inst.complete = True
          inst.content_length = 1
          inst.logger = DummyLogger()
          inst.write(b'abc')
          self.assertTrue(inst.channel.written)
          self.assertEqual(inst.logged_write_excess, True)
          self.assertEqual(len(inst.logger.logged), 1)
  
  class TestWSGITask(unittest.TestCase):
  
      def _makeOne(self, channel=None, request=None):
          if channel is None:
              channel = DummyChannel()
          if request is None:
              request = DummyParser()
          from waitress.task import WSGITask
          return WSGITask(channel, request)
  
      def test_service(self):
          inst = self._makeOne()
          def execute():
              inst.executed = True
          inst.execute = execute
          inst.complete = True
          inst.service()
          self.assertTrue(inst.start_time)
          self.assertTrue(inst.close_on_finish)
          self.assertTrue(inst.channel.written)
          self.assertEqual(inst.executed, True)
  
      def test_service_server_raises_socket_error(self):
          import socket
          inst = self._makeOne()
          def execute():
              raise socket.error
          inst.execute = execute
          self.assertRaises(socket.error, inst.service)
          self.assertTrue(inst.start_time)
          self.assertTrue(inst.close_on_finish)
          self.assertFalse(inst.channel.written)
  
      def test_execute_app_calls_start_response_twice_wo_exc_info(self):
          def app(environ, start_response):
              start_response('200 OK', [])
              start_response('200 OK', [])
          inst = self._makeOne()
          inst.channel.server.application = app
          self.assertRaises(AssertionError, inst.execute)
  
      def test_execute_app_calls_start_response_w_exc_info_complete(self):
          def app(environ, start_response):
              start_response('200 OK', [], [ValueError, ValueError(), None])
              return [b'a']
          inst = self._makeOne()
          inst.complete = True
          inst.channel.server.application = app
          inst.execute()
          self.assertTrue(inst.complete)
          self.assertEqual(inst.status, '200 OK')
          self.assertTrue(inst.channel.written)
  
      def test_execute_app_calls_start_response_w_excinf_headers_unwritten(self):
          def app(environ, start_response):
              start_response('200 OK', [], [ValueError, None, None])
              return [b'a']
          inst = self._makeOne()
          inst.wrote_header = False
          inst.channel.server.application = app
          inst.response_headers = [('a', 'b')]
          inst.execute()
          self.assertTrue(inst.complete)
          self.assertEqual(inst.status, '200 OK')
          self.assertTrue(inst.channel.written)
          self.assertFalse(('a','b') in inst.response_headers)
  
      def test_execute_app_calls_start_response_w_excinf_headers_written(self):
          def app(environ, start_response):
              start_response('200 OK', [], [ValueError, ValueError(), None])
          inst = self._makeOne()
          inst.complete = True
          inst.wrote_header = True
          inst.channel.server.application = app
          self.assertRaises(ValueError, inst.execute)
  
      def test_execute_bad_header_key(self):
          def app(environ, start_response):
              start_response('200 OK', [(None, 'a')])
          inst = self._makeOne()
          inst.channel.server.application = app
          self.assertRaises(AssertionError, inst.execute)
  
      def test_execute_bad_header_value(self):
          def app(environ, start_response):
              start_response('200 OK', [('a', None)])
          inst = self._makeOne()
          inst.channel.server.application = app
          self.assertRaises(AssertionError, inst.execute)
  
      def test_execute_hopbyhop_header(self):
          def app(environ, start_response):
              start_response('200 OK', [('Connection', 'close')])
          inst = self._makeOne()
          inst.channel.server.application = app
          self.assertRaises(AssertionError, inst.execute)
  
      def test_execute_bad_header_value_control_characters(self):
          def app(environ, start_response):
              start_response('200 OK', [('a', '\n')])
          inst = self._makeOne()
          inst.channel.server.application = app
          self.assertRaises(ValueError, inst.execute)
  
      def test_execute_bad_header_name_control_characters(self):
          def app(environ, start_response):
              start_response('200 OK', [('a\r', 'value')])
          inst = self._makeOne()
          inst.channel.server.application = app
          self.assertRaises(ValueError, inst.execute)
  
      def test_execute_bad_status_control_characters(self):
          def app(environ, start_response):
              start_response('200 OK\r', [])
          inst = self._makeOne()
          inst.channel.server.application = app
          self.assertRaises(ValueError, inst.execute)
  
      def test_preserve_header_value_order(self):
          def app(environ, start_response):
              write = start_response('200 OK', [('C', 'b'), ('A', 'b'), ('A', 'a')])
              write(b'abc')
              return []
          inst = self._makeOne()
          inst.channel.server.application = app
          inst.execute()
          self.assertTrue(b'A: b\r\nA: a\r\nC: b\r\n' in inst.channel.written)
  
      def test_execute_bad_status_value(self):
          def app(environ, start_response):
              start_response(None, [])
          inst = self._makeOne()
          inst.channel.server.application = app
          self.assertRaises(AssertionError, inst.execute)
  
      def test_execute_with_content_length_header(self):
          def app(environ, start_response):
              start_response('200 OK', [('Content-Length', '1')])
              return [b'a']
          inst = self._makeOne()
          inst.channel.server.application = app
          inst.execute()
          self.assertEqual(inst.content_length, 1)
  
      def test_execute_app_calls_write(self):
          def app(environ, start_response):
              write = start_response('200 OK', [('Content-Length', '3')])
              write(b'abc')
              return []
          inst = self._makeOne()
          inst.channel.server.application = app
          inst.execute()
          self.assertEqual(inst.channel.written[-3:], b'abc')
  
      def test_execute_app_returns_len1_chunk_without_cl(self):
          def app(environ, start_response):
              start_response('200 OK', [])
              return [b'abc']
          inst = self._makeOne()
          inst.channel.server.application = app
          inst.execute()
          self.assertEqual(inst.content_length, 3)
  
      def test_execute_app_returns_empty_chunk_as_first(self):
          def app(environ, start_response):
              start_response('200 OK', [])
              return ['', b'abc']
          inst = self._makeOne()
          inst.channel.server.application = app
          inst.execute()
          self.assertEqual(inst.content_length, None)
  
      def test_execute_app_returns_too_many_bytes(self):
          def app(environ, start_response):
              start_response('200 OK', [('Content-Length', '1')])
              return [b'abc']
          inst = self._makeOne()
          inst.channel.server.application = app
          inst.logger = DummyLogger()
          inst.execute()
          self.assertEqual(inst.close_on_finish, True)
          self.assertEqual(len(inst.logger.logged), 1)
  
      def test_execute_app_returns_too_few_bytes(self):
          def app(environ, start_response):
              start_response('200 OK', [('Content-Length', '3')])
              return [b'a']
          inst = self._makeOne()
          inst.channel.server.application = app
          inst.logger = DummyLogger()
          inst.execute()
          self.assertEqual(inst.close_on_finish, True)
          self.assertEqual(len(inst.logger.logged), 1)
  
      def test_execute_app_do_not_warn_on_head(self):
          def app(environ, start_response):
              start_response('200 OK', [('Content-Length', '3')])
              return [b'']
          inst = self._makeOne()
          inst.request.command = 'HEAD'
          inst.channel.server.application = app
          inst.logger = DummyLogger()
          inst.execute()
          self.assertEqual(inst.close_on_finish, True)
          self.assertEqual(len(inst.logger.logged), 0)
  
      def test_execute_app_returns_closeable(self):
          class closeable(list):
              def close(self):
                  self.closed = True
          foo = closeable([b'abc'])
          def app(environ, start_response):
              start_response('200 OK', [('Content-Length', '3')])
              return foo
          inst = self._makeOne()
          inst.channel.server.application = app
          inst.execute()
          self.assertEqual(foo.closed, True)
  
      def test_execute_app_returns_filewrapper_prepare_returns_True(self):
          from waitress.buffers import ReadOnlyFileBasedBuffer
          f = io.BytesIO(b'abc')
          app_iter = ReadOnlyFileBasedBuffer(f, 8192)
          def app(environ, start_response):
              start_response('200 OK', [('Content-Length', '3')])
              return app_iter
          inst = self._makeOne()
          inst.channel.server.application = app
          inst.execute()
          self.assertTrue(inst.channel.written) # header
          self.assertEqual(inst.channel.otherdata, [app_iter])
  
      def test_execute_app_returns_filewrapper_prepare_returns_True_nocl(self):
          from waitress.buffers import ReadOnlyFileBasedBuffer
          f = io.BytesIO(b'abc')
          app_iter = ReadOnlyFileBasedBuffer(f, 8192)
          def app(environ, start_response):
              start_response('200 OK', [])
              return app_iter
          inst = self._makeOne()
          inst.channel.server.application = app
          inst.execute()
          self.assertTrue(inst.channel.written) # header
          self.assertEqual(inst.channel.otherdata, [app_iter])
          self.assertEqual(inst.content_length, 3)
  
      def test_execute_app_returns_filewrapper_prepare_returns_True_badcl(self):
          from waitress.buffers import ReadOnlyFileBasedBuffer
          f = io.BytesIO(b'abc')
          app_iter = ReadOnlyFileBasedBuffer(f, 8192)
          def app(environ, start_response):
              start_response('200 OK', [])
              return app_iter
          inst = self._makeOne()
          inst.channel.server.application = app
          inst.content_length = 10
          inst.response_headers = [('Content-Length', '10')]
          inst.execute()
          self.assertTrue(inst.channel.written) # header
          self.assertEqual(inst.channel.otherdata, [app_iter])
          self.assertEqual(inst.content_length, 3)
          self.assertEqual(dict(inst.response_headers)['Content-Length'], '3')
  
      def test_get_environment_already_cached(self):
          inst = self._makeOne()
          inst.environ = object()
          self.assertEqual(inst.get_environment(), inst.environ)
  
      def test_get_environment_path_startswith_more_than_one_slash(self):
          inst = self._makeOne()
          request = DummyParser()
          request.path = '///abc'
          inst.request = request
          environ = inst.get_environment()
          self.assertEqual(environ['PATH_INFO'], '/abc')
  
      def test_get_environment_path_empty(self):
          inst = self._makeOne()
          request = DummyParser()
          request.path = ''
          inst.request = request
          environ = inst.get_environment()
          self.assertEqual(environ['PATH_INFO'], '')
  
      def test_get_environment_no_query(self):
          inst = self._makeOne()
          request = DummyParser()
          inst.request = request
          environ = inst.get_environment()
          self.assertEqual(environ['QUERY_STRING'], '')
  
      def test_get_environment_with_query(self):
          inst = self._makeOne()
          request = DummyParser()
          request.query = 'abc'
          inst.request = request
          environ = inst.get_environment()
          self.assertEqual(environ['QUERY_STRING'], 'abc')
  
      def test_get_environ_with_url_prefix_miss(self):
          inst = self._makeOne()
          inst.channel.server.adj.url_prefix = '/foo'
          request = DummyParser()
          request.path = '/bar'
          inst.request = request
          environ = inst.get_environment()
          self.assertEqual(environ['PATH_INFO'], '/bar')
          self.assertEqual(environ['SCRIPT_NAME'], '/foo')
  
      def test_get_environ_with_url_prefix_hit(self):
          inst = self._makeOne()
          inst.channel.server.adj.url_prefix = '/foo'
          request = DummyParser()
          request.path = '/foo/fuz'
          inst.request = request
          environ = inst.get_environment()
          self.assertEqual(environ['PATH_INFO'], '/fuz')
          self.assertEqual(environ['SCRIPT_NAME'], '/foo')
  
      def test_get_environ_with_url_prefix_empty_path(self):
          inst = self._makeOne()
          inst.channel.server.adj.url_prefix = '/foo'
          request = DummyParser()
          request.path = '/foo'
          inst.request = request
          environ = inst.get_environment()
          self.assertEqual(environ['PATH_INFO'], '')
          self.assertEqual(environ['SCRIPT_NAME'], '/foo')
  
      def test_get_environment_values(self):
          import sys
          inst = self._makeOne()
          request = DummyParser()
          request.headers = {
              'CONTENT_TYPE': 'abc',
              'CONTENT_LENGTH': '10',
              'X_FOO': 'BAR',
              'CONNECTION': 'close',
          }
          request.query = 'abc'
          inst.request = request
          environ = inst.get_environment()
  
          # nail the keys of environ
          self.assertEqual(sorted(environ.keys()), [
              'CONTENT_LENGTH', 'CONTENT_TYPE', 'HTTP_CONNECTION', 'HTTP_X_FOO',
              'PATH_INFO', 'QUERY_STRING', 'REMOTE_ADDR', 'REQUEST_METHOD',
              'SCRIPT_NAME', 'SERVER_NAME', 'SERVER_PORT', 'SERVER_PROTOCOL',
              'SERVER_SOFTWARE', 'wsgi.errors', 'wsgi.file_wrapper', 'wsgi.input',
              'wsgi.multiprocess', 'wsgi.multithread', 'wsgi.run_once',
              'wsgi.url_scheme', 'wsgi.version'])
  
          self.assertEqual(environ['REQUEST_METHOD'], 'GET')
          self.assertEqual(environ['SERVER_PORT'], '80')
          self.assertEqual(environ['SERVER_NAME'], 'localhost')
          self.assertEqual(environ['SERVER_SOFTWARE'], 'waitress')
          self.assertEqual(environ['SERVER_PROTOCOL'], 'HTTP/1.0')
          self.assertEqual(environ['SCRIPT_NAME'], '')
          self.assertEqual(environ['HTTP_CONNECTION'], 'close')
          self.assertEqual(environ['PATH_INFO'], '/')
          self.assertEqual(environ['QUERY_STRING'], 'abc')
          self.assertEqual(environ['REMOTE_ADDR'], '127.0.0.1')
          self.assertEqual(environ['CONTENT_TYPE'], 'abc')
          self.assertEqual(environ['CONTENT_LENGTH'], '10')
          self.assertEqual(environ['HTTP_X_FOO'], 'BAR')
          self.assertEqual(environ['wsgi.version'], (1, 0))
          self.assertEqual(environ['wsgi.url_scheme'], 'http')
          self.assertEqual(environ['wsgi.errors'], sys.stderr)
          self.assertEqual(environ['wsgi.multithread'], True)
          self.assertEqual(environ['wsgi.multiprocess'], False)
          self.assertEqual(environ['wsgi.run_once'], False)
          self.assertEqual(environ['wsgi.input'], 'stream')
          self.assertEqual(inst.environ, environ)
  
      def test_get_environment_values_w_scheme_override_untrusted(self):
          inst = self._makeOne()
          request = DummyParser()
          request.headers = {
              'CONTENT_TYPE': 'abc',
              'CONTENT_LENGTH': '10',
              'X_FOO': 'BAR',
              'X_FORWARDED_PROTO': 'https',
              'CONNECTION': 'close',
          }
          request.query = 'abc'
          inst.request = request
          environ = inst.get_environment()
          self.assertEqual(environ['wsgi.url_scheme'], 'http')
  
      def test_get_environment_values_w_scheme_override_trusted(self):
          import sys
          inst = self._makeOne()
          inst.channel.addr = ['192.168.1.1']
          inst.channel.server.adj.trusted_proxy = '192.168.1.1'
          request = DummyParser()
          request.headers = {
              'CONTENT_TYPE': 'abc',
              'CONTENT_LENGTH': '10',
              'X_FOO': 'BAR',
              'X_FORWARDED_PROTO': 'https',
              'CONNECTION': 'close',
          }
          request.query = 'abc'
          inst.request = request
          environ = inst.get_environment()
  
          # nail the keys of environ
          self.assertEqual(sorted(environ.keys()), [
              'CONTENT_LENGTH', 'CONTENT_TYPE', 'HTTP_CONNECTION', 'HTTP_X_FOO',
              'PATH_INFO', 'QUERY_STRING', 'REMOTE_ADDR', 'REQUEST_METHOD',
              'SCRIPT_NAME', 'SERVER_NAME', 'SERVER_PORT', 'SERVER_PROTOCOL',
              'SERVER_SOFTWARE', 'wsgi.errors', 'wsgi.file_wrapper', 'wsgi.input',
              'wsgi.multiprocess', 'wsgi.multithread', 'wsgi.run_once',
              'wsgi.url_scheme', 'wsgi.version'])
  
          self.assertEqual(environ['REQUEST_METHOD'], 'GET')
          self.assertEqual(environ['SERVER_PORT'], '80')
          self.assertEqual(environ['SERVER_NAME'], 'localhost')
          self.assertEqual(environ['SERVER_SOFTWARE'], 'waitress')
          self.assertEqual(environ['SERVER_PROTOCOL'], 'HTTP/1.0')
          self.assertEqual(environ['SCRIPT_NAME'], '')
          self.assertEqual(environ['HTTP_CONNECTION'], 'close')
          self.assertEqual(environ['PATH_INFO'], '/')
          self.assertEqual(environ['QUERY_STRING'], 'abc')
          self.assertEqual(environ['REMOTE_ADDR'], '192.168.1.1')
          self.assertEqual(environ['CONTENT_TYPE'], 'abc')
          self.assertEqual(environ['CONTENT_LENGTH'], '10')
          self.assertEqual(environ['HTTP_X_FOO'], 'BAR')
          self.assertEqual(environ['wsgi.version'], (1, 0))
          self.assertEqual(environ['wsgi.url_scheme'], 'https')
          self.assertEqual(environ['wsgi.errors'], sys.stderr)
          self.assertEqual(environ['wsgi.multithread'], True)
          self.assertEqual(environ['wsgi.multiprocess'], False)
          self.assertEqual(environ['wsgi.run_once'], False)
          self.assertEqual(environ['wsgi.input'], 'stream')
          self.assertEqual(inst.environ, environ)
  
      def test_get_environment_values_w_bogus_scheme_override(self):
          inst = self._makeOne()
          inst.channel.addr = ['192.168.1.1']
          inst.channel.server.adj.trusted_proxy = '192.168.1.1'
          request = DummyParser()
          request.headers = {
              'CONTENT_TYPE': 'abc',
              'CONTENT_LENGTH': '10',
              'X_FOO': 'BAR',
              'X_FORWARDED_PROTO': 'http://p02n3e.com?url=http',
              'CONNECTION': 'close',
          }
          request.query = 'abc'
          inst.request = request
          self.assertRaises(ValueError, inst.get_environment)
  
  class TestErrorTask(unittest.TestCase):
  
      def _makeOne(self, channel=None, request=None):
          if channel is None:
              channel = DummyChannel()
          if request is None:
              request = DummyParser()
              request.error = DummyError()
          from waitress.task import ErrorTask
          return ErrorTask(channel, request)
  
      def test_execute_http_10(self):
          inst = self._makeOne()
          inst.execute()
          lines = filter_lines(inst.channel.written)
          self.assertEqual(len(lines), 9)
          self.assertEqual(lines[0], b'HTTP/1.0 432 Too Ugly')
          self.assertEqual(lines[1], b'Connection: close')
          self.assertEqual(lines[2], b'Content-Length: 43')
          self.assertEqual(lines[3], b'Content-Type: text/plain')
          self.assertTrue(lines[4])
          self.assertEqual(lines[5], b'Server: waitress')
          self.assertEqual(lines[6], b'Too Ugly')
          self.assertEqual(lines[7], b'body')
          self.assertEqual(lines[8], b'(generated by waitress)')
  
      def test_execute_http_11(self):
          inst = self._makeOne()
          inst.version = '1.1'
          inst.execute()
          lines = filter_lines(inst.channel.written)
          self.assertEqual(len(lines), 8)
          self.assertEqual(lines[0], b'HTTP/1.1 432 Too Ugly')
          self.assertEqual(lines[1], b'Content-Length: 43')
          self.assertEqual(lines[2], b'Content-Type: text/plain')
          self.assertTrue(lines[3])
          self.assertEqual(lines[4], b'Server: waitress')
          self.assertEqual(lines[5], b'Too Ugly')
          self.assertEqual(lines[6], b'body')
          self.assertEqual(lines[7], b'(generated by waitress)')
  
      def test_execute_http_11_close(self):
          inst = self._makeOne()
          inst.version = '1.1'
          inst.request.headers['CONNECTION'] = 'close'
          inst.execute()
          lines = filter_lines(inst.channel.written)
          self.assertEqual(len(lines), 9)
          self.assertEqual(lines[0], b'HTTP/1.1 432 Too Ugly')
          self.assertEqual(lines[1], b'Connection: close')
          self.assertEqual(lines[2], b'Content-Length: 43')
          self.assertEqual(lines[3], b'Content-Type: text/plain')
          self.assertTrue(lines[4])
          self.assertEqual(lines[5], b'Server: waitress')
          self.assertEqual(lines[6], b'Too Ugly')
          self.assertEqual(lines[7], b'body')
          self.assertEqual(lines[8], b'(generated by waitress)')
  
      def test_execute_http_11_keep(self):
          inst = self._makeOne()
          inst.version = '1.1'
          inst.request.headers['CONNECTION'] = 'keep-alive'
          inst.execute()
          lines = filter_lines(inst.channel.written)
          self.assertEqual(len(lines), 8)
          self.assertEqual(lines[0], b'HTTP/1.1 432 Too Ugly')
          self.assertEqual(lines[1], b'Content-Length: 43')
          self.assertEqual(lines[2], b'Content-Type: text/plain')
          self.assertTrue(lines[3])
          self.assertEqual(lines[4], b'Server: waitress')
          self.assertEqual(lines[5], b'Too Ugly')
          self.assertEqual(lines[6], b'body')
          self.assertEqual(lines[7], b'(generated by waitress)')
  
  
  class DummyError(object):
      code = '432'
      reason = 'Too Ugly'
      body = 'body'
  
  class DummyTask(object):
      serviced = False
      deferred = False
      cancelled = False
  
      def __init__(self, toraise=None):
          self.toraise = toraise
  
      def service(self):
          self.serviced = True
          if self.toraise:
              raise self.toraise
  
      def defer(self):
          self.deferred = True
          if self.toraise:
              raise self.toraise
  
      def cancel(self):
          self.cancelled = True
  
  class DummyAdj(object):
      log_socket_errors = True
      ident = 'waitress'
      host = '127.0.0.1'
      port = 80
      url_prefix = ''
      trusted_proxy = None
  
  class DummyServer(object):
      server_name = 'localhost'
      effective_port = 80
  
      def __init__(self):
          self.adj = DummyAdj()
  
  class DummyChannel(object):
      closed_when_done = False
      adj = DummyAdj()
      creation_time = 0
      addr = ['127.0.0.1']
  
      def __init__(self, server=None):
          if server is None:
              server = DummyServer()
          self.server = server
          self.written = b''
          self.otherdata = []
  
      def write_soon(self, data):
          if isinstance(data, bytes):
              self.written += data
          else:
              self.otherdata.append(data)
          return len(data)
  
  class DummyParser(object):
      version = '1.0'
      command = 'GET'
      path = '/'
      query = ''
      url_scheme = 'http'
      expect_continue = False
      headers_finished = False
  
      def __init__(self):
          self.headers = {}
  
      def get_body_stream(self):
          return 'stream'
  
  def filter_lines(s):
      return list(filter(None, s.split(b'\r\n')))
  
  class DummyLogger(object):
  
      def __init__(self):
          self.logged = []
  
      def warning(self, msg):
          self.logged.append(msg)
  
      def exception(self, msg):
          self.logged.append(msg)
  