<s> import logging <EOL> import json <EOL> import operator <EOL> import traceback <EOL> import string <EOL> import os <EOL> import re <EOL> import sys <EOL> from . import parsing <EOL> from . import six <EOL> from . six import binary_type <EOL> from . six import text_type <EOL> PYTHON_MAJOR_VERSION = sys . version_info [ <NUM_LIT:0> ] <EOL> if PYTHON_MAJOR_VERSION > <NUM_LIT:2> : <EOL> from past . builtins import basestring <EOL> from past . builtins import long <EOL> """<STR_LIT>""" <EOL> logger = logging . getLogger ( '<STR_LIT>' ) <EOL> COMPARATORS = { <EOL> '<STR_LIT>' : lambda x , y : safe_length ( x ) == y , <EOL> '<STR_LIT>' : operator . lt , <EOL> '<STR_LIT>' : operator . lt , <EOL> '<STR_LIT>' : operator . lt , <EOL> '<STR_LIT>' : operator . lt , <EOL> '<STR_LIT>' : operator . eq , <EOL> '<STR_LIT>' : operator . eq , <EOL> '<STR_LIT>' : lambda x , y : operator . eq ( str ( x ) , str ( y ) ) , <EOL> '<STR_LIT>' : operator . ne , <EOL> '<STR_LIT>' : operator . ne , <EOL> '<STR_LIT>' : operator . ge , <EOL> '<STR_LIT>' : operator . ge , <EOL> '<STR_LIT>' : operator . gt , <EOL> '<STR_LIT>' : operator . gt , <EOL> '<STR_LIT>' : lambda x , y : x and operator . contains ( x , y ) , <EOL> '<STR_LIT>' : lambda x , y : y and operator . contains ( y , x ) , <EOL> '<STR_LIT>' : lambda x , y : regex_compare ( str ( x ) , str ( y ) ) , <EOL> '<STR_LIT:type>' : lambda x , y : test_type ( x , y ) <EOL> } <EOL> COMPARATORS [ '<STR_LIT>' ] = COMPARATORS [ '<STR_LIT>' ] <EOL> TYPES = { <EOL> '<STR_LIT:null>' : type ( None ) , <EOL> '<STR_LIT:none>' : type ( None ) , <EOL> '<STR_LIT>' : ( int , long , float ) , <EOL> '<STR_LIT:int>' : ( int , long ) , <EOL> '<STR_LIT:float>' : float , <EOL> '<STR_LIT>' : bool , <EOL> '<STR_LIT:string>' : basestring , <EOL> '<STR_LIT>' : list , <EOL> '<STR_LIT:list>' : list , <EOL> '<STR_LIT>' : dict , <EOL> '<STR_LIT>' : dict , <EOL> '<STR_LIT>' : ( bool , int , long , float , basestring , type ( None ) ) , <EOL> '<STR_LIT>' : ( list , dict , set ) <EOL> } <EOL> def test_type ( val , mytype ) : <EOL> """<STR_LIT>""" <EOL> typelist = TYPES . get ( mytype . lower ( ) ) <EOL> if typelist is None : <EOL> raise TypeError ( <EOL> "<STR_LIT>" . format ( mytype . lower ( ) ) ) <EOL> try : <EOL> for testtype in typelist : <EOL> if isinstance ( val , testtype ) : <EOL> return True <EOL> return False <EOL> except TypeError : <EOL> return isinstance ( val , typelist ) <EOL> VALIDATOR_TESTS = { <EOL> '<STR_LIT>' : lambda x : x is not None , <EOL> '<STR_LIT>' : lambda x : x is None <EOL> } <EOL> EXTRACTORS = dict ( ) <EOL> VALIDATORS = dict ( ) <EOL> def safe_length ( var ) : <EOL> """<STR_LIT>""" <EOL> output = - <NUM_LIT:1> <EOL> try : <EOL> output = len ( var ) <EOL> except : <EOL> pass <EOL> return output <EOL> def regex_compare ( input , regex ) : <EOL> return bool ( re . search ( regex , input ) ) <EOL> FAILURE_INVALID_RESPONSE = '<STR_LIT>' <EOL> FAILURE_CURL_EXCEPTION = '<STR_LIT>' <EOL> FAILURE_TEST_EXCEPTION = '<STR_LIT>' <EOL> FAILURE_VALIDATOR_FAILED = '<STR_LIT>' <EOL> FAILURE_VALIDATOR_EXCEPTION = '<STR_LIT>' <EOL> FAILURE_EXTRACTOR_EXCEPTION = '<STR_LIT>' <EOL> class Failure ( object ) : <EOL> """<STR_LIT>""" <EOL> message = None <EOL> failure_type = None <EOL> details = None <EOL> validator = None <EOL> def __nonzero__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return False <EOL> def __bool__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return False <EOL> def __str__ ( self ) : <EOL> return self . message <EOL> def __init__ ( self , message = "<STR_LIT>" , details = "<STR_LIT>" , failure_type = None , validator = None ) : <EOL> self . message = message <EOL> self . details = details <EOL> self . validator = validator <EOL> self . failure_type = failure_type <EOL> class AbstractExtractor ( object ) : <EOL> """<STR_LIT>""" <EOL> extractor_type = None <EOL> query = None <EOL> is_templated = False <EOL> is_body_extractor = False <EOL> is_header_extractor = False <EOL> args = None <EOL> def __str__ ( self ) : <EOL> return "<STR_LIT>" . format ( self . extractor_type , self . query , self . is_templated , self . args ) <EOL> def extract_internal ( self , query = None , body = None , headers = None , args = None ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def extract ( self , body = None , headers = None , context = None ) : <EOL> """<STR_LIT>""" <EOL> query = self . templated_query ( context = context ) <EOL> args = self . args <EOL> return self . extract_internal ( query = query , body = body , headers = headers , args = self . args ) <EOL> def templated_query ( self , context = None ) : <EOL> query = self . query <EOL> if context and self . is_templated : <EOL> query = string . Template ( query ) . safe_substitute ( <EOL> context . get_values ( ) ) <EOL> return query <EOL> def get_readable_config ( self , context = None ) : <EOL> """<STR_LIT>""" <EOL> query = self . templated_query ( context = context ) <EOL> output = '<STR_LIT>' . format ( <EOL> self . extractor_type , query , self . is_templated ) <EOL> args_string = None <EOL> if self . args : <EOL> args_string = "<STR_LIT>" + str ( self . args ) <EOL> output = output + args_string <EOL> return output <EOL> @ classmethod <EOL> def configure_base ( cls , config , extractor_base ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( config , dict ) : <EOL> try : <EOL> config = config [ '<STR_LIT>' ] <EOL> extractor_base . is_templated = True <EOL> extractor_base . query = config <EOL> except KeyError : <EOL> raise ValueError ( <EOL> "<STR_LIT>" ) <EOL> elif isinstance ( config , basestring ) : <EOL> extractor_base . query = config <EOL> extractor_base . is_templated = False <EOL> else : <EOL> raise TypeError ( <EOL> "<STR_LIT>" ) <EOL> return extractor_base <EOL> class MiniJsonExtractor ( AbstractExtractor ) : <EOL> """<STR_LIT>""" <EOL> extractor_type = '<STR_LIT>' <EOL> is_body_extractor = True <EOL> def extract_internal ( self , query = None , args = None , body = None , headers = None ) : <EOL> if PYTHON_MAJOR_VERSION > <NUM_LIT:2> and isinstance ( body , binary_type ) : <EOL> body = text_type ( body , '<STR_LIT:utf-8>' ) <EOL> try : <EOL> body = json . loads ( body ) <EOL> return self . query_dictionary ( query , body ) <EOL> except ValueError : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> @ staticmethod <EOL> def query_dictionary ( query , dictionary , delimiter = '<STR_LIT:.>' ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> stripped_query = query . strip ( delimiter ) <EOL> if stripped_query : <EOL> for x in stripped_query . split ( delimiter ) : <EOL> try : <EOL> x = int ( x ) <EOL> dictionary = dictionary [ x ] <EOL> except ValueError : <EOL> dictionary = dictionary [ x ] <EOL> except : <EOL> return None <EOL> return dictionary <EOL> @ classmethod <EOL> def parse ( cls , config ) : <EOL> base = MiniJsonExtractor ( ) <EOL> return cls . configure_base ( config , base ) <EOL> return base <EOL> class HeaderExtractor ( AbstractExtractor ) : <EOL> """<STR_LIT>""" <EOL> extractor_type = '<STR_LIT>' <EOL> is_header_extractor = True <EOL> def extract_internal ( self , query = None , args = None , body = None , headers = None ) : <EOL> low = query . lower ( ) <EOL> extracted = [ y [ <NUM_LIT:1> ] for y in filter ( lambda x : x [ <NUM_LIT:0> ] == low , headers ) ] <EOL> if len ( extracted ) == <NUM_LIT:0> : <EOL> raise ValueError ( "<STR_LIT>" . format ( query ) ) <EOL> elif len ( extracted ) == <NUM_LIT:1> : <EOL> return extracted [ <NUM_LIT:0> ] <EOL> else : <EOL> return extracted <EOL> @ classmethod <EOL> def parse ( cls , config , extractor_base = None ) : <EOL> base = HeaderExtractor ( ) <EOL> return cls . configure_base ( config , base ) <EOL> class RawBodyExtractor ( AbstractExtractor ) : <EOL> """<STR_LIT>""" <EOL> extractor_type = '<STR_LIT>' <EOL> is_header_extractor = False <EOL> is_body_extractor = True <EOL> def extract_internal ( self , query = None , args = None , body = None , headers = None ) : <EOL> return body <EOL> @ classmethod <EOL> def parse ( cls , config , extractor_base = None ) : <EOL> base = RawBodyExtractor ( ) <EOL> return base <EOL> def _get_extractor ( config_dict ) : <EOL> """<STR_LIT>""" <EOL> extractor = None <EOL> extract_config = None <EOL> for key , value in config_dict . items ( ) : <EOL> if key in EXTRACTORS : <EOL> return parse_extractor ( key , value ) <EOL> else : <EOL> raise Exception ( <EOL> '<STR_LIT>' . format ( config_dict ) ) <EOL> class AbstractValidator ( object ) : <EOL> """<STR_LIT>""" <EOL> name = None <EOL> config = None <EOL> def validate ( self , body = None , headers = None , context = None ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class ComparatorValidator ( AbstractValidator ) : <EOL> """<STR_LIT>""" <EOL> name = '<STR_LIT>' <EOL> config = None <EOL> extractor = None <EOL> comparator = None <EOL> comparator_name = "<STR_LIT>" <EOL> expected = None <EOL> isTemplateExpected = False <EOL> def get_readable_config ( self , context = None ) : <EOL> """<STR_LIT>""" <EOL> string_frags = list ( ) <EOL> string_frags . append ( <EOL> "<STR_LIT>" + self . extractor . get_readable_config ( context = context ) ) <EOL> if isinstance ( self . expected , AbstractExtractor ) : <EOL> string_frags . append ( "<STR_LIT>" + <EOL> self . expected . get_readable_config ( context = context ) ) <EOL> elif self . isTemplateExpected : <EOL> string_frags . append ( <EOL> '<STR_LIT>' . format ( self . expected ) ) <EOL> return os . linesep . join ( string_frags ) <EOL> def validate ( self , body = None , headers = None , context = None ) : <EOL> try : <EOL> extracted_val = self . extractor . extract ( <EOL> body = body , headers = headers , context = context ) <EOL> except Exception as e : <EOL> trace = traceback . format_exc ( ) <EOL> return Failure ( message = "<STR_LIT>" , details = trace , validator = self , failure_type = FAILURE_EXTRACTOR_EXCEPTION ) <EOL> expected_val = None <EOL> if isinstance ( self . expected , AbstractExtractor ) : <EOL> try : <EOL> expected_val = self . expected . extract ( <EOL> body = body , headers = headers , context = context ) <EOL> except Exception as e : <EOL> trace = traceback . format_exc ( ) <EOL> return Failure ( message = "<STR_LIT>" , details = trace , validator = self , failure_type = FAILURE_EXTRACTOR_EXCEPTION ) <EOL> elif self . isTemplateExpected and context : <EOL> expected_val = string . Template ( <EOL> self . expected ) . safe_substitute ( context . get_values ( ) ) <EOL> else : <EOL> expected_val = self . expected <EOL> if isinstance ( extracted_val , binary_type ) and isinstance ( expected_val , text_type ) : <EOL> expected_val = expected_val . encode ( '<STR_LIT:utf-8>' ) <EOL> comparison = self . comparator ( extracted_val , expected_val ) <EOL> if not comparison : <EOL> failure = Failure ( validator = self ) <EOL> failure . message = "<STR_LIT>" . format ( <EOL> self . comparator_name , extracted_val , expected_val ) <EOL> failure . details = self . get_readable_config ( context = context ) <EOL> failure . failure_type = FAILURE_VALIDATOR_FAILED <EOL> return failure <EOL> else : <EOL> return True <EOL> @ staticmethod <EOL> def parse ( config ) : <EOL> """<STR_LIT>""" <EOL> output = ComparatorValidator ( ) <EOL> config = parsing . lowercase_keys ( parsing . flatten_dictionaries ( config ) ) <EOL> output . config = config <EOL> output . extractor = _get_extractor ( config ) <EOL> if output . extractor is None : <EOL> raise ValueError ( <EOL> "<STR_LIT>" ) <EOL> if '<STR_LIT>' not in config : <EOL> output . comparator_name = '<STR_LIT>' <EOL> else : <EOL> output . comparator_name = config [ '<STR_LIT>' ] . lower ( ) <EOL> output . comparator = COMPARATORS [ output . comparator_name ] <EOL> if not output . comparator : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> try : <EOL> expected = config [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> raise ValueError ( <EOL> "<STR_LIT>" ) <EOL> if isinstance ( expected , basestring ) or isinstance ( expected , ( int , long , float , complex ) ) : <EOL> output . expected = expected <EOL> elif isinstance ( expected , dict ) : <EOL> expected = parsing . lowercase_keys ( expected ) <EOL> template = expected . get ( '<STR_LIT>' ) <EOL> if template : <EOL> if not isinstance ( template , basestring ) : <EOL> raise ValueError ( <EOL> "<STR_LIT>" ) <EOL> output . isTemplateExpected = True <EOL> output . expected = template <EOL> else : <EOL> output . expected = _get_extractor ( expected ) <EOL> if not output . expected : <EOL> raise ValueError ( <EOL> "<STR_LIT>" ) <EOL> return output <EOL> class ExtractTestValidator ( AbstractValidator ) : <EOL> """<STR_LIT>""" <EOL> name = '<STR_LIT>' <EOL> extractor = None <EOL> test_fn = None <EOL> test_name = None <EOL> config = None <EOL> def get_readable_config ( self , context = None ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT>" + self . extractor . get_readable_config ( context = context ) <EOL> @ staticmethod <EOL> def parse ( config ) : <EOL> output = ExtractTestValidator ( ) <EOL> config = parsing . lowercase_keys ( parsing . flatten_dictionaries ( config ) ) <EOL> output . config = config <EOL> extractor = _get_extractor ( config ) <EOL> output . extractor = extractor <EOL> test_name = config [ '<STR_LIT:test>' ] <EOL> output . test_name = test_name <EOL> test_fn = VALIDATOR_TESTS [ test_name ] <EOL> output . test_fn = test_fn <EOL> return output <EOL> def validate ( self , body = None , headers = None , context = None ) : <EOL> try : <EOL> extracted = self . extractor . extract ( <EOL> body = body , headers = headers , context = context ) <EOL> except Exception as e : <EOL> trace = traceback . format_exc ( ) <EOL> return Failure ( message = "<STR_LIT>" , details = trace , validator = self , failure_type = FAILURE_EXTRACTOR_EXCEPTION ) <EOL> tested = self . test_fn ( extracted ) <EOL> if tested : <EOL> return True <EOL> else : <EOL> failure = Failure ( details = self . get_readable_config ( <EOL> context = context ) , validator = self , failure_type = FAILURE_VALIDATOR_FAILED ) <EOL> failure . message = "<STR_LIT>" . format ( <EOL> self . test_name , extracted ) <EOL> return failure <EOL> def parse_extractor ( extractor_type , config ) : <EOL> """<STR_LIT>""" <EOL> parse = EXTRACTORS . get ( extractor_type . lower ( ) ) <EOL> if not parse : <EOL> raise ValueError ( <EOL> "<STR_LIT>" . format ( extractor_type ) ) <EOL> parsed = parse ( config ) <EOL> if isinstance ( parsed , AbstractExtractor ) : <EOL> return parsed <EOL> items = AbstractExtractor ( ) . __dict__ <EOL> if set ( parsed . __dict__ . keys ( ) ) . issuperset ( set ( items . keys ( ) ) ) : <EOL> return parsed <EOL> else : <EOL> raise TypeError ( <EOL> "<STR_LIT>" ) <EOL> def parse_validator ( name , config_node ) : <EOL> '''<STR_LIT>''' <EOL> name = name . lower ( ) <EOL> if name not in VALIDATORS : <EOL> raise ValueError ( <EOL> "<STR_LIT>" . format ( name ) ) <EOL> valid = VALIDATORS [ name ] ( config_node ) <EOL> if valid . name is None : <EOL> valid . name = name <EOL> if valid . config is None : <EOL> valid . config = config_node <EOL> return valid <EOL> def register_validator ( name , parse_function ) : <EOL> '''<STR_LIT>''' <EOL> name = name . lower ( ) <EOL> if name in VALIDATORS : <EOL> raise Exception ( "<STR_LIT>" . format ( name ) ) <EOL> VALIDATORS [ name ] = parse_function <EOL> def register_extractor ( extractor_name , parse_function ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( extractor_name , basestring ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> if extractor_name . lower ( ) == '<STR_LIT>' : <EOL> raise ValueError ( <EOL> "<STR_LIT>" ) <EOL> elif extractor_name . lower ( ) == '<STR_LIT:test>' : <EOL> raise ValueError ( <EOL> "<STR_LIT>" ) <EOL> elif extractor_name . lower ( ) == '<STR_LIT>' : <EOL> raise ValueError ( <EOL> "<STR_LIT>" ) <EOL> elif extractor_name in EXTRACTORS : <EOL> raise ValueError ( <EOL> "<STR_LIT>" . format ( extractor_name ) ) <EOL> EXTRACTORS [ extractor_name ] = parse_function <EOL> def register_test ( test_name , test_function ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( test_name , basestring ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> elif test_name in VALIDATOR_TESTS : <EOL> raise ValueError ( <EOL> "<STR_LIT>" . format ( test_name ) ) <EOL> VALIDATOR_TESTS [ test_name ] = test_function <EOL> def register_comparator ( comparator_name , comparator_function ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( comparator_name , basestring ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> elif comparator_name in COMPARATORS : <EOL> raise ValueError ( <EOL> "<STR_LIT>" . format ( comparator_name ) ) <EOL> COMPARATORS [ comparator_name ] = comparator_function <EOL> register_extractor ( '<STR_LIT>' , MiniJsonExtractor . parse ) <EOL> register_extractor ( '<STR_LIT>' , HeaderExtractor . parse ) <EOL> register_extractor ( '<STR_LIT>' , RawBodyExtractor . parse ) <EOL> register_validator ( '<STR_LIT>' , ComparatorValidator . parse ) <EOL> register_validator ( '<STR_LIT>' , ComparatorValidator . parse ) <EOL> register_validator ( '<STR_LIT>' , ComparatorValidator . parse ) <EOL> register_validator ( '<STR_LIT>' , ExtractTestValidator . parse ) <EOL> register_validator ( '<STR_LIT>' , ExtractTestValidator . parse ) </s>
