<s> import mock <EOL> from werkzeug . exceptions import NotFound , MethodNotAllowed <EOL> from werkzeug . routing import Map , Rule <EOL> from werkzeug . wrappers import Response <EOL> from lymph . config import Configuration <EOL> from lymph . testing import WebServiceTestCase <EOL> from lymph . web . interfaces import WebServiceInterface <EOL> from lymph . web . handlers import RequestHandler <EOL> from lymph . web . routing import HandledRule <EOL> class RuleHandler ( RequestHandler ) : <EOL> def get ( self ) : <EOL> return Response ( "<STR_LIT>" ) <EOL> class HandledRuleHandler ( RequestHandler ) : <EOL> def get ( self ) : <EOL> return Response ( "<STR_LIT>" ) <EOL> class CustomNotFound ( NotFound ) : <EOL> def __init__ ( self ) : <EOL> response = Response ( "<STR_LIT>" , status = <NUM_LIT> ) <EOL> super ( CustomNotFound , self ) . __init__ ( response = response ) <EOL> class CustomMethodNotAllowed ( MethodNotAllowed ) : <EOL> def get_body ( self , * args , ** kwargs ) : <EOL> return "<STR_LIT>" <EOL> class Web ( WebServiceInterface ) : <EOL> url_map = Map ( [ <EOL> Rule ( "<STR_LIT>" , endpoint = "<STR_LIT:test>" ) , <EOL> Rule ( "<STR_LIT>" , endpoint = RuleHandler , methods = [ '<STR_LIT>' ] ) , <EOL> Rule ( "<STR_LIT>" , endpoint = RuleHandler ) , <EOL> HandledRule ( "<STR_LIT>" , endpoint = "<STR_LIT:bar>" , handler = HandledRuleHandler ) , <EOL> Rule ( "<STR_LIT>" , endpoint = "<STR_LIT>" ) , <EOL> Rule ( "<STR_LIT>" , endpoint = <NUM_LIT> ) , <EOL> Rule ( "<STR_LIT>" , endpoint = '<STR_LIT>' ) <EOL> ] ) <EOL> def test ( self , request ) : <EOL> return Response ( "<STR_LIT>" ) <EOL> def return_none ( self , request ) : <EOL> pass <EOL> class CustomErrorHandlingWeb ( Web ) : <EOL> NotFound = CustomNotFound <EOL> MethodNotAllowed = CustomMethodNotAllowed <EOL> class WebIntegrationTest ( WebServiceTestCase ) : <EOL> service_class = Web <EOL> def test_dispatch_rule_with_string_endpoint ( self ) : <EOL> response = self . client . get ( "<STR_LIT>" ) <EOL> self . assertEqual ( response . data . decode ( "<STR_LIT:utf8>" ) , "<STR_LIT>" ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT:200> ) <EOL> def test_dispatch_rule_with_no_trailing_slash ( self ) : <EOL> response = self . client . get ( "<STR_LIT>" , follow_redirects = True ) <EOL> self . assertEqual ( response . data . decode ( "<STR_LIT:utf8>" ) , "<STR_LIT>" ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT:200> ) <EOL> def test_dispatch_rule_with_callable_endpoint ( self ) : <EOL> response = self . client . get ( "<STR_LIT>" ) <EOL> self . assertEqual ( response . data . decode ( "<STR_LIT:utf8>" ) , "<STR_LIT>" ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT:200> ) <EOL> def test_dispatch_handled_rule ( self ) : <EOL> response = self . client . get ( "<STR_LIT>" ) <EOL> self . assertEqual ( response . data . decode ( "<STR_LIT:utf8>" ) , "<STR_LIT>" ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT:200> ) <EOL> def test_dispatch_failing_rule_to_500 ( self ) : <EOL> response = self . client . get ( "<STR_LIT>" ) <EOL> self . assertEqual ( response . data . decode ( "<STR_LIT:utf8>" ) , "<STR_LIT>" ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> def test_dispatch_failing_endpoint_to_500 ( self ) : <EOL> response = self . client . get ( "<STR_LIT>" ) <EOL> self . assertEqual ( response . data . decode ( "<STR_LIT:utf8>" ) , "<STR_LIT>" ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> def test_dispatch_not_found ( self ) : <EOL> response = self . client . get ( "<STR_LIT>" ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> def test_dispatch_methond_not_allowed ( self ) : <EOL> response = self . client . post ( "<STR_LIT>" ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> response = self . client . post ( "<STR_LIT>" ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> response = self . client . post ( "<STR_LIT>" ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> def test_bad_handler_return_type ( self ) : <EOL> response = self . client . get ( "<STR_LIT>" ) <EOL> self . assertIn ( '<STR_LIT>' , response . headers ) <EOL> class CustomErrorHandlingWebIntegrationTest ( WebIntegrationTest ) : <EOL> service_class = CustomErrorHandlingWeb <EOL> def test_dispatch_not_found ( self ) : <EOL> response = self . client . get ( "<STR_LIT>" ) <EOL> self . assertEqual ( response . data . decode ( "<STR_LIT:utf8>" ) , "<STR_LIT>" ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> def test_dispatch_methond_not_allowed ( self ) : <EOL> response = self . client . post ( "<STR_LIT>" ) <EOL> self . assertEqual ( response . data . decode ( "<STR_LIT:utf8>" ) , "<STR_LIT>" ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> response = self . client . post ( "<STR_LIT>" ) <EOL> self . assertEqual ( response . data . decode ( "<STR_LIT:utf8>" ) , "<STR_LIT>" ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> response = self . client . post ( "<STR_LIT>" ) <EOL> self . assertEqual ( response . data . decode ( "<STR_LIT:utf8>" ) , "<STR_LIT>" ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> class CustomHealthCheckResponse ( Web ) : <EOL> def get_healthcheck_response ( self ) : <EOL> return <EOL> class HealthcheckTests ( WebServiceTestCase ) : <EOL> service_class = Web <EOL> def test_healthy_200 ( self ) : <EOL> response = self . client . get ( '<STR_LIT>' ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT:200> ) <EOL> def test_unhealthy_503 ( self ) : <EOL> with mock . patch . object ( Web , '<STR_LIT>' , return_value = False ) : <EOL> response = self . client . get ( '<STR_LIT>' ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> class DisabledHealtcheckInterface ( WebServiceInterface ) : <EOL> url_map = Map ( [ ] ) <EOL> class DisabledHealtcheckTest ( WebServiceTestCase ) : <EOL> service_class = DisabledHealtcheckInterface <EOL> service_config = Configuration ( { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : False <EOL> } <EOL> } ) <EOL> def test_health_404 ( self ) : <EOL> response = self . client . get ( '<STR_LIT>' ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> class CustomHealthcheckEndpointInterface ( WebServiceInterface ) : <EOL> url_map = Map ( [ ] ) <EOL> class CustomHealthcheckEndpointTest ( WebServiceTestCase ) : <EOL> service_class = CustomHealthcheckEndpointInterface <EOL> service_config = Configuration ( { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> } ) <EOL> def test_health_404 ( self ) : <EOL> response = self . client . get ( '<STR_LIT>' ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> def test_custom_endpoint_200 ( self ) : <EOL> response = self . client . get ( '<STR_LIT>' ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT:200> ) <EOL> class NoRequestTraceIdPropagation ( WebServiceTestCase ) : <EOL> service_class = Web <EOL> def test_no_trace_id_propagation_by_default ( self ) : <EOL> response = self . client . get ( '<STR_LIT>' , headers = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertNotEqual ( response . headers . get ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> class RequestTraceIdPropagation ( WebServiceTestCase ) : <EOL> service_class = Web <EOL> service_config = Configuration ( { <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } , <EOL> } ) <EOL> def test_trace_id_propagation ( self ) : <EOL> response = self . client . get ( '<STR_LIT>' , headers = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( response . headers . get ( '<STR_LIT>' ) , '<STR_LIT>' ) </s>
