def test_assign_list_of_named_addresses(self):
     addresses = [('Test User 1', '[email protected]'), ('Test User 2', '[email protected]')]
     self.addresses = addresses
     
     string_addresses = [unicode(Address(*value)) for value in addresses]
     eq_(', '.join(string_addresses), unicode(self.addresses))
     eq_(string_addresses, self.addresses)

 def test_chunked(self):
     response = self.request()
     
     self.assertEquals(response.protocol, b"HTTP/1.1")
     self.assertEquals(response.code, b"200")
     self.assertEquals(response.status, b"OK")
     self.assertEquals(response[b'content-type'], b"text/plain; charset=utf8")
     self.assertEquals(response[b'transfer-encoding'], b"chunked")
     
     request = eval(response.body)
     
     expect = {
             'CONTENT_LENGTH': None,
             'CONTENT_TYPE': None,
             'FRAGMENT': '',
             'HTTP_HOST': 'localhost',
             'PARAMETERS': unicode(),
             'PATH_INFO': b'/'.decode('iso-8859-1'),
             'QUERY_STRING': unicode(),
             'REMOTE_ADDR': '127.0.0.1',
             'REQUEST_METHOD': 'GET',
             'SCRIPT_NAME': unicode(),
             'SERVER_ADDR': '127.0.0.1',
             'SERVER_PROTOCOL': 'HTTP/1.1',
             'wsgi.multiprocess': False,
             'wsgi.multithread': False,
             'wsgi.run_once': False,
             'wsgi.url_scheme': 'http',
             'wsgi.version': (2, 0),
             'REQUEST_URI': b'http://localhost/',
             'wsgi.async': False,
             'wsgi.uri_encoding': 'utf8'
         }
     
     self.assertEquals(request, expect)

	def test_callable_bodies(self):
		message = self.build_message()
		message.plain = lambda: "plain text"
		message.rich = lambda: "rich text"
		
		assert 'plain text' in unicode(message)
		assert 'rich text' in unicode(message)

    def test_request(self):
        response = self.request(protocol=b"HTTP/1.0")
        request = eval(response.body)

        expect = {
            "CONTENT_LENGTH": None,
            "CONTENT_TYPE": None,
            "FRAGMENT": "",
            "HTTP_HOST": "localhost",
            "PARAMETERS": unicode(),
            "PATH_INFO": b"/".decode("iso-8859-1"),
            "QUERY_STRING": unicode(),
            "REMOTE_ADDR": "127.0.0.1",
            "REQUEST_METHOD": "GET",
            "SCRIPT_NAME": unicode(),
            "SERVER_ADDR": "127.0.0.1",
            "SERVER_PROTOCOL": "HTTP/1.0",
            "wsgi.multiprocess": False,
            "wsgi.multithread": False,
            "wsgi.run_once": False,
            "wsgi.url_scheme": "http",
            "wsgi.version": (2, 0),
            "REQUEST_URI": b"/",
            "wsgi.async": False,
            "wsgi.uri_encoding": "utf8",
        }

        self.assertEquals(request, expect)

