<s> from __future__ import with_statement , absolute_import <EOL> import os <EOL> import sys <EOL> import re <EOL> import tempfile <EOL> import unittest <EOL> from . import WithReader <EOL> sys . path . append ( os . path . abspath ( '<STR_LIT>' ) ) <EOL> from mysql2pgsql . lib . postgres_writer import PostgresWriter <EOL> from mysql2pgsql . lib . postgres_file_writer import PostgresFileWriter <EOL> from mysql2pgsql . lib . postgres_db_writer import PostgresDbWriter <EOL> def squeeze ( val ) : <EOL> return re . sub ( r"<STR_LIT>" , "<STR_LIT:U+0020>" , val ) . strip ( ) <EOL> class WithTables ( WithReader ) : <EOL> def setUp ( self ) : <EOL> super ( WithTables , self ) . setUp ( ) <EOL> self . table1 = next ( ( t for t in self . reader . tables if t . name == '<STR_LIT>' ) , None ) <EOL> self . table2 = next ( ( t for t in self . reader . tables if t . name == '<STR_LIT>' ) , None ) <EOL> assert self . table1 <EOL> assert self . table2 <EOL> class TestPostgresWriter ( WithTables ) : <EOL> def setUp ( self ) : <EOL> super ( self . __class__ , self ) . setUp ( ) <EOL> self . writer = PostgresWriter ( ) <EOL> assert self . writer <EOL> def test_truncate ( self ) : <EOL> trunc_cmds = self . writer . truncate ( self . table1 ) <EOL> assert len ( trunc_cmds ) == <NUM_LIT:2> <EOL> trunc_stmt , reset_seq = trunc_cmds <EOL> assert squeeze ( trunc_stmt ) == '<STR_LIT>' % self . table1 . name <EOL> if reset_seq : <EOL> self . assertRegexpMatches ( squeeze ( reset_seq ) , <EOL> "<STR_LIT>" % self . table1 . name ) <EOL> def test_write_table ( self ) : <EOL> write_table_cmds = self . writer . write_table ( self . table1 ) <EOL> assert len ( write_table_cmds ) == <NUM_LIT:2> <EOL> table_cmds , seq_cmds = write_table_cmds <EOL> assert len ( table_cmds ) == <NUM_LIT:2> <EOL> assert squeeze ( table_cmds [ <NUM_LIT:0> ] ) == '<STR_LIT>' % self . table1 . name <EOL> assert '<STR_LIT>' % self . table1 . name in table_cmds [ <NUM_LIT:1> ] <EOL> if seq_cmds : <EOL> assert len ( seq_cmds ) == <NUM_LIT:3> <EOL> self . assertRegexpMatches ( squeeze ( seq_cmds [ <NUM_LIT:0> ] ) , <EOL> '<STR_LIT>' % self . table1 . name ) <EOL> self . assertRegexpMatches ( squeeze ( seq_cmds [ <NUM_LIT:1> ] ) , <EOL> '<STR_LIT>' % self . table1 . name ) <EOL> self . assertRegexpMatches ( squeeze ( seq_cmds [ <NUM_LIT:2> ] ) , <EOL> "<STR_LIT>" % self . table1 . name ) <EOL> def test_write_indexex ( self ) : <EOL> index_cmds = self . writer . write_indexes ( self . table1 ) <EOL> assert len ( index_cmds ) == <NUM_LIT:9> <EOL> def test_write_constraints ( self ) : <EOL> constraint_cmds = self . writer . write_constraints ( self . table2 ) <EOL> assert constraint_cmds <EOL> class WithOutput ( WithTables ) : <EOL> def setUp ( self ) : <EOL> super ( WithOutput , self ) . setUp ( ) <EOL> def tearDown ( self ) : <EOL> super ( WithOutput , self ) . tearDown ( ) <EOL> class TestPostgresFileWriter ( WithOutput ) : <EOL> def setUp ( self ) : <EOL> super ( self . __class__ , self ) . setUp ( ) <EOL> self . outfile = tempfile . NamedTemporaryFile ( ) <EOL> self . writer = PostgresFileWriter ( self . outfile ) <EOL> def tearDown ( self ) : <EOL> super ( self . __class__ , self ) . tearDown ( ) <EOL> self . writer . close ( ) <EOL> def test_truncate ( self ) : <EOL> self . writer . truncate ( self . table1 ) <EOL> def test_write_table ( self ) : <EOL> self . writer . write_table ( self . table1 ) <EOL> def test_write_indexes ( self ) : <EOL> self . writer . write_indexes ( self . table1 ) <EOL> def test_write_constraints ( self ) : <EOL> self . writer . write_constraints ( self . table2 ) <EOL> def test_write_contents ( self ) : <EOL> self . writer . write_contents ( self . table1 , self . reader ) <EOL> class TestPostgresDbWriter ( WithOutput ) : <EOL> def setUp ( self ) : <EOL> super ( self . __class__ , self ) . setUp ( ) <EOL> self . writer = PostgresDbWriter ( self . config . options [ '<STR_LIT>' ] [ '<STR_LIT>' ] , True ) <EOL> def tearDown ( self ) : <EOL> super ( self . __class__ , self ) . tearDown ( ) <EOL> self . writer . close ( ) <EOL> def test_truncate ( self ) : <EOL> self . writer . truncate ( self . table1 ) <EOL> def test_write_table_indexes_and_constraints ( self ) : <EOL> self . writer . write_table ( table = self . table1 ) <EOL> self . writer . write_indexes ( self . table1 ) <EOL> self . writer . write_constraints ( self . table2 ) <EOL> def test_write_contents ( self ) : <EOL> self . writer . write_contents ( self . table1 , self . reader ) </s>
