<s> import os <EOL> import time <EOL> import logging <EOL> import warnings <EOL> from copy import deepcopy <EOL> import jsonpickle <EOL> import requests <EOL> import validictory <EOL> from yapsy . PluginManager import PluginManager <EOL> from . import COLOURS , get_verify , get_host_overrides , tcptest <EOL> from . output import Output <EOL> FORMAT = '<STR_LIT>' <EOL> THIS_DIR = os . path . dirname ( os . path . realpath ( __file__ ) ) <EOL> logging . basicConfig ( format = FORMAT , level = logging . ERROR , datefmt = "<STR_LIT>" ) <EOL> LOG = logging . getLogger ( '<STR_LIT>' ) <EOL> REQUESTS_LOG = logging . getLogger ( '<STR_LIT>' ) <EOL> REQUESTS_LOG . setLevel ( logging . ERROR ) <EOL> logging . getLogger ( '<STR_LIT>' ) . setLevel ( logging . INFO ) <EOL> manager = PluginManager ( ) <EOL> manager . setPluginPlaces ( [ THIS_DIR , "<STR_LIT>" ] ) <EOL> manager . collectPlugins ( ) <EOL> OUTPUT_WIDTH = <NUM_LIT> <EOL> SCHEMA = { <EOL> "<STR_LIT:type>" : "<STR_LIT:object>" , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:name>" : { <EOL> "<STR_LIT:type>" : "<STR_LIT:string>" <EOL> } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:type>" : "<STR_LIT:string>" , <EOL> "<STR_LIT>" : False <EOL> } , <EOL> "<STR_LIT:port>" : { <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : False <EOL> } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:type>" : "<STR_LIT:object>" , <EOL> "<STR_LIT>" : False , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : { "<STR_LIT:type>" : "<STR_LIT>" , "<STR_LIT>" : False } , <EOL> "<STR_LIT:username>" : { "<STR_LIT:type>" : "<STR_LIT>" , "<STR_LIT>" : False } , <EOL> "<STR_LIT:password>" : { "<STR_LIT:type>" : "<STR_LIT>" , "<STR_LIT>" : False } , <EOL> "<STR_LIT>" : { "<STR_LIT:type>" : "<STR_LIT>" , "<STR_LIT>" : False } , <EOL> "<STR_LIT:data>" : { "<STR_LIT:type>" : "<STR_LIT>" , "<STR_LIT>" : False } , <EOL> "<STR_LIT:file>" : { "<STR_LIT:type>" : "<STR_LIT>" , "<STR_LIT>" : False } , <EOL> "<STR_LIT>" : { "<STR_LIT:type>" : "<STR_LIT>" , "<STR_LIT>" : False } , <EOL> "<STR_LIT>" : { "<STR_LIT:type>" : "<STR_LIT>" , "<STR_LIT>" : False } , <EOL> "<STR_LIT>" : { "<STR_LIT:type>" : "<STR_LIT>" , "<STR_LIT>" : False } , <EOL> "<STR_LIT>" : { "<STR_LIT:type>" : "<STR_LIT>" , "<STR_LIT>" : False } <EOL> } <EOL> } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:type>" : "<STR_LIT:object>" , <EOL> "<STR_LIT>" : False <EOL> } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:type>" : "<STR_LIT:string>" , <EOL> "<STR_LIT>" : False , <EOL> "<STR_LIT>" : [ "<STR_LIT>" , "<STR_LIT:http>" , "<STR_LIT>" ] <EOL> } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:type>" : "<STR_LIT:string>" , <EOL> "<STR_LIT>" : False , <EOL> "<STR_LIT>" : [ "<STR_LIT:GET>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:POST>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:type>" : "<STR_LIT:None>" , <EOL> "<STR_LIT>" : False <EOL> } , <EOL> "<STR_LIT:url>" : { <EOL> "<STR_LIT:type>" : "<STR_LIT:None>" , <EOL> "<STR_LIT>" : False <EOL> } <EOL> } <EOL> } <EOL> def deepupdate ( original , update ) : <EOL> """<STR_LIT>""" <EOL> for key , value in original . items ( ) : <EOL> if key not in update : <EOL> update [ key ] = value <EOL> elif isinstance ( value , dict ) : <EOL> deepupdate ( value , update [ key ] ) <EOL> return update <EOL> class Charcoal ( object ) : <EOL> def __init__ ( self , test , host , output_format = None ) : <EOL> """<STR_LIT>""" <EOL> self . passed = <NUM_LIT:0> <EOL> self . failed = <NUM_LIT:0> <EOL> self . duration_ms = <NUM_LIT:0> <EOL> try : <EOL> validictory . validate ( test , SCHEMA ) <EOL> LOG . debug ( "<STR_LIT>" ) <EOL> except ValueError as error : <EOL> LOG . error ( "<STR_LIT>" <EOL> . format ( error ) ) <EOL> raise <EOL> try : <EOL> self . port = test [ "<STR_LIT:port>" ] <EOL> except ( AttributeError , KeyError ) : <EOL> print ( "<STR_LIT>" ) <EOL> try : <EOL> if test [ "<STR_LIT>" ] == "<STR_LIT>" : <EOL> self . port = <NUM_LIT> <EOL> else : <EOL> self . port = <NUM_LIT> <EOL> except ( AttributeError , KeyError ) : <EOL> self . port = <NUM_LIT> <EOL> self . output = Output ( output_format = output_format ) <EOL> LOG . debug ( "<STR_LIT>" . format ( test ) ) <EOL> test_defaults = dict ( inputs = dict ( allow_redirects = False , timeout = <NUM_LIT:30> ) , <EOL> method = "<STR_LIT>" , <EOL> outcomes = dict ( expect_status_code = <NUM_LIT:200> , colour_output = True ) ) <EOL> host_overrides = get_host_overrides . get_host_overrides ( host , self . port ) <EOL> if host_overrides [ '<STR_LIT>' ] is not None : <EOL> self . host = host_overrides [ '<STR_LIT>' ] <EOL> else : <EOL> self . host = host <EOL> intermediate_dict = deepupdate ( test_defaults , host_overrides ) <EOL> final_dict = deepupdate ( intermediate_dict , test ) <EOL> if "<STR_LIT>" in test and test [ "<STR_LIT>" ] : <EOL> tcptest . tcp_test ( self . host , self . port ) <EOL> try : <EOL> verify = final_dict [ "<STR_LIT>" ] [ "<STR_LIT>" ] <EOL> except ( AttributeError , KeyError ) : <EOL> verify = None <EOL> try : <EOL> proto = final_dict [ '<STR_LIT>' ] <EOL> except ( AttributeError , KeyError ) : <EOL> proto = None <EOL> ( self . verify , self . verify_specified ) = get_verify . get_verify ( verify , proto ) <EOL> self . test = deepcopy ( final_dict ) <EOL> LOG . debug ( "<STR_LIT>" . format ( self . test ) ) <EOL> if "<STR_LIT>" in self . test [ "<STR_LIT>" ] : <EOL> del self . test [ "<STR_LIT>" ] [ "<STR_LIT>" ] <EOL> self . inputs = deepcopy ( self . test [ '<STR_LIT>' ] ) <EOL> request_url_format = '<STR_LIT>' <EOL> try : <EOL> self . inputs [ '<STR_LIT:url>' ] = request_url_format . format ( protocol = self . test [ '<STR_LIT>' ] , host = self . host , <EOL> port = self . test [ '<STR_LIT:port>' ] , <EOL> uri = self . test [ '<STR_LIT>' ] ) <EOL> except KeyError : <EOL> self . inputs [ '<STR_LIT:url>' ] = request_url_format . format ( protocol = self . test [ '<STR_LIT>' ] , host = self . host , <EOL> port = self . test [ '<STR_LIT:port>' ] , <EOL> uri = '<STR_LIT>' ) <EOL> LOG . debug ( "<STR_LIT>" . format ( self . inputs [ '<STR_LIT:url>' ] ) ) <EOL> self . output . append ( "<STR_LIT:->" * OUTPUT_WIDTH ) <EOL> this_name = ( "<STR_LIT>" + str ( OUTPUT_WIDTH ) + "<STR_LIT:}>" ) . format ( self . test [ '<STR_LIT:name>' ] [ : OUTPUT_WIDTH ] ) <EOL> self . output . append ( this_name ) <EOL> this_url = ( "<STR_LIT>" + str ( OUTPUT_WIDTH ) + "<STR_LIT:}>" ) . format ( self . inputs [ "<STR_LIT:url>" ] [ : OUTPUT_WIDTH ] ) <EOL> self . output . append ( this_url ) <EOL> self . output . append ( self . __repr__ ( ) ) <EOL> self . output . append ( "<STR_LIT:->" * OUTPUT_WIDTH ) <EOL> self . output . append ( self . inputs , sec = '<STR_LIT>' ) <EOL> self . run ( ) <EOL> def run ( self ) : <EOL> with warnings . catch_warnings ( ) : <EOL> warnings . simplefilter ( "<STR_LIT:error>" ) <EOL> try : <EOL> LOG . debug ( "<STR_LIT>" . format ( type ( self . verify ) , self . verify ) ) <EOL> start = int ( round ( time . time ( ) * <NUM_LIT:1000> ) ) <EOL> if self . test [ "<STR_LIT>" ] != '<STR_LIT>' : <EOL> self . req = getattr ( requests , self . test [ '<STR_LIT>' ] . lower ( ) ) ( verify = self . verify , ** self . inputs ) <EOL> else : <EOL> tcptest . tcp_test ( self . host , self . port ) <EOL> end = int ( round ( time . time ( ) * <NUM_LIT:1000> ) ) <EOL> self . duration_ms = end - start <EOL> except ( RuntimeWarning , requests . exceptions . SSLError ) : <EOL> warnings . simplefilter ( "<STR_LIT:ignore>" ) <EOL> start = int ( round ( time . time ( ) * <NUM_LIT:1000> ) ) <EOL> try : <EOL> self . req = getattr ( requests , self . test [ '<STR_LIT>' ] . lower ( ) ) ( verify = self . verify , ** self . inputs ) <EOL> except requests . exceptions . SSLError : <EOL> message , status = self . fail_test ( "<STR_LIT>" ) <EOL> self . add_output ( "<STR_LIT>" , message , status ) <EOL> return <EOL> end = int ( round ( time . time ( ) * <NUM_LIT:1000> ) ) <EOL> self . duration_ms = end - start <EOL> if not self . verify_specified : <EOL> message , status = self . fail_test ( "<STR_LIT>" ) <EOL> self . add_output ( "<STR_LIT>" , message , status ) <EOL> else : <EOL> if self . verify : <EOL> message , status = self . fail_test ( "<STR_LIT>" ) <EOL> self . add_output ( "<STR_LIT>" , message , status ) <EOL> else : <EOL> message , status = self . warn_test ( "<STR_LIT>" ) <EOL> self . add_output ( "<STR_LIT>" , message , status ) <EOL> if self . test [ "<STR_LIT>" ] != '<STR_LIT>' : <EOL> self . output . append ( dict ( self . req . headers ) , sec = '<STR_LIT>' ) <EOL> self . output . append ( self . req . status_code , sec = '<STR_LIT>' ) <EOL> if '<STR_LIT>' in self . test : <EOL> try : <EOL> req_content = self . req . content . decode ( ) <EOL> except UnicodeDecodeError : <EOL> req_content = self . req . content <EOL> self . output . append ( req_content ) <EOL> for plugin_info in manager . getAllPlugins ( ) : <EOL> for outcome in self . test [ '<STR_LIT>' ] : <EOL> if plugin_info . name == outcome : <EOL> manager . activatePluginByName ( plugin_info . name ) <EOL> message , status = plugin_info . plugin_object . run ( self ) <EOL> self . add_output ( plugin_info . name , message , status ) <EOL> manager . deactivatePluginByName ( plugin_info . name ) <EOL> def __str__ ( self ) : <EOL> return str ( self . output ) <EOL> def __repr__ ( self ) : <EOL> LOG . debug ( self . inputs ) <EOL> if "<STR_LIT>" in self . inputs : <EOL> if self . inputs [ "<STR_LIT>" ] != '<STR_LIT>' : <EOL> header_list = [ '<STR_LIT>' . format ( k , v ) for k , v in list ( self . inputs [ "<STR_LIT>" ] . items ( ) ) ] <EOL> header = "<STR_LIT>" + ( "<STR_LIT>" . join ( header_list ) ) <EOL> else : <EOL> header = self . inputs [ "<STR_LIT>" ] <EOL> else : <EOL> header = '<STR_LIT>' <EOL> if "<STR_LIT:data>" in self . inputs : <EOL> data = '<STR_LIT>' + str ( self . inputs [ "<STR_LIT:data>" ] ) <EOL> else : <EOL> data = '<STR_LIT>' <EOL> if not self . verify and self . test [ "<STR_LIT>" ] == "<STR_LIT>" : <EOL> curl_insecure = '<STR_LIT>' <EOL> else : <EOL> curl_insecure = '<STR_LIT>' <EOL> command = '<STR_LIT>' <EOL> output = command . format ( method = str ( self . test [ '<STR_LIT>' ] ) . upper ( ) , headers = header , data = data , <EOL> uri = self . inputs [ '<STR_LIT:url>' ] , curl_insecure = curl_insecure ) <EOL> return output <EOL> def pass_test ( self , message ) : <EOL> self . passed += <NUM_LIT:1> <EOL> status = "<STR_LIT>" <EOL> if self . test [ "<STR_LIT>" ] [ "<STR_LIT>" ] : <EOL> status = COLOURS . to_green ( status ) <EOL> return message , status <EOL> def fail_test ( self , message ) : <EOL> self . failed += <NUM_LIT:1> <EOL> status = "<STR_LIT>" <EOL> if self . test [ "<STR_LIT>" ] [ "<STR_LIT>" ] : <EOL> status = COLOURS . to_red ( status ) <EOL> return message , status <EOL> def warn_test ( self , message ) : <EOL> self . passed += <NUM_LIT:1> <EOL> status = "<STR_LIT>" <EOL> if self . test [ "<STR_LIT>" ] [ "<STR_LIT>" ] : <EOL> status = COLOURS . to_yellow ( status ) <EOL> return message , status <EOL> def add_output ( self , name , message , status ) : <EOL> test_out = name + "<STR_LIT>" + message + "<STR_LIT:.>" * ( OUTPUT_WIDTH - len ( name ) - len ( message ) - <NUM_LIT:8> ) + status . rjust ( <NUM_LIT:8> ) <EOL> self . output . append ( test_out ) <EOL> def _to_json ( self ) : <EOL> return jsonpickle . encode ( self ) </s>
