<s> from django_nose . tools import assert_equal , assert_false , assert_raises , assert_true <EOL> from mock import ANY , patch , PropertyMock <EOL> from pontoon . base . models import ChangedEntityLocale , Project , Repository <EOL> from pontoon . base . tests import ( <EOL> ChangedEntityLocaleFactory , <EOL> CONTAINS , <EOL> ProjectFactory , <EOL> RepositoryFactory , <EOL> TestCase , <EOL> TranslationFactory , <EOL> ) <EOL> from pontoon . base . utils import aware_datetime <EOL> from pontoon . sync . core import serial_task <EOL> from pontoon . sync . models import ProjectSyncLog , RepositorySyncLog , SyncLog <EOL> from pontoon . sync . tasks import sync_project , sync_project_repo <EOL> from pontoon . sync . tests import ( <EOL> FAKE_CHECKOUT_PATH , <EOL> FakeCheckoutTestCase , <EOL> ProjectSyncLogFactory , <EOL> SyncLogFactory , <EOL> ) <EOL> class SyncProjectTests ( TestCase ) : <EOL> def setUp ( self ) : <EOL> super ( SyncProjectTests , self ) . setUp ( ) <EOL> self . db_project = ProjectFactory . create ( ) <EOL> self . sync_log = SyncLogFactory . create ( ) <EOL> self . mock_pull_changes = self . patch ( <EOL> '<STR_LIT>' , return_value = True ) <EOL> self . mock_project_needs_sync = self . patch_object ( <EOL> Project , '<STR_LIT>' , new_callable = PropertyMock , return_value = True ) <EOL> self . mock_sync_project_repo = self . patch ( '<STR_LIT>' ) <EOL> self . mock_perform_sync_project = self . patch ( '<STR_LIT>' , return_value = [ [ ] , [ ] ] ) <EOL> def test_missing_project ( self ) : <EOL> """<STR_LIT>""" <EOL> with patch ( '<STR_LIT>' ) as mock_log : <EOL> with assert_raises ( Project . DoesNotExist ) : <EOL> sync_project ( <NUM_LIT> , self . sync_log . pk ) <EOL> mock_log . error . assert_called_with ( CONTAINS ( '<STR_LIT>' ) ) <EOL> assert_false ( self . mock_perform_sync_project . called ) <EOL> def test_missing_log ( self ) : <EOL> """<STR_LIT>""" <EOL> with patch ( '<STR_LIT>' ) as mock_log : <EOL> with assert_raises ( SyncLog . DoesNotExist ) : <EOL> sync_project ( self . db_project . pk , <NUM_LIT> ) <EOL> mock_log . error . assert_called_with ( CONTAINS ( '<STR_LIT>' ) ) <EOL> assert_false ( self . mock_perform_sync_project . called ) <EOL> def test_db_changed_no_repo_changed ( self ) : <EOL> """<STR_LIT>""" <EOL> self . mock_pull_changes . return_value = False <EOL> self . mock_project_needs_sync . return_value = True <EOL> sync_project ( self . db_project . pk , self . sync_log . pk ) <EOL> assert_true ( self . mock_perform_sync_project . called ) <EOL> def test_no_changes_skip ( self ) : <EOL> """<STR_LIT>""" <EOL> self . mock_pull_changes . return_value = False <EOL> self . mock_project_needs_sync . return_value = False <EOL> with patch ( '<STR_LIT>' ) as mock_log : <EOL> sync_project ( self . db_project . pk , self . sync_log . pk ) <EOL> assert_false ( self . mock_perform_sync_project . called ) <EOL> mock_log . info . assert_called_with ( <EOL> CONTAINS ( '<STR_LIT>' , self . db_project . slug ) <EOL> ) <EOL> assert_true ( ProjectSyncLog . objects . get ( project = self . db_project ) . skipped ) <EOL> def test_no_changes_force ( self ) : <EOL> """<STR_LIT>""" <EOL> self . mock_pull_changes . return_value = False <EOL> self . mock_project_needs_sync . return_value = False <EOL> sync_project ( self . db_project . pk , self . sync_log . pk , force = True ) <EOL> assert_true ( self . mock_perform_sync_project . called ) <EOL> def test_no_pull ( self ) : <EOL> """<STR_LIT>""" <EOL> sync_project ( self . db_project . pk , self . sync_log . pk , no_pull = True ) <EOL> assert_false ( self . mock_pull_changes . called ) <EOL> def test_create_project_log ( self ) : <EOL> assert_false ( ProjectSyncLog . objects . exists ( ) ) <EOL> repo = RepositoryFactory . create ( ) <EOL> self . db_project . repositories = [ repo ] <EOL> self . db_project . save ( ) <EOL> sync_project ( self . db_project . pk , self . sync_log . pk ) <EOL> log = ProjectSyncLog . objects . get ( project = self . db_project ) <EOL> assert_equal ( self . mock_sync_project_repo . delay . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] , repo . pk ) <EOL> assert_equal ( self . mock_sync_project_repo . delay . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:2> ] , log . pk ) <EOL> class SyncProjectRepoTests ( FakeCheckoutTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( SyncProjectRepoTests , self ) . setUp ( ) <EOL> self . project_sync_log = ProjectSyncLogFactory . create ( ) <EOL> self . mock_pull_changes = self . patch ( <EOL> '<STR_LIT>' , return_value = True ) <EOL> self . mock_commit_changes = self . patch ( '<STR_LIT>' ) <EOL> self . mock_repo_checkout_path = self . patch_object ( <EOL> Repository , '<STR_LIT>' , new_callable = PropertyMock , <EOL> return_value = FAKE_CHECKOUT_PATH ) <EOL> def test_clear_changed_entities ( self ) : <EOL> """<STR_LIT>""" <EOL> self . now = aware_datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:2> ) <EOL> changed1 , changed2 , changed_after = ChangedEntityLocaleFactory . create_batch ( <NUM_LIT:3> , <EOL> locale = self . translated_locale , <EOL> entity__resource = self . main_db_resource , <EOL> when = aware_datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> ) <EOL> changed_after . when = aware_datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:3> ) <EOL> changed_after . save ( ) <EOL> sync_project_repo ( self . db_project . pk , self . repository . pk , <EOL> self . project_sync_log . pk , self . now ) <EOL> with assert_raises ( ChangedEntityLocale . DoesNotExist ) : <EOL> changed1 . refresh_from_db ( ) <EOL> with assert_raises ( ChangedEntityLocale . DoesNotExist ) : <EOL> changed2 . refresh_from_db ( ) <EOL> changed_after . refresh_from_db ( ) <EOL> def test_reset_project_has_changed ( self ) : <EOL> """<STR_LIT>""" <EOL> self . db_project . has_changed = True <EOL> self . db_project . save ( ) <EOL> sync_project ( self . db_project . pk , self . project_sync_log . sync_log . pk ) <EOL> self . db_project . refresh_from_db ( ) <EOL> assert_false ( self . db_project . has_changed ) <EOL> def test_no_commit ( self ) : <EOL> """<STR_LIT>""" <EOL> sync_project_repo ( self . db_project . pk , self . repository . pk , <EOL> self . project_sync_log . pk , self . now , <EOL> no_commit = True ) <EOL> assert_false ( self . mock_commit_changes . called ) <EOL> def test_remove_duplicate_approvals ( self ) : <EOL> """<STR_LIT>""" <EOL> self . main_vcs_translation . strings [ None ] = '<STR_LIT>' <EOL> self . main_vcs_translation . fuzzy = False <EOL> duplicate_translation = TranslationFactory . create ( <EOL> entity = self . main_db_entity , <EOL> locale = self . translated_locale , <EOL> string = '<STR_LIT>' , <EOL> approved = True , <EOL> approved_date = aware_datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:3> ) <EOL> ) <EOL> ChangedEntityLocale . objects . filter ( entity = self . main_db_entity ) . delete ( ) <EOL> with patch ( '<STR_LIT>' , return_value = self . vcs_project ) : <EOL> sync_project_repo ( self . db_project . pk , self . repository . pk , <EOL> self . project_sync_log . pk , self . now ) <EOL> assert_equal ( self . main_db_entity . translation_set . filter ( approved = True ) . count ( ) , <NUM_LIT:1> ) <EOL> new_translation = self . main_db_entity . translation_set . get ( <EOL> string = '<STR_LIT>' <EOL> ) <EOL> assert_false ( new_translation . approved ) <EOL> assert_true ( new_translation . approved_date is None ) <EOL> duplicate_translation . refresh_from_db ( ) <EOL> assert_true ( duplicate_translation . approved ) <EOL> assert_equal ( duplicate_translation . approved_date , aware_datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:3> ) ) <EOL> def test_create_project_log ( self ) : <EOL> assert_false ( RepositorySyncLog . objects . exists ( ) ) <EOL> repo = RepositoryFactory . create ( ) <EOL> self . db_project . repositories = [ repo ] <EOL> self . db_project . save ( ) <EOL> sync_project_repo ( self . db_project . pk , self . repository . pk , <EOL> self . project_sync_log . pk , self . now ) <EOL> log = RepositorySyncLog . objects . get ( repository = self . repository ) <EOL> assert_equal ( log . repository , self . repository ) <EOL> class UserError ( Exception ) : <EOL> pass <EOL> class SyncExecutionTests ( TestCase ) : <EOL> def test_serial_task ( self ) : <EOL> """<STR_LIT>""" <EOL> @ serial_task ( <NUM_LIT:100> ) <EOL> def test_task ( self , callback ) : <EOL> return callback ( ) <EOL> def execute_second_inner_task ( ) : <EOL> return test_task . delay ( lambda : None ) <EOL> first_call = test_task . delay ( execute_second_inner_task ) <EOL> second_call = first_call . get ( ) <EOL> assert_true ( first_call . successful ( ) ) <EOL> assert_true ( second_call . failed ( ) ) <EOL> assert_raises ( RuntimeError , second_call . get ) <EOL> def test_release_lock_after_timeout ( self ) : <EOL> """<STR_LIT>""" <EOL> with patch ( '<STR_LIT>' ) as mock_cache : <EOL> @ serial_task ( <NUM_LIT:3> ) <EOL> def timeout_task ( self ) : <EOL> return <NUM_LIT> <EOL> first_call = timeout_task . delay ( ) <EOL> assert_true ( first_call . successful ( ) ) <EOL> assert_equal ( first_call . get ( ) , <NUM_LIT> ) <EOL> mock_cache . add . assert_called_with ( ANY , ANY , timeout = <NUM_LIT:3> ) <EOL> def test_parametrized_serial_task ( self ) : <EOL> """<STR_LIT>""" <EOL> with patch ( '<STR_LIT>' ) as mock_cache : <EOL> @ serial_task ( <NUM_LIT:3> , lock_key = "<STR_LIT>" ) <EOL> def task_lock_key ( self , param ) : <EOL> return param <EOL> first_call = task_lock_key . delay ( <NUM_LIT> ) <EOL> second_call = task_lock_key . delay ( <NUM_LIT> ) <EOL> assert_true ( first_call . successful ( ) ) <EOL> assert_true ( second_call . successful ( ) ) <EOL> assert_true ( first_call . get ( ) , <NUM_LIT> ) <EOL> assert_true ( second_call . get ( ) , <NUM_LIT> ) <EOL> mock_cache . add . assert_any_call ( CONTAINS ( '<STR_LIT>' ) , ANY , timeout = <NUM_LIT:3> ) <EOL> mock_cache . add . assert_any_call ( CONTAINS ( '<STR_LIT>' ) , ANY , timeout = <NUM_LIT:3> ) <EOL> def test_exception_during_sync ( self ) : <EOL> """<STR_LIT>""" <EOL> @ serial_task ( <NUM_LIT:100> ) <EOL> def exception_task ( self ) : <EOL> raise UserError <EOL> first_call = exception_task . delay ( ) <EOL> second_call = exception_task . delay ( ) <EOL> assert_true ( first_call . failed ( ) ) <EOL> assert_true ( second_call . failed ( ) ) <EOL> assert_raises ( UserError , first_call . get ) <EOL> assert_raises ( UserError , second_call . get ) </s>
