<s> import os <EOL> import shutil <EOL> import tempfile <EOL> import functools <EOL> from test_server import TestServer <EOL> from unittest import TestCase <EOL> import logging <EOL> from six . moves . urllib . request import urlopen <EOL> import socket <EOL> import time <EOL> from contextlib import contextmanager <EOL> from tempfile import mkdtemp , mkstemp <EOL> from shutil import rmtree <EOL> from grab import Grab <EOL> logger = logging . getLogger ( '<STR_LIT>' ) <EOL> TEST_DIR = os . path . dirname ( os . path . realpath ( __file__ ) ) <EOL> TEST_SERVER_PORT = <NUM_LIT> <EOL> ADDRESS = '<STR_LIT:localhost>' <EOL> EXTRA_PORT1 = TEST_SERVER_PORT + <NUM_LIT:1> <EOL> EXTRA_PORT2 = TEST_SERVER_PORT + <NUM_LIT:2> <EOL> GLOBAL = { <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : None , <EOL> } <EOL> @ contextmanager <EOL> def temp_dir ( root_dir = None ) : <EOL> dir_ = mkdtemp ( dir = root_dir ) <EOL> yield dir_ <EOL> rmtree ( dir_ ) <EOL> @ contextmanager <EOL> def temp_file ( root_dir = None ) : <EOL> fd , file_ = mkstemp ( dir = root_dir ) <EOL> yield file_ <EOL> os . close ( fd ) <EOL> os . unlink ( file_ ) <EOL> def build_grab ( * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> kwargs . setdefault ( '<STR_LIT>' , GLOBAL [ '<STR_LIT>' ] ) <EOL> return Grab ( * args , ** kwargs ) <EOL> def build_spider ( cls , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> kwargs . setdefault ( '<STR_LIT>' , GLOBAL [ '<STR_LIT>' ] ) <EOL> if kwargs [ '<STR_LIT>' ] : <EOL> kwargs . setdefault ( '<STR_LIT>' , <NUM_LIT:2> ) <EOL> else : <EOL> kwargs [ '<STR_LIT>' ] = <NUM_LIT:1> <EOL> return cls ( ** kwargs ) <EOL> class BaseGrabTestCase ( TestCase ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> cls . server = start_server ( ) <EOL> def setUp ( self ) : <EOL> self . server . reset ( ) <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> cls . server . stop ( ) <EOL> def multiprocess_mode ( mode ) : <EOL> def wrapper_builder ( func ) : <EOL> def wrapper ( self , * args , ** kwargs ) : <EOL> if mode != GLOBAL [ '<STR_LIT>' ] : <EOL> logger . debug ( '<STR_LIT>' <EOL> '<STR_LIT>' % ( <EOL> func . __module__ , <EOL> self . __class__ . __name__ , <EOL> func . __name__ , mode ) ) <EOL> else : <EOL> return func ( self , * args , ** kwargs ) <EOL> return wrapper <EOL> return wrapper_builder <EOL> def start_server ( ) : <EOL> logger . debug ( '<STR_LIT>' % ( ADDRESS , TEST_SERVER_PORT ) ) <EOL> server = TestServer ( address = ADDRESS , port = TEST_SERVER_PORT , <EOL> extra_ports = [ EXTRA_PORT1 , EXTRA_PORT2 ] ) <EOL> server . start ( ) <EOL> '''<STR_LIT>''' <EOL> return server <EOL> def stop_server ( ) : <EOL> logger . debug ( '<STR_LIT>' ) <EOL> def exclude_transport ( * names ) : <EOL> def decorator ( func ) : <EOL> def caller ( * args , ** kwargs ) : <EOL> if GLOBAL [ '<STR_LIT>' ] in names : <EOL> func_name = '<STR_LIT>' % ( func . __module__ , func . __name__ ) <EOL> logger . debug ( '<STR_LIT>' <EOL> % ( func_name , GLOBAL [ '<STR_LIT>' ] ) ) <EOL> return None <EOL> else : <EOL> return func ( * args , ** kwargs ) <EOL> return caller <EOL> return decorator <EOL> def only_transport ( * names ) : <EOL> def decorator ( func ) : <EOL> def caller ( * args , ** kwargs ) : <EOL> if GLOBAL [ '<STR_LIT>' ] in names : <EOL> return func ( * args , ** kwargs ) <EOL> else : <EOL> func_name = '<STR_LIT>' % ( func . __module__ , func . __name__ ) <EOL> logger . debug ( '<STR_LIT>' <EOL> % ( func_name , GLOBAL [ '<STR_LIT>' ] ) ) <EOL> return None <EOL> return caller <EOL> return decorator </s>
