<s> import pytest <EOL> import psycopg2 <EOL> import psycopg2 . extras <EOL> from pgcli . main import format_output <EOL> from pgcli . pgexecute import register_json_typecasters <EOL> POSTGRES_USER , POSTGRES_HOST = '<STR_LIT>' , '<STR_LIT:localhost>' <EOL> def db_connection ( dbname = None ) : <EOL> conn = psycopg2 . connect ( user = POSTGRES_USER , host = POSTGRES_HOST , database = dbname ) <EOL> conn . autocommit = True <EOL> return conn <EOL> try : <EOL> conn = db_connection ( ) <EOL> CAN_CONNECT_TO_DB = True <EOL> SERVER_VERSION = conn . server_version <EOL> json_types = register_json_typecasters ( conn , lambda x : x ) <EOL> JSON_AVAILABLE = '<STR_LIT>' in json_types <EOL> JSONB_AVAILABLE = '<STR_LIT>' in json_types <EOL> except : <EOL> CAN_CONNECT_TO_DB = JSON_AVAILABLE = JSONB_AVAILABLE = False <EOL> SERVER_VERSION = <NUM_LIT:0> <EOL> dbtest = pytest . mark . skipif ( <EOL> not CAN_CONNECT_TO_DB , <EOL> reason = "<STR_LIT>" ) <EOL> requires_json = pytest . mark . skipif ( <EOL> not JSON_AVAILABLE , <EOL> reason = '<STR_LIT>' ) <EOL> requires_jsonb = pytest . mark . skipif ( <EOL> not JSONB_AVAILABLE , <EOL> reason = '<STR_LIT>' ) <EOL> def create_db ( dbname ) : <EOL> with db_connection ( ) . cursor ( ) as cur : <EOL> try : <EOL> cur . execute ( '''<STR_LIT>''' ) <EOL> except : <EOL> pass <EOL> def drop_tables ( conn ) : <EOL> with conn . cursor ( ) as cur : <EOL> cur . execute ( '''<STR_LIT>''' ) <EOL> def run ( executor , sql , join = False , expanded = False , pgspecial = None , <EOL> exception_formatter = None ) : <EOL> "<STR_LIT>" <EOL> results = executor . run ( sql , pgspecial , exception_formatter ) <EOL> formatted = [ ] <EOL> for title , rows , headers , status , sql , success in results : <EOL> formatted . extend ( format_output ( title , rows , headers , status , '<STR_LIT>' , <EOL> expanded = expanded ) ) <EOL> if join : <EOL> formatted = '<STR_LIT:\n>' . join ( formatted ) <EOL> return formatted </s>
