<s> """<STR_LIT>""" <EOL> import socket <EOL> import struct <EOL> import pytest <EOL> import requests <EOL> import hug <EOL> from hug import use <EOL> class TestService ( object ) : <EOL> """<STR_LIT>""" <EOL> service = use . Service ( version = <NUM_LIT:1> , timeout = <NUM_LIT:100> , raise_on = ( <NUM_LIT> , ) ) <EOL> def test_init ( self ) : <EOL> """<STR_LIT>""" <EOL> assert self . service . version == <NUM_LIT:1> <EOL> assert self . service . raise_on == ( <NUM_LIT> , ) <EOL> assert self . service . timeout == <NUM_LIT:100> <EOL> def test_request ( self ) : <EOL> """<STR_LIT>""" <EOL> with pytest . raises ( NotImplementedError ) : <EOL> self . service . request ( '<STR_LIT:POST>' , '<STR_LIT>' ) <EOL> def test_get ( self ) : <EOL> """<STR_LIT>""" <EOL> with pytest . raises ( NotImplementedError ) : <EOL> self . service . get ( '<STR_LIT>' ) <EOL> def test_post ( self ) : <EOL> """<STR_LIT>""" <EOL> with pytest . raises ( NotImplementedError ) : <EOL> self . service . post ( '<STR_LIT>' ) <EOL> def test_delete ( self ) : <EOL> """<STR_LIT>""" <EOL> with pytest . raises ( NotImplementedError ) : <EOL> self . service . delete ( '<STR_LIT>' ) <EOL> def test_put ( self ) : <EOL> """<STR_LIT>""" <EOL> with pytest . raises ( NotImplementedError ) : <EOL> self . service . put ( '<STR_LIT>' ) <EOL> def test_trace ( self ) : <EOL> """<STR_LIT>""" <EOL> with pytest . raises ( NotImplementedError ) : <EOL> self . service . trace ( '<STR_LIT>' ) <EOL> def test_patch ( self ) : <EOL> """<STR_LIT>""" <EOL> with pytest . raises ( NotImplementedError ) : <EOL> self . service . patch ( '<STR_LIT>' ) <EOL> def test_options ( self ) : <EOL> """<STR_LIT>""" <EOL> with pytest . raises ( NotImplementedError ) : <EOL> self . service . options ( '<STR_LIT>' ) <EOL> def test_head ( self ) : <EOL> """<STR_LIT>""" <EOL> with pytest . raises ( NotImplementedError ) : <EOL> self . service . head ( '<STR_LIT>' ) <EOL> def test_connect ( self ) : <EOL> """<STR_LIT>""" <EOL> with pytest . raises ( NotImplementedError ) : <EOL> self . service . connect ( '<STR_LIT>' ) <EOL> class TestHTTP ( object ) : <EOL> """<STR_LIT>""" <EOL> service = use . HTTP ( '<STR_LIT>' , raise_on = ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> url_service = use . HTTP ( '<STR_LIT>' , raise_on = ( <NUM_LIT> , <NUM_LIT> ) , json_transport = False ) <EOL> def test_init ( self ) : <EOL> """<STR_LIT>""" <EOL> assert self . service . endpoint == '<STR_LIT>' <EOL> assert self . service . raise_on == ( <NUM_LIT> , <NUM_LIT> ) <EOL> def test_request ( self ) : <EOL> """<STR_LIT>""" <EOL> response = self . url_service . request ( '<STR_LIT:GET>' , '<STR_LIT>' , query = '<STR_LIT>' ) <EOL> assert response <EOL> assert response . data <EOL> with pytest . raises ( requests . HTTPError ) : <EOL> response = self . service . request ( '<STR_LIT:GET>' , '<STR_LIT>' , query = '<STR_LIT>' ) <EOL> with pytest . raises ( requests . HTTPError ) : <EOL> self . url_service . request ( '<STR_LIT:GET>' , '<STR_LIT>' , query = '<STR_LIT>' ) <EOL> class TestLocal ( object ) : <EOL> """<STR_LIT>""" <EOL> service = use . Local ( __name__ ) <EOL> def test_init ( self ) : <EOL> """<STR_LIT>""" <EOL> assert isinstance ( self . service . api , hug . API ) <EOL> def test_request ( self ) : <EOL> """<STR_LIT>""" <EOL> assert self . service . get ( '<STR_LIT>' ) . data == '<STR_LIT>' <EOL> assert self . service . get ( '<STR_LIT>' ) . status_code == <NUM_LIT> <EOL> assert self . service . get ( '<STR_LIT>' ) . status_code == <NUM_LIT> <EOL> self . service . raise_on = ( <NUM_LIT> , <NUM_LIT> ) <EOL> with pytest . raises ( requests . HTTPError ) : <EOL> assert self . service . get ( '<STR_LIT>' ) <EOL> with pytest . raises ( requests . HTTPError ) : <EOL> assert self . service . get ( '<STR_LIT>' ) <EOL> class TestSocket ( object ) : <EOL> """<STR_LIT>""" <EOL> on_unix = getattr ( socket , '<STR_LIT>' , False ) <EOL> tcp_service = use . Socket ( connect_to = ( '<STR_LIT>' , <NUM_LIT> ) , proto = '<STR_LIT>' , timeout = <NUM_LIT> ) <EOL> udp_service = use . Socket ( connect_to = ( '<STR_LIT>' , <NUM_LIT> ) , proto = '<STR_LIT>' , timeout = <NUM_LIT> ) <EOL> def test_init ( self ) : <EOL> """<STR_LIT>""" <EOL> assert isinstance ( self . tcp_service , use . Service ) <EOL> def test_protocols ( self ) : <EOL> """<STR_LIT>""" <EOL> protocols = sorted ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> if self . on_unix : <EOL> assert sorted ( self . tcp_service . protocols ) == protocols <EOL> else : <EOL> protocols . remove ( '<STR_LIT>' ) <EOL> protocols . remove ( '<STR_LIT>' ) <EOL> assert sorted ( self . tcp_service . protocols ) == protocols <EOL> def test_streams ( self ) : <EOL> if self . on_unix : <EOL> assert set ( self . tcp_service . streams ) == set ( ( '<STR_LIT>' , '<STR_LIT>' , ) ) <EOL> else : <EOL> assert set ( self . tcp_service . streams ) == set ( ( '<STR_LIT>' , ) ) <EOL> def test_datagrams ( self ) : <EOL> if self . on_unix : <EOL> assert set ( self . tcp_service . datagrams ) == set ( ( '<STR_LIT>' , '<STR_LIT>' , ) ) <EOL> else : <EOL> assert set ( self . tcp_service . datagrams ) == set ( ( '<STR_LIT>' , ) ) <EOL> def test_inet ( self ) : <EOL> assert set ( self . tcp_service . inet ) == set ( ( '<STR_LIT>' , '<STR_LIT>' , ) ) <EOL> def test_unix ( self ) : <EOL> if self . on_unix : <EOL> assert set ( self . tcp_service . unix ) == set ( ( '<STR_LIT>' , '<STR_LIT>' , ) ) <EOL> else : <EOL> assert set ( self . tcp_service . unix ) == set ( ) <EOL> def test_connection ( self ) : <EOL> assert self . tcp_service . connection . connect_to == ( '<STR_LIT>' , <NUM_LIT> ) <EOL> assert self . tcp_service . connection . proto == '<STR_LIT>' <EOL> assert set ( self . tcp_service . connection . sockopts ) == set ( ) <EOL> def test_settimeout ( self ) : <EOL> self . tcp_service . settimeout ( <NUM_LIT> ) <EOL> assert self . tcp_service . timeout == <NUM_LIT> <EOL> def test_connection_sockopts_unit ( self ) : <EOL> self . tcp_service . connection . sockopts . clear ( ) <EOL> self . tcp_service . setsockopt ( socket . SOL_SOCKET , socket . SO_KEEPALIVE , <NUM_LIT:1> ) <EOL> assert self . tcp_service . connection . sockopts == { ( socket . SOL_SOCKET , socket . SO_KEEPALIVE , <NUM_LIT:1> ) } <EOL> def test_connection_sockopts_batch ( self ) : <EOL> self . tcp_service . setsockopt ( ( ( socket . SOL_SOCKET , socket . SO_KEEPALIVE , <NUM_LIT:1> ) , <EOL> ( socket . SOL_SOCKET , socket . SO_REUSEADDR , <NUM_LIT:1> ) ) ) <EOL> assert self . tcp_service . connection . sockopts == { ( socket . SOL_SOCKET , socket . SO_KEEPALIVE , <NUM_LIT:1> ) , <EOL> ( socket . SOL_SOCKET , socket . SO_REUSEADDR , <NUM_LIT:1> ) } <EOL> def test_datagram_request ( self ) : <EOL> """<STR_LIT>""" <EOL> packet = struct . pack ( "<STR_LIT>" , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> for name in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> header = b"<STR_LIT>" <EOL> header += bytes ( str ( len ( name ) ) , "<STR_LIT:utf-8>" ) + b"<STR_LIT:s>" <EOL> query = struct . pack ( header , len ( name ) , name . encode ( '<STR_LIT:utf-8>' ) ) <EOL> packet = packet + query <EOL> dns_query = packet + struct . pack ( "<STR_LIT>" , <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> assert len ( self . udp_service . request ( dns_query . decode ( "<STR_LIT:utf-8>" ) , buffer_size = <NUM_LIT> ) . data . read ( ) ) > <NUM_LIT:0> <EOL> @ hug . get ( ) <EOL> def hello_world ( ) : <EOL> return '<STR_LIT>' <EOL> @ hug . get ( ) <EOL> def exception ( response ) : <EOL> response . status = hug . HTTP_500 <EOL> @ hug . get ( ) <EOL> def validation_error ( data ) : <EOL> return data </s>
