<s> """<STR_LIT>""" <EOL> from unittest import skipUnless <EOL> from twisted . internet import reactor <EOL> from twisted . python . filepath import FilePath <EOL> from eliot import Message <EOL> from ... common import loop_until <EOL> from . testtools import make_dataset_integration_testcase <EOL> try : <EOL> from pg8000 import connect , InterfaceError , ProgrammingError <EOL> PG8000_INSTALLED = True <EOL> except ImportError : <EOL> PG8000_INSTALLED = False <EOL> def get_postgres_connection ( host , port , database = None ) : <EOL> """<STR_LIT>""" <EOL> def connect_to_postgres ( ) : <EOL> try : <EOL> return connect ( host = host , user = u"<STR_LIT>" , port = port , <EOL> database = database ) <EOL> except ( InterfaceError , ProgrammingError ) as e : <EOL> Message . new ( <EOL> message_type = u"<STR_LIT>" , <EOL> exception = unicode ( e . __class__ ) , reason = unicode ( e ) ) . write ( ) <EOL> return False <EOL> d = loop_until ( reactor , connect_to_postgres ) <EOL> return d <EOL> def insert_data ( test_case , host , port ) : <EOL> """<STR_LIT>""" <EOL> d = get_postgres_connection ( host , port ) <EOL> def create_database ( connection ) : <EOL> connection . autocommit = True <EOL> cursor = connection . cursor ( ) <EOL> cursor . execute ( "<STR_LIT>" ) <EOL> cursor . close ( ) <EOL> connection . close ( ) <EOL> d . addCallback ( create_database ) <EOL> d . addCallback ( <EOL> lambda _ : get_postgres_connection ( host , port , u"<STR_LIT>" ) ) <EOL> def add_data ( connection ) : <EOL> cursor = connection . cursor ( ) <EOL> cursor . execute ( <EOL> "<STR_LIT>" ) <EOL> cursor . execute ( <EOL> "<STR_LIT>" ) <EOL> connection . commit ( ) <EOL> connection . close ( ) <EOL> d . addCallback ( add_data ) <EOL> return d <EOL> def assert_inserted ( test_case , host , port ) : <EOL> """<STR_LIT>""" <EOL> d = get_postgres_connection ( host , port , u"<STR_LIT>" ) <EOL> def assert_data ( connection ) : <EOL> cursor = connection . cursor ( ) <EOL> cursor . execute ( "<STR_LIT>" ) <EOL> fetched_data = cursor . fetchone ( ) [ <NUM_LIT:0> ] <EOL> test_case . assertEqual ( fetched_data , <NUM_LIT> ) <EOL> d . addCallback ( assert_data ) <EOL> return d <EOL> class PostgresIntegrationTests ( make_dataset_integration_testcase ( <EOL> u"<STR_LIT>" , FilePath ( b'<STR_LIT>' ) , <NUM_LIT> , <EOL> insert_data , assert_inserted , <EOL> ) ) : <EOL> """<STR_LIT>""" <EOL> @ skipUnless ( PG8000_INSTALLED , "<STR_LIT>" ) <EOL> def setUp ( self ) : <EOL> super ( PostgresIntegrationTests , self ) . setUp ( ) </s>
