<s> import os <EOL> import glob <EOL> import inspect <EOL> from flexmock import flexmock , flexmock_teardown <EOL> from . . import OratorTestCase <EOL> from orator . migrations import Migrator , DatabaseMigrationRepository , Migration <EOL> from orator import DatabaseManager <EOL> from orator . connections import Connection <EOL> from orator . utils import PY3K <EOL> class MigratorTestCase ( OratorTestCase ) : <EOL> def setUp ( self ) : <EOL> if PY3K : <EOL> self . orig = inspect . getargspec <EOL> inspect . getargspec = lambda fn : inspect . getfullargspec ( fn ) [ : <NUM_LIT:4> ] <EOL> def tearDown ( self ) : <EOL> flexmock_teardown ( ) <EOL> if PY3K : <EOL> inspect . getargspec = self . orig <EOL> def test_migrations_are_run_up_when_outstanding_migrations_exist ( self ) : <EOL> resolver_mock = flexmock ( DatabaseManager ) <EOL> resolver_mock . should_receive ( '<STR_LIT>' ) . and_return ( { } ) <EOL> resolver = flexmock ( DatabaseManager ( { } ) ) <EOL> connection = flexmock ( ) <EOL> connection . should_receive ( '<STR_LIT>' ) . twice ( ) . and_return ( connection ) <EOL> resolver . should_receive ( '<STR_LIT>' ) . and_return ( connection ) <EOL> migrator = flexmock ( <EOL> Migrator ( <EOL> flexmock ( <EOL> DatabaseMigrationRepository ( <EOL> resolver , <EOL> '<STR_LIT>' <EOL> ) <EOL> ) , <EOL> resolver <EOL> ) <EOL> ) <EOL> g = flexmock ( glob ) <EOL> g . should_receive ( '<STR_LIT>' ) . with_args ( os . path . join ( os . getcwd ( ) , '<STR_LIT>' ) ) . and_return ( [ <EOL> os . path . join ( os . getcwd ( ) , '<STR_LIT>' ) , <EOL> os . path . join ( os . getcwd ( ) , '<STR_LIT>' ) , <EOL> os . path . join ( os . getcwd ( ) , '<STR_LIT>' ) <EOL> ] ) <EOL> migrator . get_repository ( ) . should_receive ( '<STR_LIT>' ) . once ( ) . and_return ( [ '<STR_LIT>' ] ) <EOL> migrator . get_repository ( ) . should_receive ( '<STR_LIT>' ) . once ( ) . and_return ( <NUM_LIT:1> ) <EOL> migrator . get_repository ( ) . should_receive ( '<STR_LIT>' ) . once ( ) . with_args ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> migrator . get_repository ( ) . should_receive ( '<STR_LIT>' ) . once ( ) . with_args ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> bar_mock = flexmock ( MigrationStub ( ) ) <EOL> bar_mock . set_connection ( connection ) <EOL> bar_mock . should_receive ( '<STR_LIT>' ) . once ( ) <EOL> baz_mock = flexmock ( MigrationStub ( ) ) <EOL> baz_mock . set_connection ( connection ) <EOL> baz_mock . should_receive ( '<STR_LIT>' ) . once ( ) <EOL> migrator . should_receive ( '<STR_LIT>' ) . with_args ( os . getcwd ( ) , '<STR_LIT>' ) . once ( ) . and_return ( bar_mock ) <EOL> migrator . should_receive ( '<STR_LIT>' ) . with_args ( os . getcwd ( ) , '<STR_LIT>' ) . once ( ) . and_return ( baz_mock ) <EOL> migrator . run ( os . getcwd ( ) ) <EOL> def test_migrations_are_run_up_directly_if_transactional_is_false ( self ) : <EOL> resolver_mock = flexmock ( DatabaseManager ) <EOL> resolver_mock . should_receive ( '<STR_LIT>' ) . and_return ( { } ) <EOL> resolver = flexmock ( DatabaseManager ( { } ) ) <EOL> connection = flexmock ( ) <EOL> connection . should_receive ( '<STR_LIT>' ) . never ( ) <EOL> resolver . should_receive ( '<STR_LIT>' ) . and_return ( connection ) <EOL> migrator = flexmock ( <EOL> Migrator ( <EOL> flexmock ( <EOL> DatabaseMigrationRepository ( <EOL> resolver , <EOL> '<STR_LIT>' <EOL> ) <EOL> ) , <EOL> resolver <EOL> ) <EOL> ) <EOL> g = flexmock ( glob ) <EOL> g . should_receive ( '<STR_LIT>' ) . with_args ( os . path . join ( os . getcwd ( ) , '<STR_LIT>' ) ) . and_return ( [ <EOL> os . path . join ( os . getcwd ( ) , '<STR_LIT>' ) , <EOL> os . path . join ( os . getcwd ( ) , '<STR_LIT>' ) , <EOL> os . path . join ( os . getcwd ( ) , '<STR_LIT>' ) <EOL> ] ) <EOL> migrator . get_repository ( ) . should_receive ( '<STR_LIT>' ) . once ( ) . and_return ( [ '<STR_LIT>' ] ) <EOL> migrator . get_repository ( ) . should_receive ( '<STR_LIT>' ) . once ( ) . and_return ( <NUM_LIT:1> ) <EOL> migrator . get_repository ( ) . should_receive ( '<STR_LIT>' ) . once ( ) . with_args ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> migrator . get_repository ( ) . should_receive ( '<STR_LIT>' ) . once ( ) . with_args ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> bar_mock = flexmock ( MigrationStub ( ) ) <EOL> bar_mock . transactional = False <EOL> bar_mock . set_connection ( connection ) <EOL> bar_mock . should_receive ( '<STR_LIT>' ) . once ( ) <EOL> baz_mock = flexmock ( MigrationStub ( ) ) <EOL> baz_mock . transactional = False <EOL> baz_mock . set_connection ( connection ) <EOL> baz_mock . should_receive ( '<STR_LIT>' ) . once ( ) <EOL> migrator . should_receive ( '<STR_LIT>' ) . with_args ( os . getcwd ( ) , '<STR_LIT>' ) . once ( ) . and_return ( bar_mock ) <EOL> migrator . should_receive ( '<STR_LIT>' ) . with_args ( os . getcwd ( ) , '<STR_LIT>' ) . once ( ) . and_return ( baz_mock ) <EOL> migrator . run ( os . getcwd ( ) ) <EOL> def test_up_migration_can_be_pretended ( self ) : <EOL> resolver_mock = flexmock ( DatabaseManager ) <EOL> resolver_mock . should_receive ( '<STR_LIT>' ) . and_return ( { } ) <EOL> resolver = flexmock ( DatabaseManager ( { } ) ) <EOL> connection = flexmock ( Connection ( None ) ) <EOL> connection . should_receive ( '<STR_LIT>' ) . replace_with ( lambda callback : callback ( None ) ) <EOL> resolver . should_receive ( '<STR_LIT>' ) . with_args ( None ) . and_return ( connection ) <EOL> migrator = flexmock ( <EOL> Migrator ( <EOL> flexmock ( <EOL> DatabaseMigrationRepository ( <EOL> resolver , <EOL> '<STR_LIT>' <EOL> ) <EOL> ) , <EOL> resolver <EOL> ) <EOL> ) <EOL> g = flexmock ( glob ) <EOL> g . should_receive ( '<STR_LIT>' ) . with_args ( os . path . join ( os . getcwd ( ) , '<STR_LIT>' ) ) . and_return ( [ <EOL> os . path . join ( os . getcwd ( ) , '<STR_LIT>' ) , <EOL> os . path . join ( os . getcwd ( ) , '<STR_LIT>' ) , <EOL> os . path . join ( os . getcwd ( ) , '<STR_LIT>' ) <EOL> ] ) <EOL> migrator . get_repository ( ) . should_receive ( '<STR_LIT>' ) . once ( ) . and_return ( [ '<STR_LIT>' ] ) <EOL> migrator . get_repository ( ) . should_receive ( '<STR_LIT>' ) . once ( ) . and_return ( <NUM_LIT:1> ) <EOL> bar_mock = flexmock ( MigrationStub ( ) ) <EOL> bar_mock . should_receive ( '<STR_LIT>' ) . once ( ) . and_return ( None ) <EOL> bar_mock . should_receive ( '<STR_LIT>' ) . once ( ) <EOL> baz_mock = flexmock ( MigrationStub ( ) ) <EOL> baz_mock . should_receive ( '<STR_LIT>' ) . once ( ) . and_return ( None ) <EOL> baz_mock . should_receive ( '<STR_LIT>' ) . once ( ) <EOL> migrator . should_receive ( '<STR_LIT>' ) . with_args ( os . getcwd ( ) , '<STR_LIT>' ) . once ( ) . and_return ( bar_mock ) <EOL> migrator . should_receive ( '<STR_LIT>' ) . with_args ( os . getcwd ( ) , '<STR_LIT>' ) . once ( ) . and_return ( baz_mock ) <EOL> migrator . run ( os . getcwd ( ) , True ) <EOL> def test_nothing_is_done_when_no_migrations_outstanding ( self ) : <EOL> resolver_mock = flexmock ( DatabaseManager ) <EOL> resolver_mock . should_receive ( '<STR_LIT>' ) . and_return ( None ) <EOL> resolver = flexmock ( DatabaseManager ( { } ) ) <EOL> migrator = flexmock ( <EOL> Migrator ( <EOL> flexmock ( <EOL> DatabaseMigrationRepository ( <EOL> resolver , <EOL> '<STR_LIT>' <EOL> ) <EOL> ) , <EOL> resolver <EOL> ) <EOL> ) <EOL> g = flexmock ( glob ) <EOL> g . should_receive ( '<STR_LIT>' ) . with_args ( os . path . join ( os . getcwd ( ) , '<STR_LIT>' ) ) . and_return ( [ <EOL> os . path . join ( os . getcwd ( ) , '<STR_LIT>' ) <EOL> ] ) <EOL> migrator . get_repository ( ) . should_receive ( '<STR_LIT>' ) . once ( ) . and_return ( [ '<STR_LIT>' ] ) <EOL> migrator . run ( os . getcwd ( ) ) <EOL> def test_last_batch_of_migrations_can_be_rolled_back ( self ) : <EOL> resolver_mock = flexmock ( DatabaseManager ) <EOL> resolver_mock . should_receive ( '<STR_LIT>' ) . and_return ( { } ) <EOL> resolver = flexmock ( DatabaseManager ( { } ) ) <EOL> connection = flexmock ( ) <EOL> connection . should_receive ( '<STR_LIT>' ) . twice ( ) . and_return ( connection ) <EOL> resolver . should_receive ( '<STR_LIT>' ) . and_return ( connection ) <EOL> migrator = flexmock ( <EOL> Migrator ( <EOL> flexmock ( <EOL> DatabaseMigrationRepository ( <EOL> resolver , <EOL> '<STR_LIT>' <EOL> ) <EOL> ) , <EOL> resolver <EOL> ) <EOL> ) <EOL> foo_migration = MigrationStub ( '<STR_LIT:foo>' ) <EOL> bar_migration = MigrationStub ( '<STR_LIT:bar>' ) <EOL> migrator . get_repository ( ) . should_receive ( '<STR_LIT>' ) . once ( ) . and_return ( [ <EOL> foo_migration , <EOL> bar_migration <EOL> ] ) <EOL> bar_mock = flexmock ( MigrationStub ( ) ) <EOL> bar_mock . set_connection ( connection ) <EOL> bar_mock . should_receive ( '<STR_LIT>' ) . once ( ) <EOL> foo_mock = flexmock ( MigrationStub ( ) ) <EOL> foo_mock . set_connection ( connection ) <EOL> foo_mock . should_receive ( '<STR_LIT>' ) . once ( ) <EOL> migrator . should_receive ( '<STR_LIT>' ) . with_args ( os . getcwd ( ) , '<STR_LIT:bar>' ) . once ( ) . and_return ( bar_mock ) <EOL> migrator . should_receive ( '<STR_LIT>' ) . with_args ( os . getcwd ( ) , '<STR_LIT:foo>' ) . once ( ) . and_return ( foo_mock ) <EOL> migrator . get_repository ( ) . should_receive ( '<STR_LIT>' ) . once ( ) . with_args ( bar_migration ) <EOL> migrator . get_repository ( ) . should_receive ( '<STR_LIT>' ) . once ( ) . with_args ( foo_migration ) <EOL> migrator . rollback ( os . getcwd ( ) ) <EOL> def test_last_batch_of_migrations_can_be_rolled_back_directly_if_transactional_is_false ( self ) : <EOL> resolver_mock = flexmock ( DatabaseManager ) <EOL> resolver_mock . should_receive ( '<STR_LIT>' ) . and_return ( { } ) <EOL> resolver = flexmock ( DatabaseManager ( { } ) ) <EOL> connection = flexmock ( ) <EOL> connection . should_receive ( '<STR_LIT>' ) . never ( ) <EOL> resolver . should_receive ( '<STR_LIT>' ) . and_return ( connection ) <EOL> migrator = flexmock ( <EOL> Migrator ( <EOL> flexmock ( <EOL> DatabaseMigrationRepository ( <EOL> resolver , <EOL> '<STR_LIT>' <EOL> ) <EOL> ) , <EOL> resolver <EOL> ) <EOL> ) <EOL> foo_migration = MigrationStub ( '<STR_LIT:foo>' ) <EOL> bar_migration = MigrationStub ( '<STR_LIT:bar>' ) <EOL> migrator . get_repository ( ) . should_receive ( '<STR_LIT>' ) . once ( ) . and_return ( [ <EOL> foo_migration , <EOL> bar_migration <EOL> ] ) <EOL> bar_mock = flexmock ( MigrationStub ( ) ) <EOL> bar_mock . transactional = False <EOL> bar_mock . set_connection ( connection ) <EOL> bar_mock . should_receive ( '<STR_LIT>' ) . once ( ) <EOL> foo_mock = flexmock ( MigrationStub ( ) ) <EOL> foo_mock . transactional = False <EOL> foo_mock . set_connection ( connection ) <EOL> foo_mock . should_receive ( '<STR_LIT>' ) . once ( ) <EOL> migrator . should_receive ( '<STR_LIT>' ) . with_args ( os . getcwd ( ) , '<STR_LIT:bar>' ) . once ( ) . and_return ( bar_mock ) <EOL> migrator . should_receive ( '<STR_LIT>' ) . with_args ( os . getcwd ( ) , '<STR_LIT:foo>' ) . once ( ) . and_return ( foo_mock ) <EOL> migrator . get_repository ( ) . should_receive ( '<STR_LIT>' ) . once ( ) . with_args ( bar_migration ) <EOL> migrator . get_repository ( ) . should_receive ( '<STR_LIT>' ) . once ( ) . with_args ( foo_migration ) <EOL> migrator . rollback ( os . getcwd ( ) ) <EOL> def test_rollback_migration_can_be_pretended ( self ) : <EOL> resolver_mock = flexmock ( DatabaseManager ) <EOL> resolver_mock . should_receive ( '<STR_LIT>' ) . and_return ( { } ) <EOL> resolver = flexmock ( DatabaseManager ( { } ) ) <EOL> connection = flexmock ( Connection ( None ) ) <EOL> connection . should_receive ( '<STR_LIT>' ) . replace_with ( lambda callback : callback ( None ) ) <EOL> resolver . should_receive ( '<STR_LIT>' ) . with_args ( None ) . and_return ( connection ) <EOL> migrator = flexmock ( <EOL> Migrator ( <EOL> flexmock ( <EOL> DatabaseMigrationRepository ( <EOL> resolver , <EOL> '<STR_LIT>' <EOL> ) <EOL> ) , <EOL> resolver <EOL> ) <EOL> ) <EOL> foo_migration = MigrationStub ( '<STR_LIT:foo>' ) <EOL> bar_migration = MigrationStub ( '<STR_LIT:bar>' ) <EOL> migrator . get_repository ( ) . should_receive ( '<STR_LIT>' ) . once ( ) . and_return ( [ <EOL> foo_migration , <EOL> bar_migration <EOL> ] ) <EOL> bar_mock = flexmock ( MigrationStub ( ) ) <EOL> bar_mock . should_receive ( '<STR_LIT>' ) . once ( ) <EOL> foo_mock = flexmock ( MigrationStub ( ) ) <EOL> foo_mock . should_receive ( '<STR_LIT>' ) . once ( ) <EOL> migrator . should_receive ( '<STR_LIT>' ) . with_args ( os . getcwd ( ) , '<STR_LIT:bar>' ) . once ( ) . and_return ( bar_mock ) <EOL> migrator . should_receive ( '<STR_LIT>' ) . with_args ( os . getcwd ( ) , '<STR_LIT:foo>' ) . once ( ) . and_return ( foo_mock ) <EOL> migrator . rollback ( os . getcwd ( ) , True ) <EOL> def test_nothing_is_rolled_back_when_nothing_in_repository ( self ) : <EOL> resolver = flexmock ( DatabaseManager ) <EOL> resolver . should_receive ( '<STR_LIT>' ) . and_return ( None ) <EOL> migrator = flexmock ( <EOL> Migrator ( <EOL> flexmock ( <EOL> DatabaseMigrationRepository ( <EOL> resolver , <EOL> '<STR_LIT>' <EOL> ) <EOL> ) , <EOL> resolver <EOL> ) <EOL> ) <EOL> migrator . get_repository ( ) . should_receive ( '<STR_LIT>' ) . once ( ) . and_return ( [ ] ) <EOL> migrator . rollback ( os . getcwd ( ) ) <EOL> class MigrationStub ( Migration ) : <EOL> def __init__ ( self , migration = None ) : <EOL> self . migration = migration <EOL> def up ( self ) : <EOL> pass <EOL> def down ( self ) : <EOL> pass <EOL> def __getitem__ ( self , item ) : <EOL> return self . migration </s>
