<s> """<STR_LIT>""" <EOL> import sys <EOL> import logging <EOL> import pyrax <EOL> from modularodm import Q <EOL> from website . app import init_app <EOL> from website . addons . osfstorage import model <EOL> from scripts import utils as script_utils <EOL> from scripts . osfstorage import settings as storage_settings <EOL> TEST_CONTAINER_NAME = '<STR_LIT>' <EOL> PROD_CONTAINER_NAME = '<STR_LIT>' <EOL> test_container = None <EOL> prod_container = None <EOL> logger = logging . getLogger ( __name__ ) <EOL> logging . basicConfig ( level = logging . INFO ) <EOL> def migrate_version ( version ) : <EOL> if version . location [ '<STR_LIT>' ] != TEST_CONTAINER_NAME : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> key = test_container . get_object ( version . location [ '<STR_LIT:object>' ] ) <EOL> key . copy ( prod_container ) <EOL> logger . info ( '<STR_LIT>' . format ( <EOL> version . _id , <EOL> PROD_CONTAINER_NAME ) <EOL> ) <EOL> version . location [ '<STR_LIT>' ] = PROD_CONTAINER_NAME <EOL> version . save ( ) <EOL> def get_targets ( ) : <EOL> query = Q ( '<STR_LIT>' , '<STR_LIT>' , TEST_CONTAINER_NAME ) <EOL> return model . OsfStorageFileVersion . find ( query ) <EOL> def main ( dry_run ) : <EOL> versions = get_targets ( ) <EOL> for version in versions : <EOL> logger . info ( '<STR_LIT>' . format ( version . _id ) ) <EOL> if not dry_run : <EOL> migrate_version ( version ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> init_app ( set_backends = True , routes = False ) <EOL> dry_run = '<STR_LIT>' in sys . argv <EOL> if not dry_run : <EOL> script_utils . add_file_logger ( logger , __file__ ) <EOL> pyrax . settings . set ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> pyrax . set_credentials ( <EOL> storage_settings . USERNAME , <EOL> storage_settings . API_KEY , <EOL> region = storage_settings . REGION <EOL> ) <EOL> test_container = pyrax . cloudfiles . get_container ( TEST_CONTAINER_NAME ) <EOL> prod_container = pyrax . cloudfiles . get_container ( PROD_CONTAINER_NAME ) <EOL> main ( dry_run = dry_run ) <EOL> import mock <EOL> from nose . tools import * <EOL> from tests . base import OsfTestCase <EOL> from website . addons . osfstorage . tests . factories import FileVersionFactory <EOL> class TestMigrateContainer ( OsfTestCase ) : <EOL> def tearDown ( self ) : <EOL> super ( TestMigrateContainer , self ) . tearDown ( ) <EOL> model . OsfStorageFileVersion . remove ( ) <EOL> def test_get_targets ( self ) : <EOL> versions = [ FileVersionFactory ( ) for _ in range ( <NUM_LIT:5> ) ] <EOL> versions [ <NUM_LIT:0> ] . location [ '<STR_LIT>' ] = TEST_CONTAINER_NAME <EOL> versions [ <NUM_LIT:0> ] . save ( ) <EOL> targets = get_targets ( ) <EOL> assert_equal ( len ( targets ) , <NUM_LIT:1> ) <EOL> assert_equal ( targets [ <NUM_LIT:0> ] , versions [ <NUM_LIT:0> ] ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_migrate_version ( self , mock_test_container , mock_prod_container ) : <EOL> mock_test_object = mock . Mock ( ) <EOL> mock_test_container . get_object . return_value = mock_test_object <EOL> version = FileVersionFactory ( ) <EOL> version . location [ '<STR_LIT>' ] = TEST_CONTAINER_NAME <EOL> version . save ( ) <EOL> migrate_version ( version ) <EOL> mock_test_container . get_object . assert_called_with ( version . location [ '<STR_LIT:object>' ] ) <EOL> mock_test_object . copy . assert_called_with ( mock_prod_container ) <EOL> version . reload ( ) <EOL> assert_equal ( version . location [ '<STR_LIT>' ] , PROD_CONTAINER_NAME ) <EOL> assert_equal ( len ( get_targets ( ) ) , <NUM_LIT:0> ) <EOL> def test_dry_run ( self ) : <EOL> versions = [ FileVersionFactory ( ) for _ in range ( <NUM_LIT:5> ) ] <EOL> versions [ <NUM_LIT:0> ] . location [ '<STR_LIT>' ] = TEST_CONTAINER_NAME <EOL> versions [ <NUM_LIT:0> ] . save ( ) <EOL> main ( dry_run = True ) <EOL> assert_equal ( len ( get_targets ( ) ) , <NUM_LIT:1> ) </s>
