<s> from __future__ import absolute_import , with_statement <EOL> import unittest <EOL> from random import randint <EOL> from gevent_fastcgi . const import ( <EOL> FCGI_STDIN , <EOL> FCGI_STDOUT , <EOL> FCGI_STDERR , <EOL> FCGI_DATA , <EOL> FCGI_RECORD_TYPES , <EOL> FCGI_RECORD_HEADER_LEN , <EOL> FCGI_MAX_CONTENT_LEN , <EOL> ) <EOL> from gevent_fastcgi . base import Record , Connection , PartialRead <EOL> from . . utils import binary_data , MockSocket <EOL> class ConnectionTests ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . sock = MockSocket ( ) <EOL> def tearDown ( self ) : <EOL> del self . sock <EOL> def test_read_write ( self ) : <EOL> record_type = FCGI_DATA <EOL> request_id = randint ( <NUM_LIT:1> , <NUM_LIT> ) <EOL> data = binary_data ( ) <EOL> record = Record ( record_type , data , request_id ) <EOL> conn = Connection ( self . sock ) <EOL> conn . write_record ( record ) <EOL> self . sock . flip ( ) <EOL> record = conn . read_record ( ) <EOL> assert record . type == record_type <EOL> assert record . content == data <EOL> assert record . request_id == request_id <EOL> assert conn . read_record ( ) is None <EOL> def test_read_write_long_content ( self ) : <EOL> data = binary_data ( FCGI_MAX_CONTENT_LEN + <NUM_LIT:1> ) <EOL> conn = Connection ( self . sock ) <EOL> with self . assertRaises ( ValueError ) : <EOL> conn . write_record ( Record ( FCGI_STDERR , data , <NUM_LIT:1> ) ) <EOL> def test_partial_read ( self ) : <EOL> conn = Connection ( self . sock ) <EOL> data = binary_data ( FCGI_RECORD_HEADER_LEN - <NUM_LIT:1> ) <EOL> self . sock . sendall ( data ) <EOL> self . sock . flip ( ) <EOL> self . assertRaises ( PartialRead , conn . read_record ) </s>
