<s> """<STR_LIT>""" <EOL> __all__ = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] <EOL> import codecs <EOL> import os <EOL> from testtools import try_import <EOL> from testtools . compat import _b <EOL> from testtools . content_type import ContentType , UTF8_TEXT <EOL> from testtools . testresult import TestResult <EOL> functools = try_import ( '<STR_LIT>' ) <EOL> _join_b = _b ( "<STR_LIT>" ) . join <EOL> DEFAULT_CHUNK_SIZE = <NUM_LIT> <EOL> def _iter_chunks ( stream , chunk_size ) : <EOL> """<STR_LIT>""" <EOL> chunk = stream . read ( chunk_size ) <EOL> while chunk : <EOL> yield chunk <EOL> chunk = stream . read ( chunk_size ) <EOL> class Content ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , content_type , get_bytes ) : <EOL> """<STR_LIT>""" <EOL> if None in ( content_type , get_bytes ) : <EOL> raise ValueError ( "<STR_LIT>" % ( <EOL> content_type , get_bytes ) ) <EOL> self . content_type = content_type <EOL> self . _get_bytes = get_bytes <EOL> def __eq__ ( self , other ) : <EOL> return ( self . content_type == other . content_type and <EOL> _join_b ( self . iter_bytes ( ) ) == _join_b ( other . iter_bytes ( ) ) ) <EOL> def iter_bytes ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _get_bytes ( ) <EOL> def iter_text ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . content_type . type != "<STR_LIT:text>" : <EOL> raise ValueError ( "<STR_LIT>" % self . content_type ) <EOL> return self . _iter_text ( ) <EOL> def _iter_text ( self ) : <EOL> """<STR_LIT>""" <EOL> encoding = self . content_type . parameters . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> try : <EOL> decoder = codecs . getincrementaldecoder ( encoding ) ( ) <EOL> for bytes in self . iter_bytes ( ) : <EOL> yield decoder . decode ( bytes ) <EOL> final = decoder . decode ( _b ( '<STR_LIT>' ) , True ) <EOL> if final : <EOL> yield final <EOL> except AttributeError : <EOL> bytes = '<STR_LIT>' . join ( self . iter_bytes ( ) ) <EOL> yield bytes . decode ( encoding ) <EOL> def __repr__ ( self ) : <EOL> return "<STR_LIT>" % ( <EOL> self . content_type , _join_b ( self . iter_bytes ( ) ) ) <EOL> class TracebackContent ( Content ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , err , test ) : <EOL> """<STR_LIT>""" <EOL> if err is None : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> content_type = ContentType ( '<STR_LIT:text>' , '<STR_LIT>' , <EOL> { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT:utf8>" } ) <EOL> self . _result = TestResult ( ) <EOL> value = self . _result . _exc_info_to_unicode ( err , test ) <EOL> super ( TracebackContent , self ) . __init__ ( <EOL> content_type , lambda : [ value . encode ( "<STR_LIT:utf8>" ) ] ) <EOL> def text_content ( text ) : <EOL> """<STR_LIT>""" <EOL> return Content ( UTF8_TEXT , lambda : [ text . encode ( '<STR_LIT:utf8>' ) ] ) <EOL> def maybe_wrap ( wrapper , func ) : <EOL> """<STR_LIT>""" <EOL> if functools is not None : <EOL> wrapper = functools . update_wrapper ( wrapper , func ) <EOL> return wrapper <EOL> def content_from_file ( path , content_type = None , chunk_size = DEFAULT_CHUNK_SIZE , <EOL> buffer_now = False ) : <EOL> """<STR_LIT>""" <EOL> if content_type is None : <EOL> content_type = UTF8_TEXT <EOL> def reader ( ) : <EOL> stream = open ( path , '<STR_LIT:rb>' ) <EOL> for chunk in _iter_chunks ( stream , chunk_size ) : <EOL> yield chunk <EOL> stream . close ( ) <EOL> return content_from_reader ( reader , content_type , buffer_now ) <EOL> def content_from_stream ( stream , content_type = None , <EOL> chunk_size = DEFAULT_CHUNK_SIZE , buffer_now = False ) : <EOL> """<STR_LIT>""" <EOL> if content_type is None : <EOL> content_type = UTF8_TEXT <EOL> reader = lambda : _iter_chunks ( stream , chunk_size ) <EOL> return content_from_reader ( reader , content_type , buffer_now ) <EOL> def content_from_reader ( reader , content_type , buffer_now ) : <EOL> """<STR_LIT>""" <EOL> if content_type is None : <EOL> content_type = UTF8_TEXT <EOL> if buffer_now : <EOL> contents = list ( reader ( ) ) <EOL> reader = lambda : contents <EOL> return Content ( content_type , reader ) <EOL> def attach_file ( detailed , path , name = None , content_type = None , <EOL> chunk_size = DEFAULT_CHUNK_SIZE , buffer_now = True ) : <EOL> """<STR_LIT>""" <EOL> if name is None : <EOL> name = os . path . basename ( path ) <EOL> content_object = content_from_file ( <EOL> path , content_type , chunk_size , buffer_now ) <EOL> detailed . addDetail ( name , content_object ) </s>
