<s> import tempfile <EOL> import unittest <EOL> from fbtftp . base_server import BaseServer <EOL> '''<STR_LIT>''' <EOL> RRQ = b'<STR_LIT>' <EOL> TEST_PAYLOADS = ( <EOL> RRQ + b'<STR_LIT>' , <EOL> RRQ + b'<STR_LIT>' , <EOL> RRQ + b'<STR_LIT>' , <EOL> RRQ + b'<STR_LIT>' , <EOL> RRQ + b'<STR_LIT>' , <EOL> RRQ + b'<STR_LIT>' , <EOL> RRQ + b'<STR_LIT>' , <EOL> ) <EOL> class MockSocketListener : <EOL> def __init__ ( self , network_queue ) : <EOL> self . _network_queue = network_queue <EOL> def recvfrom ( self , blocksize ) : <EOL> data = self . _network_queue . pop ( <NUM_LIT:0> ) <EOL> peer = '<STR_LIT>' <EOL> return data , peer <EOL> def close ( self ) : <EOL> pass <EOL> class StaticServer ( BaseServer ) : <EOL> def __init__ ( <EOL> self , address , port , retries , timeout , root , stats_callback , <EOL> network_queue <EOL> ) : <EOL> super ( ) . __init__ ( address , port , retries , timeout ) <EOL> self . _root = root <EOL> self . _listener = MockSocketListener ( network_queue ) <EOL> class TestServerMalformedPacket ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . tmpdir = tempfile . TemporaryDirectory ( ) <EOL> self . host = '<STR_LIT>' <EOL> self . port = <NUM_LIT:0> <EOL> self . timeout = <NUM_LIT:2> <EOL> def testMalformedPackets ( self ) : <EOL> for payload in TEST_PAYLOADS : <EOL> server = StaticServer ( <EOL> self . host , self . port , <NUM_LIT:2> , <NUM_LIT:2> , self . tmpdir , None , [ payload ] <EOL> ) <EOL> server . on_new_data ( ) <EOL> server . close ( ) <EOL> del server </s>