def echo(chunked, request):
    prune(request, True)
    
    result = unicode(pformat(request)).encode('utf8')
    headers = [(b'Content-Type', b'text/plain; charset=utf8')]
    
    if not chunked:
        headers.append((b'Content-Length', unicode(len(result)).encode('ascii')))
    
    return b'200 OK', headers, (result.split(b"\n") if chunked else [result])

	def test_mime_generation_rich_embedded(self):
		message = self.build_message()
		message.plain = "Hello world."
		message.rich = "Farewell cruel world."
		
		message.attach("hello.txt", b"Fnord.", "text", "plain", True)
		
		assert 'Hello world.' in unicode(message)
		assert 'Farewell cruel world.' in unicode(message)
		assert 'hello.txt' in unicode(message)
		assert 'Rm5vcmQu' in unicode(message)  # Fnord. in base64

	def test_mime_attachments(self):
		message = self.build_message()
		message.plain = "Hello world."
		message.rich = "Farewell cruel world."
		
		message.attach("hello.txt", b"Fnord.")
		
		assert 'Hello world.' in unicode(message)
		assert 'Farewell cruel world.' in unicode(message)
		assert 'hello.txt' in unicode(message)
		assert 'Rm5vcmQu' in unicode(message)  # Fnord. in base64
		assert 'text/plain\n' in unicode(message)

	def test_language_specification_but_no_charset_for_attachment_name(self):
		message = self.build_message()
		message.plain = "Hello world."
		message.rich = "Farewell cruel world."
		message.attach("☃.txt", b"unicode snowman", filename_language='en-us')
		
		assert 'Hello world.' in unicode(message)
		assert 'Farewell cruel world.' in unicode(message)
		if sys.version_info < (3, 0):
			assert 'filename*="utf-8\'en-us\'%E2%98%83.txt"' in unicode(message) # ☃ is encoded in ASCII as \xe2\x98\x83, which is URL encoded as %E2%98%83
		else:
			assert 'filename*=utf-8\'en-us\'%E2%98%83.txt' in unicode(message) # ☃ is encoded in ASCII as \xe2\x98\x83, which is URL encoded as %E2%98%83
		assert 'dW5pY29kZSBzbm93bWFu' in unicode(message)  # unicode snowman in base64

 def render(self, safe=False):
     parts = []
     
     parts.append((self.scheme + "://") if self.scheme else "")
     parts.append(self.user or "")
     parts.append((":" + self.password) if self.user else ("@" if self.user else ""))
     parts.append(self.host or "")
     parts.append((":" + str(self.port)) if self.port else "")
     parts.append(unicode(self.path) or "/")
     parts.append((";" + unicode(self.params)) if self.params else "")
     parts.append(("?" + unicode(self.query)) if self.query else "")
     parts.append(("#" + quote_plus(self.fragment)) if self.fragment else "")
     
     return "".join(parts)

 def __init__(self, name_or_email, email=None, encoding='utf-8'):
     if email is None:
         if isinstance(name_or_email, AddressList):
             if not 0 < len(name_or_email) < 2:
                 raise ValueError("AddressList to convert must only contain a single Address.")
             
             name_or_email = unicode(name_or_email[0])
         
         if isinstance(name_or_email, (tuple, list)):
             self.name = unicodestr(name_or_email[0], encoding)
             self.address = unicodestr(name_or_email[1], encoding)
         
         elif isinstance(name_or_email, bytes):
             self.name, self.address = parseaddr(unicodestr(name_or_email, encoding))
         
         elif isinstance(name_or_email, unicode):
             self.name, self.address = parseaddr(name_or_email)
         
         else:
             raise TypeError('Expected string, tuple or list, got {0} instead'.format(
                     repr(type(name_or_email))
                 ))
     else:
         self.name = unicodestr(name_or_email, encoding)
         self.address = unicodestr(email, encoding)
     
     email, err = EmailValidator().validate_email(self.address)
     
     if err:
         raise ValueError('"{0}" is not a valid e-mail address: {1}'.format(email, err))

 def __init__(self, server, protocol, client):
     self.server = server
     self.protocol = protocol
     self.client = client
     
     env = dict()
     env['REMOTE_ADDR'] = client.address[0]
     env['SERVER_NAME'] = protocol._name
     env['SERVER_ADDR'] = protocol._addr
     env['SERVER_PORT'] = protocol._port
     env['SCRIPT_NAME'] = unicode()
     
     env['wsgi.input'] = IO()
     env['wsgi.errors'] = errorlog
     env['wsgi.version'] = (2, 0)
     env['wsgi.multithread'] = getattr(server, 'threaded', False) # TODO: Temporary hack until marrow.server 1.0 release.
     env['wsgi.multiprocess'] = server.fork != 1
     env['wsgi.run_once'] = False
     env['wsgi.url_scheme'] = 'http'
     env['wsgi.async'] = False # TODO
     
     if self.server.threaded is not False:
         env['wsgi.executor'] = self.server.executor # pimp out the concurrent.futures thread pool executor
     
     # env['wsgi.script_name'] = b''
     # env['wsgi.path_info'] = b''
     
     self.environ = None
     self.environ_template = env
     
     self.finished = False
     self.pipeline = protocol.options.get('pipeline', True) # TODO
     
     client.read_until(dCRLF, self.headers)

	def test_can_set_encoding_for_message_explicitely(self):
		message = self.build_message()
		assert 'iso-8859-1' not in unicode(message).lower()
		message.encoding = 'ISO-8859-1'
		msg = email.message_from_string(str(message))
		assert msg['Content-Type'] == 'text/plain; charset="iso-8859-1"'
		assert msg['Content-Transfer-Encoding'] == 'quoted-printable'

 def __call__(self, request):
     result = b"Hello " + self.name + b"!"
     
     return b'200 OK', [
             (b'Content-Type', b'text/plain; charset=utf8'),
             (b'Content-Length', unicode(len(result)).encode('ascii'))
         ], [result]

 def __init__(self, value=None, separator='/', encoded=False):
     self.separator = unicode(separator)
     
     super(Path, self).__init__()
     
     if value is not None:
         self._assign(value, encoded)

 def test_path_slicing(self):
     class MockOb(object):
         path = Path()
 
     instance = MockOb()
 
     instance.path = '/foo/bar/baz'
     
     self.assertEqual(unicode(instance.path[1:]), unicode('foo/bar/baz'))
     self.assertEqual(unicode(instance.path[2:]), unicode('bar/baz'))
     self.assertEqual(unicode(instance.path[0:2]), unicode('/foo'))
     self.assertEqual(unicode(instance.path[::2]), unicode('/bar'))

 def mock_headers(self, length=None, disable=False, mime=b"text/plain"):
     headers = [(b'Content-Type', mime)] if mime else []
     
     if length is not None:
         headers.append((b'Content-Length', unicode(length).encode('ascii')))
     
     if disable:
         headers.append((b'Content-Encoding', b'mock'))
     
     return headers

 def _assign(self, value, encoded=False):
     self.clear()
     
     if value is None:
         return
     
     if isinstance(value, (binary, unicode)):
         self.extend((unquote_plus(i) if encoded else i) for i in unicode(value).split(self.separator))
         return
     
     self.extend(value)

	def _add_headers_to_message(self, message, headers):
		for header in headers:
			if header[1] is None or (isinstance(header[1], list) and not header[1]):
				continue
			
			name, value = header
			
			if isinstance(value, (Address, AddressList)):
				value = unicode(value)
			
			message[name] = value

def body_echo(request):
    prune(request, True)
    
    if 'wsgi.input' not in request:
        request['wsgi.input'] = b""
    
    headers = [(b'Content-Type', b'text/plain; charset=utf8')]
    
    if 'HTTP_CONTENT_LENGTH' in request:
        headers.append((b'Content-Length', unicode(len(request['wsgi.input'])).encode('ascii')))
    
    return b'200 OK', headers, [request['wsgi.input']]

	def test_missing_values(self):
		message = Message()
		with pytest.raises(ValueError):
			unicode(message)
		
		message.author = "[email protected]"
		with pytest.raises(ValueError):
			unicode(message)
		
		message.subject = "Attn: Bob Dole"
		with pytest.raises(ValueError):
			unicode(message)
		
		message.to = "[email protected]"
		with pytest.raises(ValueError):
			unicode(message)
		
		message.plain = "Testing!"
		
		try:
			unicode(message)
		except ValueError:
			assert False, "Message should be valid."